68b05a031c
Much of what a scanner finds is not speech. Doorbells, tyre-pressure sensors, weather stations, remote controls, paging and packet radio all carry something a receiver can read, and until now the answer was "OOK / ASK data burst" and a WAV file. Now the bits come out. The observation the whole thing is built on is that whatever the modulation, a data signal is the same shape once it has been sliced: a train of alternating runs whose lengths carry the information. On-off keying gives that directly -- the carrier is up or it is down -- and two-level FSK gives exactly the same thing from the discriminator, one tone or the other. So both reduce to a run-length train and everything after that is shared. What the runs mean is the line code, and it is worked out from the runs alone rather than configured, because each code makes a different prediction about which of the two histograms is the bimodal one: PWM (EV1527, PT2262, and nearly every 433 MHz remote), PPM, Manchester, and plain NRZ. Four-level FSK is recognised as such and read as symbols rather than sliced down the middle, which produces bits that mean nothing; where a frame sync word appears the system is named outright. Two protocols carry their own framing and checksums and so are read in full. POCSAG paging: all three rates tried because nothing in the signal says which it is, every codeword checked and single-bit errors corrected against the BCH code, and the address, function letter and message text reported. AX.25 as APRS uses it: the frame check has to come out right before a frame is reported at all, and the sender's callsign goes onto the map with everyone else's. The hard half is refusing what is not data. Noise sliced at a threshold produces runs and runs produce bits, so three things guard against it: the runs have to quantise to the line code's own grid; most of the bursts in a capture have to decode the same way, because one lucky window in eight is a coincidence and that is exactly what SSB voice produced; and, much the strongest, the packet has to repeat, because bits that come back identical six times did not come from noise. A reading with none of that behind it is reported as nothing at all rather than as a bit string with a low number beside it that somebody will read anyway. Across 27 recordings of speech, music, static, a bare carrier, Morse and PSK it returns nothing 27 times. A firm decode also outranks the content check, which is statistical: a burst of keying demodulated as FM audio is a buzz and the speech detector likes a buzz, but a frame whose own checksum came out right is not a statistic. Such a capture is kept and filed as data, not as voice. What comes out is written to a _data.txt beside the recording, shown on the live display and in the line-per-hit output, and takes the place of the transcript at the top of saunterbrowse -- where it is searchable, so "which page mentioned engine 4" is a question that can be asked. `bandsaunter analyze` decodes a file you already have. The simulator gained two honest transmitters to test against: a pulse-width remote that repeats a real payload, and a pager that sends real POCSAG batches with real BCH check bits. Random keying exercises the classifier but leaves a decoder nothing to get right. The POCSAG encoder lives next to the decoder rather than in the test helpers, so a bug shared by both cannot hide. Fixed along the way: - Rich reads a square bracket as markup, and a decoded page is arbitrary text off the air. "[/x]" in a message ended the live display with a MarkupError; so did typing "[/" at saunterbrowse's search prompt. Everything that did not come from this program is escaped now. - Otsu returned the first bin of a plateau. Two populations with nothing between them -- silence and full carrier, which is what on-off keying is -- make every threshold in the gap equally good, and taking the first put it hard against the lower population with the hysteresis band outside the data entirely, so nothing sliced at all. - Estimating the symbol clock by counting along a cumulative grid is a fixed point: a unit two per cent small produces two per cent more symbols and reproduces itself exactly. Rounding each run on its own converges instead, because every run votes independently. The grid is then the right way to extract the bits, where rounding runs one at a time drifts. - A clipped first repeat used to truncate every other repeat to its length. The consensus is taken over the commonest length now. 761 -> 869 tests.
1296 lines
52 KiB
Python
1296 lines
52 KiB
Python
"""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 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 load_default
|
||
from .kml import DEFAULT_KML_NAME, KmlLog
|
||
|
||
__all__ = ["main", "Capture", "scan_directory", "Browser", "Player"]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# One recording on disk
|
||
# ---------------------------------------------------------------------------
|
||
|
||
# 0146.880000MHz--2026-08-22_12_54_59-nfm.wav -- the name a scan writes.
|
||
_STEM = re.compile(
|
||
r"^(?P<mhz>\d+\.\d+)MHz"
|
||
r"--(?P<date>\d{4}-\d{2}-\d{2})_(?P<h>\d{2})_(?P<m>\d{2})_(?P<s>\d{2})"
|
||
r"-(?P<mode>.+)$")
|
||
|
||
# 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 _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")
|
||
|
||
|
||
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):
|
||
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()
|
||
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.message = ""
|
||
self.show_help = False
|
||
self.reading = False # full-screen transcript
|
||
self.read_top = 0
|
||
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))
|
||
|
||
# -- 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))
|
||
|
||
def _footer(self) -> Text:
|
||
# Both 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.searching:
|
||
return Text.from_markup(
|
||
f"[bold]search:[/bold] {escape(self.query)}[blink]_[/blink]"
|
||
" [bright_black]enter to accept, esc to clear"
|
||
"[/bright_black]")
|
||
if self.message:
|
||
return Text.from_markup(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 Text.from_markup(
|
||
f"[bold red]♪[/bold red] {bar} {_dur(done)}/{_dur(total)}"
|
||
" [bright_black]space stop[/bright_black]")
|
||
return Text.from_markup(
|
||
"[bright_black]↑↓[/bright_black] move "
|
||
"[bright_black]⏎[/bright_black] play "
|
||
"[bright_black]space[/bright_black] stop "
|
||
"[bright_black]/[/bright_black] search "
|
||
"[bright_black]t[/bright_black] read "
|
||
"[bright_black]s[/bright_black] sort by "
|
||
f"{self.sort} "
|
||
"[bright_black]r[/bright_black] reload "
|
||
"[bright_black]?[/bright_black] keys "
|
||
"[bright_black]q[/bright_black] quit")
|
||
|
||
def _help(self) -> Panel:
|
||
t = Table(box=None, show_header=False, pad_edge=False)
|
||
t.add_column(style="bold cyan", width=14)
|
||
t.add_column()
|
||
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, identification or "
|
||
"anything that was said"),
|
||
("", ""),
|
||
("callsigns", "found in the transcript and looked up "
|
||
"automatically; --no-lookup keeps it offline"),
|
||
("Esc", "clear the filter"),
|
||
("s", "sort by time, frequency or length"),
|
||
("r", "re-read the directory"),
|
||
("o", "print the file's path and quit"),
|
||
("? 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")
|
||
t.add_row("", "")
|
||
t.add_row("player", player)
|
||
return Panel(t, title="keys", title_align="left",
|
||
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):
|
||
if self.show_help:
|
||
return Align.center(self._help(), vertical="middle")
|
||
if self.reading:
|
||
return 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.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
|
||
if cap is not None and cap.transcript:
|
||
self.reading = True
|
||
self.read_top = 0
|
||
else:
|
||
self.message = "no transcript for this recording"
|
||
elif key == "o":
|
||
cap = self.current
|
||
if cap is not None:
|
||
self.message = str(cap.path)
|
||
return False
|
||
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
|
||
|
||
# -- 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
|
||
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())
|