0f7e47e55e
Reported as bandsaunter using "completely different" identities from a script
that had been watching the same sensors for years. They are the same
identities. Their list against this program's:
14645 = 3935 8677 = 21E5 14717 = 397D
3542 = 0DD6 10011 = 271B
Five of their eight, matching exactly; the other three simply were not
transmitting during the three seconds of capture I had. One base is
hexadecimal and the other decimal, and nothing anywhere said so.
The hexadecimal is not arbitrary -- the identity is a bit field with the
channel packed into the top two bits, and that shape is visible in hex and
invisible in decimal, which is why the decoder carries it that way. But
rtl_433 and everything built on it prints these in decimal, so anybody who
comes to this with their own sensors already written down has the other form,
and being handed a list that looks unrelated to theirs is a poor welcome.
So both are shown wherever a person reads: the live display when there is room
for the column, the report, the sensor list and the spreadsheet. `bandsaunter
sensors` is the table for correlating two lists and now has them side by side,
with a line saying which is which and why.
Either may be typed at --name. One case needs care rather than cleverness:
"3935" is a valid identity in both bases and they are different sensors, so
when both are out there it says the identity is ambiguous and asks for the
whole key instead of picking whichever the code reaches first.
A first attempt at the lookup converted a decimal identity to hexadecimal and
matched on that as well as comparing decimals directly. Both worked, so
neither could be tested apart from the other; the conversion was removed
rather than given a test written backwards from it.
Full suite 2391 passed, checked against five deliberately broken builds.
Built as 2026-09-20_01.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016PsWPTweCT6pwxKngvVxcg
947 lines
40 KiB
Python
Executable file
947 lines
40 KiB
Python
Executable file
"""Live terminal display for a running scan."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import select
|
|
import sys
|
|
import termios
|
|
import time
|
|
import tty
|
|
from collections import deque
|
|
from dataclasses import dataclass
|
|
|
|
import numpy as np
|
|
from rich.console import Console, Group
|
|
from rich.markup import escape
|
|
from rich.panel import Panel
|
|
from rich.table import Table
|
|
from rich.text import Text
|
|
|
|
from .bandplan import band_label, fmt_hz, shorten_band
|
|
from .flightlog import in_speed, speed_label
|
|
from .recorder import HitRecord
|
|
from .scanner import Detection, Scanner
|
|
|
|
__all__ = ["ScanDisplay", "AircraftDisplay", "WeatherDisplay", "KeyReader",
|
|
"print_hit", "print_band_table"]
|
|
|
|
_SPARK = " ▁▂▃▄▅▆▇█"
|
|
|
|
# Width of the band column in the line-per-hit output, which has no live
|
|
# layout to shrink and so fixes it once.
|
|
_PLAIN_BAND = 20
|
|
|
|
# How many terminal lines each panel of the display occupies. The layout has
|
|
# to know this exactly: rich moves the cursor back over as many lines as it
|
|
# wrote last time, so a model that is one line out overflows the terminal by
|
|
# one line on every refresh, and the top of the frame marches down the screen.
|
|
# The sweep panel used to be counted as one line shorter than it is, which
|
|
# cancelled against the hit list being counted one line taller -- so the sum
|
|
# came out right whenever both were drawn, and wrong on a terminal too short
|
|
# for the hit list. test_resize.py checks these against what is rendered.
|
|
_H_RECEIVER = 3 # one line of text, two borders
|
|
_H_SWEEP = 4 # bar and state, two borders
|
|
_H_SPECTRUM = 1 # the sparkline row inside the sweep panel
|
|
_H_RECORD = 3
|
|
_H_FOOTER = 3
|
|
_H_HITS_CHROME = 3 # two borders and the column headings
|
|
|
|
|
|
def _one_line(markup: str) -> Text:
|
|
"""A line of markup that will never wrap onto a second row.
|
|
|
|
Every panel in this display is counted as a fixed number of lines by
|
|
:meth:`ScanDisplay._layout`, which decides what fits on the terminal. One
|
|
line wrapping makes the whole frame a row taller than that arithmetic
|
|
allows for, the frame no longer fits where it was drawn, and each refresh
|
|
leaves another copy of the header behind -- which is what a narrow window
|
|
used to look like after a few seconds.
|
|
"""
|
|
text = Text.from_markup(markup)
|
|
text.no_wrap = True
|
|
text.overflow = "ellipsis"
|
|
return text
|
|
|
|
|
|
def _sparkline(values: np.ndarray, width: int = 60,
|
|
lo: float | None = None, hi: float | None = None) -> str:
|
|
"""Compress a spectrum into one row of block characters."""
|
|
v = np.asarray(values, dtype=np.float64)
|
|
if v.size == 0:
|
|
return " " * width
|
|
if v.size > width:
|
|
# Max-reduce rather than average: a narrow carrier must stay visible.
|
|
edges = np.linspace(0, v.size, width + 1).astype(int)
|
|
v = np.array([v[edges[i]:edges[i + 1]].max() if edges[i + 1] > edges[i]
|
|
else v[min(edges[i], v.size - 1)] for i in range(width)])
|
|
lo = float(np.percentile(v, 5)) if lo is None else lo
|
|
hi = float(np.percentile(v, 99.5)) if hi is None else hi
|
|
if hi <= lo:
|
|
hi = lo + 1.0
|
|
idx = np.clip((v - lo) / (hi - lo) * (len(_SPARK) - 1), 0,
|
|
len(_SPARK) - 1).astype(int)
|
|
return "".join(_SPARK[i] for i in idx)
|
|
|
|
|
|
class KeyReader:
|
|
"""Non-blocking single-key input, restoring the terminal on exit."""
|
|
|
|
def __init__(self, enabled: bool = True):
|
|
self.enabled = enabled and sys.stdin.isatty()
|
|
self._old = None
|
|
|
|
def __enter__(self):
|
|
if self.enabled:
|
|
try:
|
|
self._old = termios.tcgetattr(sys.stdin)
|
|
tty.setcbreak(sys.stdin.fileno())
|
|
except (termios.error, ValueError):
|
|
self.enabled = False
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
if self._old is not None:
|
|
try:
|
|
termios.tcsetattr(sys.stdin, termios.TCSADRAIN, self._old)
|
|
except (termios.error, ValueError):
|
|
pass
|
|
return False
|
|
|
|
def get(self) -> str | None:
|
|
if not self.enabled:
|
|
return None
|
|
try:
|
|
r, _, _ = select.select([sys.stdin], [], [], 0)
|
|
if r:
|
|
return sys.stdin.read(1)
|
|
except (OSError, ValueError):
|
|
return None
|
|
return None
|
|
|
|
|
|
@dataclass
|
|
class _RecState:
|
|
frequency: float = 0.0
|
|
mode: str = ""
|
|
elapsed: float = 0.0
|
|
snr: float = 0.0
|
|
present: bool = False
|
|
quiet_for: float = 0.0
|
|
active: bool = False
|
|
note: str = "" # e.g. "TRUNK: Motorola SMARTNET / SmartZone"
|
|
|
|
|
|
class ScanDisplay:
|
|
"""Renders scanner callbacks into a live dashboard.
|
|
|
|
Attach with ``display.attach(scanner)``; the scanner then drives it.
|
|
"""
|
|
|
|
def __init__(self, scanner: Scanner, console: Console | None = None,
|
|
max_hits: int = 12, show_spectrum: bool = True):
|
|
self.scanner = scanner
|
|
self.console = console or Console()
|
|
self.hits: deque[HitRecord] = deque(maxlen=max_hits)
|
|
self.messages: deque[str] = deque(maxlen=4)
|
|
self.show_spectrum = show_spectrum
|
|
|
|
self._spark = ""
|
|
self._span = ""
|
|
self._band = ""
|
|
self._peak = 0.0
|
|
self._step_i = 0
|
|
self._n_steps = 1
|
|
self._rec = _RecState()
|
|
self._last_detection: Detection | None = None
|
|
self._dirty = True
|
|
self._size = self.console.size
|
|
|
|
# -- callbacks --------------------------------------------------------
|
|
def attach(self, scanner: Scanner | None = None) -> None:
|
|
s = scanner or self.scanner
|
|
cb = s.cb
|
|
cb.on_step = self.on_step
|
|
cb.on_detection = self.on_detection
|
|
cb.on_record_start = self.on_record_start
|
|
cb.on_record_tick = self.on_record_tick
|
|
cb.on_record_end = self.on_record_end
|
|
cb.on_record_note = self.on_record_note
|
|
cb.on_status = self.on_status
|
|
cb.on_error = self.on_error
|
|
|
|
def on_step(self, i, n, step, psd_db, freqs):
|
|
self._step_i, self._n_steps = i, n
|
|
if self.show_spectrum:
|
|
inband = (freqs >= step.low) & (freqs <= step.high)
|
|
if np.any(inband):
|
|
self._spark = _sparkline(psd_db[inband], width=self._spark_width())
|
|
self._peak = float(psd_db[inband].max())
|
|
self._span = f"{fmt_hz(step.low)} - {fmt_hz(step.high)}"
|
|
# The band at the middle of the step, so the sweep line says what is
|
|
# being listened to and not only where.
|
|
self._band = band_label(0.5 * (step.low + step.high), 24)
|
|
self._dirty = True
|
|
|
|
def on_detection(self, det: Detection):
|
|
self._last_detection = det
|
|
self._dirty = True
|
|
|
|
def on_record_start(self, det, rec):
|
|
self._rec = _RecState(frequency=rec.frequency, mode=rec.mode,
|
|
active=True)
|
|
self._dirty = True
|
|
|
|
def on_record_tick(self, rec, elapsed, present, snr, quiet_for=0.0):
|
|
self._rec.elapsed = elapsed
|
|
self._rec.present = present
|
|
self._rec.snr = snr
|
|
self._rec.quiet_for = quiet_for
|
|
self._rec.mode = rec.mode
|
|
self._dirty = True
|
|
|
|
def on_record_note(self, note: str):
|
|
self._rec.note = note
|
|
self._dirty = True
|
|
|
|
def on_record_end(self, hit: HitRecord):
|
|
self._rec.active = False
|
|
if hit.kept:
|
|
self.hits.appendleft(hit)
|
|
self._dirty = True
|
|
|
|
def on_status(self, msg: str):
|
|
# Escaped on the way in. A status line can carry text straight off
|
|
# the air -- a decoded pager message, a callsign, a Morse decode --
|
|
# and rich reads square brackets as markup: "[/x]" in a message is
|
|
# not a style, it is an exception in the middle of the display.
|
|
# Errors keep their markup, which is written here rather than by
|
|
# whatever raised.
|
|
self.messages.appendleft(escape(msg))
|
|
self._dirty = True
|
|
|
|
def on_error(self, exc: Exception):
|
|
self.messages.appendleft(f"[red]{type(exc).__name__}: {exc}[/red]")
|
|
self._dirty = True
|
|
|
|
# -- rendering ---------------------------------------------------------
|
|
def resized(self) -> bool:
|
|
"""True once each time the terminal has changed size.
|
|
|
|
Live rendering works by moving the cursor back over the frame it drew
|
|
last time and overwriting it, which is only correct while the frame is
|
|
still where it was put. Resizing the window reflows everything on the
|
|
screen, so the arithmetic no longer describes anything, and what is
|
|
left behind is half of the old frame in pieces. The caller clears the
|
|
screen when this returns true; polled rather than handled as a signal,
|
|
because the display is redrawn several times a second anyway and a
|
|
signal handler that runs in the middle of a write has to be right
|
|
about far more than this does.
|
|
"""
|
|
size = self.console.size
|
|
if size == self._size:
|
|
return False
|
|
self._size = size
|
|
self._dirty = True
|
|
return True
|
|
|
|
def _spark_width(self) -> int:
|
|
return max(20, min(120, self.console.width - 24))
|
|
|
|
def _header(self) -> Panel:
|
|
d = self.scanner.device
|
|
st = d.status() if d else {}
|
|
gain = st.get("gain", "?")
|
|
gain = f"{gain:.1f} dB" if isinstance(gain, (int, float)) else str(gain)
|
|
ds = st.get("direct_sampling", 0)
|
|
# Ordered by what has to survive a narrow window. The flags come
|
|
# first among the optional parts because "SIMULATED" disappearing off
|
|
# the end of the line is how somebody comes to believe they are
|
|
# listening to the air.
|
|
head = [f"[bold]{st.get('tuner', '?')}[/bold]"]
|
|
if ds:
|
|
head.append("[yellow]direct sampling[/yellow]")
|
|
if st.get("simulated"):
|
|
head.append("[magenta]SIMULATED[/magenta]")
|
|
optional = [f"{st.get('sample_rate', 0)/1e6:.3f} MS/s",
|
|
f"gain {gain}",
|
|
f"{st.get('ppm', 0):+d} ppm"]
|
|
room = max(10, self.console.width - 4)
|
|
bits = head + optional
|
|
while len(optional) > 0 and \
|
|
len(Text.from_markup(" ".join(bits)).plain) > room:
|
|
optional.pop()
|
|
bits = head + optional
|
|
return Panel(_one_line(" ".join(bits)),
|
|
title="receiver", border_style="blue", padding=(0, 1))
|
|
|
|
def _sweep_panel(self, show_spectrum: bool = True) -> Panel:
|
|
s = self.scanner.stats
|
|
frac = (self._step_i + 1) / max(1, self._n_steps)
|
|
bar_w = max(10, min(40, self.console.width - 60))
|
|
filled = int(frac * bar_w)
|
|
bar = "[green]" + "━" * filled + "[/green]" + \
|
|
"[grey37]" + "━" * (bar_w - filled) + "[/grey37]"
|
|
lines = [
|
|
_one_line(
|
|
f"{bar} step {self._step_i + 1}/{self._n_steps} "
|
|
f"[bold]{self._span}[/bold]"
|
|
+ (f" [magenta]{self._band}[/magenta]" if self._band else "")),
|
|
]
|
|
if show_spectrum and self.show_spectrum and self._spark:
|
|
lines.append(_one_line(
|
|
f"[cyan]{self._spark}[/cyan] peak {self._peak:6.1f} dBFS"))
|
|
state = s.state
|
|
colour = {"recording": "red", "sweeping": "green",
|
|
"paused": "yellow"}.get(state, "white")
|
|
lines.append(_one_line(
|
|
f"[{colour}]{state}[/{colour}] cycle {s.cycles + 1} "
|
|
f"hits {s.recordings} dropped {s.discarded} "
|
|
f"detections {s.detections} up {_dur(s.elapsed)}"))
|
|
return Panel(Group(*lines), title="sweep", border_style="blue",
|
|
padding=(0, 1))
|
|
|
|
def _record_panel(self) -> Panel | None:
|
|
r = self._rec
|
|
if not r.active:
|
|
return None
|
|
cfg = self.scanner.cfg
|
|
limit = cfg.record_seconds
|
|
bar_w = max(10, min(30, self.console.width - 70))
|
|
frac = min(1.0, r.elapsed / limit) if limit else 0.0
|
|
filled = int(frac * bar_w)
|
|
bar = ("[red]" + "█" * filled + "[/red]" +
|
|
"[grey37]" + "░" * (bar_w - filled) + "[/grey37]") if limit \
|
|
else "[red]recording[/red]"
|
|
hang = cfg.hang_seconds
|
|
if r.present:
|
|
sq = "[green]SIGNAL[/green]"
|
|
else:
|
|
# Show the gap counting down, so it is obvious the recording is
|
|
# being held open across a pause rather than stuck.
|
|
sq = f"[yellow]gap {min(r.quiet_for, hang):4.1f}/{hang:g}s[/yellow]"
|
|
limit_s = f"/{limit:g}s" if limit else ""
|
|
band = band_label(r.frequency, 22)
|
|
where = f" [magenta]{band}[/magenta]" if band else ""
|
|
if r.note:
|
|
# A control channel is not being recorded so much as identified
|
|
# and abandoned; saying "REC" while that happens is a lie.
|
|
return Panel(
|
|
_one_line(
|
|
f"[bold black on yellow] {escape(r.note)} "
|
|
f"[/bold black on yellow] "
|
|
f"{fmt_hz(r.frequency)}{where} [{r.mode}] "
|
|
f"SNR {r.snr:5.1f} dB [yellow]skipping[/yellow]"),
|
|
border_style="yellow", padding=(0, 1))
|
|
return Panel(
|
|
_one_line(
|
|
f"[bold red]REC[/bold red] {fmt_hz(r.frequency)}{where} "
|
|
f"[{r.mode}] {bar} {r.elapsed:5.1f}{limit_s} "
|
|
f"{sq} SNR {r.snr:5.1f} dB"),
|
|
border_style="red", padding=(0, 1))
|
|
|
|
def _layout(self) -> tuple[bool, bool, int, bool]:
|
|
"""Decide what fits: ``(receiver, spectrum, hit rows, footer)``.
|
|
|
|
On a short terminal the optional parts are given up in order -- the
|
|
spectrum row, then the hit list, then the key hints, and last of all
|
|
the receiver panel, which says nothing that changes. What is never
|
|
given up is the sweep line and, while one is running, the recording.
|
|
|
|
A frame taller than the terminal cannot be redrawn where it was drawn:
|
|
rich moves the cursor back over exactly as many lines as it wrote last
|
|
time, and if the frame did not fit, those are not the lines it is
|
|
looking at. Every refresh then leaves another copy of the top of the
|
|
frame behind, which is how a header ends up printed down the screen.
|
|
"""
|
|
height = self.console.size.height or 24
|
|
# One line short of the window: the cursor has to sit somewhere after
|
|
# the frame without the terminal scrolling to make room for it.
|
|
budget = max(6, height - 1)
|
|
spectrum = bool(self.show_spectrum and self._spark)
|
|
|
|
for want_receiver in (True, False):
|
|
for want_spectrum in ((True, False) if spectrum else (False,)):
|
|
for want_footer in (True, False):
|
|
base = self._fixed_height(want_receiver, want_spectrum,
|
|
want_footer)
|
|
rows = budget - base - _H_HITS_CHROME
|
|
if rows >= 1:
|
|
return want_receiver, want_spectrum, rows, want_footer
|
|
if budget - base >= 0:
|
|
return want_receiver, want_spectrum, 0, want_footer
|
|
# Narrower than anything can be drawn in. The sweep line alone, which
|
|
# rich will crop; there is nothing further to give up.
|
|
return False, False, 0, False
|
|
|
|
def _fixed_height(self, receiver: bool, spectrum: bool,
|
|
footer: bool) -> int:
|
|
"""Every line of the frame except the hit list."""
|
|
return ((_H_RECEIVER if receiver else 0)
|
|
+ _H_SWEEP + (_H_SPECTRUM if spectrum else 0)
|
|
+ (_H_RECORD if self._rec.active else 0)
|
|
+ (_H_FOOTER if footer else 0))
|
|
|
|
def _layout_height(self) -> int:
|
|
"""How many lines the frame will occupy, from the layout alone.
|
|
|
|
The number the overwriting depends on: rich moves the cursor back over
|
|
exactly this many lines before drawing the next frame, so if what is
|
|
actually rendered is even one line taller -- because a panel wrapped --
|
|
the frame creeps down the screen a row at a time.
|
|
"""
|
|
receiver, spectrum, rows, footer = self._layout()
|
|
base = self._fixed_height(receiver, spectrum, footer)
|
|
if rows <= 0:
|
|
return base
|
|
# The hit list is as tall as it has hits to show, up to the room it
|
|
# was given -- and one row regardless, for the "nothing yet" line.
|
|
drawn = max(1, min(rows, len(self.hits)))
|
|
return base + drawn + _H_HITS_CHROME
|
|
|
|
def _hit_capacity(self) -> int:
|
|
"""How many hit rows fit without pushing the display off the screen.
|
|
|
|
A display taller than the terminal cannot be redrawn in place, so each
|
|
refresh scrolls another copy of it into the scrollback and the header
|
|
appears over and over. The number of hits grows as the scan runs,
|
|
which is why it starts fine and degrades.
|
|
"""
|
|
return self._layout()[2]
|
|
|
|
def _band_width(self) -> int:
|
|
"""How much of the row the band column may take, or 0 for none.
|
|
|
|
It is the first thing given up on a narrow terminal: what a signal
|
|
was identified as matters more than which allocation it sat in, and
|
|
a band name squeezed to four characters says nothing.
|
|
"""
|
|
# time, frequency, duration, SNR, and enough left for the
|
|
# identification, which is the column that must not be squeezed.
|
|
spare = self.console.width - 8 - 15 - 6 - 7 - 30
|
|
if spare < 10:
|
|
return 0
|
|
return max(10, min(22, spare))
|
|
|
|
@staticmethod
|
|
def _band_of(hit: HitRecord, width: int) -> str:
|
|
"""What band a hit sat in, preferring what was recorded with it.
|
|
|
|
Taken from the record rather than recomputed so that a hit kept from
|
|
an older run, or a band plan that has since been edited, still shows
|
|
the band it was actually filed under.
|
|
"""
|
|
name = hit.band or (hit.band_labels[0] if hit.band_labels else "") \
|
|
or band_label(hit.frequency)
|
|
return shorten_band(name, width)
|
|
|
|
def _hits_table(self) -> Panel:
|
|
band_w = self._band_width()
|
|
t = Table(box=None, expand=True, pad_edge=False, show_edge=False)
|
|
t.add_column("time", style="grey62", width=8)
|
|
t.add_column("frequency", style="bold cyan", width=15, justify="right")
|
|
if band_w:
|
|
t.add_column("band", style="magenta", width=band_w,
|
|
no_wrap=True, overflow="ellipsis")
|
|
t.add_column("dur", width=6, justify="right")
|
|
t.add_column("SNR", width=7, justify="right")
|
|
# One line per hit, always: _hit_capacity counts rows, and a row
|
|
# that wraps to two pushes the display past the bottom of the
|
|
# terminal, where it can no longer be redrawn in place.
|
|
t.add_column("identified as", ratio=1, no_wrap=True,
|
|
overflow="ellipsis")
|
|
capacity = self._hit_capacity()
|
|
shown = list(self.hits)[:capacity]
|
|
for h in shown:
|
|
extra = ""
|
|
if h.image_kind:
|
|
extra = (f' [magenta]{escape(h.image_kind)} '
|
|
f'{escape(h.image_mode)}[/magenta]')
|
|
elif h.morse_text:
|
|
extra = f' [yellow]"{escape(h.morse_text.strip()[:32])}"[/yellow]'
|
|
elif h.data_messages:
|
|
# What was decoded is the most interesting thing about a data
|
|
# capture, and far more so than its baud rate.
|
|
extra = (" [bright_cyan]"
|
|
+ escape(h.data_messages[0][:40]) + "[/bright_cyan]")
|
|
elif h.data_encoding:
|
|
extra = f" [cyan]{escape(h.data_headline()[:40])}[/cyan]"
|
|
elif h.ctcss_hz:
|
|
extra = f" [grey62]CTCSS {h.ctcss_hz:.1f}[/grey62]"
|
|
elif h.baud:
|
|
extra = f" [grey62]{h.baud:.0f} baud[/grey62]"
|
|
conf = h.confidence
|
|
colour = "green" if conf >= 0.7 else "yellow" if conf >= 0.45 else "grey62"
|
|
row = [
|
|
time.strftime("%H:%M:%S", time.localtime(h.started_at)),
|
|
fmt_hz(h.frequency),
|
|
]
|
|
if band_w:
|
|
row.append(self._band_of(h, band_w))
|
|
row += [
|
|
f"{h.duration:.1f}s",
|
|
f"{h.snr_db:.1f}",
|
|
Text.from_markup(
|
|
f"[{colour}]{h.classification or 'unclassified'}[/{colour}]{extra}"),
|
|
]
|
|
t.add_row(*row)
|
|
if not self.hits:
|
|
# One cell short of the row and the placeholder lands in a
|
|
# narrow numeric column and wraps, making the panel taller
|
|
# than the layout budgeted for.
|
|
blanks = [""] * (5 if band_w else 4)
|
|
t.add_row(*blanks, Text("no signals recorded yet", style="grey42"))
|
|
hidden = len(self.hits) - len(shown)
|
|
title = "recorded signals"
|
|
if hidden > 0:
|
|
title += f" [grey62]({hidden} more above)[/grey62]"
|
|
return Panel(t, title=title, border_style="blue", padding=(0, 1))
|
|
|
|
def _footer(self) -> Panel:
|
|
keys = ("[bold]q[/bold] quit [bold]p[/bold] pause "
|
|
"[bold]s[/bold] skip [bold]l[/bold] lock out "
|
|
"[bold]+/-[/bold] threshold")
|
|
msg = self.messages[0] if self.messages else ""
|
|
return Panel(_one_line(f"{keys} {msg}"),
|
|
border_style="grey37", padding=(0, 1))
|
|
|
|
def render(self):
|
|
receiver, spectrum, rows, footer = self._layout()
|
|
parts = []
|
|
if receiver:
|
|
parts.append(self._header())
|
|
parts.append(self._sweep_panel(show_spectrum=spectrum))
|
|
rec = self._record_panel()
|
|
if rec is not None:
|
|
parts.append(rec)
|
|
if rows > 0:
|
|
parts.append(self._hits_table())
|
|
if footer:
|
|
parts.append(self._footer())
|
|
return Group(*parts)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Aircraft, while they are overhead
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# Altitude in colour, low warm to high cold: the same convention the map uses,
|
|
# so a height can be read off either without a key.
|
|
_ALTITUDE_BANDS = ((1_500, "bright_red"), (5_000, "red"), (10_000, "dark_orange"),
|
|
(18_000, "yellow"), (24_000, "green"),
|
|
(30_000, "bright_cyan"), (36_000, "cyan"),
|
|
(99_000, "bright_blue"))
|
|
|
|
# How fresh a report is, in colour. Past the last of these the aircraft is
|
|
# about to be dropped from the display.
|
|
_AGE_STYLES = ((5.0, "green"), (15.0, "yellow"), (1e9, "red"))
|
|
|
|
_COMPASS = ("N", "NNE", "NE", "ENE", "E", "ESE", "SE", "SSE",
|
|
"S", "SSW", "SW", "WSW", "W", "WNW", "NW", "NNW")
|
|
|
|
|
|
def altitude_style(feet: float) -> str:
|
|
"""The colour a height is drawn in."""
|
|
for limit, style in _ALTITUDE_BANDS:
|
|
if feet < limit:
|
|
return style
|
|
return _ALTITUDE_BANDS[-1][1]
|
|
|
|
|
|
def age_style(seconds: float) -> str:
|
|
for limit, style in _AGE_STYLES:
|
|
if seconds < limit:
|
|
return style
|
|
return "red"
|
|
|
|
|
|
def compass(degrees: float) -> str:
|
|
"""A heading as a point of the compass, which is easier to read at a glance."""
|
|
return _COMPASS[int((degrees % 360.0) / 22.5 + 0.5) % 16]
|
|
|
|
|
|
class AircraftDisplay:
|
|
"""One line per aircraft, updated in place while listening.
|
|
|
|
An aircraft appears when its first frame arrives, its counter climbs as
|
|
more come in, and it disappears once nothing has been heard from it for
|
|
``hold`` seconds -- at which point everything below moves up. The order
|
|
is the order they were first heard, so a row does not jump about under
|
|
the eye while it is being read.
|
|
|
|
Everything the frames say is here; everything a register says is here as
|
|
it arrives, on a second line under the aircraft it belongs to, because
|
|
the two are different kinds of knowledge and should not be mistaken for
|
|
each other.
|
|
"""
|
|
|
|
def __init__(self, console: Console, book=None, hold: float = 45.0,
|
|
title: str = "", unit: str = "knots"):
|
|
self.console = console
|
|
self.book = book
|
|
self.hold = hold
|
|
self.title = title
|
|
# Aircraft broadcast knots; this is only what they are shown in.
|
|
self.unit = unit
|
|
self.frames = 0
|
|
self.started = time.time()
|
|
self.log_path = None
|
|
self.registry = None
|
|
self.gone = 0
|
|
|
|
# -- what the listener tells it ---------------------------------------
|
|
def update(self, registry, frames: int, log_path=None) -> None:
|
|
self.registry = registry
|
|
self.frames = frames
|
|
if log_path is not None:
|
|
self.log_path = log_path
|
|
|
|
def showing(self, now: float | None = None) -> list:
|
|
"""The aircraft still worth drawing, oldest first heard at the top."""
|
|
if self.registry is None:
|
|
return []
|
|
now = time.time() if now is None else now
|
|
alive = [craft for craft in self.registry.aircraft.values()
|
|
if now - craft.last_seen <= self.hold]
|
|
return sorted(alive, key=lambda c: (c.first_seen, c.icao))
|
|
|
|
# -- drawing ----------------------------------------------------------
|
|
def render(self, now: float | None = None, width: int | None = None):
|
|
"""The whole display.
|
|
|
|
``width`` is the terminal it is going to be drawn on; it decides how
|
|
many columns there is room for, and defaults to the console this
|
|
display was built with, which is the one the live view uses.
|
|
"""
|
|
now = time.time() if now is None else now
|
|
width = self.console.size.width if width is None else width
|
|
flying = self.showing(now)
|
|
parts = [self._header(flying, now)]
|
|
if flying:
|
|
parts.append(self._table(flying, now, width))
|
|
else:
|
|
parts.append(Panel(_one_line(
|
|
"[grey62]nothing heard yet — ADS-B needs an aerial cut for "
|
|
"1090 MHz[/grey62]"), border_style="grey37", padding=(0, 1)))
|
|
return Group(*parts)
|
|
|
|
def _header(self, flying, now: float) -> Panel:
|
|
elapsed = max(0.001, now - self.started)
|
|
heard = len(self.registry.aircraft) if self.registry is not None else 0
|
|
rate = self.frames / elapsed
|
|
where = f" [grey62]{self.log_path.name}[/grey62]" if self.log_path \
|
|
else ""
|
|
return Panel(_one_line(
|
|
f"[bold cyan]1090 MHz[/bold cyan] "
|
|
f"[bold]{len(flying)}[/bold] overhead "
|
|
f"[grey62]{heard} seen[/grey62] "
|
|
f"[bold]{self.frames}[/bold] frames "
|
|
f"[grey62]{rate:.0f}/s {_dur(elapsed)}[/grey62]{where} "
|
|
f"[grey62]control-C to stop[/grey62]"),
|
|
border_style="blue", padding=(0, 1))
|
|
|
|
def _table(self, flying, now: float, width: int) -> Table:
|
|
"""As many columns as the terminal has room for, widest first.
|
|
|
|
A narrow terminal keeps what only the aircraft can say -- who, how
|
|
high, how fast, how many frames -- and drops what a website said,
|
|
because that can be read afterwards and the aeroplane cannot.
|
|
"""
|
|
t = Table(box=None, header_style="bold", pad_edge=False, expand=False)
|
|
t.add_column("callsign", width=9, no_wrap=True)
|
|
t.add_column("ICAO", width=6, style="grey62", no_wrap=True)
|
|
if width >= 96:
|
|
t.add_column("aircraft", width=17, overflow="ellipsis",
|
|
no_wrap=True)
|
|
t.add_column("altitude", width=10, justify="right", no_wrap=True)
|
|
# Wide enough for the heading it is given: "speed km/h" is longer
|
|
# than "speed kt", and a truncated unit is a wrong unit.
|
|
speeds = f"speed {speed_label(self.unit)}"
|
|
t.add_column(speeds, width=max(9, len(speeds)), justify="right",
|
|
no_wrap=True)
|
|
t.add_column("track", width=8, no_wrap=True)
|
|
if width >= 114:
|
|
t.add_column("position", width=19, no_wrap=True)
|
|
t.add_column("frames", width=6, justify="right", no_wrap=True)
|
|
t.add_column("last", width=5, justify="right", no_wrap=True)
|
|
if width >= 132:
|
|
t.add_column("operator / route", overflow="ellipsis", no_wrap=True,
|
|
style="grey58")
|
|
for craft in flying:
|
|
self._row(t, craft, now, width)
|
|
return t
|
|
|
|
def _row(self, t: Table, craft, now: float, width: int) -> None:
|
|
entry = self.book.get(craft.icao, craft.callsign) \
|
|
if self.book is not None else None
|
|
age = max(0.0, now - craft.last_seen)
|
|
cells = [Text(craft.callsign or "—",
|
|
style="bold white" if craft.callsign else "grey62"),
|
|
Text(craft.icao)]
|
|
if width >= 96:
|
|
cells.append(Text(self._kind(entry), style="grey62"))
|
|
cells += [self._altitude(craft),
|
|
Text(f"{in_speed(craft.ground_speed_kt, self.unit):.0f}"
|
|
if craft.ground_speed_kt else "", style="white"),
|
|
self._track(craft)]
|
|
if width >= 114:
|
|
cells.append(Text(f"{craft.latitude:8.4f},{craft.longitude:9.4f}"
|
|
if craft.located else " no fix yet",
|
|
style="cyan" if craft.located else "grey37"))
|
|
cells.append(Text(f"{craft.messages}", style="bold"))
|
|
cells.append(Text(f"{age:.0f}s", style=age_style(age)))
|
|
if width >= 132:
|
|
cells.append(Text(self._told(entry)))
|
|
t.add_row(*cells)
|
|
|
|
@staticmethod
|
|
def _kind(entry) -> str:
|
|
"""The airframe in as few characters as say anything: type and mark."""
|
|
if entry is None:
|
|
return ""
|
|
if entry.type_code and entry.registration:
|
|
return f"{entry.type_code} {entry.registration}"
|
|
return entry.type_code or entry.registration or ""
|
|
|
|
@staticmethod
|
|
def _altitude(craft) -> Text:
|
|
if not craft.altitude_ft:
|
|
return Text("")
|
|
arrow = ""
|
|
if craft.vertical_rate_fpm > 100:
|
|
arrow = "[green]↑[/green]"
|
|
elif craft.vertical_rate_fpm < -100:
|
|
arrow = "[red]↓[/red]"
|
|
style = altitude_style(craft.altitude_ft)
|
|
return Text.from_markup(
|
|
f"[{style}]{craft.altitude_ft:,}[/{style}]{arrow}")
|
|
|
|
@staticmethod
|
|
def _track(craft) -> Text:
|
|
if not craft.ground_speed_kt and not craft.track_deg:
|
|
return Text("")
|
|
return Text(f"{craft.track_deg:03.0f}° {compass(craft.track_deg)}",
|
|
style="grey62")
|
|
|
|
@staticmethod
|
|
def _told(entry) -> str:
|
|
"""What a register says about this aircraft, if anything yet."""
|
|
if entry is None:
|
|
return ""
|
|
bits = [x for x in (entry.operator or entry.airline, entry.route,
|
|
entry.country) if x]
|
|
return " · ".join(bits)
|
|
|
|
|
|
def _dur(seconds: float) -> str:
|
|
seconds = int(seconds)
|
|
h, rem = divmod(seconds, 3600)
|
|
m, s = divmod(rem, 60)
|
|
return f"{h}:{m:02d}:{s:02d}" if h else f"{m}:{s:02d}"
|
|
|
|
|
|
def print_hit(console: Console, hit: HitRecord) -> None:
|
|
"""One-line-per-hit output for non-interactive runs."""
|
|
if not hit.kept:
|
|
return
|
|
stamp = time.strftime("%H:%M:%S", time.localtime(hit.started_at))
|
|
conf = f"{hit.confidence*100:.0f}%"
|
|
band = ScanDisplay._band_of(hit, _PLAIN_BAND)
|
|
# Padded even when empty, so the columns line up down the page whether or
|
|
# not a particular frequency falls in a band the plan knows.
|
|
where = f" [magenta]{band:<{_PLAIN_BAND}}[/magenta]"
|
|
console.print(
|
|
f"[grey62]{stamp}[/grey62] "
|
|
f"[bold cyan]{fmt_hz(hit.frequency):>14}[/bold cyan]{where}"
|
|
f" {hit.duration:5.1f}s SNR {hit.snr_db:5.1f} dB "
|
|
f"[green]{hit.classification or 'unclassified'}[/green] ({conf})",
|
|
highlight=False)
|
|
indent = 26 + _PLAIN_BAND + 2
|
|
if hit.image_path:
|
|
console.print(f'{"":>{indent}}[magenta]picture: '
|
|
f'{escape(hit.image_kind)} {escape(hit.image_mode)} '
|
|
f'{hit.image_width}x{hit.image_height} → '
|
|
f'{escape(hit.image_path)}[/magenta]',
|
|
highlight=False, soft_wrap=True)
|
|
if hit.morse_text:
|
|
console.print(f'{"":>{indent}}[yellow]Morse @ {hit.morse_wpm:.0f} WPM: '
|
|
f'"{escape(hit.morse_text.strip())}"[/yellow]',
|
|
highlight=False)
|
|
for line in hit.data_messages[:4]:
|
|
console.print(f'{"":>{indent}}[bright_cyan]{escape(line)}'
|
|
f'[/bright_cyan]', highlight=False)
|
|
if hit.data_hex and not hit.data_messages:
|
|
console.print(f'{"":>{indent}}[cyan]{escape(hit.data_headline())}'
|
|
f'[/cyan] [grey54]{hit.data_hex[:48]}[/grey54]',
|
|
highlight=False)
|
|
if hit.reasons:
|
|
console.print(f'{"":>{indent}}[grey54]{escape(hit.reasons[0])}'
|
|
f'[/grey54]', highlight=False)
|
|
|
|
|
|
def print_band_table(console: Console, presets, title: str = "band plan") -> None:
|
|
t = Table(title=title, box=None, header_style="bold")
|
|
t.add_column("key", style="cyan")
|
|
t.add_column("name")
|
|
t.add_column("range", justify="right")
|
|
t.add_column("mode", justify="center")
|
|
t.add_column("notes", style="grey62", overflow="fold")
|
|
for p in presets:
|
|
extent = (f"{len(p.expand())} ranges, {fmt_hz(p.start)}-{fmt_hz(p.stop)}"
|
|
if p.is_group else f"{fmt_hz(p.start)} - {fmt_hz(p.stop)}")
|
|
t.add_row(p.key, p.name, extent, p.mode, p.note)
|
|
console.print(t)
|
|
|
|
|
|
class WeatherDisplay:
|
|
"""One line per weather sensor, updated in place while listening.
|
|
|
|
A sensor appears when its first message arrives and stays until nothing
|
|
has been heard from it for ``hold`` seconds -- half an hour by default,
|
|
which is long compared with the sixteen seconds between messages and so
|
|
means a sensor that vanishes has really stopped rather than been missed.
|
|
|
|
The line shows the *latest value of every quantity*, not the latest
|
|
message. A 5-in-1 has more to say than fits in one message and sends two
|
|
kinds alternately, so its last message is either the wind and the rain or
|
|
the temperature and the humidity, never both; showing the newest of each
|
|
means the line is the whole sensor rather than half of it flickering.
|
|
|
|
Rows are numbered and the numbers do not move, because naming a sensor
|
|
means reading a row and then typing its number, and a list that reorders
|
|
itself between those two moments is a list that gets things named wrong.
|
|
"""
|
|
|
|
def __init__(self, console: Console, book=None, hold: float = 1800.0,
|
|
imperial: bool = False, frequency: float = 433.92e6):
|
|
self.console = console
|
|
self.book = book
|
|
self.hold = hold
|
|
self.imperial = imperial
|
|
self.frequency = frequency
|
|
self.messages = 0
|
|
self.started = time.time()
|
|
self.log_path = None
|
|
self.garden = None
|
|
|
|
# -- what the listener tells it ---------------------------------------
|
|
def update(self, garden, messages: int, log_path=None) -> None:
|
|
self.garden = garden
|
|
self.messages = messages
|
|
if log_path is not None:
|
|
self.log_path = log_path
|
|
|
|
def showing(self, now: float | None = None) -> list:
|
|
if self.garden is None:
|
|
return []
|
|
return self.garden.showing(self.hold, now)
|
|
|
|
# -- drawing ----------------------------------------------------------
|
|
def render(self, now: float | None = None, width: int | None = None):
|
|
now = time.time() if now is None else now
|
|
width = self.console.size.width if width is None else width
|
|
here = self.showing(now)
|
|
parts = [self._header(here, now)]
|
|
if here:
|
|
parts.append(self._table(here, now, width))
|
|
else:
|
|
parts.append(Panel(_one_line(
|
|
"[grey62]nothing heard yet — these are a few milliwatts at "
|
|
"433.92 MHz, and a quarter-wave whip is 17 cm[/grey62]"),
|
|
border_style="grey37", padding=(0, 1)))
|
|
return Group(*parts)
|
|
|
|
def _header(self, here, now: float) -> Panel:
|
|
elapsed = max(0.001, now - self.started)
|
|
nameless = sum(1 for s in here if not self._name(s))
|
|
where = f" [grey62]{self.log_path.name}[/grey62]" if self.log_path \
|
|
else ""
|
|
# The offer to name something is in the header rather than at the
|
|
# bottom because the bottom of this display moves as sensors arrive.
|
|
naming = f"[bold cyan]n[/bold cyan] [grey62]to name" \
|
|
f"{f' ({nameless} unnamed)' if nameless else ''}[/grey62]"
|
|
return Panel(_one_line(
|
|
f"[bold cyan]{self.frequency / 1e6:g} MHz[/bold cyan] "
|
|
f"[bold]{len(here)}[/bold] sensor{'s' if len(here) != 1 else ''} "
|
|
f"[bold]{self.messages}[/bold] message"
|
|
f"{'s' if self.messages != 1 else ''} "
|
|
f"[grey62]{_dur(elapsed)}[/grey62]{where} {naming} "
|
|
f"[grey62]control-C to stop[/grey62]"),
|
|
border_style="blue", padding=(0, 1))
|
|
|
|
def _name(self, station) -> str:
|
|
return self.book.name_for(station.key) if self.book is not None else ""
|
|
|
|
def _table(self, here, now: float, width: int) -> Table:
|
|
"""As many columns as the terminal has room for, widest first.
|
|
|
|
A narrow terminal keeps the number, the name and what the sensor
|
|
said, and drops the model and the reception statistics: the first
|
|
three are why anyone is looking, and the rest can be read in the
|
|
report afterwards.
|
|
"""
|
|
t = Table(box=None, header_style="bold", pad_edge=False, expand=False)
|
|
t.add_column("#", style="grey62", width=2, justify="right")
|
|
t.add_column("name", width=14, no_wrap=True)
|
|
t.add_column("id", width=5, style="grey62", no_wrap=True)
|
|
if width >= 100:
|
|
# The same identity in decimal, which is the form other tools
|
|
# for this band print and the form anybody's own list is in.
|
|
t.add_column("dec", width=6, style="grey62", no_wrap=True,
|
|
justify="right")
|
|
if width >= 92:
|
|
t.add_column("model", width=16, style="grey62", no_wrap=True)
|
|
t.add_column("readings", overflow="fold")
|
|
if width >= 68:
|
|
# Widest-first, but this one is kept on a narrow terminal: it is
|
|
# what somebody moving an aerial about is watching, and they are
|
|
# not doing it on a wide window.
|
|
t.add_column("signal", width=7, justify="right")
|
|
t.add_column("batt", width=4, justify="center")
|
|
if width >= 84:
|
|
t.add_column("msgs", width=5, justify="right", style="grey62")
|
|
t.add_column("ago", width=5, justify="right", style="grey62")
|
|
for i, station in enumerate(here, 1):
|
|
name = self._name(station)
|
|
row = [str(i),
|
|
Text(name, style="bold") if name
|
|
else Text("unnamed", style="yellow"),
|
|
station.sensor]
|
|
if width >= 100:
|
|
row.append(station.number)
|
|
if width >= 92:
|
|
row.append(station.model or "")
|
|
row.append(self._readings(station, now))
|
|
if width >= 68:
|
|
from .weather import signal_text
|
|
|
|
row.append(Text.from_markup(signal_text(station.snr)))
|
|
row.append(Text("low", style="bold red") if station.battery_low
|
|
else Text("ok", style="green"))
|
|
if width >= 84:
|
|
row.append(f"{station.messages:,}")
|
|
row.append(_dur(max(0.0, now - station.last)))
|
|
t.add_row(*row)
|
|
return t
|
|
|
|
def _readings(self, station, now: float) -> Text:
|
|
"""Everything the sensor is currently saying, newest values first.
|
|
|
|
A quantity that has not been reported for a while is dimmed rather
|
|
than dropped. The 5-in-1 alternates its two messages, so half of
|
|
what it says is always a message old and dropping that would make
|
|
the line flicker; but a quantity that has been stale for minutes
|
|
while the sensor is otherwise fine is worth seeing greyed out.
|
|
"""
|
|
from .acurite import format_measure
|
|
|
|
out = Text()
|
|
for name, measure in station.values.items():
|
|
if out.plain:
|
|
out.append(" ")
|
|
stale = now - station.times.get(name, now) > 90.0
|
|
out.append(f"{name} ", style="grey62")
|
|
out.append(format_measure(measure, self.imperial),
|
|
style="grey58" if stale else "bold white")
|
|
if station.unread and not station.values:
|
|
out.append("framed, not understood", style="grey62")
|
|
return out
|