4239635f74
The program had no licence file at all, and pyproject claimed MIT into the void. It is now the GNU General Public License, version 3 or later: LICENSE holds the text verbatim, pyproject declares it with the OSI classifier, both .deb builds write /usr/share/doc/<pkg>/copyright in the machine-readable format Policy requires, both manuals carry a COPYING section, and --version prints the GNU notice on both programs. INSTALL.md is the step-by-step: what you need, the Debian package, the virtual environment for everywhere else, how to check it worked, every optional dependency with what it buys and what happens without it, and the errors people actually hit first -- PEP 668 at the top, because on Debian a plain "pip install ." refuses and reads as a broken program. Speech transcription gets its own four steps, because it is the only part with a real download in it: the recogniser into the environment bandsaunter runs from, checking it took, the model (base.en, 148 MB, from Hugging Face into ~/.cache/huggingface, fetched deliberately rather than in the middle of a scan), then turning it on. With the sizes of every model, the offline routes, and what to do when --engines says no although pip says yes. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016PsWPTweCT6pwxKngvVxcg
1951 lines
80 KiB
Python
1951 lines
80 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 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_notice
|
||
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<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",
|
||
"image": "bold magenta",
|
||
}
|
||
|
||
|
||
@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 image_path(self) -> str:
|
||
"""The picture this recording turned out to be, if it was one.
|
||
|
||
The sidecar is asked first and the directory second: a recording
|
||
copied somewhere else keeps its sidecar but not the absolute path in
|
||
it, and the picture is beside the .wav either way.
|
||
"""
|
||
saved = str(self.meta.get("image_path", "")).strip()
|
||
if saved and Path(saved).is_file():
|
||
return saved
|
||
beside = self.path.with_suffix(".png")
|
||
if beside.is_file():
|
||
return str(beside)
|
||
return saved
|
||
|
||
@property
|
||
def image_kind(self) -> str:
|
||
return str(self.meta.get("image_kind", "")).strip()
|
||
|
||
@property
|
||
def image_mode(self) -> str:
|
||
return str(self.meta.get("image_mode", "")).strip()
|
||
|
||
@property
|
||
def image_size(self) -> str:
|
||
width = int(self.meta.get("image_width") or 0)
|
||
height = int(self.meta.get("image_height") or 0)
|
||
return f"{width}x{height}" if width and height else ""
|
||
|
||
@property
|
||
def waterfall(self) -> str:
|
||
"""The drawing of the signal itself, where one was made.
|
||
|
||
Not a picture off the air like SSTV or a satellite pass: a picture
|
||
*of* the capture, drawn for the ones that produced no readable
|
||
words, which is most of them.
|
||
"""
|
||
saved = str(self.meta.get("waterfall_path", "")).strip()
|
||
if saved and Path(saved).is_file():
|
||
return saved
|
||
beside = self.path.with_name(self.path.stem + "_waterfall.png")
|
||
return str(beside) if beside.is_file() else ""
|
||
|
||
@property
|
||
def picture_headline(self) -> str:
|
||
"""Whatever picture this capture has, however it came to exist.
|
||
|
||
A decoded transmission and a drawing of the signal are said
|
||
differently on purpose: one is a picture somebody sent, the other is
|
||
a view of one. Both open in the same viewer, so both belong here.
|
||
"""
|
||
return self.image_headline or ("waterfall" if self.waterfall else "")
|
||
|
||
@property
|
||
def image_headline(self) -> str:
|
||
"""What kind of picture it is. Pictures off the air only."""
|
||
if not (self.image_kind or self.image_path):
|
||
return ""
|
||
bits = [self.image_kind or "image"]
|
||
if self.image_mode:
|
||
bits.append(self.image_mode)
|
||
if self.image_size:
|
||
bits.append(self.image_size)
|
||
if self.meta.get("image_complete") is False:
|
||
bits.append("partial")
|
||
return " ".join(bits)
|
||
|
||
@property
|
||
def images(self) -> list[str]:
|
||
"""Every picture file belonging to this recording.
|
||
|
||
More than one for a weather satellite: the pass is written whole, and
|
||
each of the satellite's two sensors is written again on its own,
|
||
because that is what anybody actually looks at.
|
||
"""
|
||
found: list[str] = []
|
||
main = self.image_path
|
||
if main:
|
||
found.append(main)
|
||
for path in (self.meta.get("image_paths") or []):
|
||
if str(path) not in found and Path(str(path)).is_file():
|
||
found.append(str(path))
|
||
stem = self.path.stem
|
||
for extra in sorted(self.path.parent.glob(glob.escape(stem) + "_*.png")):
|
||
if str(extra) not in found:
|
||
found.append(str(extra))
|
||
return found
|
||
|
||
@property
|
||
def morse(self) -> str:
|
||
"""What a CW capture was keying, if anything."""
|
||
return str(self.meta.get("morse_text", "")).strip()
|
||
|
||
@property
|
||
def morse_complete(self) -> str:
|
||
"""The part of it a station can be identified from.
|
||
|
||
A capture that opened partway through an ident lost the start of the
|
||
word it opened on, and what is left of that word can read as a whole
|
||
one: "K1AA" caught halfway through is "K1A", which belongs to
|
||
somebody else. Recordings made before this was written down carry no
|
||
such distinction, so for those it is the whole text or nothing.
|
||
"""
|
||
# Presence, not truthiness. An empty value means the decoder looked
|
||
# and found nothing safe to identify from, which is a different thing
|
||
# from a sidecar that predates the question being asked.
|
||
if "morse_complete" in self.meta:
|
||
return str(self.meta["morse_complete"]).strip()
|
||
return self.morse
|
||
|
||
@property
|
||
def callsigns(self) -> list[str]:
|
||
"""Callsign-shaped runs in everything this capture said.
|
||
|
||
Not only the transcript. A beacon, a repeater or an unattended
|
||
transmitter identifies itself in Morse and says nothing else at all,
|
||
and an APRS packet carries the sender's callsign in its first field --
|
||
those are the same people as the ones on the net, and belong in the
|
||
same list and on the same map.
|
||
|
||
Packet lines are only searched where they look like AX.25. A hex
|
||
dump is a string of two-character groups, and enough of those in a row
|
||
will join into something callsign-shaped that nobody transmitted.
|
||
"""
|
||
if self._calls is None:
|
||
# Only the transcript has spacing a recogniser invented; the
|
||
# gaps in Morse and in a packet header were put there by the
|
||
# sender, so nothing is joined across those.
|
||
sources = [(self.transcript, True), (self.morse_complete, False)]
|
||
sources += [(line, False) for line in self.decoded if ">" in line]
|
||
seen: set[str] = set()
|
||
found: list[str] = []
|
||
for text, join in sources:
|
||
for call in (find_callsigns(text, join_words=join)
|
||
if text else []):
|
||
if call not in seen:
|
||
seen.add(call)
|
||
found.append(call)
|
||
self._calls = found
|
||
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 ones 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)
|
||
# A weather satellite pass is written three times over -- the whole
|
||
# frame and one file per sensor -- and all three belong to it.
|
||
try:
|
||
found.update(p for p in
|
||
self.path.parent.glob(glob.escape(stem) + "_*.png")
|
||
if p.is_file())
|
||
except OSError:
|
||
pass
|
||
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 = ("date/time", "frequency", "duration")
|
||
|
||
# What --sort used to be called. Kept working because it is the sort of thing
|
||
# people put in a shell alias and never look at again.
|
||
SORT_ALIASES = {"time": "date/time", "date": "date/time"}
|
||
|
||
# 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}
|
||
|
||
|
||
# How a moment is written everywhere in the browser: the date first so that a
|
||
# column of them reads down in order, then the clock the way a person says it.
|
||
# Zero-padded on purpose -- an unpadded hour puts a ragged edge down the middle
|
||
# of the listing, and a column that does not line up is a column nobody reads.
|
||
STAMP = "%y-%m-%d %I:%M:%S %p"
|
||
STAMP_WIDTH = 20
|
||
|
||
|
||
def _stamp(when: datetime | None) -> str:
|
||
"""One capture's date and time, or an empty string when it has neither."""
|
||
if when is None:
|
||
return ""
|
||
# Lower case because AM in capitals shouts, and the field beside it is a
|
||
# frequency: the eye should land on the number, not on the meridiem.
|
||
return when.strftime(STAMP).lower()
|
||
|
||
|
||
# The widest frequency there is room for -- 1090.000001 MHz -- and the
|
||
# narrowest worth reserving.
|
||
FREQ_WIDTH = (9, 15)
|
||
|
||
|
||
def _freq_width(captures) -> int:
|
||
"""How wide the frequency column has to be for these recordings.
|
||
|
||
Fixed at the width of the widest frequency in the whole list rather than
|
||
the widest on screen: a column that changes width as the list scrolls
|
||
under it makes the whole listing appear to twitch. Anything given up
|
||
here goes to the end of the line, which is where what was said is.
|
||
"""
|
||
widest = max((len(fmt_hz(c.frequency)) if c.frequency
|
||
else min(FREQ_WIDTH[1], len(c.path.stem))
|
||
for c in captures), default=0)
|
||
return max(FREQ_WIDTH[0], min(FREQ_WIDTH[1], widest))
|
||
|
||
|
||
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 = SORTS[0]
|
||
self.freq_width = FREQ_WIDTH[0]
|
||
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:
|
||
# Newest first, and strictly by the whole moment -- year, month,
|
||
# day, then hour, minute, second -- because started_at is one
|
||
# number counting from the epoch rather than a formatted string.
|
||
# The filename breaks a tie, which only happens when two signals
|
||
# landed in the same second.
|
||
self.view.sort(key=lambda c: (-c.started_at, c.path.name))
|
||
self.index = max(0, min(self.index, len(self.view) - 1))
|
||
self.freq_width = _freq_width(self.view)
|
||
|
||
@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. So is
|
||
# what a station keyed: "/W1AW" should find the CW ident as readily
|
||
# as the net that mentioned it.
|
||
if any(q in line.lower() for line in cap.decoded):
|
||
return True
|
||
if q in cap.morse.lower():
|
||
return True
|
||
# And by what kind of picture it is: "/sstv" and "/apt" are how
|
||
# anybody would look for the ones worth keeping.
|
||
if cap.picture_headline and q in cap.picture_headline.lower():
|
||
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 _morse_lines(self, pad: int = 3) -> list[str]:
|
||
"""What a CW capture keyed, wrapped to the width it is drawn at.
|
||
|
||
Morse gets a panel of its own rather than the "no transcript" notice
|
||
it used to share with silence and static. A station that identifies
|
||
itself in CW has said the one thing worth reading, and the callsigns
|
||
found in it belong under it exactly as they do under speech.
|
||
"""
|
||
cap = self.current
|
||
if cap is None or cap.transcript or not cap.morse:
|
||
return []
|
||
width = max(20, (self.console.size.width or 80) - 2 - 2 * pad)
|
||
text = Text(cap.morse, style="bold white")
|
||
return [line.plain.rstrip()
|
||
for line in text.wrap(self.console, width)]
|
||
|
||
@staticmethod
|
||
def _picture_panel_wins(cap) -> bool:
|
||
"""Whether the top panel is a picture rather than words.
|
||
|
||
A picture off the air always wins: it *is* the transmission. A
|
||
waterfall wins only when there is nothing else, because it says less
|
||
about a capture than a decoded pager message or a Morse ident does
|
||
-- it is a view of the signal, not a reading of it.
|
||
"""
|
||
if cap is None:
|
||
return False
|
||
if cap.image_headline:
|
||
return True
|
||
return bool(cap.waterfall) and not (cap.transcript or cap.morse_complete
|
||
or cap.morse or cap.decoded)
|
||
|
||
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
|
||
if self._picture_panel_wins(cap):
|
||
body = len(self._image_lines(cap))
|
||
else:
|
||
body = (len(self._transcript_lines()) or len(self._morse_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(f"{when:%a} {_stamp(when)}", 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 self._picture_panel_wins(cap):
|
||
return self._image_panel(cap, height)
|
||
morse = self._morse_lines()
|
||
if morse:
|
||
calls = self._callsign_lines()
|
||
# Never below one: on a very short window the callsigns can eat
|
||
# the whole panel, and a negative slice would take lines off the
|
||
# wrong end of the message.
|
||
room = max(1, height - 4 - (len(calls) + 1 if calls else 0))
|
||
shown = morse[:room - 1] if len(morse) > room else morse
|
||
body = Text("\n".join(shown), style="bold white")
|
||
if len(morse) > len(shown):
|
||
body.append(f"\n… {len(morse) - len(shown)} 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")
|
||
return Panel(body, title="Morse", title_align="left",
|
||
border_style="yellow", padding=(1, 3), 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)
|
||
|
||
def _image_lines(self, cap: Capture, pad: int = 3) -> list[str]:
|
||
"""What the picture is and where it was written.
|
||
|
||
The path in full and on a line of its own, because it is the thing
|
||
somebody came here for: the browser cannot draw a PNG in a terminal,
|
||
so the most useful thing it can do is say exactly what to open.
|
||
"""
|
||
width = max(20, (self.console.size.width or 80) - 2 - 2 * pad)
|
||
lines = [cap.picture_headline[:width], ""]
|
||
files = cap.images
|
||
if not files:
|
||
return lines + ["the picture file is not beside the recording"]
|
||
for path in files:
|
||
# Wrapped rather than cut off. Half a path is not something
|
||
# anybody can open, and the path is the entire reason this panel
|
||
# exists: the browser cannot draw a PNG in a terminal, so saying
|
||
# exactly what to open is the most useful thing it can do.
|
||
text = Text(path)
|
||
lines += [line.plain.rstrip()
|
||
for line in text.wrap(self.console, width)]
|
||
return lines
|
||
|
||
def _image_panel(self, cap: Capture, height: int) -> Panel:
|
||
body = Text()
|
||
lines = self._image_lines(cap)
|
||
body.append(lines[0] + "\n", style="bold magenta")
|
||
for line in lines[1:]:
|
||
body.append(line + "\n", style="bold white" if line else "")
|
||
calls = self._callsign_lines()
|
||
if calls:
|
||
body.append("\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")
|
||
title = "waterfall" if not cap.image_headline else "picture"
|
||
return Panel(body, title=title, title_align="left",
|
||
border_style="magenta", 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
|
||
# Morse that decoded never reaches here: it has a panel of its own.
|
||
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")
|
||
if cap.image_kind:
|
||
line.append(f" {len(cap.images)} file(s)", style="magenta")
|
||
elif cap.waterfall:
|
||
# Not in the summary column: "waterfall" says less about a
|
||
# capture than "OOK / ASK data burst" does, and the summary has
|
||
# room for one of them.
|
||
line.append(" waterfall", style="magenta")
|
||
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=self.freq_width, justify="right") # frequency
|
||
t.add_column(width=STAMP_WIDTH) # date and 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.image_headline
|
||
or cap.transcript.replace("\n", " ")
|
||
or cap.morse
|
||
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(_stamp(when), 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 date/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.picture_headline:
|
||
lines = self._image_lines(cap, pad=4)
|
||
title_word = "picture"
|
||
if not lines and cap is not None and cap.morse:
|
||
lines = self._morse_lines(pad=4)
|
||
title_word = "Morse"
|
||
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" · {_stamp(when)}"
|
||
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.morse
|
||
or cap.decoded or cap.picture_headline):
|
||
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:
|
||
# The picture where there is one: on a capture that turned
|
||
# out to be an image, the PNG is what anybody wants to pipe
|
||
# into something else, not the audio it arrived as. A
|
||
# waterfall counts, on a capture that has nothing else --
|
||
# there is no listening to a data burst.
|
||
self.message = cap.image_path or cap.waterfall or 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.",
|
||
# Raw, so the licence notice --version prints keeps its line breaks
|
||
# instead of being reflowed into a paragraph.
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
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 + tuple(SORT_ALIASES),
|
||
default=SORTS[0], metavar="ORDER",
|
||
help="initial order: %s (default: %s, newest first)"
|
||
% (", ".join(SORTS), SORTS[0]))
|
||
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=version_notice("saunterbrowse"))
|
||
return p
|
||
|
||
|
||
def _first_line(cap: Capture) -> str:
|
||
"""The most informative thing about a capture, in one line."""
|
||
if cap.image_headline:
|
||
return f"{cap.image_headline} {cap.image_path}".strip()
|
||
if cap.transcript:
|
||
return cap.transcript.splitlines()[0]
|
||
if cap.morse:
|
||
return cap.morse
|
||
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 = SORT_ALIASES.get(args.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())
|