"""saunterbrowse -- read and listen to what a scan collected. A scan leaves a directory of recordings, each with a JSON sidecar holding what the classifier made of it and, for speech, a transcript. Reading that by hand means opening files one at a time and guessing which is worth playing. This browses them instead: arrow keys move, Enter plays, and the transcript -- the thing you actually want to read -- gets the top of the screen. """ from __future__ import annotations import argparse import glob import json import os import re import select import shutil import signal import subprocess import sys import termios import time import tty import wave from dataclasses import dataclass, field from datetime import datetime from pathlib import Path from rich.align import Align from rich.console import Console, Group from rich.layout import Layout from rich.live import Live from rich.markup import escape from rich.panel import Panel from rich.table import Table from rich.text import Text from . import __version__ from .bandplan import band_names, fmt_hz, shorten_band from .callsign import CallsignBook, HEADING, find_callsigns from .config import DEFAULT_CONFIG_DIR, load_default, remember_lockouts from .kml import DEFAULT_KML_NAME, KmlLog from .ranges import Lockout __all__ = ["main", "Capture", "scan_directory", "Browser", "Player", "FILING"] # --------------------------------------------------------------------------- # One recording on disk # --------------------------------------------------------------------------- # 0146.880000MHz--2026-08-22_12_54_59-nfm.wav -- the name a scan writes. _STEM = re.compile( r"^(?P\d+\.\d+)MHz" r"--(?P\d{4}-\d{2}-\d{2})_(?P\d{2})_(?P\d{2})_(?P\d{2})" r"-(?P.+)$") # Colour per content category, so the list can be read at a glance rather than # word by word. Voice is what most people are looking for, so it is the one # that stands out. CATEGORY_STYLE = { "voice": "bold green", "cw": "bold yellow", "digital": "cyan", "trunk": "bright_black", "carrier": "magenta", "noise": "bright_black", } @dataclass class Capture: """A .wav in the recordings directory, plus whatever sits beside it. The sidecars are read on demand and cached. A directory of ten thousand recordings would take a noticeable moment to open otherwise, and only the dozen rows actually on screen are ever needed. """ path: Path frequency: float = 0.0 started_at: float = 0.0 mode: str = "" combined: bool = False # one growing file per frequency, not a hit _meta: dict | None = field(default=None, init=False, repr=False) _transcript: str | None = field(default=None, init=False, repr=False) _duration: float | None = field(default=None, init=False, repr=False) _calls: list | None = field(default=None, init=False, repr=False) _decoded: list | None = field(default=None, init=False, repr=False) # -- lazily read sidecars --------------------------------------------- @property def meta(self) -> dict: """The ``hit`` block of the JSON sidecar, or an empty dict.""" if self._meta is None: self._meta = {} side = self.path.with_suffix(".json") try: doc = json.loads(side.read_text()) self._meta = doc.get("hit") or doc except (OSError, ValueError): pass return self._meta @property def transcript(self) -> str: """The speech in this recording, if a recogniser found any. Two places hold it and they can disagree: the sidecar records what was recognised at the time, while the .txt file is what a later re-run wrote. The file wins, being the more recent of the two. """ if self._transcript is None: self._transcript = "" for candidate in (self.path.with_name(self.path.stem + "_transcription.txt"),): try: self._transcript = candidate.read_text().strip() break except OSError: continue if not self._transcript: self._transcript = str(self.meta.get("transcript", "")).strip() return self._transcript @property def duration(self) -> float: """Seconds of audio, from the WAV header rather than the sidecar. The header is the only source that is true for a combined file, which grows every time another transmission is appended to it. """ if self._duration is None: self._duration = 0.0 try: with wave.open(str(self.path)) as w: rate = w.getframerate() if rate: self._duration = w.getnframes() / float(rate) except (wave.Error, OSError): pass return self._duration @property def decoded(self) -> list[str]: """What a data capture turned out to say, a line at a time. The file beside the recording wins over the sidecar for the same reason the transcript does: it is what the most recent run wrote. """ if self._decoded is None: lines: list[str] = [] try: text = self.path.with_name( self.path.stem + "_data.txt").read_text() lines = [ln for ln in text.splitlines() if ln.strip() and not ln.startswith("#")] except OSError: saved = self.meta.get("data_messages") or [] lines = [str(m) for m in saved if str(m).strip()] for key in ("data_hex", "data_bits"): value = str(self.meta.get(key, "")).strip() if value and not saved: lines.append(value) self._decoded = lines return self._decoded @property def data_headline(self) -> str: """What kind of data it was, never what the data said.""" protocol = str(self.meta.get("data_protocol", "")).strip() encoding = str(self.meta.get("data_encoding", "")).strip() if not (protocol or encoding): return "decoded data" if self.decoded else "" bits = [protocol or f"{encoding} data"] baud = self.meta.get("baud") if baud: bits.append(f"{float(baud):.0f} baud") repeats = int(self.meta.get("data_repeats") or 0) if repeats > 1: bits.append(f"x{repeats}") checks = self.meta.get("data_checks") or [] if checks: bits.append(str(checks[0])) return " ".join(bits) @property def callsigns(self) -> list[str]: """Callsign-shaped runs in the transcript, found once and kept.""" if self._calls is None: self._calls = find_callsigns(self.transcript) return self._calls # -- derived ---------------------------------------------------------- @property def bands(self) -> list[str]: """What the frequency is, most specific first. Preferring the sidecar means a recording keeps the band it was filed under even if the band plan is edited later; falling back to the frequency means recordings from before bands were recorded at all still get named. """ saved = self.meta.get("bands") or self.meta.get("band_labels") or [] names = [str(b) for b in saved if str(b).strip()] return names or (band_names(self.frequency) if self.frequency else []) @property def band(self) -> str: bands = self.bands return bands[0] if bands else "" @property def category(self) -> str: if self.combined: return "combined" return str(self.meta.get("category", "")) or "unknown" @property def classification(self) -> str: if self.combined: return "every transmission heard on this frequency" return str(self.meta.get("classification", "")) or "unclassified" @property def when(self) -> datetime | None: return (datetime.fromtimestamp(self.started_at) if self.started_at else None) def size_bytes(self) -> int: try: return self.path.stat().st_size except OSError: return 0 def files(self) -> list[Path]: """Every file that belongs to this recording, the .wav included. A capture is not one file. Beside it a scan writes the JSON sidecar, sometimes the raw IQ and its SigMF description, the transcript, and the decoded data. They are one thing, and moving or deleting the recording without them leaves sidecars describing a recording nobody has any more. Matched on the stem followed by a dot, never on the stem followed by anything: two signals can land in the same second on the same frequency, in which case the second is filed as ``...-nfm_2``, and a looser pattern would sweep that up with its neighbour. """ stem = self.path.stem found = {self.path} try: found.update(p for p in self.path.parent.glob(glob.escape(stem) + ".*") if p.is_file()) except OSError: pass # The two that do not follow the pattern, because a transcript is # named after the recording rather than sharing its extension. for tail in ("_transcription.txt", "_data.txt"): side = self.path.with_name(stem + tail) if side.is_file(): found.add(side) return sorted(found) def _from_name(path: Path) -> Capture: """Read what the filename alone says, without touching any sidecar.""" m = _STEM.match(path.stem) if not m: # A combined per-frequency file is named for its frequency and nothing # else. Anything else unparseable is still listed: a recording the # browser cannot name is better shown than hidden. freq = 0.0 bare = re.match(r"^(\d+\.\d+)MHz$", path.stem) if bare: freq = float(bare.group(1)) * 1e6 try: mtime = path.stat().st_mtime except OSError: mtime = 0.0 return Capture(path=path, frequency=freq, started_at=mtime, mode="", combined=bool(bare)) when = datetime.strptime( f"{m['date']} {m['h']}:{m['m']}:{m['s']}", "%Y-%m-%d %H:%M:%S") return Capture(path=path, frequency=float(m["mhz"]) * 1e6, started_at=when.timestamp(), mode=m["mode"]) def scan_directory(directory: Path) -> list[Capture]: """Every recording in one directory, newest first.""" try: names = sorted(p for p in Path(directory).iterdir() if p.suffix.lower() == ".wav" and p.is_file()) except OSError: return [] caps = [_from_name(p) for p in names] caps.sort(key=lambda c: c.started_at, reverse=True) return caps # --------------------------------------------------------------------------- # Playback # --------------------------------------------------------------------------- # In preference order. The first that exists is used; the flags are whatever # each one needs to play a file once and exit without opening a window. PLAYERS: tuple[tuple[str, tuple[str, ...]], ...] = ( ("pw-play", ()), ("paplay", ()), ("aplay", ("-q",)), ("play", ("-q",)), ("ffplay", ("-nodisp", "-autoexit", "-loglevel", "quiet")), ("mpv", ("--no-video", "--really-quiet")), ) class Player: """Plays one file at a time, in a child process. A child rather than an audio library: the recordings are ordinary WAVs, every desktop already has something that plays them, and a browser that cannot start is worse than one that cannot play. """ def __init__(self, command: list[str] | None = None): self.command = command if command is not None else self._find() self.proc: subprocess.Popen | None = None self.playing: Capture | None = None self.started_at = 0.0 self.error = "" @staticmethod def _find() -> list[str] | None: for name, args in PLAYERS: found = shutil.which(name) if found: return [found, *args] return None @property def available(self) -> bool: return bool(self.command) def play(self, cap: Capture) -> None: self.stop() if not self.command: self.error = ("no audio player found -- install one of: " + ", ".join(n for n, _ in PLAYERS)) return try: self.proc = subprocess.Popen( [*self.command, str(cap.path)], stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True) except OSError as exc: self.error = f"{self.command[0]}: {exc}" return self.error = "" self.playing = cap self.started_at = time.time() def stop(self) -> None: if self.proc is not None and self.proc.poll() is None: # The whole group, not just the child. Several of these players # are wrapper scripts that exec or fork the real one, and # signalling only the script leaves the sound playing with nothing # on screen to stop it. start_new_session put them in a group of # their own precisely so this is safe. try: os.killpg(os.getpgid(self.proc.pid), signal.SIGTERM) except (OSError, ProcessLookupError): try: self.proc.terminate() except OSError: pass self.proc = None self.playing = None @property def active(self) -> bool: if self.proc is None: return False if self.proc.poll() is not None: self.proc = None self.playing = None return False return True @property def elapsed(self) -> float: return time.time() - self.started_at if self.active else 0.0 # --------------------------------------------------------------------------- # Keyboard # --------------------------------------------------------------------------- # An arrow key is not a character: the terminal sends "\x1b[A", three bytes, # and reading them one at a time turns one keypress into three commands. _ESCAPES = { "[A": "up", "[B": "down", "[C": "right", "[D": "left", "[H": "home", "[F": "end", "[5~": "pgup", "[6~": "pgdn", "OA": "up", "OB": "down", "OC": "right", "OD": "left", "OH": "home", "OF": "end", "[1~": "home", "[4~": "end", } class Keyboard: """Non-blocking key reader that understands escape sequences. Reads the file descriptor rather than the stream. ``sys.stdin.read(1)`` goes through a buffered text wrapper, which in cbreak mode blocks until it has enough bytes to be sure of a character -- so the first keypress never returned and the program hung with the screen drawn. """ NAMED = {"\r": "enter", "\n": "enter", " ": "space", "\x7f": "backspace", "\x08": "backspace", "\x03": "q"} def __init__(self, stream=None): self.stream = stream or sys.stdin self.fd: int | None = None self.enabled = False self._old = None self._pending = "" def __enter__(self): try: if self.stream.isatty(): self.fd = self.stream.fileno() self._old = termios.tcgetattr(self.fd) tty.setcbreak(self.fd) self.enabled = True except (termios.error, ValueError, OSError, AttributeError): self.enabled = False return self def __exit__(self, *exc): if self._old is not None and self.fd is not None: try: termios.tcsetattr(self.fd, termios.TCSADRAIN, self._old) except (termios.error, ValueError, OSError): pass return False def _fill(self, timeout: float) -> bool: """Take whatever is waiting on the terminal into the buffer.""" if self.fd is None: return False try: if not select.select([self.fd], [], [], timeout)[0]: return False data = os.read(self.fd, 64) except (OSError, ValueError): return False if not data: return False self._pending += data.decode("utf8", "replace") return True def get(self, timeout: float = 0.1) -> str: """One key, named. Empty string when nothing was pressed.""" if not self.enabled: time.sleep(timeout) return "" if not self._pending and not self._fill(timeout): return "" return self._take() def _take(self) -> str: ch = self._pending[0] if ch != "\x1b": self._pending = self._pending[1:] return self.NAMED.get(ch, ch) # Escape, or the first byte of a sequence. A terminal sends the rest # in the same breath, so a short wait separates a pressed Esc from an # arrow without making Esc feel slow. if len(self._pending) == 1: self._fill(0.02) body = self._pending[1:] for n in (3, 2): if body[:n] in _ESCAPES: self._pending = self._pending[1 + n:] return _ESCAPES[body[:n]] self._pending = self._pending[1:] return "escape" # --------------------------------------------------------------------------- # The browser # --------------------------------------------------------------------------- SORTS = ("time", "frequency", "duration") # The three subdirectories a recording can be filed into, and the key that # does it. Upper case on purpose: j and k are under the same fingers, and a # key that moves a file is not one to hit while scrolling. They are ordinary # directories under the recordings directory, so a scan never looks in them # again and the browser opens one by being pointed at it. FILING: tuple[tuple[str, str, str], ...] = ( ("S", "saved", "keep this one"), ("I", "investigate", "come back to this one"), ("N", "noise", "not a signal worth keeping"), ) _FILING_KEYS = {key: name for key, name, _ in FILING} def _dur(seconds: float) -> str: seconds = max(0.0, float(seconds)) if seconds >= 3600: return f"{int(seconds // 3600)}h{int(seconds % 3600 // 60):02d}m" if seconds >= 60: return f"{int(seconds // 60)}m{int(seconds % 60):02d}s" return f"{seconds:.1f}s" def _size(n: int) -> str: for unit, scale in (("G", 1 << 30), ("M", 1 << 20), ("k", 1 << 10)): if n >= scale: return f"{n / scale:.1f} {unit}B" return f"{n} B" class Browser: """The whole application: state, rendering and the key loop.""" def __init__(self, directory: Path, console: Console | None = None, player: Player | None = None, book: CallsignBook | None = None, config_dir: Path | None = None): self.directory = Path(directory) self.console = console or Console() self.player = player if player is not None else Player() self.book = book if book is not None else CallsignBook() # Where the lock-out key writes. Passed in rather than reached for, # so that nothing can write to the settings a real scan runs from # without having been told to. self.config_dir = Path(config_dir) if config_dir else None self.captures: list[Capture] = [] self.view: list[Capture] = [] self.index = 0 self.top = 0 # first row shown in the list self.sort = "time" self.query = "" self.searching = False self.confirm = "" # an action waiting to be agreed to self.message = "" self.show_help = False self.reading = False # full-screen transcript self.read_top = 0 self._undo: list[tuple[Path, Path]] = [] # the last move, to reverse self._undo_path: Path | None = None # where its .wav came from self._undo_label = "" self._size = self.console.size self.reload() # -- state ------------------------------------------------------------ def reload(self) -> None: keep = self.current.path if self.view else None self.captures = scan_directory(self.directory) self.apply() if keep is not None: for i, cap in enumerate(self.view): if cap.path == keep: self.index = i break def apply(self) -> None: """Re-filter and re-sort, keeping the cursor inside the result.""" q = self.query.lower().strip() if q: self.view = [c for c in self.captures if self._matches(c, q)] else: self.view = list(self.captures) if self.sort == "frequency": self.view.sort(key=lambda c: (c.frequency, c.started_at)) elif self.sort == "duration": self.view.sort(key=lambda c: c.duration, reverse=True) else: self.view.sort(key=lambda c: c.started_at, reverse=True) self.index = max(0, min(self.index, len(self.view) - 1)) @staticmethod def _matches(cap: Capture, q: str) -> bool: """Search the name, the identification and the words that were said. Searching the transcript is the point of it: "did anyone mention the repeater" is a question about content, not about filenames. """ if q in cap.path.name.lower(): return True if q in cap.classification.lower() or q in cap.category.lower(): return True # By band as well as by frequency: "70 cm" is how an operator thinks # of a range, and remembering 420-450 MHz is not the point. if any(q in band.lower() for band in cap.bands): return True # And in what a data capture said: a pager message is as searchable # as a spoken one, and the reason for searching is the same. if any(q in line.lower() for line in cap.decoded): return True return q in cap.transcript.lower() @property def current(self) -> Capture | None: if not self.view: return None return self.view[max(0, min(self.index, len(self.view) - 1))] def move(self, delta: int) -> None: if not self.view: return self.index = max(0, min(len(self.view) - 1, self.index + delta)) # -- managing what is on disk ----------------------------------------- # # Everything here works on the whole capture -- the recording and every # sidecar written beside it -- and leaves the cursor on the row it was # on, which is now the next recording. Going through a night's scan is # one key per recording, and a cursor that jumped after each one would # make that impossible. def _forget(self, cap: Capture) -> None: """Drop a capture from the lists without re-reading the directory.""" for group in (self.captures, self.view): try: group.remove(cap) except ValueError: pass self.index = max(0, min(self.index, len(self.view) - 1)) self.top = max(0, min(self.top, max(0, len(self.view) - 1))) @staticmethod def _plan_move(cap: Capture, target: Path) -> list[tuple[Path, Path]]: """Where each of a capture's files should go, avoiding collisions. The whole set is renamed together or not at all: a recording called one thing and a transcript called another is a pair nothing will ever put back together. """ files = cap.files() stem = cap.path.stem tails = [f.name[len(stem):] for f in files] candidate, n = stem, 2 while any((target / (candidate + tail)).exists() for tail in tails): candidate, n = f"{stem}_{n}", n + 1 return [(f, target / (candidate + tail)) for f, tail in zip(files, tails)] def _release(self, cap: Capture) -> None: """Stop playing this recording, if it is the one being played. Moving or deleting a file out from under a player leaves sound coming out of a recording that is no longer there, and a progress bar counting up against a name nothing on disk answers to. """ if (self.player.playing is not None and self.player.playing.path == cap.path): self.player.stop() def _file_into(self, name: str) -> None: """Move the highlighted capture into a subdirectory of its own.""" cap = self.current if cap is None: return self._release(cap) target = self.directory / name try: target.mkdir(parents=True, exist_ok=True) except OSError as exc: self.message = f"cannot make {name}/: {exc}" return done: list[tuple[Path, Path]] = [] for src, dst in self._plan_move(cap, target): try: src.rename(dst) except OSError as exc: # Put back whatever has already moved. Half a capture in # each of two directories is worse than none moved at all. for was, now in reversed(done): try: now.rename(was) except OSError: pass self.message = f"could not move {src.name}: {exc}" return done.append((src, dst)) self._undo, self._undo_path = done, cap.path self._undo_label = f"{cap.path.name} back from {name}/" self._forget(cap) plural = "" if len(done) == 1 else "s" self.message = (f"{cap.path.name} → {name}/ " f"({len(done)} file{plural}) u to undo") def _undo_move(self) -> None: """Reverse the last filing. Not a delete: that one is gone.""" if not self._undo: self.message = "nothing to put back" return moves, self._undo = self._undo, [] home, self._undo_path = self._undo_path, None failed = 0 for was, now in reversed(moves): try: now.rename(was) except OSError: failed += 1 # A full re-read, because the capture has to go back into the list in # the place the current sort order puts it, not the place it was. self.reload() if home is not None: for i, cap in enumerate(self.view): if cap.path == home: self.index = i break self.message = (f"{failed} file(s) could not be put back" if failed else f"put {self._undo_label}") self._undo_label = "" def _delete(self) -> None: """Delete the highlighted capture and everything written beside it.""" cap = self.current if cap is None: return self._release(cap) gone = 0 for path in cap.files(): try: path.unlink() gone += 1 except OSError as exc: self.message = f"could not delete {path.name}: {exc}" if gone: self.reload() return # There is nothing to undo a delete with, and leaving the last move # under u would put the wrong thing back. self._undo, self._undo_path, self._undo_label = [], None, "" self._forget(cap) others = gone - 1 self.message = f"deleted {cap.path.name}" + ( f" and {others} sidecar{'' if others == 1 else 's'}" if others else "") def _mask(self) -> None: """Lock this frequency out, so no later scan stops on it again. The same list the scanner's own lock-out key writes to, and the same file, so a birdie masked here while reading last night's recordings is gone from tonight's. """ cap = self.current if cap is None: return if not cap.frequency: self.message = ("nothing in this recording's name says what " "frequency it was on") return directory = self.config_dir or DEFAULT_CONFIG_DIR cfg, _ = load_default(directory) width = max(1.0, cfg.lockout_width) for existing in cfg.lockout: low, high = Lockout.coerce(existing).interval(width) if low <= cap.frequency <= high: self.message = (f"{fmt_hz(cap.frequency)} is already locked " "out") return entry = Lockout(cap.frequency) cfg.lockout.append(entry) written = remember_lockouts(cfg, directory) if written is None: self.message = f"could not write the lock-out to {directory}" return # "The next scan", not "scans": one already running read its settings # when it started and will not see this. self.message = (f"locked out {entry.describe()} ±{fmt_hz(width / 2)} " f"in {written} — the next scan will not stop there") # -- rendering -------------------------------------------------------- def _callsign_lines(self, pad: int = 3) -> list[str]: """The DETECTED CALLSIGNS block, wrapped to the panel's width. Kept out of _transcript_lines so that scrolling the reader never scrolls the callsigns off the bottom: they belong to the whole transcript, not to the part of it currently on screen. """ cap = self.current if cap is None or not cap.callsigns: return [] width = max(24, (self.console.size.width or 80) - 2 - 2 * pad) entries = self.book.get_all(cap.callsigns) out = [HEADING] label = max(len(e.call) for e in entries) for entry in entries: line = f" {entry.call:<{label}} {entry.summary()}" # As much of the detail as fits, dropping whole items from the # end. All-or-nothing threw away the licence class and the grid # square on any terminal narrower than the widest entry. for item in entry.details(): if len(line) + len(item) + 5 > width: break line += " · " + item out.append(line[:width]) return out def _transcript_lines(self, pad: int = 3) -> list[str]: """The transcript wrapped to the width it will be drawn at. The width has to be the panel's own, not the screen's: wrapping to one width and drawing at a narrower one wraps everything a second time, which strands the last word or two of each line on a line of its own. """ cap = self.current if cap is None or not cap.transcript: return [] width = max(20, (self.console.size.width or 80) - 2 - 2 * pad) text = Text(cap.transcript, style="bold white") return [line.plain.rstrip() for line in text.wrap(self.console, width)] def _transcript_height(self) -> int: """Tall enough for the words, but never more than a third of the screen. A fixed panel is wrong in both directions: an eight-line frame around one sentence is mostly empty, and the same frame around a four-minute net swallows the rest of it without saying so. """ screen = self.console.size.height or 24 calls = self._callsign_lines() # The callsigns are the answer to a question the transcript raised, so # they get room of their own rather than eating into it: a taller # ceiling when there are any, and never fewer than enough to show them. ceiling = max(6, min(20 if calls else 16, screen // (2 if calls else 3))) cap = self.current body = len(self._transcript_lines()) or ( len(cap.decoded) + 2 if cap is not None and cap.decoded else 0) needed = body + len(calls) + 4 if calls: needed += 1 # the blank line above floor = min(ceiling, len(calls) + 6) return max(6, floor, min(ceiling, needed)) def _rows(self) -> int: """How many list rows fit under everything above them.""" height = self.console.size.height or 24 # header 3, transcript n, details 4, footer 1, the list's own borders 2 return max(3, height - 3 - self._transcript_height() - 4 - 1 - 2) def _header(self) -> Panel: cap = self.current if cap is None: body = Text("no recordings here", style="bright_black") return Panel(Align.center(body), border_style="bright_black", padding=(0, 1)) left = Text() left.append(fmt_hz(cap.frequency) if cap.frequency else cap.path.stem, style="bold cyan") if cap.mode: left.append(" ") left.append(cap.mode.upper(), style="bold white") when = cap.when if when is not None: left.append(" ") left.append(when.strftime("%a %d %b %H:%M:%S"), style="white") left.append(" ") left.append(_dur(cap.duration), style="bright_black") snr = cap.meta.get("snr_db") if snr: left.append(f" SNR {float(snr):.1f} dB", style="bright_black") left.append(" ") left.append(cap.category, style=CATEGORY_STYLE.get(cap.category, "white")) return Panel(left, border_style="blue", padding=(0, 1), title=f"[bright_black]{self.index + 1} of " f"{len(self.view)}[/bright_black]", title_align="right") def _transcript_panel(self) -> Panel: """The top of the screen, and the reason this program exists.""" cap = self.current height = self._transcript_height() if cap is None: return Panel("", border_style="bright_black", height=height) if not cap.transcript and cap.decoded: # A data capture has no words, but it does have content, and it # belongs in the same place a transcript would be: at the top, # where the reader is already looking. body = Text() headline = cap.data_headline if headline: body.append(headline + "\n\n", style="bold cyan") room = max(1, height - 4 - (2 if headline else 0)) shown = cap.decoded[:room] body.append("\n".join(shown), style="bold white") if len(cap.decoded) > len(shown): body.append(f"\n… {len(cap.decoded) - len(shown)} more line(s)" " — press t to read it all", style="yellow") return Panel(body, title="decoded", title_align="left", border_style="cyan", padding=(1, 3), height=height) if not cap.transcript: inner = Align.center(Text(self._why_no_transcript(cap), style="bright_black", justify="center"), vertical="middle") return Panel(inner, title="no transcript", title_align="left", border_style="bright_black", padding=(1, 3), height=height) lines = self._transcript_lines() calls = self._callsign_lines() # Whatever else is squeezed, the callsigns stay: they are the part of # this panel that cannot be recovered by listening to the recording. room = height - 4 - (len(calls) + 1 if calls else 0) overflows = len(lines) > room shown = lines[:room - 1] if overflows else lines hidden = len(lines) - len(shown) body = Text("\n".join(shown), style="bold white") if hidden > 0: # Say so. Silently cutting the end off a transmission is how a # reader ends up believing they have read all of it. body.append(f"\n… {hidden} more line(s) — press t to read it all", style="not bold yellow") if calls: body.append("\n\n") body.append(calls[0], style="not bold bold cyan") for line in calls[1:]: body.append("\n") body.append(line, style="not bold white") block = body if not calls and not hidden and len(shown) + 2 < height: block = Align.left(body, vertical="middle") return Panel(block, title="transcript", title_align="left", border_style="green", padding=(1, 3), height=height) @staticmethod def _why_no_transcript(cap: Capture) -> str: """Say which of the several reasons applies, rather than just 'none'. An empty panel is the same shape whether the recording was Morse, was silent, or was never offered to a recogniser, and those want different things done about them. """ if cap.combined: return ("one file per frequency\n" "transcripts for these are appended to a .txt beside it") cat = cap.category if cap.meta.get("morse_text"): return f'Morse, decoded as:\n"{cap.meta["morse_text"].strip()}"' if cat == "cw": return "Morse -- nothing for a speech recogniser to hear" if cat in ("digital", "trunk"): return "data, not speech" if cat == "carrier": return "an unmodulated carrier -- silence by definition" if cat == "voice": return ("speech, but no transcript beside it\n" "run: bandsaunter transcribe --engines") if not cap.meta: return "no sidecar beside this recording" return "nothing was transcribed for this recording" def _details(self) -> Panel: cap = self.current if cap is None: return Panel("", border_style="bright_black", height=4) line = Text() line.append(cap.classification, style=CATEGORY_STYLE.get(cap.category, "white")) conf = cap.meta.get("confidence") if conf: line.append(f" {float(conf) * 100:.0f}%", style="bright_black") if cap.meta.get("ctcss_hz"): line.append(f" CTCSS {float(cap.meta['ctcss_hz']):.1f} Hz", style="yellow") # Only where it means something. The estimator returns a figure for # every capture, and "120 baud" beside a conversation is noise. if cap.meta.get("baud") and cap.category in ("digital", "trunk", "cw"): line.append(f" {float(cap.meta['baud']):.0f} baud", style="cyan") if cap.meta.get("morse_wpm"): line.append(f" {float(cap.meta['morse_wpm']):.0f} WPM", style="yellow") checks = cap.meta.get("data_checks") or [] if checks: # A checksum that came out right is the strongest thing anyone # can say about a decode, so it is said on the front line. line.append(f" {checks[0]} ✓", style="green") second = Text() bands = cap.bands if bands: second.append(bands[0], style="magenta") if len(bands) > 1: second.append(" " + " · ".join(bands[1:3]), style="bright_black") elif cap.meta.get("range_label"): second.append(str(cap.meta["range_label"]), style="bright_black") else: second.append(f"{cap.path.name} {_size(cap.size_bytes())}", style="bright_black") return Panel(Group(line, second), border_style="bright_black", padding=(0, 1), height=4) def _list(self) -> Panel: rows = self._rows() # Keep the cursor in view with a little context either side, so moving # never lands on the very edge of the window. margin = 2 if rows > 6 else 0 if self.index < self.top + margin: self.top = max(0, self.index - margin) elif self.index >= self.top + rows - margin: self.top = self.index - rows + 1 + margin self.top = max(0, min(self.top, max(0, len(self.view) - rows))) t = Table(box=None, expand=True, pad_edge=False, show_edge=False, show_header=False) t.add_column(width=2) # cursor / playing marker t.add_column(width=15, justify="right") # frequency t.add_column(width=8) # time t.add_column(width=5) # mode t.add_column(width=7, justify="right") # duration # One line per recording, always. A wrapped row would push the ones # below it off the bottom, and the cursor arithmetic counts rows. t.add_column(ratio=1, overflow="ellipsis", no_wrap=True) for i in range(self.top, min(len(self.view), self.top + rows)): cap = self.view[i] here = i == self.index playing = (self.player.playing is not None and self.player.playing.path == cap.path) mark = "▶" if playing else ("›" if here else " ") base = "on grey19 " if here else "" cat = CATEGORY_STYLE.get(cap.category, "white") when = cap.when summary = (cap.transcript.replace("\n", " ") or (cap.decoded[0] if cap.decoded else "") or cap.classification) t.add_row( Text(mark, style=base + ("bold red" if playing else "bold cyan")), Text(fmt_hz(cap.frequency) if cap.frequency else cap.path.stem[:15], style=base + ("bold cyan" if here else "cyan")), Text(when.strftime("%H:%M:%S") if when else "", style=base + "bright_black"), Text(cap.mode or "", style=base + "white"), Text(_dur(cap.duration), style=base + "bright_black"), Text(summary, style=base + (cat if not cap.transcript else "white"), no_wrap=True, overflow="ellipsis"), ) if not self.view: hint = ("nothing here yet" if not self.captures else f"nothing matches '{self.query}'") t.add_row("", "", "", "", "", Text(hint, style="bright_black")) title = f"recordings in {self.directory}" if self.query: title += f" filter: {self.query}" return Panel(t, title=title, title_align="left", border_style="blue", padding=(0, 1)) @staticmethod def _line(markup: str) -> Text: """One line, and never more, whatever it was asked to hold. The footer sits in a region exactly one row tall. A line that wrapped would push the frame past the bottom of the window, and the whole display depends on the frame being the height it says it is. """ text = Text.from_markup(markup) text.no_wrap = True text.overflow = "ellipsis" return text def _keyline(self) -> Text: """The keys, as many of them as the window is wide enough to hold. Dropped in a deliberate order rather than truncated: an ellipsis in the middle of the line hides whichever keys happen to be at the end, and "q quit" is the one nobody can afford not to be told. """ items = [("↑↓", "move"), ("⏎", "play"), ("space", "stop"), ("/", "search"), ("t", "read"), ("S I N", "file"), ("d", "delete"), ("m", "mask"), ("s", f"sort by {self.sort}"), ("r", "reload"), ("?", "keys"), ("q", "quit")] # Everything given up here is still on the page ? opens, which is why # ? is among the last to go. expendable = ["r", "s", "space", "m", "d", "S I N", "t", "/", "⏎", "↑↓", "?"] width = self.console.size.width or 80 shown = items while expendable and sum(len(k) + len(w) + 4 for k, w in shown) - 3 > width: gone = expendable.pop(0) shown = [item for item in shown if item[0] != gone] text = Text(no_wrap=True, overflow="ellipsis") for i, (key, what) in enumerate(shown): if i: text.append(" ") text.append(key, style="bright_black") text.append(" " + what) return text def _footer(self) -> Text: # All of these carry text that did not come from this program -- what # the user typed, a filename, a player's error -- and rich reads a # square bracket as markup. Typing "[/" at the search prompt used to # end the session with a MarkupError. if self.confirm: return self._line( f"[bold yellow]{escape(self.confirm)}[/bold yellow] " "[bold]y[/bold][bright_black]/[/bright_black][bold]n[/bold]") if self.searching: return self._line( f"[bold]search:[/bold] {escape(self.query)}[blink]_[/blink]" " [bright_black]enter to accept, esc to clear" "[/bright_black]") if self.message: return self._line(f"[yellow]{escape(self.message)}[/yellow]") if self.player.active and self.player.playing is not None: total = self.player.playing.duration done = self.player.elapsed width = max(10, min(40, (self.console.size.width or 80) - 46)) filled = int(width * min(1.0, done / total)) if total else 0 bar = ("[red]" + "━" * filled + "[/red]" + "[grey37]" + "━" * (width - filled) + "[/grey37]") return self._line( f"[bold red]♪[/bold red] {bar} {_dur(done)}/{_dur(total)}" " [bright_black]space stop[/bright_black]") return self._keyline() def _help(self) -> Panel: """Every key, on one screen, in a window the size a terminal opens at. Eighty by twenty-four is still what a new terminal is, and a list of keys that has to be scrolled to reach "q" is a list that failed at the one job it has. So it is kept to twenty rows that fit in sixty columns: the three filing keys share a line, because the directories they name say what each one does, and the player goes in the border rather than costing a row. """ t = Table(box=None, show_header=False, pad_edge=False) t.add_column(style="bold cyan", width=14) t.add_column() filing = " ".join(key for key, _, _ in FILING) names = ", ".join(f"{name}/" for _, name, _ in FILING[:-1]) names += f" or {FILING[-1][1]}/" for key, what in ( ("↑ ↓ / k j", "move through the recordings"), ("PgUp PgDn", "a screenful at a time"), ("Home End", "first and last"), ("Enter", "play the highlighted recording"), ("space", "stop playing"), ("t", "read the whole transcript, full screen"), ("/", "filter by frequency, name, class or anything said"), ("Esc", "clear the filter"), ("s", "sort by time, frequency or length"), ("r", "re-read the directory"), ("o", "print the file's path and quit"), ("", ""), (filing, f"file it into {names}"), ("u", "put the last one filed back"), ("d", "delete it and its sidecars for good; asks first"), ("m", "lock this frequency out of every later scan"), ("", ""), ("callsigns", "found in transcripts and looked up for you"), ("? h", "this list"), ("q", "quit")): t.add_row(key, what) player = (" ".join(self.player.command) if self.player.command else "none found -- install pw-play, paplay, aplay, " "sox or ffmpeg") return Panel(t, title="keys", title_align="left", subtitle=f"[bright_black]player: {escape(player)}" "[/bright_black]", subtitle_align="right", border_style="cyan", padding=(1, 2)) def _reader(self) -> Panel: """The whole transcript, with nothing else competing for the screen.""" cap = self.current screen = self.console.size.height or 24 calls = self._callsign_lines(pad=4) room = max(3, screen - 4 - (len(calls) + 1 if calls else 0)) lines = self._transcript_lines(pad=4) title_word = "transcript" if not lines and cap is not None and cap.decoded: # A long paging capture is as worth reading in full as a long # net, and there is nowhere else to read it. lines = ([cap.data_headline, ""] if cap.data_headline else []) \ + list(cap.decoded) title_word = "decoded" self.read_top = max(0, min(self.read_top, max(0, len(lines) - room))) shown = lines[self.read_top:self.read_top + room] body = Text("\n".join(shown), style="white") calls = self._callsign_lines(pad=4) if calls and self.read_top + room >= len(lines): # Only once the reader has reached the end of the words: the # callsigns belong under the transcript, not floating beside the # middle of it. body.append("\n\n") body.append(calls[0], style="bold cyan") for line in calls[1:]: body.append("\n" + line) where = (f"{self.read_top + 1}-{self.read_top + len(shown)}" f" of {len(lines)}" if len(lines) > room else "") title = title_word if cap is not None and cap.frequency: title += f" · {fmt_hz(cap.frequency)}" when = cap.when if when is not None: title += f" · {when.strftime('%d %b %H:%M:%S')}" return Panel(body, title=title, title_align="left", subtitle=f"[bright_black]{where} ↑↓ scroll " f"⏎ play t or esc back[/bright_black]", subtitle_align="right", border_style="green", padding=(1, 4)) def render(self): # Every screen is wrapped in a layout, which fills the terminal # exactly. Nothing erases the alternate screen between frames -- the # cursor is sent home and the new frame written over the old one -- so # a frame shorter than the screen leaves the tail of the last one # visible below it. Pressing t on a full window used to leave most of # the recording list under the reader. if self.show_help: return Layout(Align.center(self._help(), vertical="middle")) if self.reading: return Layout(self._reader()) layout = Layout() # One height, computed once: the panel and the region it sits in have # to agree, or the difference shows up as a gap in the middle of the # screen. tall = self._transcript_height() layout.split_column( Layout(self._header(), size=3), Layout(self._transcript_panel(), size=tall), Layout(self._details(), size=4), Layout(self._list(), ratio=1), Layout(self._footer(), size=1), ) return layout # -- key handling ----------------------------------------------------- def handle(self, key: str) -> bool: """Act on one key. Returns False when it is time to stop.""" if not key: return True self.message = "" if self.confirm: return self._handle_confirm(key) if self.searching: return self._handle_search(key) if self.show_help and key not in ("?", "h", "q"): self.show_help = False return True if self.reading: return self._handle_reader(key) rows = self._rows() if key in ("q", "escape") and not self.query: return False if key == "escape": self.query = "" self.apply() elif key == "q": return False elif key in ("up", "k"): self.move(-1) elif key in ("down", "j"): self.move(1) elif key in ("pgup", "left"): self.move(-rows) elif key in ("pgdn", "right"): self.move(rows) elif key == "home": self.index = 0 elif key == "end": self.index = max(0, len(self.view) - 1) elif key == "enter": self._play() elif key == "space": if self.player.active: self.player.stop() else: self._play() elif key == "s": self.sort = SORTS[(SORTS.index(self.sort) + 1) % len(SORTS)] self.apply() self.message = f"sorted by {self.sort}" elif key == "r": self.reload() self.message = f"{len(self.captures)} recording(s)" elif key == "/": self.searching = True self.query = "" elif key in ("?", "h"): self.show_help = not self.show_help elif key == "t": cap = self.current # Either kind of content: the decoded panel says "press t to read # it all" when a long paging capture overflows it, and the key has # to mean what the panel says it means. if cap is not None and (cap.transcript or cap.decoded): self.reading = True self.read_top = 0 else: self.message = "nothing to read for this recording" elif key == "o": cap = self.current if cap is not None: self.message = str(cap.path) return False elif key in _FILING_KEYS: self._file_into(_FILING_KEYS[key]) elif key == "u": self._undo_move() elif key == "m": self._mask() elif key == "d": cap = self.current if cap is None: return True # Asked rather than done. Filing is reversible and is not worth # a question every time; this one is not reversible at all, and # d is next to s and f on the same row of the keyboard. others = len(cap.files()) - 1 self.confirm = "delete " + cap.path.name + ( f" and {others} sidecar file{'' if others == 1 else 's'}?" if others else "?") return True def _handle_confirm(self, key: str) -> bool: """One question, one answer. Anything but yes means no.""" question, self.confirm = self.confirm, "" if key not in ("y", "Y"): self.message = "left alone" return True if question.startswith("delete "): self._delete() return True def _handle_reader(self, key: str) -> bool: rows = max(3, (self.console.size.height or 24) - 6) if key in ("t", "escape", "q", "space"): self.reading = False elif key in ("up", "k"): self.read_top -= 1 elif key in ("down", "j"): self.read_top += 1 elif key in ("pgup", "left"): self.read_top -= rows elif key in ("pgdn", "right"): self.read_top += rows elif key == "home": self.read_top = 0 elif key == "end": self.read_top = 1 << 20 # clamped when the panel is drawn elif key == "enter": self._play() self.read_top = max(0, self.read_top) return True def _handle_search(self, key: str) -> bool: if key == "enter": self.searching = False elif key == "escape": self.searching = False self.query = "" elif key == "backspace": self.query = self.query[:-1] elif len(key) == 1 and key.isprintable(): self.query += key else: return True self.index = 0 self.top = 0 self.apply() return True def _play(self) -> None: cap = self.current if cap is None: return self.player.play(cap) if self.player.error: self.message = self.player.error def resized(self) -> bool: """True once each time the terminal has changed size. Polled rather than handled as a signal: the screen is redrawn on every key and on every timeout anyway, and a signal handler that runs in the middle of a write would have to be right about far more than this does. """ size = self.console.size if size == self._size: return False self._size = size return True # -- main loop -------------------------------------------------------- def run(self) -> int: with Keyboard() as keys, Live(self.render(), console=self.console, screen=True, auto_refresh=False, transient=False) as live: while True: key = keys.get(0.15 if self.player.active else 0.4) if not self.handle(key): break if self.resized(): # What is on the screen was drawn for a window that no # longer exists, and nothing here erases before it draws. self.console.clear() live.update(self.render(), refresh=True) self.player.stop() if self.message: self.console.print(self.message) return 0 # --------------------------------------------------------------------------- # Entry point # --------------------------------------------------------------------------- def default_directory() -> Path: """Where a scan writes, so the browser opens on it without being told.""" env = os.environ.get("BANDSAUNTER_OUTPUT") if env: return Path(env).expanduser() try: cfg, _ = load_default() return Path(cfg.output_dir).expanduser() except Exception: return Path("~/bandsaunter").expanduser() def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser( prog="saunterbrowse", description="Read and listen to what a bandsaunter scan collected.", epilog="Arrow keys move, Enter plays, / searches what was said.") p.add_argument("directory", nargs="?", default=None, help="where the recordings are " "(default: the scanner's output directory)") p.add_argument("--sort", choices=SORTS, default="time", help="initial order (default: time, newest first)") p.add_argument("--filter", default="", metavar="TEXT", help="start with only recordings matching this") p.add_argument("--player", default=None, metavar="CMD", help="command to play a .wav, given the path as its last " "argument (default: the first of " + ", ".join(n for n, _ in PLAYERS) + " installed)") p.add_argument("--list", action="store_true", help="print one line per recording and exit, " "without opening the browser") p.add_argument("--callsigns", action="store_true", help="print every callsign heard, with who it belongs to, " "and exit") p.add_argument("--kml", nargs="?", const=DEFAULT_KML_NAME, default=None, metavar="FILE", help="write a map of where the stations heard are " f"licensed, and exit (default: {DEFAULT_KML_NAME} " "in the recordings directory)") p.add_argument("--no-lookup", dest="lookup", action="store_false", help="do not contact the licence database; callsigns are " "still found, and described from their prefix alone") p.add_argument("-V", "--version", action="version", version=f"saunterbrowse {__version__}") return p def _first_line(cap: Capture) -> str: """The most informative thing about a capture, in one line.""" if cap.transcript: return cap.transcript.splitlines()[0] if cap.decoded: # A pager message speaks for itself; a packet of hex does not, so # that one is introduced by what kind of packet it is. first = cap.decoded[0] headline = cap.data_headline return first if " " in first.strip(" 0123456789ABCDEF") or not headline \ else f"{headline} {first}" return cap.classification def _write_kml(console: Console, browser: "Browser", book: CallsignBook, where: str) -> int: """Build a map from the transcripts already on disk. The same file the scanner writes, so running this over an old directory and then scanning again continues one map rather than starting a second. """ heard: dict[str, list[Capture]] = {} for cap in browser.view: for call in cap.callsigns: heard.setdefault(call, []).append(cap) if not heard: console.print("[yellow]no callsigns in any transcript here — " "nothing to map[/yellow]") return 1 book.get_all(heard) book.wait(15.0) book.save() path = Path(where).expanduser() if not path.is_absolute() and path.parent == Path("."): path = browser.directory / path.name log = KmlLog(path, title="bandsaunter — stations heard", description=f"Callsigns heard in {browser.directory}, " "placed where their licences say they are.") for call, caps in sorted(heard.items()): entry = book.get(call) for cap in caps: log.add(entry, cap.frequency, cap.started_at or 0.0, cap.band, cap.path.name) written = log.save() if written is None: console.print(f"[red]could not write {path}[/red]") if not log.readable: console.print("[bright_black]There is a file there already that " "is not readable as KML, and overwriting it would " "throw it away.[/bright_black]") return 2 placed = sum(1 for c in log.contacts.values() if c.located) console.print(f"[green]{written}[/green]") console.print(f"[bright_black]{len(log)} station(s), {placed} with a " f"position.[/bright_black]") missing = len(log) - placed if missing: console.print(f"[bright_black]{missing} had no licence on file — " "listed in the map under 'no location on file'." "[/bright_black]") return 0 def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) console = Console() directory = (Path(args.directory).expanduser() if args.directory else default_directory()) if not directory.is_dir(): console.print(f"[red]no such directory: {directory}[/red]") console.print("[bright_black]Pass one, or run a scan first so there " "is something to browse.[/bright_black]") return 2 player = Player(args.player.split() if args.player else None) book = CallsignBook(online=args.lookup) browser = Browser(directory, console=console, player=player, book=book) browser.sort = args.sort browser.query = args.filter browser.apply() if args.callsigns: # One entry per callsign, with every frequency and time it was heard # on: the same operator turns up across a night, and a list that # repeated them would bury that. heard: dict[str, list[Capture]] = {} for cap in browser.view: for call in cap.callsigns: heard.setdefault(call, []).append(cap) if not heard: console.print("[yellow]no callsigns in any transcript here" "[/yellow]") return 1 book.get_all(heard) book.wait(15.0) book.save() for call, caps in sorted(heard.items()): entry = book.get(call) console.print(f"[bold cyan]{call}[/bold cyan] {entry.summary()}", highlight=False, soft_wrap=True) detail = entry.details() if detail: console.print(f"[bright_black]{'':>{len(call)}} " f"{' · '.join(detail)}[/bright_black]", highlight=False, soft_wrap=True) for cap in caps: when = cap.when.strftime("%Y-%m-%d %H:%M:%S") if cap.when else "" console.print(f"[bright_black]{'':>{len(call)}} heard on " f"{fmt_hz(cap.frequency)} at {when}" f"[/bright_black]", highlight=False, soft_wrap=True) return 0 if args.kml is not None: return _write_kml(console, browser, book, args.kml) if args.list: for cap in browser.view: when = cap.when.strftime("%Y-%m-%d %H:%M:%S") if cap.when else "" line = (f"{fmt_hz(cap.frequency):>14} {shorten_band(cap.band, 20):<20} " f"{when} " f"{_dur(cap.duration):>7} {cap.category:<8} " f"{_first_line(cap)}") console.print(line, highlight=False, soft_wrap=True) return 0 if not browser.captures: console.print(f"[yellow]no recordings in {directory}[/yellow]") return 1 if not console.is_terminal: console.print("[red]saunterbrowse needs a terminal.[/red] " "[bright_black]Use --list to print the recordings " "instead.[/bright_black]") return 2 try: return browser.run() except KeyboardInterrupt: browser.player.stop() return 0 finally: book.save() if __name__ == "__main__": sys.exit(main())