8a789e57e1
Two things, both found by measuring rather than by reading the code. The voice-activity filter inside the recogniser is off. It was costing words: across a night of land-mobile captures it dropped 5-15% of what the same model finds without it -- 491 against 507, 339 against 384, 263 against 310 -- because a single-word over between two transmissions looks to a VAD exactly like the noise it exists to remove, and on a scanner those short replies are the ones worth having. Turning it off has a cost, and the cost is that Whisper hands back "You" for five seconds of hiss as confidently as it hands back a sentence. So the whole capture is now asked once whether anything in it rises above its own noise. Digital silence measures 0.0 dB of contrast and hiss at any level 0.7, while the quietest real capture of that night measures 8.9 and most measure 10-27; the bar sits at 3, an order of magnitude clear of both. It can veto a capture but never trim one, which is the whole difference between it and the filter it replaces. The second thing: callsigns like WQVF960 were being missed entirely. The shape being matched was the amateur one -- prefix, district digit, suffix -- and everything else the FCC licenses is written the other way round, the letters first and then the digits. On the GMRS and business channels that is most of what is said: nine callsigns across five transcripts of one evening went by unrecognised, and now do not. The shape is written as the three allocations that exist rather than as "letters then digits", which claims KN95, WD40 and KC135. Its letters are checked against the word list even when they arrive as a single token, which the amateur shape does not need -- no English word has a digit in the middle of it, but "west 120" and "word 100" fit this one exactly. Lookups now fall back to hamdb.org when callook has nothing. Not a spare copy: callook holds United States amateur licences only, so DL1ABC and VE3ABC are INVALID there and resolve perfectly well from the other. And a GMRS callsign is not looked up at all -- every database reachable without an account is an amateur register, so reporting WQVF960 as "unlisted" would blame the callsign for the absence of a source. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016PsWPTweCT6pwxKngvVxcg
347 lines
13 KiB
Python
347 lines
13 KiB
Python
"""A station that identifies itself, and what happens to it after that.
|
|
|
|
Most stations on the air never say a word. A repeater, a beacon or an
|
|
unattended transmitter sends its callsign in Morse and stops, and until this
|
|
existed the text was decoded, written into the sidecar, and read by nobody:
|
|
the callsign book and the map were built where the transcripts were, and a
|
|
CW ident produces no transcript.
|
|
|
|
So these are about the whole path -- decode, find the callsign, look it up,
|
|
say so, put it on the map -- and about the two places it is allowed to say
|
|
nothing rather than say the wrong thing.
|
|
"""
|
|
import json
|
|
import wave
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pytest
|
|
from rich.console import Console
|
|
|
|
from bandsaunter.browse import Browser, Player
|
|
from bandsaunter.callsign import CallsignBook
|
|
from bandsaunter.config import ScanConfig
|
|
from bandsaunter.morse import decode_morse
|
|
from bandsaunter.quality import assess
|
|
from bandsaunter.ranges import parse_range_list
|
|
from bandsaunter.scanner import Scanner
|
|
from bandsaunter.simulator import SimulatedDevice, default_transmitters
|
|
|
|
from morse_gen import morse_audio
|
|
|
|
FS = 16000
|
|
|
|
|
|
class StubBook(CallsignBook):
|
|
"""Answers every lookup from itself, so no test goes to the network."""
|
|
|
|
def __init__(self, tmp, **kw):
|
|
self.asked: list[str] = []
|
|
super().__init__(cache=Path(tmp) / "calls.json", **kw)
|
|
|
|
def _request(self, url, call):
|
|
self.asked.append(call)
|
|
return {"status": "VALID", "current": {"callsign": call},
|
|
"name": "Newington Radio Club",
|
|
"address": {"line2": "Newington, CT"},
|
|
"location": {"latitude": "41.71", "longitude": "-72.72",
|
|
"gridsquare": "FN31pr"}}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# The scan
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _scanner(tmp_path, mhz: str, **over) -> Scanner:
|
|
cfg = ScanConfig(ranges=parse_range_list(mhz),
|
|
output_dir=str(tmp_path), transcribe=False,
|
|
record_seconds=over.pop("record_seconds", 10),
|
|
hang_seconds=1.5,
|
|
max_runtime_seconds=over.pop("seconds", 45),
|
|
**over)
|
|
scanner = Scanner(cfg, device=SimulatedDevice(realtime=False).open())
|
|
scanner.prepare()
|
|
scanner.callsigns = StubBook(tmp_path)
|
|
return scanner
|
|
|
|
|
|
def test_the_book_and_the_map_exist_without_a_speech_recogniser(tmp_path):
|
|
"""They used to be built inside the transcription branch.
|
|
|
|
Callsigns arrive in Morse and in packets as well as in speech, neither of
|
|
which involves a recogniser, so a machine with none installed found none
|
|
of them -- which is most of the machines this runs on.
|
|
"""
|
|
scanner = _scanner(tmp_path, "147.0M-147.1M")
|
|
assert scanner.transcriber is None
|
|
assert scanner.callsigns is not None
|
|
assert scanner.kml is not None
|
|
|
|
|
|
def test_a_repeater_identifying_in_morse_reaches_the_map(tmp_path):
|
|
scanner = _scanner(tmp_path, "147.0M-147.1M", seconds=45)
|
|
scanner.run()
|
|
assert scanner.hits, "the ident was not recorded at all"
|
|
assert any(h.morse_text for h in scanner.hits), \
|
|
"recorded it and read nothing out of it"
|
|
assert "K1AA" in scanner.heard
|
|
assert "K1AA" in scanner.kml.contacts
|
|
contact = scanner.kml.contacts["K1AA"]
|
|
assert contact.name == "Newington Radio Club"
|
|
assert contact.located
|
|
|
|
|
|
def test_the_map_is_written_out(tmp_path):
|
|
scanner = _scanner(tmp_path, "147.0M-147.1M", seconds=45)
|
|
scanner.run()
|
|
written = (tmp_path / scanner.cfg.kml_file).read_text()
|
|
assert "K1AA" in written and "<Point>" in written
|
|
|
|
|
|
def test_a_beacon_is_identified_from_the_words_that_survived(tmp_path):
|
|
"""A continuous beacon is always caught partway through.
|
|
|
|
Every capture of one begins and ends in the middle of the message, so
|
|
what can be said about it is whatever lies between two word gaps -- which
|
|
means the capture has to be longer than one repeat before any word is
|
|
certain to be bounded on both sides. This beacon repeats every eight
|
|
seconds; a ten-second capture of it is identifiable only by luck, and a
|
|
twenty-second one always.
|
|
"""
|
|
scanner = _scanner(tmp_path, "144.05M-144.15M", seconds=50,
|
|
record_seconds=20)
|
|
scanner.run()
|
|
assert "W1AW" in scanner.heard
|
|
assert set(scanner.heard) == {"W1AW"}, \
|
|
f"invented a station: {sorted(scanner.heard)}"
|
|
|
|
|
|
def test_the_hit_keeps_both_the_text_and_the_part_it_can_be_identified_from(
|
|
tmp_path):
|
|
scanner = _scanner(tmp_path, "144.05M-144.15M", seconds=30,
|
|
record_seconds=20)
|
|
scanner.run()
|
|
cw = [h for h in scanner.hits if h.morse_text]
|
|
assert cw
|
|
for hit in cw:
|
|
assert hit.morse_complete in hit.morse_text or not hit.morse_complete
|
|
|
|
|
|
def test_the_sidecar_carries_them(tmp_path):
|
|
scanner = _scanner(tmp_path, "147.0M-147.1M", seconds=45)
|
|
scanner.run()
|
|
sidecars = [json.loads(p.read_text()) for p in tmp_path.glob("*.json")]
|
|
hits = [doc.get("hit", doc) for doc in sidecars]
|
|
assert any(h.get("morse_text") for h in hits)
|
|
assert any(h.get("morse_complete") for h in hits)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Every capture is offered to the decoder, not only the ones that looked keyed
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_morse_under_a_label_that_is_not_cw_is_still_decoded(tmp_path):
|
|
"""A burst of CW is a second or two of a capture the classifier named
|
|
after whatever filled the rest of it, or after nothing at all."""
|
|
seen = {}
|
|
scanner = _scanner(tmp_path, "147.0M-147.1M", seconds=45)
|
|
real = scanner._decode_cw
|
|
|
|
def watch(iq, demod):
|
|
got = real(iq, demod)
|
|
seen.setdefault("calls", 0)
|
|
seen["calls"] += 1
|
|
return got
|
|
|
|
scanner._decode_cw = watch
|
|
scanner.run()
|
|
assert seen.get("calls"), "nothing was offered to the CW decoder"
|
|
|
|
|
|
def _on_the_air(mode: str, seconds: float = 3.0, freq: float = 147.06e6):
|
|
"""A real signal, classified and demodulated the way the scanner does.
|
|
|
|
Not a stand-in feature object: what `assess` does with a Morse decode
|
|
depends on measurements taken off the signal, and a hand-written set of
|
|
them would only ever prove that the test agreed with itself.
|
|
"""
|
|
from bandsaunter.classify import classify
|
|
from bandsaunter.demod import make_demodulator
|
|
from bandsaunter.simulator import VirtualTransmitter
|
|
|
|
fs = 240_000
|
|
voice = mode == "nfm"
|
|
tx = VirtualTransmitter(freq, mode, 0.5, 12_500 if voice else 500,
|
|
f"test {mode}", message="DE W1AW", wpm=20)
|
|
iq = tx.generate(0.0, int(fs * seconds), fs)
|
|
# A receiver always has some. Without it the key-up stretches of a CW
|
|
# signal are exactly zero, which is not a thing any aerial produces.
|
|
rng = np.random.default_rng(3)
|
|
iq = iq + (0.004 * (rng.standard_normal(iq.size)
|
|
+ 1j * rng.standard_normal(iq.size))).astype("complex64")
|
|
cls = classify(iq, fs, freq_hz=freq)
|
|
demod = make_demodulator("nfm" if voice else "cw", fs,
|
|
12_500.0 if voice else 800.0, FS)
|
|
return cls, demod.process(iq), demod.audio_rate
|
|
|
|
|
|
def test_speech_is_not_relabelled_by_a_morse_decode():
|
|
"""Both can be true at once, and the conversation is what was recorded.
|
|
|
|
Every capture is offered to the CW decoder now, not only the ones that
|
|
looked keyed, so a decode can land on a capture full of speech. The text
|
|
is recorded either way; the label follows the speech.
|
|
"""
|
|
cls, audio, rate = _on_the_air("nfm")
|
|
spoken = assess(cls, audio, rate, morse=None)
|
|
if spoken.category != "voice":
|
|
pytest.skip("the speech detector did not hear the synthetic talker")
|
|
|
|
_, keyed, keyed_rate = _on_the_air("cw")
|
|
morse = decode_morse(keyed, keyed_rate)
|
|
assert morse.is_morse, "the fixture did not produce a Morse decode"
|
|
assert assess(cls, audio, rate, morse=morse).category == "voice"
|
|
|
|
|
|
def test_a_keyed_carrier_is_still_labelled_cw():
|
|
"""The families that were always decoded keep their behaviour exactly."""
|
|
cls, audio, rate = _on_the_air("cw")
|
|
morse = decode_morse(audio, rate)
|
|
assert morse.is_morse
|
|
assert assess(cls, audio, rate, morse=morse).category == "cw"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Reading it back
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _capture(directory: Path, meta: dict) -> Path:
|
|
stem = "0147.060000MHz--2026-08-29_10_00_00-cw"
|
|
with wave.open(str(directory / f"{stem}.wav"), "wb") as w:
|
|
w.setnchannels(1)
|
|
w.setsampwidth(2)
|
|
w.setframerate(FS)
|
|
w.writeframes(b"\0\0" * FS)
|
|
(directory / f"{stem}.json").write_text(json.dumps({"hit": meta}))
|
|
return directory / f"{stem}.wav"
|
|
|
|
|
|
def _browser(directory) -> Browser:
|
|
console = Console(width=100, height=30, force_terminal=True)
|
|
return Browser(directory, console=console, player=Player([]),
|
|
book=StubBook(directory))
|
|
|
|
|
|
def frame(browser) -> str:
|
|
with browser.console.capture() as cap:
|
|
browser.console.print(browser.render())
|
|
return cap.get()
|
|
|
|
|
|
@pytest.fixture
|
|
def keyed(tmp_path):
|
|
_capture(tmp_path, {"frequency": 147.06e6, "category": "cw",
|
|
"classification": "CW / Morse at 20 WPM",
|
|
"morse_text": "E DE K1AA",
|
|
"morse_complete": "DE K1AA",
|
|
"morse_wpm": 20.0, "confidence": 0.9})
|
|
return tmp_path
|
|
|
|
|
|
def test_the_morse_gets_the_top_of_the_screen(keyed):
|
|
shown = frame(_browser(keyed))
|
|
assert "Morse" in shown and "DE K1AA" in shown
|
|
|
|
|
|
def test_the_callsign_is_listed_under_it(keyed):
|
|
browser = _browser(keyed)
|
|
browser.book.get_all(browser.current.callsigns)
|
|
browser.book.wait(5.0)
|
|
shown = frame(browser)
|
|
assert "K1AA" in shown and "Newington Radio Club" in shown
|
|
|
|
|
|
def test_a_cw_capture_can_be_searched_for_by_what_it_keyed(keyed):
|
|
browser = _browser(keyed)
|
|
browser.query = "k1aa"
|
|
browser.apply()
|
|
assert len(browser.view) == 1
|
|
|
|
|
|
def test_the_reader_opens_on_it(keyed):
|
|
browser = _browser(keyed)
|
|
assert browser.handle("t")
|
|
assert browser.reading
|
|
assert "DE K1AA" in frame(browser)
|
|
|
|
|
|
def test_only_the_part_it_can_be_identified_from_is_searched(tmp_path):
|
|
"""The sidecar keeps both, and the callsign comes out of the safe one."""
|
|
_capture(tmp_path, {"frequency": 147.06e6, "category": "cw",
|
|
"morse_text": "K1A", "morse_complete": "",
|
|
"classification": "CW / Morse at 20 WPM"})
|
|
browser = _browser(tmp_path)
|
|
assert browser.current.morse == "K1A"
|
|
assert browser.current.callsigns == []
|
|
|
|
|
|
def test_an_older_recording_falls_back_to_the_whole_text(tmp_path):
|
|
"""Sidecars written before this distinction existed carry only the text."""
|
|
_capture(tmp_path, {"frequency": 147.06e6, "category": "cw",
|
|
"morse_text": "VVV DE W1AW",
|
|
"classification": "CW / Morse at 20 WPM"})
|
|
assert _browser(tmp_path).current.callsigns == ["W1AW"]
|
|
|
|
|
|
def test_a_hex_dump_is_not_searched_for_callsigns(tmp_path):
|
|
"""Enough two-character groups in a row join into something shaped like
|
|
a callsign that nobody transmitted."""
|
|
_capture(tmp_path, {"frequency": 147.06e6, "category": "digital",
|
|
"data_messages": ["4A 3F 1B 22 9C 04"],
|
|
"classification": "OOK data"})
|
|
assert _browser(tmp_path).current.callsigns == []
|
|
|
|
|
|
def test_a_packet_header_is(tmp_path):
|
|
_capture(tmp_path, {"frequency": 144.39e6, "category": "digital",
|
|
"data_messages": ["W1AW-1>APRS,TCPIP*:=4123.45N/"
|
|
"07234.56W-"],
|
|
"classification": "AX.25 / APRS"})
|
|
assert _browser(tmp_path).current.callsigns == ["W1AW"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# The demo band is invented; the callsigns in it are not
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_the_simulator_identifies_itself_the_way_a_repeater_does():
|
|
idents = [t for t in default_transmitters()
|
|
if t.mode == "cw" and t.period_seconds]
|
|
assert idents, "nothing in the demo band sends a short CW ident"
|
|
assert idents[0].on_seconds < 8.0
|
|
|
|
|
|
def test_a_simulated_run_neither_looks_up_nor_maps_anything(monkeypatch,
|
|
tmp_path):
|
|
"""W1AW is the ARRL's own station, and the demo band is made up.
|
|
|
|
Looking it up would put a licence nobody heard on the same map a real
|
|
scan writes.
|
|
"""
|
|
from bandsaunter import cli
|
|
seen = {}
|
|
|
|
class Stop(Exception):
|
|
pass
|
|
|
|
def fake_scanner(cfg, **kw):
|
|
seen["cfg"] = cfg
|
|
raise Stop
|
|
|
|
monkeypatch.setattr(cli, "Scanner", fake_scanner)
|
|
monkeypatch.setattr(cli, "console", Console(file=open("/dev/null", "w")))
|
|
with pytest.raises(Stop):
|
|
cli.main(["scan", "--simulate", "--no-config", "-r", "144M-148M",
|
|
"-o", str(tmp_path)])
|
|
assert seen["cfg"].callsign_lookup is False
|
|
assert seen["cfg"].kml_file == ""
|