"""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, 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=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 "" 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. """ scanner = _scanner(tmp_path, "144.05M-144.15M", seconds=30) 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) 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 == ""