diff --git a/README.md b/README.md index 2237ae5..19ca95d 100644 --- a/README.md +++ b/README.md @@ -803,6 +803,25 @@ Only voice is transcribed — running a recogniser over Morse or a data burst costs seconds and produces nothing. CW is decoded separately and appears in the metadata as text already. +**One transcript per transmission, and none is ever overwritten.** The +timestamp is part of the name, so two overs on the same frequency cannot land +on the same file — a second transmission on 146.52 MHz writes +`...12_19_44-nfm_transcription.txt` beside the first, not over it. + +With `--combine` there is one recording per frequency, so there is one +transcript per frequency too, and it works the other way: each over is +**appended** with the time it was heard, and an unattended receiver keeps +adding to it night after night. + +``` +[2026-08-21 12:18:38] Net control, this is W1AW, standing by. +[2026-08-21 12:19:44] Roger, copy that, back to you. +``` + +Both behaviours have tests that run a real scan and check the files, including +one that runs a second scan into the same directory and asserts the earlier +text is still at the top. + **A capture with nothing recognisable in it produces no file.** Music, a carrier with an open mic, a fragment too short to make out: nothing is written, rather than a directory of placeholders. The transcript is also @@ -982,11 +1001,65 @@ or symbol rate where there is one, and the bands the frequency falls in. | `space` | stop playing | | `t` | read the whole transcript full screen, scrolling | | `/` | filter — by frequency, filename, identification, **or anything that was said** | +| — | callsigns are found and looked up automatically; no key needed | | `s` | sort by time, frequency or length | | `r` | re-read the directory, picking up what a running scan has written | | `o` | print the file's path and quit | | `q` | quit | +### Detected callsigns + +Under the transcript, every callsign heard in it is listed with the name and +location on its licence: + +``` +╭─ transcript ──────────────────────────────────────────────────────────╮ +│ │ +│ Alright, moving on. There is an HF net at 1.30pm on 7.242 │ +│ megahertz. Are there any announcements? Alright, KU 0W. │ +│ │ +│ DETECTED CALLSIGNS: │ +│ KU0W Rod R Gowdy — Tucson, AZ · Extra · DM42lj · 85742 │ +│ │ +╰───────────────────────────────────────────────────────────────────────╯ +``` + +Note what the recogniser actually wrote: **"KU 0W"**, with a space. Speech +recognisers are poor at callsigns — they are not words, they are said one +character at a time — so a callsign arrives broken wherever the speaker +paused, and an operator who spells it out gets *"kilo uniform zero whiskey"* +written down verbatim. All three forms read back to `KU0W`. + +The other half of the problem is not inventing them. A browser that reports +callsigns nobody said is worse than one that reports none, so a run of words +is only accepted when none of its parts is an ordinary English word — *"or 3. +Can you open 4"* fits the shape once the punctuation is gone, and is not a +callsign. A single token said in one breath is trusted, because `W1BOY` is a +perfectly good callsign. Across 126 real transcripts from an overnight scan, +that turns three candidates into the one that was actually said. + +```bash +saunterbrowse --callsigns # everyone who identified themselves, and where +saunterbrowse --no-lookup # find them, but contact nothing +``` + +Lookups use the FCC's own licence data via [callook.info](https://callook.info), +which needs no account or key. The callsign is the only thing sent; results are +cached in `~/.cache/bandsaunter/callsigns.json`, so the same net is looked up +once however many nights you record it, and a lookup never delays the display — +the entry reads `looking up…` and fills itself in. + +`--no-lookup` contacts nothing. Callsigns are still found and still described +from their own structure: the prefix is allocated by the ITU and the digit is +the US licensing district, so `VE3ABC` is Canada and `N7XYZ` is US district 7 +with no database at all. Outside the US that structural description is all +there is — callook.info holds US licences only. + +US amateur licence records are public by law and include the licensee's +address; that is what is shown. + +### Searching what was said + Searching the transcripts is the point of it: *"did anyone mention the repeater"* is a question about content, not about filenames. diff --git a/bandsaunter/__init__.py b/bandsaunter/__init__.py index e6fec0c..a8cf52b 100755 --- a/bandsaunter/__init__.py +++ b/bandsaunter/__init__.py @@ -9,7 +9,7 @@ and transcribing speech. # 2026-08-21_02 is the second build made on the 21st. The revision is padded # to two digits so versions sort as text. VERSION_DATE = "2026-08-22" -VERSION_REVISION = 3 +VERSION_REVISION = 4 __version__ = f"{VERSION_DATE}_{VERSION_REVISION:02d}" diff --git a/bandsaunter/browse.py b/bandsaunter/browse.py index ff33479..e368ab4 100644 --- a/bandsaunter/browse.py +++ b/bandsaunter/browse.py @@ -37,6 +37,7 @@ from rich.text import Text from . import __version__ from .bandplan import fmt_hz +from .callsign import CallsignBook, HEADING, find_callsigns from .config import load_default __all__ = ["main", "Capture", "scan_directory", "Browser", "Player"] @@ -82,6 +83,7 @@ class Capture: _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) # -- lazily read sidecars --------------------------------------------- @property @@ -136,6 +138,13 @@ class Capture: pass return self._duration + @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 category(self) -> str: @@ -411,10 +420,12 @@ class Browser: """The whole application: state, rendering and the key loop.""" def __init__(self, directory: Path, console: Console | None = None, - player: Player | 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 @@ -479,6 +490,32 @@ class Browser: 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. @@ -502,9 +539,16 @@ class Browser: net swallows the rest of it without saying so. """ screen = self.console.size.height or 24 - ceiling = max(6, min(16, screen // 3)) - needed = len(self._transcript_lines()) + 4 # borders and padding - return max(6, min(ceiling, needed)) + 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))) + needed = len(self._transcript_lines()) + 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.""" @@ -556,10 +600,11 @@ class Browser: height=height) lines = self._transcript_lines() - room = height - 4 # borders and vertical padding + 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 - # The notice needs a line of its own, or the panel crops the very - # thing that was there to say something had been cropped. shown = lines[:room - 1] if overflows else lines hidden = len(lines) - len(shown) body = Text("\n".join(shown), style="bold white") @@ -568,11 +613,15 @@ class Browser: # 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") - block = body - elif len(shown) + 2 < height: + 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") - else: - block = body return Panel(block, title="transcript", title_align="left", border_style="green", padding=(1, 3), height=height) @@ -737,6 +786,9 @@ class Browser: ("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"), @@ -756,11 +808,21 @@ class Browser: """The whole transcript, with nothing else competing for the screen.""" cap = self.current screen = self.console.size.height or 24 - room = max(3, screen - 4) + calls = self._callsign_lines(pad=4) + room = max(3, screen - 4 - (len(calls) + 1 if calls else 0)) lines = self._transcript_lines(pad=4) 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 = "transcript" @@ -958,6 +1020,12 @@ def build_parser() -> argparse.ArgumentParser: 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("--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 @@ -975,11 +1043,44 @@ def main(argv: list[str] | None = None) -> int: return 2 player = Player(args.player.split() if args.player else None) - browser = Browser(directory, console=console, player=player) + 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.list: for cap in browser.view: when = cap.when.strftime("%Y-%m-%d %H:%M:%S") if cap.when else "" @@ -1002,6 +1103,8 @@ def main(argv: list[str] | None = None) -> int: except KeyboardInterrupt: browser.player.stop() return 0 + finally: + book.save() if __name__ == "__main__": diff --git a/bandsaunter/callsign.py b/bandsaunter/callsign.py new file mode 100644 index 0000000..d446024 --- /dev/null +++ b/bandsaunter/callsign.py @@ -0,0 +1,536 @@ +"""Finding callsigns in a transcript, and saying who they belong to. + +Speech recognisers are poor at callsigns. They are not words, they are said +one character at a time, and the recogniser reaches for whatever English is +nearest: ``KU0W`` comes out as "KU 0W", ``K7RA`` as "K7 RA", and an operator +who spells it out in the phonetic alphabet gets "kilo uniform zero whiskey" +written down verbatim. So finding one is not a matter of a single regular +expression over the text as written. + +What comes out is a shape -- one or two letters, a digit, one to three +letters -- which is checked against the licence database. A callsign that +resolves to a licence is real; one that does not is reported as unverified +rather than silently dropped, because a mangled callsign is still a signal +that someone identified themselves. +""" + +from __future__ import annotations + +import json +import os +import re +import threading +import time +import urllib.parse +import urllib.request +from dataclasses import dataclass +from pathlib import Path + +__all__ = ["Callsign", "CallsignBook", "find_callsigns", "describe_prefix", + "PHONETIC", "LOOKUP_URL"] + +# The FCC's own licence data, served as JSON without an account or a key. +# US callsigns only; everything else resolves to what the prefix alone says. +LOOKUP_URL = "https://callook.info/{call}/json" + +PHONETIC = { + "alpha": "A", "alfa": "A", "bravo": "B", "charlie": "C", "delta": "D", + "echo": "E", "foxtrot": "F", "fox": "F", "golf": "G", "hotel": "H", + "india": "I", "juliet": "J", "juliett": "J", "julliet": "J", + "kilo": "K", "lima": "L", "mike": "M", "november": "N", "oscar": "O", + "papa": "P", "quebec": "Q", "romeo": "R", "sierra": "S", "tango": "T", + "uniform": "U", "victor": "V", "whiskey": "W", "whisky": "W", + "xray": "X", "x-ray": "X", "yankee": "Y", "zulu": "Z", + "zero": "0", "one": "1", "two": "2", "three": "3", "four": "4", + "five": "5", "fife": "5", "six": "6", "seven": "7", "eight": "8", + "nine": "9", "niner": "9", +} + +# A prefix, a digit, then one to three letters. The prefix is one or two +# letters (W, KU, VE) or a digit and a letter (2E, 4X, 9A) -- the latter is why +# a plain "letters then digit" pattern is not enough, and why every British +# 2E0 callsign would otherwise be missed. +SHAPE = re.compile(r"^(?:[A-Z]{1,2}|[0-9][A-Z])[0-9][A-Z]{1,3}$") + +# Things that fit the shape and are never callsigns. Every entry here has +# been checked against SHAPE: a list of words that could not match anyway +# would only suggest the filter was doing more than it is. +NOT_CALLSIGNS = frozenset({"B2B", "B4U", "H2O", "M4A", "P2P", "Y2K", "W2K", + "H2S", "N2O", "C2C", "D2D", "F2F"}) + +# Ordinary English words, short enough to be mistaken for part of a callsign. +# They only disqualify a *join*: "or 3. Can you open 4" became OR3CAN, and +# "CC1 boy" became CC1BOY, because both fit the shape once the punctuation is +# gone. Nobody spells a callsign with English words -- they say the letters, +# or the phonetic alphabet, which is converted before this is consulted -- so +# a component that is a word means the run is not a callsign. A single token +# is still trusted: W1BOY is a perfectly good callsign, said in one breath. +COMMON_WORDS = frozenset(""" +a i o an as at be by do go he hi if in is it me my no of on or so to up us we +add age ago air all and any are arm art ask bad bag bar bed bet big bit box +boy bus but buy can car cat cop cup cut dad day did die dog dry due eat egg +end eye far few fit fix fly for fun gas get god got gun guy had has hat her +him his hit hot how ice its job key kid law lay leg let lie lot low man map +may men met mom nor not now odd off oil old one out own pay per pop put ran +red rid row run sat saw say sea see set she sir sit six son sun tax tea ten +the tie tip toe ton too top try two use van war was way wet who why win yes +yet you +able also away back bad been beer bell best bill blue boat body book both +call came care case city club cold come cost dark data date days dead deal +does done door down draw drop each easy else even ever face fact fall feel +feet fell file fill find fire fish five flat food foot form four free from +full game gave give goes gold gone good grew grey guys half hall hand hard +have head hear held help here hers high hold hole home hope hour huge idea +into item join just keep kept kind knew know land last late lead left less +life like line list live long look lose lost love made mail main make many +mark mean meet mile mind mine miss mode more most move much must name near +neck need news next nice nine none note okay once only open over page paid +part pass past path pick plan play plus poor post pull pure push race rain +read real rest ride ring rise risk road rock role roll room rule safe said +sale same save says seem seen self sell send sent ship shop shot show shut +side sign site size skin slow snow sold some song soon sort soul stay step +stop such sure take talk tall team tell test text than that them then they +thin this thus time tiny told took town tree trip true turn type unit upon +used user very view vote wait walk wall want ward warm wash wave ways weak +wear week well went were west what when whom wide wife wild will wind wine +wire wish with wood word work yard yeah year your zone +""".upper().split()) + +# Single letters that are English words far more often than they are the start +# of a callsign. They are only rejected as the *first* token of a join -- +# "a B4U" must not become AB4U -- and never when the speaker said them in the +# phonetic alphabet, where "alpha" is unambiguous. +LONE_WORDS = frozenset({"A", "I", "O"}) + + +# --------------------------------------------------------------------------- +# What a callsign says about itself +# --------------------------------------------------------------------------- + +# ITU prefix blocks, coarsely. A full table runs to several hundred entries +# and most of it never appears in a scanner's transcripts; this covers what a +# receiver in North America actually hears, and says "unknown" rather than +# guessing for the rest. +_PREFIXES: tuple[tuple[tuple[str, ...], str], ...] = ( + (("K", "N", "W", "AA", "AB", "AC", "AD", "AE", "AF", "AG", "AI", "AJ", + "AK", "AL", "KG", "KH", "KL", "KP", "NH", "NL", "NP", "WH", "WL", + "WP"), "United States"), + (("VA", "VE", "VO", "VY", "CF", "CG", "CH", "CI", "CJ", "CK", "CY", + "CZ", "XJ", "XK", "XL", "XM", "XN", "XO"), "Canada"), + (("XE", "XF", "4A", "6D", "6E"), "Mexico"), + (("G", "M", "2E", "GM", "GW", "GI", "GD", "GJ", "GU"), "United Kingdom"), + (("VK", "AX"), "Australia"), + (("ZL", "ZM"), "New Zealand"), + (("JA", "JE", "JF", "JG", "JH", "JI", "JJ", "JK", "JL", "JM", "JN", + "JO", "JP", "JQ", "JR", "JS", "7J", "7K", "7L", "7M", "7N"), "Japan"), + (("DL", "DK", "DJ", "DB", "DC", "DD", "DF", "DG", "DH", "DO"), "Germany"), + (("F",), "France"), + (("I",), "Italy"), + (("EA", "EB", "EC"), "Spain"), + (("PY", "PP", "PQ", "PR", "PS", "PT", "PU", "PV", "PW"), "Brazil"), + (("LU",), "Argentina"), + (("CE", "CA", "CB", "CC"), "Chile"), + (("HK",), "Colombia"), + (("CO", "CM"), "Cuba"), + (("UA", "UB", "R", "RA", "RK", "RN", "RU", "RV", "RW", "RX", "RZ"), + "Russia"), + (("BY", "BA", "BD", "BG", "BH", "BI"), "China"), + (("HL", "DS"), "South Korea"), + (("VU",), "India"), + (("ZS", "ZR", "ZT", "ZU"), "South Africa"), + (("4X", "4Z"), "Israel"), + (("SM", "SA", "SB", "SC", "SD", "SE", "SF", "SG", "SH", "SI", "SJ", + "SK", "SL", "8S"), "Sweden"), + (("LA", "LB", "LC", "LD", "LE", "LF", "LG", "LH", "LI", "LJ", "LK", + "LL", "LM", "LN"), "Norway"), + (("OH", "OF", "OG", "OI"), "Finland"), + (("OZ", "5P", "5Q", "OU", "OV"), "Denmark"), + (("PA", "PB", "PC", "PD", "PE", "PF", "PG", "PH", "PI"), "Netherlands"), + (("ON", "OO", "OP", "OQ", "OR", "OS", "OT"), "Belgium"), + (("HB", "HE"), "Switzerland"), + (("OE",), "Austria"), + (("SP", "SN", "SO", "SQ", "SR", "3Z"), "Poland"), + (("CT", "CQ", "CR", "CS"), "Portugal"), + (("EI", "EJ"), "Ireland"), +) + +# US call districts. A callsign's digit says where the licence was issued, +# which for older licences is often not where the operator now lives -- so +# this is phrased as the district, not as an address. +_US_DISTRICTS = { + "0": "district 0 (CO IA KS MN MO NE ND SD)", + "1": "district 1 (New England)", + "2": "district 2 (NY NJ)", + "3": "district 3 (DE MD PA)", + "4": "district 4 (Southeast)", + "5": "district 5 (South Central)", + "6": "district 6 (California)", + "7": "district 7 (Northwest and Mountain)", + "8": "district 8 (MI OH WV)", + "9": "district 9 (IL IN WI)", +} + + +def describe_prefix(call: str) -> tuple[str, str]: + """``(country, district)`` from the callsign's own structure. + + Needs no database and no network: the prefix is allocated by the ITU and + the digit is the licensing district, so a callsign carries this much about + itself wherever it is heard. + """ + call = call.upper() + country = "" + for prefixes, name in _PREFIXES: + for p in sorted(prefixes, key=len, reverse=True): + if not call.startswith(p): + continue + # A one-letter prefix has to be followed by the rest of a real + # prefix -- an optional second letter, then the district digit. + # Without that, "I" claims every callsign beginning with I, and + # with too strict a version ("K" then a digit) it fails to claim + # KU0W, whose prefix is two letters long. + if len(p) == 1 and not re.match(r"^[A-Z]?[0-9]", call[1:]): + continue + country = name + break + if country: + break + district = "" + digit = next((c for c in call if c.isdigit()), "") + if country == "United States" and digit in _US_DISTRICTS: + district = _US_DISTRICTS[digit] + return country, district + + +# --------------------------------------------------------------------------- +# Finding them in text +# --------------------------------------------------------------------------- + +def _tokens(text: str) -> list[tuple[str, bool]]: + """``(text, was_phonetic)`` per word, punctuation dropped. + + Whether a token was spelled out matters: "alpha" is a letter beyond doubt, + while a bare "a" is almost always the English article. + """ + out = [] + for raw in re.split(r"[^0-9A-Za-z\-]+", text): + if not raw: + continue + spoken = PHONETIC.get(raw.lower()) + out.append((spoken, True) if spoken else (raw.upper(), False)) + return out + + +# A callsign is at most six characters, so a run of more than six tokens can +# never be one however it was broken up. +MAX_SPAN = 6 + + +def find_callsigns(text: str, max_found: int = 12) -> list[str]: + """Every callsign-shaped run in ``text``, in the order they appear. + + Candidates are built from runs of adjacent tokens, because a recogniser + breaks a callsign wherever the speaker paused -- "KU 0W" and "K7 RA" are + one callsign each, written as two words -- and an operator who spells it + out phonetically produces one token per character: "kilo uniform zero + whiskey" is four. Longest first, so "KU 0W" wins over the "U0W" hiding + inside it. + """ + tokens = _tokens(text) + found: list[str] = [] + seen: set[str] = set() + i = 0 + while i < len(tokens): + for span in range(MAX_SPAN, 0, -1): + if i + span > len(tokens): + continue + run = tokens[i:i + span] + if span > 1: + if run[0][0] in LONE_WORDS and not run[0][1]: + continue # "a B4U" is not AB4U + # Only words of two letters or more. A callsign spelled out + # one character at a time -- "W 1 A W" -- has a bare "A" in + # the middle of it, and rejecting that would lose exactly the + # case this join exists to catch. The first position is + # guarded separately, above. + if any(len(word) > 1 and word in COMMON_WORDS and not phonetic + for word, phonetic in run): + continue # "or 3. Can you" is not OR3CAN + joined = "".join(word for word, _ in run) + if not SHAPE.match(joined) or joined in NOT_CALLSIGNS: + continue + if joined not in seen: + seen.add(joined) + found.append(joined) + i += span + break + else: + i += 1 + if len(found) >= max_found: + break + return found + + +# --------------------------------------------------------------------------- +# Looking them up +# --------------------------------------------------------------------------- + +@dataclass +class Callsign: + """What is known about one callsign.""" + + call: str + name: str = "" + location: str = "" # town and state, as licensed + country: str = "" + district: str = "" + oper_class: str = "" + grid: str = "" + licence_type: str = "" # PERSON, CLUB, ... + expires: str = "" + previous: str = "" + trustee: str = "" + postcode: str = "" + status: str = "pending" # pending / found / unlisted / offline + fetched_at: float = 0.0 + + @property + def known(self) -> bool: + return self.status == "found" + + def summary(self) -> str: + """One line: who and where, falling back to what the prefix says.""" + if self.known: + where = self.location or self.country + bits = [b for b in (self.name, where) if b] + return " — ".join(bits) if bits else self.call + if self.status == "pending": + return "looking up…" + # Not in the database: say what the callsign says about itself, which + # is real information, rather than nothing at all. + bits = [b for b in (self.country, self.district) if b] + tail = ", ".join(bits) if bits else "no matching licence" + if self.status == "offline": + return f"not looked up ({tail})" if bits else "not looked up" + return f"unlisted ({tail})" if bits else "no matching licence" + + def details(self) -> list[str]: + """The rest of it, for anywhere with room to show more than one line.""" + out = [] + if self.oper_class: + out.append(self.oper_class.title()) + if self.licence_type and self.licence_type.upper() != "PERSON": + out.append(self.licence_type.title()) + if self.grid: + out.append(self.grid) + if self.postcode: + out.append(self.postcode) + if self.previous: + out.append(f"ex {self.previous}") + if self.trustee: + out.append(f"trustee {self.trustee}") + if self.expires: + out.append(f"expires {self.expires}") + return out + + +# Words the licence database writes in capitals that are not names, and would +# be wrong in title case: "PAUL PAKES COOK, III." must not become "Iii.". +_KEEP_CAPS = frozenset({"II", "III", "IV", "V", "VI", "VII", "VIII", "JR", + "SR", "MD", "DDS", "PHD", "HQ", "ARRL", "USA", "US", + "LLC", "INC", "LTD", "ARC", "EMA", "RACES", "ARES"}) + + +def person_case(name: str) -> str: + """Licence records are all capitals; make them readable without lying. + + Straight title case turns "III." into "Iii." and "ARRL" into "Arrl", so + anything that is an acronym, a Roman numeral or a suffix is left alone. + A name that was not in capitals to begin with is not touched at all -- + somebody else's idea of how their name is spelled is not ours to correct. + """ + if not name or not name.isupper(): + return name + out = [] + for word in name.split(): + bare = word.strip(".,") + out.append(word if bare in _KEEP_CAPS else word.title()) + return " ".join(out) + + +def split_postcode(line: str) -> tuple[str, str]: + """``"TUCSON, AZ 85742"`` -> ``("Tucson, AZ", "85742")``. + + The town and state are the useful half; the postcode is detail that + belongs further down the entry, not in the one line that has to say where + somebody is. + """ + m = re.match(r"^(.*?)[ ,]+(\d{5}(?:-\d{4})?)$", line.strip()) + if not m: + return person_case(line.strip()), "" + town = m.group(1).strip().rstrip(",") + # State abbreviations stay in capitals; the town is title-cased. + parts = [p.strip() for p in town.split(",")] + if len(parts) == 2 and len(parts[1]) == 2: + town = f"{person_case(parts[0])}, {parts[1].upper()}" + else: + town = person_case(town) + return town, m.group(2) + + +def _cache_path() -> Path: + root = os.environ.get("XDG_CACHE_HOME") or "~/.cache" + return Path(root).expanduser() / "bandsaunter" / "callsigns.json" + + +class CallsignBook: + """Resolves callsigns, in the background, once each. + + Rendering must never wait on a network request, so a lookup returns + immediately with whatever is known and fills itself in later; the display + redraws several times a second anyway. Results are cached on disk, so the + same net logged night after night is looked up once. + """ + + def __init__(self, online: bool = True, cache: Path | None = None, + timeout: float = 5.0, url: str = LOOKUP_URL, + max_age: float = 30 * 86_400): + self.online = online + self.timeout = timeout + self.url = url + self.max_age = max_age + self.cache_path = Path(cache) if cache is not None else _cache_path() + self._lock = threading.Lock() + self._entries: dict[str, Callsign] = {} + self._threads: list[threading.Thread] = [] + self._dirty = False + self._load() + + # -- cache ------------------------------------------------------------ + def _load(self) -> None: + try: + raw = json.loads(self.cache_path.read_text()) + except (OSError, ValueError): + return + now = time.time() + for call, body in (raw or {}).items(): + try: + entry = Callsign(**body) + except TypeError: + continue # written by a version with other fields + if entry.status in ("found", "unlisted") and \ + now - entry.fetched_at < self.max_age: + self._entries[call] = entry + + def save(self) -> None: + """Write the cache. Failing to is never worth an error.""" + with self._lock: + if not self._dirty: + return + body = {c: e.__dict__ for c, e in self._entries.items() + if e.status in ("found", "unlisted")} + self._dirty = False + try: + self.cache_path.parent.mkdir(parents=True, exist_ok=True) + tmp = self.cache_path.with_suffix(".tmp") + tmp.write_text(json.dumps(body, indent=1, sort_keys=True)) + tmp.replace(self.cache_path) + except OSError: + pass + + # -- lookup ----------------------------------------------------------- + def get(self, call: str) -> Callsign: + """What is known about ``call`` right now, starting a fetch if needed.""" + call = call.upper() + with self._lock: + entry = self._entries.get(call) + if entry is not None: + return entry + country, district = describe_prefix(call) + entry = Callsign(call=call, country=country, district=district, + status="pending" if self.online else "offline") + self._entries[call] = entry + if self.online: + thread = threading.Thread(target=self._fetch, args=(entry,), + daemon=True) + self._threads.append(thread) + thread.start() + return entry + + def get_all(self, calls) -> list[Callsign]: + return [self.get(c) for c in calls] + + def wait(self, timeout: float = 10.0) -> None: + """Block until the outstanding lookups finish. For scripts, not the UI.""" + deadline = time.time() + timeout + for thread in list(self._threads): + thread.join(max(0.0, deadline - time.time())) + + def _fetch(self, entry: Callsign) -> None: + try: + body = self._request(entry.call) + except Exception: + # Offline, blocked, rate-limited, or the service moved. The + # prefix still says something, and a browser that cannot look a + # callsign up must still show the transcript. + with self._lock: + entry.status = "offline" + return + self._apply(entry, body) + with self._lock: + self._dirty = True + + def _request(self, call: str) -> dict: + req = urllib.request.Request( + self.url.format(call=urllib.parse.quote(call)), + headers={"User-Agent": "bandsaunter"}) + with urllib.request.urlopen(req, timeout=self.timeout) as response: + return json.loads(response.read(64_000).decode("utf8", "replace")) + + @staticmethod + def _apply(entry: Callsign, body: dict) -> None: + entry.fetched_at = time.time() + if not isinstance(body, dict) or body.get("status") != "VALID": + entry.status = "unlisted" + return + address = body.get("address") or {} + location = body.get("location") or {} + other = body.get("otherInfo") or {} + current = body.get("current") or {} + previous = body.get("previous") or {} + trustee = body.get("trustee") or {} + entry.name = person_case(str(body.get("name") or "")) + entry.location, entry.postcode = split_postcode( + str(address.get("line2") or "")) + entry.grid = str(location.get("gridsquare") or "") + entry.oper_class = str(current.get("operClass") or "") + entry.licence_type = str(body.get("type") or "") + entry.expires = str(other.get("expiryDate") or "") + entry.previous = str(previous.get("callsign") or "") + entry.trustee = str(trustee.get("callsign") or "") + if not entry.country: + entry.country = "United States" + entry.status = "found" + + +# --------------------------------------------------------------------------- +# Rendering +# --------------------------------------------------------------------------- + +HEADING = "DETECTED CALLSIGNS:" + + +def report(entries: list[Callsign], width: int = 78) -> list[str]: + """The block that goes at the bottom of a transcript, as plain lines.""" + if not entries: + return [] + out = [HEADING] + pad = max((len(e.call) for e in entries), default=4) + for entry in entries: + line = f" {entry.call:<{pad}} {entry.summary()}" + extra = entry.details() + if extra: + line += " · " + " · ".join(extra) + out.append(line[:width] if width and len(line) > width + else line) + return out diff --git a/packaging/bandsaunter.1 b/packaging/bandsaunter.1 index d480756..3d5c372 100644 --- a/packaging/bandsaunter.1 +++ b/packaging/bandsaunter.1 @@ -1,5 +1,5 @@ .\" Generated by packaging/make-man.py -- do not edit by hand. -.TH BANDSAUNTER 1 "2026-08-22" "bandsaunter 2026-08-22_03" "User Commands" +.TH BANDSAUNTER 1 "2026-08-22" "bandsaunter 2026-08-22_04" "User Commands" .SH NAME bandsaunter \- scan, record and identify radio signals with an RTL-SDR .SH SYNOPSIS @@ -877,6 +877,30 @@ to have each one written into the lock\-out list as it is found, so the scanner stops looking at it at all; with .B \-\-save\-lockouts on, that list survives a restart. +.SH TRANSCRIPTS +Anything the content check identifies as voice is passed to a speech +recogniser, and the words are written to a +.I _transcription.txt +beside the recording. Only voice: running a recogniser over Morse or a data +burst costs seconds and produces nothing. +.PP +One transcript per transmission, and none is ever overwritten \[em] the +timestamp is part of the name, so two overs on one frequency cannot land on +the same file. +.PP +With +.B \-\-combine +there is one recording per frequency, so there is one transcript per +frequency, and each over is appended to it with the time it was heard. An +unattended receiver keeps adding to that file night after night rather than +starting it over. +.PP +A capture with nothing recognisable in it produces no file at all, rather +than a directory of placeholders. +.PP +.BR saunterbrowse (1) +reads these back, and lists any callsigns it finds in them with the licence +they belong to. .SH HF RECEPTION These receivers cannot normally tune below about 24 MHz. Below that they can sample the antenna directly instead, which opens up shortwave: broadcast, diff --git a/packaging/make-browse-man.py b/packaging/make-browse-man.py index 524d531..874ec35 100755 --- a/packaging/make-browse-man.py +++ b/packaging/make-browse-man.py @@ -30,6 +30,8 @@ saunterbrowse \- read and listen to what a bandsaunter scan collected .RB [ \-\-player .IR CMD ] .RB [ \-\-list ] +.RB [ \-\-callsigns ] +.RB [ \-\-no\-lookup ] .SH DESCRIPTION A scan leaves a directory of recordings. Beside each one is a JSON file holding what the classifier made of it, and for speech a transcript of what @@ -118,6 +120,14 @@ The command used to play a recording. The path is appended as its last argument. By default the first of these that is installed is used: {players}. .TP +.B \-\-callsigns +Print every callsign heard in the directory, with the licence it belongs to +and where and when it was heard, then exit. +.TP +.B \-\-no\-lookup +Do not contact the licence database. Callsigns are still found and still +described from their prefix; only the name and address are missing. +.TP .B \-\-list Print one line per recording and exit, without drawing anything. This is what to use over a pipe, in a script, or anywhere there is no terminal. @@ -133,6 +143,46 @@ message says which packages would fix it. .PP Over ssh there is usually no sound server at the far end. The browser and its transcripts work regardless; only Enter has nothing to do. +.SH DETECTED CALLSIGNS +Under the transcript, headed +.BR "DETECTED CALLSIGNS:" , +is every callsign heard in it, with the name and location on the licence. +.PP +Finding them is not a matter of one regular expression over the text as +written. A speech recogniser is poor at callsigns \[em] they are not words, +they are said one character at a time \[em] so it breaks them wherever the +speaker paused and writes the phonetic alphabet down verbatim. "KU 0W", +"K7 RA" and "kilo uniform zero whiskey" are all one callsign each, and all +three are read back correctly. +.PP +False positives are the thing to avoid: a browser that invents callsigns is +worse than one that finds none. So a run of words is only accepted when none +of its parts is an ordinary English word \[em] "or 3. Can you open 4" fits the +shape once the punctuation is gone, and is not a callsign \[em] while a single +token said in one breath is trusted, because W1BOY is a perfectly good +callsign. +.PP +The lookup uses the FCC's own licence data, published at callook.info, which +needs no account and no key. The callsign is the only thing sent, results are +cached under +.I ~/.cache/bandsaunter/ +so the same net is looked up once however many nights it is recorded, and a +lookup never delays the display: the entry says "looking up" and fills itself +in. +.PP +.B \-\-no\-lookup +contacts nothing at all. Callsigns are still found, and still described from +their own structure \[em] the prefix is allocated by the ITU and the digit is +the US licensing district, so a callsign says which country and which region +it belongs to without any database. Outside the United States that is all +there is; callook.info holds US licences only. +.PP +.B \-\-callsigns +prints every callsign in the directory, who it belongs to, and each frequency +and time it was heard on, then exits. +.PP +US amateur licence records are public by law, and include the licensee's +address. That is what is shown. .SH TRANSCRIPTS A transcript appears only where a recogniser produced one, which means the capture was judged to be speech and @@ -189,6 +239,14 @@ saunterbrowse \-\-list | grep \-i "mile marker" .RE .PP Search the transcripts from a script. +.PP +.RS +.EX +saunterbrowse \-\-callsigns +.EE +.RE +.PP +Everyone who identified themselves, and where they were heard. .SH EXIT STATUS 0 on a clean exit, 1 when the directory holds no recordings, 2 when it does not exist or there is no terminal to draw on. diff --git a/packaging/make-man.py b/packaging/make-man.py index 2dcc798..0f9499b 100755 --- a/packaging/make-man.py +++ b/packaging/make-man.py @@ -328,6 +328,30 @@ to have each one written into the lock\-out list as it is found, so the scanner stops looking at it at all; with .B \-\-save\-lockouts on, that list survives a restart. +.SH TRANSCRIPTS +Anything the content check identifies as voice is passed to a speech +recogniser, and the words are written to a +.I _transcription.txt +beside the recording. Only voice: running a recogniser over Morse or a data +burst costs seconds and produces nothing. +.PP +One transcript per transmission, and none is ever overwritten \[em] the +timestamp is part of the name, so two overs on one frequency cannot land on +the same file. +.PP +With +.B \-\-combine +there is one recording per frequency, so there is one transcript per +frequency, and each over is appended to it with the time it was heard. An +unattended receiver keeps adding to that file night after night rather than +starting it over. +.PP +A capture with nothing recognisable in it produces no file at all, rather +than a directory of placeholders. +.PP +.BR saunterbrowse (1) +reads these back, and lists any callsigns it finds in them with the licence +they belong to. .SH HF RECEPTION These receivers cannot normally tune below about 24 MHz. Below that they can sample the antenna directly instead, which opens up shortwave: broadcast, diff --git a/packaging/saunterbrowse.1 b/packaging/saunterbrowse.1 index 7778bdc..b81b439 100644 --- a/packaging/saunterbrowse.1 +++ b/packaging/saunterbrowse.1 @@ -1,5 +1,5 @@ .\" Generated by packaging/make-browse-man.py -- do not edit by hand. -.TH SAUNTERBROWSE 1 "2026-08-22" "bandsaunter 2026-08-22_03" "User Commands" +.TH SAUNTERBROWSE 1 "2026-08-22" "bandsaunter 2026-08-22_04" "User Commands" .SH NAME saunterbrowse \- read and listen to what a bandsaunter scan collected .SH SYNOPSIS @@ -12,6 +12,8 @@ saunterbrowse \- read and listen to what a bandsaunter scan collected .RB [ \-\-player .IR CMD ] .RB [ \-\-list ] +.RB [ \-\-callsigns ] +.RB [ \-\-no\-lookup ] .SH DESCRIPTION A scan leaves a directory of recordings. Beside each one is a JSON file holding what the classifier made of it, and for speech a transcript of what @@ -100,6 +102,14 @@ The command used to play a recording. The path is appended as its last argument. By default the first of these that is installed is used: pw-play, paplay, aplay, play, ffplay, mpv. .TP +.B \-\-callsigns +Print every callsign heard in the directory, with the licence it belongs to +and where and when it was heard, then exit. +.TP +.B \-\-no\-lookup +Do not contact the licence database. Callsigns are still found and still +described from their prefix; only the name and address are missing. +.TP .B \-\-list Print one line per recording and exit, without drawing anything. This is what to use over a pipe, in a script, or anywhere there is no terminal. @@ -115,6 +125,46 @@ message says which packages would fix it. .PP Over ssh there is usually no sound server at the far end. The browser and its transcripts work regardless; only Enter has nothing to do. +.SH DETECTED CALLSIGNS +Under the transcript, headed +.BR "DETECTED CALLSIGNS:" , +is every callsign heard in it, with the name and location on the licence. +.PP +Finding them is not a matter of one regular expression over the text as +written. A speech recogniser is poor at callsigns \[em] they are not words, +they are said one character at a time \[em] so it breaks them wherever the +speaker paused and writes the phonetic alphabet down verbatim. "KU 0W", +"K7 RA" and "kilo uniform zero whiskey" are all one callsign each, and all +three are read back correctly. +.PP +False positives are the thing to avoid: a browser that invents callsigns is +worse than one that finds none. So a run of words is only accepted when none +of its parts is an ordinary English word \[em] "or 3. Can you open 4" fits the +shape once the punctuation is gone, and is not a callsign \[em] while a single +token said in one breath is trusted, because W1BOY is a perfectly good +callsign. +.PP +The lookup uses the FCC's own licence data, published at callook.info, which +needs no account and no key. The callsign is the only thing sent, results are +cached under +.I ~/.cache/bandsaunter/ +so the same net is looked up once however many nights it is recorded, and a +lookup never delays the display: the entry says "looking up" and fills itself +in. +.PP +.B \-\-no\-lookup +contacts nothing at all. Callsigns are still found, and still described from +their own structure \[em] the prefix is allocated by the ITU and the digit is +the US licensing district, so a callsign says which country and which region +it belongs to without any database. Outside the United States that is all +there is; callook.info holds US licences only. +.PP +.B \-\-callsigns +prints every callsign in the directory, who it belongs to, and each frequency +and time it was heard on, then exits. +.PP +US amateur licence records are public by law, and include the licensee's +address. That is what is shown. .SH TRANSCRIPTS A transcript appears only where a recogniser produced one, which means the capture was judged to be speech and @@ -171,6 +221,14 @@ saunterbrowse \-\-list | grep \-i "mile marker" .RE .PP Search the transcripts from a script. +.PP +.RS +.EX +saunterbrowse \-\-callsigns +.EE +.RE +.PP +Everyone who identified themselves, and where they were heard. .SH EXIT STATUS 0 on a clean exit, 1 when the directory holds no recordings, 2 when it does not exist or there is no terminal to draw on. diff --git a/tests/test_browse.py b/tests/test_browse.py index 4c568d3..4a68676 100644 --- a/tests/test_browse.py +++ b/tests/test_browse.py @@ -19,6 +19,7 @@ from rich.console import Console from bandsaunter.browse import (Browser, Player, PLAYERS, Keyboard, main, scan_directory) +from bandsaunter.callsign import CallsignBook # -- fixtures ---------------------------------------------------------------- @@ -51,6 +52,7 @@ def make_capture(directory: Path, mhz: float, when: str, mode: str, @pytest.fixture def library(tmp_path): """A small recordings directory covering the cases that render differently.""" + # The transcript names a callsign, as a real one from a net would. make_capture(tmp_path, 146.52, "2026-08-22_10_00_00", "nfm", 4.0, transcript="Net control, this is W1AW, standing by.", meta={"category": "voice", "snr_db": 21.5, @@ -68,6 +70,20 @@ def library(tmp_path): return tmp_path +@pytest.fixture(autouse=True) +def no_network(monkeypatch): + """No test in this file may contact the licence database. + + One did, silently, and passed -- it was only visible because the assertion + it failed printed a real operator's address. A stub that is forgotten + should fail loudly rather than work. + """ + def refuse(self, call): + raise AssertionError(f"a test tried to look up {call} for real") + + monkeypatch.setattr(CallsignBook, "_request", refuse) + + def browser(directory, width=100, height=30, player=None) -> Browser: console = Console(width=width, height=height, force_terminal=True) return Browser(directory, console=console, @@ -655,3 +671,172 @@ def test_the_output_directory_is_found_without_being_told(monkeypatch, monkeypatch.setenv("BANDSAUNTER_OUTPUT", str(tmp_path)) from bandsaunter.browse import default_directory assert default_directory() == tmp_path + + +# -- detected callsigns ------------------------------------------------------ + +class StubBook(CallsignBook): + """Answers lookups from a dict, so no test touches the network.""" + + def __init__(self, answers=None, tmp=None, **kw): + self.answers = answers or {} + self.requested = [] + super().__init__(cache=(tmp or Path("/nonexistent")) / "c.json", **kw) + + def _request(self, call): + self.requested.append(call) + if call not in self.answers: + raise OSError("offline") + return self.answers[call] + + +KU0W = { + "status": "VALID", "type": "PERSON", "name": "ROD R GOWDY", + "current": {"callsign": "KU0W", "operClass": "EXTRA"}, + "previous": {"callsign": ""}, "trustee": {"callsign": ""}, + "address": {"line2": "TUCSON, AZ 85742"}, + "location": {"gridsquare": "DM42lj"}, + "otherInfo": {"expiryDate": "08/09/2034"}, +} + + +@pytest.fixture +def net(tmp_path): + """A directory whose transcript names a callsign, and a stubbed lookup.""" + make_capture(tmp_path, 146.52, "2026-08-22_10_00_00", "nfm", 4.0, + transcript="Net control, this is KU 0W, standing by.", + meta={"category": "voice", "classification": "NFM voice"}) + return tmp_path + + +def net_browser(directory, tmp, answers=None, **kw): + b = browser(directory, **kw) + b.book = StubBook(answers if answers is not None else {"KU0W": KU0W}, + tmp=tmp) + return b + + +def test_callsigns_are_listed_under_the_transcript(net, tmp_path): + b = net_browser(net, tmp_path) + b.book.get("KU0W") + b.book.wait(5) + out = frame(b) + head, _, rest = out.partition("recordings in") + assert "DETECTED CALLSIGNS:" in head + assert "KU0W" in head + assert "Rod R Gowdy" in head and "Tucson, AZ" in head + # Under the words, not above them. + assert head.index("standing by") < head.index("DETECTED CALLSIGNS:") + + +def test_a_callsign_broken_by_the_recogniser_is_still_found(net, tmp_path): + """The transcript says "KU 0W"; the licence is under KU0W.""" + cap = scan_directory(net)[0] + assert cap.callsigns == ["KU0W"] + + +def test_the_lookup_is_shown_as_pending_before_it_lands(net, tmp_path): + class Slow(StubBook): + def _request(self, call): + time.sleep(1.0) + return super()._request(call) + + b = browser(net) + b.book = Slow({"KU0W": KU0W}, tmp=tmp_path) + assert "looking up" in frame(b) + + +def test_a_transcript_with_no_callsign_has_no_block(tmp_path): + make_capture(tmp_path, 146.52, "2026-08-22_10_00_00", "nfm", + transcript="the meeting is at seven on the fourth Saturday", + meta={"category": "voice"}) + assert "DETECTED CALLSIGNS" not in frame(browser(tmp_path)) + + +def test_a_callsign_that_cannot_be_looked_up_still_appears(net, tmp_path): + """Offline, the prefix still says which country and district it is.""" + b = net_browser(net, tmp_path, answers={}) + b.book.get("KU0W") + b.book.wait(5) + out = frame(b) + assert "KU0W" in out and "United States" in out + + +def test_the_callsigns_survive_a_transcript_too_long_for_the_panel(tmp_path): + """Whatever else is squeezed, the callsigns stay: they are the part of the + panel that cannot be recovered by listening to the recording.""" + make_capture(tmp_path, 146.52, "2026-08-22_10_00_00", "nfm", + transcript="KU0W " + " ".join(f"word{i}" for i in range(400)), + meta={"category": "voice"}) + b = net_browser(tmp_path, tmp_path, height=22) + b.book.get("KU0W") + b.book.wait(5) + out = frame(b) + assert "DETECTED CALLSIGNS:" in out and "Rod R Gowdy" in out + assert "more line(s)" in out + + +def test_the_reader_shows_them_at_the_end_of_the_words(tmp_path): + make_capture(tmp_path, 146.52, "2026-08-22_10_00_00", "nfm", + transcript="KU0W " + " ".join(f"word{i}" for i in range(400)), + meta={"category": "voice"}) + b = net_browser(tmp_path, tmp_path, height=22) + b.book.get("KU0W") + b.book.wait(5) + b.handle("t") + assert "DETECTED CALLSIGNS" not in frame(b), "shown beside the middle" + b.handle("end") + assert "DETECTED CALLSIGNS:" in frame(b) + + +def test_the_frame_still_fits_with_callsigns(net, tmp_path): + for height in (16, 20, 24, 40): + b = net_browser(net, tmp_path, height=height) + b.book.get("KU0W") + b.book.wait(5) + lines = frame(b).rstrip("\n").split("\n") + assert len(lines) <= height, f"{len(lines)} lines in {height}" + + +def test_callsigns_are_found_once_per_recording(net): + cap = scan_directory(net)[0] + assert cap._calls is None + first = cap.callsigns + assert cap._calls is not None + assert cap.callsigns is first + + +# -- the callsign command line ----------------------------------------------- + +def test_callsigns_flag_reports_where_each_was_heard(net, capsys, monkeypatch): + monkeypatch.setattr("bandsaunter.browse.CallsignBook", + lambda **kw: StubBook({"KU0W": KU0W}, tmp=net, **kw)) + assert main(["--callsigns", str(net)]) == 0 + out = capsys.readouterr().out + assert "KU0W" in out and "Rod R Gowdy" in out + assert "heard on" in out and "146.52 MHz" in out + + +def test_callsigns_flag_says_so_when_there_are_none(tmp_path, capsys): + make_capture(tmp_path, 856.5625, "2026-08-22_11_00_00", "fsk", 2.0, + meta={"category": "trunk"}) + assert main(["--callsigns", str(tmp_path)]) == 1 + assert "no callsigns" in capsys.readouterr().out + + +def test_no_lookup_contacts_nothing(net, capsys, monkeypatch): + """The flag has to mean it: nothing leaves the machine.""" + made = [] + + class Watching(StubBook): + def _request(self, call): + made.append(call) + return super()._request(call) + + monkeypatch.setattr("bandsaunter.browse.CallsignBook", + lambda **kw: Watching({"KU0W": KU0W}, tmp=net, **kw)) + assert main(["--callsigns", "--no-lookup", str(net)]) == 0 + out = capsys.readouterr().out + assert made == [], "a request was made with --no-lookup" + assert "KU0W" in out + assert "United States" in out, "the prefix should still be described" diff --git a/tests/test_callsign.py b/tests/test_callsign.py new file mode 100644 index 0000000..d493d60 --- /dev/null +++ b/tests/test_callsign.py @@ -0,0 +1,358 @@ +"""Finding callsigns in a transcript, and saying who they belong to. + +The hard part is not the regular expression, it is everything the speech +recogniser does to a callsign on the way: it breaks it wherever the speaker +paused, and it writes the phonetic alphabet down as words. So most of these +are about text that came out of a real recogniser rather than text anyone +would type. + +Nothing here touches the network. The lookup is tested against a stub, and +one test asserts that no request is made when it is switched off. +""" +import json +import time + +import pytest + +from bandsaunter.callsign import (Callsign, CallsignBook, HEADING, + NOT_CALLSIGNS, SHAPE, describe_prefix, + find_callsigns, person_case, report, + split_postcode) + + +@pytest.fixture(autouse=True) +def no_network(monkeypatch): + """Nothing here may contact the licence database for real. + + Every lookup test uses a stub; this makes forgetting one fail loudly + rather than quietly pass with somebody's real address in the output. + """ + def refuse(self, call): + raise AssertionError(f"a test tried to look up {call} for real") + + monkeypatch.setattr(CallsignBook, "_request", refuse) + + +# -- finding them ------------------------------------------------------------ + +@pytest.mark.parametrize("text,want", [ + ("This is W1AW calling CQ", ["W1AW"]), + ("K7RA this is N7XYZ over", ["K7RA", "N7XYZ"]), + ("VE3ABC and G0ABC and 2E0XYZ", ["VE3ABC", "G0ABC", "2E0XYZ"]), + ("net control, W1AW, W1AW again", ["W1AW"]), +]) +def test_a_callsign_written_properly_is_found(text, want): + assert find_callsigns(text) == want + + +@pytest.mark.parametrize("text,want", [ + ("Alright, KU 0W. And the HF net will be a couple minutes.", ["KU0W"]), + ("N7 XYZ this is K7 RA", ["N7XYZ", "K7RA"]), + ("W 1 A W standing by", ["W1AW"]), +]) +def test_a_callsign_broken_across_words_is_still_one_callsign(text, want): + """A recogniser breaks it wherever the speaker paused. "KU 0W" is what + came out of a real recording of somebody saying KU0W.""" + assert find_callsigns(text) == want + + +@pytest.mark.parametrize("text,want", [ + ("kilo uniform zero whiskey, are you there", ["KU0W"]), + ("whiskey one alpha whiskey this is kilo seven romeo alpha", + ["W1AW", "K7RA"]), + ("november seven x-ray yankee zulu", ["N7XYZ"]), + ("kilo uniform zero whisky", ["KU0W"]), # both spellings + ("niner", []), +]) +def test_the_phonetic_alphabet_is_read_back(text, want): + """Spelled out, a callsign arrives as one token per character.""" + assert find_callsigns(text) == want + + +@pytest.mark.parametrize("text", [ + "7 on alleviate or 3. Can you open 4 and close the rest.", + "CC1 boy", + "I bought 3 of them for $5 and a B4U model MP3 player", + "the 4th Saturday at 1.30pm on 7.242 megahertz", + "Check out communication of group activities and training.", + "we need 2 of them and 4 spare", + "it was on channel 3 for a while", +]) +def test_ordinary_speech_produces_no_callsigns(text): + """Every one of these came out of a real transcript or is one word away + from something that did. A browser that invents callsigns is worse than + one that finds none.""" + assert find_callsigns(text) == [] + + +def test_a_join_is_refused_when_a_part_is_an_english_word(): + """"or 3. Can you" fits the shape once the punctuation is gone. Nobody + spells a callsign with English words.""" + assert find_callsigns("or 3 can") == [] + assert find_callsigns("CC1 boy") == [] + # ... but a single token said in one breath is trusted. + assert find_callsigns("W1BOY is on frequency") == ["W1BOY"] + + +def test_a_lone_article_is_not_a_prefix(): + assert find_callsigns("a B4U player") == [] + assert find_callsigns("alpha bravo four uniform") == ["AB4U"] + + +def test_the_same_callsign_is_reported_once(): + assert find_callsigns("W1AW W1AW W1AW") == ["W1AW"] + + +def test_they_come_back_in_the_order_they_were_said(): + assert find_callsigns("N7XYZ this is W1AW") == ["N7XYZ", "W1AW"] + + +def test_the_count_is_capped(): + text = " ".join(f"W{i}AB" for i in range(0, 9)) + " K7RA N7XYZ VE3ABC" + assert len(find_callsigns(text, max_found=4)) == 4 + + +def test_empty_text_is_not_an_error(): + assert find_callsigns("") == [] + assert find_callsigns(" \n ") == [] + + +def test_every_excluded_word_could_actually_have_matched(): + """A list of words that could not match anyway would only suggest the + filter was doing more work than it is.""" + for word in NOT_CALLSIGNS: + assert SHAPE.match(word), f"{word} never matched in the first place" + + +# -- what the callsign says about itself ------------------------------------- + +@pytest.mark.parametrize("call,country", [ + ("W1AW", "United States"), ("KU0W", "United States"), + ("N7XYZ", "United States"), ("AA1AB", "United States"), + ("VE3ABC", "Canada"), ("XE1ABC", "Mexico"), + ("G0ABC", "United Kingdom"), ("2E0XYZ", "United Kingdom"), + ("JA1ABC", "Japan"), ("VK2DEF", "Australia"), + ("IK4ABC", "Italy"), ("DL1ABC", "Germany"), + ("4X4ABC", "Israel"), ("ZZ9ZZZ", ""), +]) +def test_the_prefix_names_the_country_without_a_database(call, country): + """This works offline and for callsigns no database holds.""" + assert describe_prefix(call)[0] == country + + +def test_a_one_letter_prefix_does_not_claim_everything(): + """"I" is Italy, but only when what follows is the rest of a prefix.""" + assert describe_prefix("IK4ABC")[0] == "Italy" + assert describe_prefix("KU0W")[0] == "United States" # not Italy-shaped + + +def test_the_digit_gives_the_us_district(): + assert "New England" in describe_prefix("W1AW")[1] + assert "California" in describe_prefix("K6ABC")[1] + assert describe_prefix("VE3ABC")[1] == "", "districts are a US idea" + + +# -- tidying the licence record ---------------------------------------------- + +@pytest.mark.parametrize("raw,want", [ + ("ROD R GOWDY", "Rod R Gowdy"), + ("PAUL PAKES COOK, III.", "Paul Pakes Cook, III."), + ("ARRL HQ OPERATORS CLUB", "ARRL HQ Operators Club"), + ("JOHN SMITH JR", "John Smith JR"), + ("de Vries, Anna", "de Vries, Anna"), # already mixed: left alone +]) +def test_licence_names_are_made_readable_without_being_corrupted(raw, want): + assert person_case(raw) == want + + +@pytest.mark.parametrize("raw,town,code", [ + ("TUCSON, AZ 85742", "Tucson, AZ", "85742"), + ("SEATTLE, WA 98105-3505", "Seattle, WA", "98105-3505"), + ("NEWINGTON, CT 06111", "Newington, CT", "06111"), + ("SOMEWHERE ABROAD", "Somewhere Abroad", ""), +]) +def test_the_postcode_is_separated_from_the_place(raw, town, code): + """The town and state say where somebody is; the postcode is detail.""" + assert split_postcode(raw) == (town, code) + + +# -- looking them up --------------------------------------------------------- + +VALID = { + "status": "VALID", "type": "PERSON", "name": "ROD R GOWDY", + "current": {"callsign": "KU0W", "operClass": "EXTRA"}, + "previous": {"callsign": "KK7QPA"}, + "trustee": {"callsign": ""}, + "address": {"line1": "5455 W OASIS RD", "line2": "TUCSON, AZ 85742"}, + "location": {"gridsquare": "DM42lj"}, + "otherInfo": {"expiryDate": "08/09/2034"}, +} + + +class StubBook(CallsignBook): + """A book whose lookups are answered from a dict, not the network.""" + + def __init__(self, answers, **kw): + self.answers = answers + self.requested = [] + super().__init__(**kw) + + def _request(self, call): + self.requested.append(call) + if call not in self.answers: + raise OSError("no route to host") + return self.answers[call] + + +def book(tmp_path, answers=None, **kw): + return StubBook(answers if answers is not None else {"KU0W": VALID}, + cache=tmp_path / "cache.json", **kw) + + +def test_a_lookup_fills_in_the_name_and_the_place(tmp_path): + b = book(tmp_path) + b.get("KU0W") + b.wait(5) + entry = b.get("KU0W") + assert entry.known + assert entry.name == "Rod R Gowdy" + assert entry.location == "Tucson, AZ" + assert entry.grid == "DM42lj" + assert entry.oper_class == "EXTRA" + assert entry.previous == "KK7QPA" + + +def test_rendering_never_waits_for_the_network(tmp_path): + """A lookup returns at once with what is known, and fills itself in. + + The stub is deliberately slow: a fast one would finish before the + assertion and the test would pass whether or not the call blocked. + """ + class Slow(StubBook): + def _request(self, call): + time.sleep(1.0) + return super()._request(call) + + b = Slow({"KU0W": VALID}, cache=tmp_path / "cache.json") + started = time.time() + entry = b.get("KU0W") + assert time.time() - started < 0.3, "the lookup blocked the caller" + assert entry.status == "pending" + assert entry.summary() == "looking up…" + b.wait(10) + assert b.get("KU0W").known + + +def test_an_unknown_callsign_says_what_the_prefix_says(tmp_path): + """Not in the database is not nothing: the prefix is still information.""" + b = book(tmp_path, {"G0ABC": {"status": "INVALID"}}) + b.get("G0ABC") + b.wait(5) + summary = b.get("G0ABC").summary() + assert "unlisted" in summary and "United Kingdom" in summary + + +def test_a_lookup_that_fails_is_not_an_error(tmp_path): + """Offline, blocked or rate-limited, the transcript still has to show.""" + b = book(tmp_path, {}) + b.get("W1AW") + b.wait(5) + entry = b.get("W1AW") + assert entry.status == "offline" + assert "United States" in entry.summary() + + +def test_nothing_is_requested_when_lookup_is_switched_off(tmp_path): + b = book(tmp_path, online=False) + entry = b.get("KU0W") + b.wait(2) + assert b.requested == [], "a request was made with lookup off" + assert entry.status == "offline" + assert "United States" in entry.summary() + + +def test_a_callsign_is_looked_up_once(tmp_path): + b = book(tmp_path) + for _ in range(5): + b.get("KU0W") + b.wait(5) + assert b.requested == ["KU0W"] + + +def test_results_are_cached_between_runs(tmp_path): + """The same net, logged night after night, is looked up once.""" + first = book(tmp_path) + first.get("KU0W") + first.wait(5) + first.save() + assert (tmp_path / "cache.json").exists() + + second = book(tmp_path) + entry = second.get("KU0W") + assert entry.known, "the cache was not read" + assert second.requested == [], "a cached callsign was looked up again" + + +def test_a_stale_cache_entry_is_looked_up_again(tmp_path): + first = book(tmp_path) + first.get("KU0W") + first.wait(5) + first.save() + body = json.loads((tmp_path / "cache.json").read_text()) + body["KU0W"]["fetched_at"] = time.time() - 400 * 86_400 + (tmp_path / "cache.json").write_text(json.dumps(body)) + + second = book(tmp_path) + second.get("KU0W") + second.wait(5) + assert second.requested == ["KU0W"] + + +def test_a_cache_from_another_version_is_ignored_not_fatal(tmp_path): + (tmp_path / "cache.json").write_text( + json.dumps({"KU0W": {"call": "KU0W", "somethingelse": 1}})) + b = book(tmp_path) + b.get("KU0W") + b.wait(5) + assert b.requested == ["KU0W"], "an unusable cache entry was trusted" + assert b.get("KU0W").known + + +def test_an_unreadable_cache_is_ignored_not_fatal(tmp_path): + (tmp_path / "cache.json").write_text("{not json") + b = book(tmp_path) + b.get("KU0W") + b.wait(5) + assert b.get("KU0W").known + + +def test_only_resolved_entries_are_cached(tmp_path): + """Caching a failure would make one flaky lookup permanent.""" + b = book(tmp_path, {}) + b.get("W1AW") + b.wait(5) + b.save() + body = json.loads((tmp_path / "cache.json").read_text()) \ + if (tmp_path / "cache.json").exists() else {} + assert "W1AW" not in body + + +# -- the block that goes under the transcript -------------------------------- + +def test_the_report_is_headed_the_way_it_was_asked_for(): + entry = Callsign(call="W1AW", name="ARRL", location="Newington, CT", + status="found") + lines = report([entry]) + assert lines[0] == HEADING == "DETECTED CALLSIGNS:" + assert "W1AW" in lines[1] and "Newington" in lines[1] + + +def test_nothing_found_means_no_block_at_all(): + assert report([]) == [] + + +def test_the_report_carries_the_extra_detail(): + entry = Callsign(call="KU0W", name="Rod R Gowdy", location="Tucson, AZ", + oper_class="EXTRA", grid="DM42lj", status="found") + line = report([entry], width=0)[1] + assert "Extra" in line and "DM42lj" in line diff --git a/tests/test_transcribe.py b/tests/test_transcribe.py index a7b7a10..3798968 100644 --- a/tests/test_transcribe.py +++ b/tests/test_transcribe.py @@ -5,6 +5,7 @@ from datetime import datetime from pathlib import Path import numpy as np +import re import pytest from bandsaunter import transcribe as tr @@ -424,3 +425,69 @@ def test_a_missing_vendor_directory_is_not_added(tmp_path, monkeypatch): monkeypatch.setattr(tr, "VENDOR_DIR", tmp_path / "absent") tr._add_vendor_path() assert str(tmp_path / "absent") not in sys.path + + +# --------------------------------------------------------------------------- +# Two transmissions on one frequency +# --------------------------------------------------------------------------- + +def test_each_transmission_gets_a_transcript_of_its_own(tmp_path, fake_engine, + monkeypatch): + """The default is one file per transmission, named after it. Nothing is + overwritten because nothing is shared: the timestamp is in the name, so + two overs on one frequency cannot land on one file.""" + monkeypatch.setattr("bandsaunter.scanner.available_engine", lambda: "fake") + scanner, hits = _scan(tmp_path, [V(146_520_000, "nfm", 0.4, 12_500, "v")], + record_seconds=2.0, max_cycles=3, + revisit_seconds=0.05) + assert len(hits) >= 2, "only one transmission was captured" + texts = sorted(tmp_path.glob("*_transcription.txt")) + assert len(texts) == len(hits), [p.name for p in texts] + for path in texts: + assert "transcribed text" in path.read_text() + # One transmission, one line. More than one would mean two captures + # had collided on a single name. + assert len(path.read_text().strip().split("\n")) == 1, path.name + + +def test_combining_appends_every_over_to_one_file(tmp_path, fake_engine, + monkeypatch): + """With --combine there is one recording per frequency, so there is one + transcript per frequency, and each over is added to the end of it with the + time it was heard.""" + monkeypatch.setattr("bandsaunter.scanner.available_engine", lambda: "fake") + scanner, hits = _scan(tmp_path, [V(146_520_000, "nfm", 0.4, 12_500, "v")], + combine_by_frequency=True, + announce_timestamps=False, record_seconds=2.0, + max_cycles=3, revisit_seconds=0.05) + assert len(hits) >= 2 + texts = list(tmp_path.glob("*_transcription.txt")) + assert len(texts) == 1, [p.name for p in texts] + lines = [ln for ln in texts[0].read_text().strip().split("\n") if ln] + assert len(lines) == len(hits) + for line in lines: + assert re.match(r"^\[\d{4}-\d\d-\d\d \d\d:\d\d:\d\d\] ", line), line + + +def test_a_later_run_never_truncates_an_earlier_transcript(tmp_path, + fake_engine, + monkeypatch): + """The question this answers: can a later transmission on the same + frequency wipe out an earlier one's words? The combined file is opened + for append, and it is the only transcript two captures ever share, so an + unattended receiver adds to it night after night rather than starting it + over.""" + monkeypatch.setattr("bandsaunter.scanner.available_engine", lambda: "fake") + args = dict(combine_by_frequency=True, announce_timestamps=False, + record_seconds=2.0, max_cycles=2, revisit_seconds=0.05) + tx = [V(146_520_000, "nfm", 0.4, 12_500, "v")] + + _scan(tmp_path, tx, **args) + combined = next(iter(tmp_path.glob("*_transcription.txt"))) + before = combined.read_text() + assert before.strip() + + _scan(tmp_path, tx, **args) # a second run into the same directory + after = combined.read_text() + assert after.startswith(before), "the earlier transcript was overwritten" + assert len(after) > len(before), "the later over was not added"