"""The FT8 section: the code, the radio, the options and the front ends. The important thing about this file is where its truth comes from. An encoder tested against its own decoder agrees with it about anything they are both wrong about -- this project learned that expensively on 433 MHz -- so the decoding here is checked against signals that came off the air and were decoded by somebody else's program, not against anything written here. Those recordings live outside the repository, and the tests that use them skip when they are absent. What does not skip is REAL: whole codewords lifted off the air, which pin the checksum, the parity and every field of the message format to reality in forty-four characters apiece. """ import math import time import wave from pathlib import Path import numpy as np import pytest from bandsaunter import ft8, ft8code as code, ft8log, ft8sim, ft8wave from bandsaunter import ft8tables as tables REFERENCE = Path("/mnt/global/bandsaunter/ft8ref") needs_recordings = pytest.mark.skipif( not REFERENCE.is_dir(), reason="the off-air recordings are not here") # Whole codewords, off the air, with what they say. A hundred and # seventy-four bits each: seventy-seven of message, fourteen of checksum and # eighty-three of parity, so one of these exercises the lot. REAL = [ ("B79A7EA67655CA92DB4D342FDD55B071779A631CFB4C", "PA3EPP SP8NFO KN09", "standard"), ("0000002046E853111D48153FD733DA62C110A5E053FC", "CQ F4FSY JN25", "standard"), ("E20AFC7590F9DA9FA7C9EEEA8A1A96D06F6855C82B58", "VK4BLE OH1EDK -20", "standard"), ("70D84EAC55E5FB9FA70CD7148B9180DD1CD36A7EFC6C", "ET3RFG/R IN3ADG -23", "standard"), ("1E6191F629601212F10D6992973A9B5AEA0596877498", "2M0OGG RA6ABO KN96", "standard"), ("70AFEC9338AD521FA50DE077F5C94CAD362B2CC3A364", "ES5GI DD3SF 73", "standard"), ("00000024519BF311248987ED8D2C4AA596CE262E1E68", "CQ IK4LZH JN54", "standard"), ("4B9ACCF4519BF31FAA4F0C50A7B6CEA7F9C51BED8E80", "9A9TT IK4LZH -10", "standard"), ("0B13C7046826781FA50F539DFE6D9FAF63F7C21EC268", "R2EA IZ4OUL 73", "standard"), ("CBEC04F45AC9D49FA509770E71FEDE65EE2EFE66E8A8", "SA5QED IQ5PJ 73", "standard"), ] def bits_of_hex(text: str) -> np.ndarray: raw = np.unpackbits(np.frombuffer(bytes.fromhex(text), dtype=np.uint8)) return raw[:code.BITS].astype(np.uint8) # --------------------------------------------------------------------------- # The two tables the code is defined by # --------------------------------------------------------------------------- def test_the_generator_and_the_parity_matrix_describe_the_same_code(): """The two published tables are not derivable from one another -- the generator's parity half runs to fifty-odd bits a row and the sparse one to six or seven -- so that they agree is checked rather than assumed. If they disagreed, every codeword this builds would fail every check and nothing would ever decode, which is a long way to find out.""" rng = np.random.default_rng(11) for _ in range(50): message = rng.integers(0, 2, code.PAYLOAD).astype(np.uint8) assert code.parity_holds(code.encode(message)) def test_every_codeword_bit_is_checked_three_times(): """A regular column weight, which is what the message passing assumes when it gathers three slots per bit instead of scattering.""" assert code.SLOTS_OF_BIT.shape == (code.BITS, 3) assert sorted({len(c) for c in tables.CHECKS}) == [6, 7] def test_a_broken_codeword_fails_its_parity(): rng = np.random.default_rng(3) word = code.encode(rng.integers(0, 2, code.PAYLOAD).astype(np.uint8)) for flip in (0, 40, 90, 173): bad = word.copy() bad[flip] ^= 1 assert not code.parity_holds(bad) # --------------------------------------------------------------------------- # Reality # --------------------------------------------------------------------------- @pytest.mark.parametrize("packed,text,kind", REAL) def test_a_codeword_off_the_air_says_what_it_said(packed, text, kind): """The whole chain against something nobody here made up.""" word = bits_of_hex(packed) assert code.parity_holds(word), "the parity of a real transmission" assert code.crc_holds(word[:code.PAYLOAD]), "the checksum of a real one" message = code.unpack(word[:code.MESSAGE_BITS]) assert message.text == text assert message.kind == kind def test_a_real_codeword_is_repaired_from_a_damaged_copy(): """What the error correction is for, on a real transmission rather than a made-up one. Ten of the hundred and seventy-four bits, each of them *confidently* wrong, which is the hard case: a bit the detector is unsure about costs the decoding far less than one it is certain about and mistaken. Real errors arrive uncertain, which is why the same code reads signals off the air with far more than ten bits astray.""" word = bits_of_hex(REAL[0][0]) rng = np.random.default_rng(5) belief = np.where(word == 1, -4.0, 4.0).astype(np.float32) for flip in rng.choice(code.BITS, 10, replace=False): belief[flip] = -belief[flip] got = code.repair(belief) assert got is not None and (got == word).all() assert code.unpack(got[:code.MESSAGE_BITS]).text == REAL[0][1] def test_noise_does_not_decode_into_a_message(): """The checksum's job. The error correction will converge on *a* codeword given enough noise, and the fourteen bits are what stop that becoming a line in somebody's log.""" rng = np.random.default_rng(17) lied = 0 for _ in range(200): belief = rng.standard_normal(code.BITS).astype(np.float32) * 2.0 got = code.repair(belief) if got is not None and code.crc_holds(got[:code.PAYLOAD]): lied += 1 # One in sixteen thousand gets through by arithmetic; two hundred tries # should see none, and seeing several would mean the CRC is not being # checked at all. assert lied == 0, f"{lied} runs of noise passed the checksum" # --------------------------------------------------------------------------- # The checksum and the coding # --------------------------------------------------------------------------- def test_the_checksum_covers_the_message_zero_extended(): """Eighty-two bits, not seventy-seven -- which is not the same answer and is the sort of thing that decodes nothing at all.""" payload = np.zeros(code.MESSAGE_BITS, dtype=np.uint8) payload[3] = payload[70] = 1 block = code.with_crc(payload) assert block.size == code.PAYLOAD assert code.crc_holds(block) broken = block.copy() broken[10] ^= 1 assert not code.crc_holds(broken) def test_the_tones_carry_the_bits_and_the_sync_is_where_it_should_be(): rng = np.random.default_rng(2) word = code.encode(rng.integers(0, 2, code.PAYLOAD).astype(np.uint8)) tones = code.tones_of(word) assert tones.size == code.TONES assert (code.bits_of(tones) == word).all() for start in (0, 36, 72): assert list(tones[start:start + 7]) == list(tables.COSTAS) def test_a_tone_mistaken_for_its_neighbour_costs_one_bit(): """Which is the whole reason for the Gray mapping, and is worth a test because getting the map backwards still round-trips perfectly.""" for tone in range(7): a = code.UNGRAY[tone] b = code.UNGRAY[tone + 1] assert bin(int(a) ^ int(b)).count("1") == 1 # --------------------------------------------------------------------------- # Messages # --------------------------------------------------------------------------- @pytest.mark.parametrize("text", [ "CQ DL1UDO JO31", "CQ DX Z33Z KN11", "CQ JA OH1LWZ KP11", "VK4BLE OH8JK R-17", "JR5MJS OH8NW 73", "LZ1CWK DC8VA RR73", "G4CUS SP4FCA +10", "2M0OGG RA6ABO KN96", "YO7CGS A41ZZ -11", "R2EA IZ4OUL R-08", "W2XYZ K1ABC RRR", ]) def test_a_standard_message_survives_the_whole_chain(text): payload = code.pack(text) word = code.encode(code.with_crc(payload)) assert code.parity_holds(word) back = code.bits_of(code.tones_of(word)) assert (back == word).all() got = code.repair(np.where(word == 1, -4.0, 4.0).astype(np.float32)) assert got is not None and (got == word).all() assert code.unpack(got[:code.MESSAGE_BITS]).text == text def test_the_digit_of_a_callsign_is_found_rather_than_guessed_at(): """Z33Z has a digit in the second place that belongs to the prefix, so a packer that assumes the second character is the separating digit turns it into a hash and loses the callsign.""" assert code.is_standard_call("Z33Z") assert code.is_standard_call("K1ABC") assert code.is_standard_call("DL1UDO") assert code.is_standard_call("2M0OGG") assert not code.is_standard_call("ET3RFG/R") assert not code.is_standard_call("") def test_free_text_is_not_dressed_up_as_two_callsigns(): """A packer that falls back to hashing any word it does not recognise always succeeds, and would send HELLO WORLD as two hashed callsigns.""" got = code.unpack(code.pack("HELLO WORLD")) assert got.kind == "free text" assert got.text == "HELLO WORLD" def test_free_text_is_thirteen_characters_and_says_so_by_truncating(): got = code.unpack(code.pack("TNX FER QSO 73")) assert got.kind == "free text" assert len(got.text) <= 13 @pytest.mark.parametrize("text,grid,report,calling", [ ("CQ DL1UDO JO31", "JO31", "", True), ("VK4BLE OH8JK R-17", "", "R-17", False), ("JR5MJS OH8NW 73", "", "73", False), ("W2XYZ K1ABC RRR", "", "RRR", False), ]) def test_what_can_be_picked_out_of_a_message(text, grid, report, calling): got = code.unpack(code.pack(text)) assert got.grid == grid and got.report == report assert got.calling is calling def test_a_hashed_callsign_stays_unknown_until_it_has_been_heard_in_full(): """A wrong callsign in a log is worse than a missing one, so a hash that has not been heard spelled out stays <...> rather than guessed.""" book = code.CallBook() assert book.look_up(code.hash_of("GM4ABC", 22), 22) == "<...>" book.remember("GM4ABC") assert book.look_up(code.hash_of("GM4ABC", 22), 22) == "GM4ABC" assert book.look_up(code.hash_of("GM4ABD", 22), 22) == "<...>" def test_a_callsign_too_short_to_mean_anything_is_not_remembered(): book = code.CallBook() for rubbish in ("", "A", "CQ", "DE", "QRZ"): book.remember(rubbish) assert book.by_hash == {} # --------------------------------------------------------------------------- # The radio # --------------------------------------------------------------------------- def test_a_transmission_is_found_where_it_was_put(): """The decoder's own clock and dial, against a signal whose time and frequency are known to the sample.""" for hertz in (500.0, 1200.0, 2400.0): for offset in (0.2, 0.5, 1.4): audio = ft8wave.transmit("CQ DL1UDO JO31", hertz=hertz, offset=offset) got = ft8wave.listen_to(audio, 12000.0) assert len(got) == 1, (hertz, offset) assert got[0].text == "CQ DL1UDO JO31" assert abs(got[0].hertz - hertz) < 3.5 # Reported against the nominal start of sending, half a second # into the slot, which is what every FT8 program reports. assert abs(got[0].offset - (offset - 0.5)) < 0.12 def test_the_nominal_start_is_half_a_second_into_the_slot(): """A station whose clock is right reports as dt 0.0, not dt 0.5: the transmission is 12.64 seconds long in a slot of fifteen and everybody begins half a second in.""" audio = ft8wave.transmit("CQ DL1UDO JO31", offset=ft8wave.NOMINAL_START) got = ft8wave.listen_to(audio, 12000.0) assert got and abs(got[0].offset) < 0.1 def test_several_transmissions_in_one_slot_all_come_out(): """Which is the whole point of the mode: they overlap in time entirely and are told apart by frequency alone.""" said = [("CQ DL1UDO JO31", 500.0), ("VK4BLE OH8JK R-17", 1000.0), ("JR5MJS OH8NW 73", 1600.0), ("CQ DX Z33Z KN11", 2300.0)] audio = np.zeros(int(12000 * code.SLOT_S), dtype=np.float32) for text, hertz in said: audio += ft8wave.transmit(text, hertz=hertz, offset=0.5) got = {d.text for d in ft8wave.listen_to(audio, 12000.0)} assert got == {t for t, _hz in said} def test_a_slot_of_noise_decodes_to_nothing(): rng = np.random.default_rng(9) noise = rng.standard_normal(int(12000 * code.SLOT_S)).astype(np.float32) assert ft8wave.listen_to(noise, 12000.0) == [] def test_it_hears_a_signal_well_below_the_noise(): """The claim the mode is built on. Fifteen decibels under is a signal an operator hears nothing of at all.""" heard = 0 for seed in range(5): audio = ft8wave.transmit("CQ DL1UDO JO31", hertz=1200.0, offset=0.5, snr_db=-15.0, seed=seed) got = ft8wave.listen_to(audio, 12000.0) heard += any(d.text == "CQ DL1UDO JO31" for d in got) assert heard == 5 def test_the_signal_report_tracks_the_signal(): """Not to a decibel -- it is an estimate -- but a stronger signal has to read stronger, or the number is worse than not printing one.""" readings = [] for want in (-15.0, -5.0, 5.0): audio = ft8wave.transmit("CQ DL1UDO JO31", hertz=1200.0, offset=0.5, snr_db=want, seed=1) got = ft8wave.listen_to(audio, 12000.0) assert got readings.append(got[0].snr_db) assert readings[0] < readings[1] < readings[2] assert all(abs(r - w) < 8.0 for r, w in zip(readings, (-15.0, -5.0, 5.0))) def test_the_waterfall_is_half_a_symbol_and_half_a_tone(): """A transmission a quarter of a symbol late, or a quarter of a tone off, still has to be caught.""" fall = ft8wave.waterfall(np.zeros(12000 * 15, dtype=np.float32), 12000.0) assert fall.hz_per_bin == pytest.approx(code.SYMBOL_HZ / ft8wave.FREQ_STEPS) assert fall.seconds_per_step == pytest.approx(code.SYMBOL_S / ft8wave.TIME_STEPS) def test_audio_too_short_to_hold_a_symbol_is_refused_rather_than_guessed(): with pytest.raises(ValueError): ft8wave.waterfall(np.zeros(10, dtype=np.float32), 12000.0) # --------------------------------------------------------------------------- # Against the air # --------------------------------------------------------------------------- def _recording(name): with wave.open(str(REFERENCE / f"{name}.wav")) as handle: raw = handle.readframes(handle.getnframes()) rate = float(handle.getframerate()) audio = np.frombuffer(raw, dtype=np.int16).astype(np.float32) / 32768.0 want = {} for line in (REFERENCE / f"{name}.txt").read_text().splitlines(): parts = line.split(None, 4) if len(parts) >= 5: want[" ".join(parts[4].lstrip("~ ").split())] = ( float(parts[1]), float(parts[2]), float(parts[3])) return audio, rate, want @needs_recordings @pytest.mark.parametrize("name,least", [ ("191111_110145", 2), ("websdr_test3", 7), ("20m_busy_test_02", 14), ]) def test_it_decodes_signals_that_came_off_the_air(name, least): """The only test here whose answers were not written by this program.""" audio, rate, want = _recording(name) got = {d.text: d for d in ft8wave.listen_to(audio, rate)} matched = [w for w in want if any(w.startswith(t) for t in got)] assert len(matched) >= least, f"decoded {len(matched)} of {len(want)}" # And nothing it decoded may be something nobody else heard: a false # line in a log is worse than a missing one. unknown = [t for t in got if not any(w.startswith(t) for w in want)] assert len(unknown) <= 1, unknown @needs_recordings def test_the_time_and_frequency_agree_with_what_else_heard_them(): audio, rate, want = _recording("websdr_test3") errors = [] for d in ft8wave.listen_to(audio, rate): for text, (snr, dt, hz) in want.items(): if text.startswith(d.text): errors.append((d.offset - dt, d.hertz - hz)) assert len(errors) >= 6 assert max(abs(dt) for dt, _hz in errors) < 0.2 assert max(abs(hz) for _dt, hz in errors) < 4.0 # --------------------------------------------------------------------------- # Slots and grids # --------------------------------------------------------------------------- def test_slots_are_quarter_minutes_of_utc(): """Not of the local clock and not of when this program started: the protocol is built on every station on earth agreeing which one it is.""" assert ft8.slot_start(1_700_000_007.0) == 1_700_000_000.0 - \ (1_700_000_000.0 % 15) for when in (0.0, 1_700_000_000.0, 1_700_000_014.9): assert ft8.slot_start(when) % ft8.SLOT == 0 assert 0 <= when - ft8.slot_start(when) < ft8.SLOT assert ft8.next_slot(when) == ft8.slot_start(when) + ft8.SLOT @pytest.mark.parametrize("grid,lat,lon", [ ("IO91", 51.5, -1.0), ("FN31", 41.5, -73.0), ("JN54", 44.5, 11.0), ("RE78", -41.5, 175.0), ]) def test_a_grid_square_is_where_it_says(grid, lat, lon): got = ft8.grid_at(grid) assert got == pytest.approx((lat, lon), abs=0.01) def test_a_grid_that_is_not_one_is_refused(): for rubbish in ("", "IO", "9191", "ZZ99", "IO9", "ABCD"): assert ft8.grid_at(rubbish) is None def test_the_distance_between_two_squares_is_the_great_circle_one(): km, bearing = ft8.grid_away("IO91", "FN31") assert km == pytest.approx(5390, abs=60) assert bearing == pytest.approx(288, abs=3) assert ft8.grid_away("IO91", "IO91")[0] < 1.0 assert ft8.grid_away("IO91", "") is None # --------------------------------------------------------------------------- # The options # --------------------------------------------------------------------------- def test_every_option_names_a_field_that_exists(): fields = set(ft8.Ft8Options().__dict__) for option in ft8.OPTIONS: assert option.key in fields, f"{option.key} is not an option" def test_every_field_is_an_option_somebody_can_reach(): assert set(ft8.Ft8Options().__dict__) == {o.key for o in ft8.OPTIONS} def test_every_option_is_in_a_group_and_says_what_it_does(): for option in ft8.OPTIONS: assert option.group in ft8.OPTION_GROUPS assert option.help and option.detail and option.flags if option.kind == "bool": assert option.off_flags, f"{option.key} cannot be turned off" for group in ft8.OPTION_GROUPS: assert ft8.in_group(group) def test_every_option_flag_is_one_the_parser_accepts(): from bandsaunter.cli import build_parser known = set() for action in build_parser()._subparsers._group_actions[0].choices[ "ft8"]._actions: known.update(action.option_strings) for option in ft8.OPTIONS: for flag in option.flags + option.off_flags: assert flag in known, f"{flag} is in the table and not the parser" def test_every_command_line_option_is_reachable_from_the_menu(): """Asked for explicitly, and worth a test rather than a promise: the menu builds itself from the groups, so an option in no group would be settable from the command line and invisible in the menu.""" from bandsaunter import tui text = Path(tui.__file__).read_text() assert "def ft8_menu(" in text assert "ft8_menu(console, cfg)" in text, "not reachable from the front page" reachable = set() for group in ft8.OPTION_GROUPS: reachable |= {o.key for o in ft8.in_group(group)} assert reachable == {o.key for o in ft8.OPTIONS} def test_the_options_survive_being_saved_and_read_back(tmp_path): options = ft8.Ft8Options(band="20m", frequency=14_074_000.0, grid="IO91", units="imperial", slots=12) ft8.save_options(options, tmp_path) assert ft8.load_options(tmp_path) == options def test_a_settings_file_with_a_mistake_in_it_costs_the_defaults(tmp_path): (tmp_path / "ft8.yaml").write_text("band: [not a band\n") assert ft8.load_options(tmp_path) == ft8.Ft8Options() @pytest.mark.parametrize("options,broken", [ (ft8.Ft8Options(rate=48_000.0), "sample rate"), (ft8.Ft8Options(seconds=-1.0), "negative"), (ft8.Ft8Options(hold=0.0), "display"), (ft8.Ft8Options(frequency=10.0), "frequency"), (ft8.Ft8Options(lowest=2000.0, highest=500.0), "above the bottom"), (ft8.Ft8Options(lowest=1000.0, highest=1050.0), "narrower"), (ft8.Ft8Options(most=0), "candidate"), (ft8.Ft8Options(rounds=0), "error correction"), (ft8.Ft8Options(grid="nowhere"), "grid square"), ]) def test_a_setting_that_cannot_work_is_refused(options, broken): assert any(broken in err for err in options.validate()) def test_the_defaults_are_all_workable(): assert ft8.Ft8Options().validate() == [] def test_choosing_a_band_sets_the_frequency_with_it(): options = ft8.defaults() ft8.use_band(options, "20m") assert options.frequency == 14_074_000.0 and options.band == "20m" ft8.use_band(options, "nonsense") assert options.frequency == 144_174_000.0 def test_the_band_list_says_which_ones_a_plain_receiver_can_reach(): """Almost all the activity is on shortwave, which this kind of receiver cannot hear without help, and finding that out by listening to silence for ten minutes is the wrong way to learn it.""" assert "upconverter" in ft8.band_text(ft8.Ft8Options(band="20m")) assert "upconverter" not in ft8.band_text(ft8.Ft8Options(band="2m")) assert ft8.defaults().band == "2m" def test_the_command_line_beats_the_saved_settings(tmp_path, monkeypatch): from bandsaunter.cli import build_parser ft8.save_options(ft8.Ft8Options(band="2m", grid="AA00"), tmp_path) monkeypatch.setattr(ft8, "options_path", lambda directory=None: tmp_path / "ft8.yaml") args = build_parser().parse_args(["ft8", "--band", "20m", "--grid", "IO91"]) assert args.band == "20m" and args.grid == "IO91" # --------------------------------------------------------------------------- # End to end # --------------------------------------------------------------------------- class _Clock: """A clock the test winds, so a run of slots takes no real time.""" def __init__(self, at=1_700_000_000.0): self.at = at def now(self): return self.at def sleep(self, seconds): self.at += seconds def _run(slots=2, **kw): clock = _Clock() band = ft8sim.SimulatedBand(rate=240_000.0, realtime=True, stations=12, clock=clock.now, sleep=clock.sleep) options = ft8.defaults() options.slots = slots for key, value in kw.items(): setattr(options, key, value) heard = ft8.Heard() heard.started = clock.now() total = ft8.pump(band, options, heard, None, clock.now(), clock=clock.now) return heard, total def test_a_made_up_band_comes_out_the_other_end(): """The whole path: samples, sideband, slot boundaries, sync, decoding, and into the book of stations.""" heard, total = _run(slots=2) assert heard.slots == 2 assert total >= 8, f"only {total} decodes in two slots" assert len(heard.band.stations) >= 6 for station in heard.band.all(): assert station.decodes >= 1 assert station.call and station.call != "<...>" def test_the_slot_boundaries_are_found_rather_than_assumed(): """A receiver that dated its samples a block out would slice every slot late and cut the first half-second -- three symbols and part of the opening sync -- off every transmission on the band.""" heard, _total = _run(slots=2) assert heard.band.decodes # Everything on this band starts within half a second of the nominal # start, so everything should be reported within about that of zero. worst = max(abs(d.offset) for d in heard.band.decodes) assert worst < 1.0, f"the worst decode was {worst:.2f} s out" def test_callsigns_only_leaves_out_the_free_text(): heard, _ = _run(slots=1, calls_only=True) assert all(d.kind in ("standard", "non-standard") for d in heard.band.decodes) def test_the_search_can_be_narrowed_and_the_narrowing_is_obeyed(): heard, _ = _run(slots=1, lowest=800.0, highest=1600.0) assert heard.band.decodes assert all(800.0 <= d.hertz <= 1600.0 for d in heard.band.decodes) def test_a_station_that_calls_cq_is_counted_as_calling(): heard, _ = _run(slots=2) calling = [s for s in heard.band.all() if s.calling] assert calling, "nobody called CQ on a band that is mostly CQ" assert all(s.grid for s in calling), "a CQ says where it is" # --------------------------------------------------------------------------- # Writing it down # --------------------------------------------------------------------------- def _decode(text="CQ DL1UDO JO31", **kw): fields = dict(text=text, kind="standard", calls=("DL1UDO",), grid="JO31", report="", calling=True, hertz=1200.0, offset=0.2, snr_db=-7.0, at=1_700_000_000.0) fields.update(kw) return ft8wave.Decode(**fields) def test_a_log_is_written_and_read_back(tmp_path): log = ft8log.open_log(tmp_path, 1_700_000_000.0, 14_074_000.0, "20m") assert log is not None for i in range(3): log.append(_decode(hertz=1000.0 + i * 100)) log.close() rows = ft8log.read_log(log.path) assert len(rows) == 3 assert rows[0]["text"] == "CQ DL1UDO JO31" assert rows[1]["hertz"] == 1100.0 def test_a_half_written_log_line_is_skipped_rather_than_raising(tmp_path): path = tmp_path / "ft8-part.txt" path.write_text("# header\n2023-11-14 22:13:20 -7 0.2 1200.0 ~ OK\n" "2023-11-14 22:13:35 -7 0.2\n") rows = ft8log.read_log(path) assert len(rows) == 1 and rows[0]["text"] == "OK" def test_a_csv_has_a_row_for_every_decode(tmp_path): where = ft8log.write_csv(tmp_path / "x.csv", [_decode(), _decode("W2XYZ K1ABC RRR")]) assert where is not None body = where.read_text().splitlines() assert len(body) == 3 and body[0].startswith("utc,") def test_an_adif_says_these_were_heard_rather_than_worked(tmp_path): """Nothing here transmits, so nothing here is a contact. An ADIF that let a logging program treat these as worked would put claims into somebody's log that they cannot make.""" where = ft8log.write_adif(tmp_path / "x.adi", [_decode(), _decode("W2XYZ K1ABC RRR", calls=("W2XYZ", "K1ABC"))], band="20m", frequency=14_074_000.0, my_grid="IO91") body = where.read_text() assert "heard only, not worked" in body assert "not contacts" in body assert "DL1UDO" in body assert "FT8" in body assert "IO91" in body def test_an_adif_keeps_one_record_a_station_rather_than_one_a_decode(tmp_path): many = [_decode(snr_db=s) for s in (-20.0, -3.0, -11.0)] where = ft8log.write_adif(tmp_path / "y.adi", many) body = where.read_text() assert body.count("") == 1 # And the one kept is the best report, that being the one worth passing on. assert "-3" in body # --------------------------------------------------------------------------- # The display # --------------------------------------------------------------------------- def test_the_display_says_what_it_is_waiting_for_before_anything_arrives(): from rich.console import Console from bandsaunter.ui import Ft8Display console = Console(width=160, force_terminal=False) display = Ft8Display(console, band="20m — 14.074 MHz") display.update(ft8.Heard()) with console.capture() as caught: console.print(display.render()) shown = caught.get() assert "nothing decoded yet" in shown # And the one thing that silently stops FT8 working is worth saying. assert "clock" in shown def test_the_display_shows_the_last_slot_and_the_running_total(): from rich.console import Console from bandsaunter.ui import Ft8Display console = Console(width=110, force_terminal=False) display = Ft8Display(console, grid="IO91", band="20m — 14.074 MHz") heard = ft8.Heard() for text in ("CQ DL1UDO JO31", "W2XYZ K1ABC RRR"): found = _decode(text, calls=tuple(text.split()[1:2] or ["DL1UDO"]), grid="JO31" if "CQ" in text else "") heard.band.add(found) heard.decodes += 1 heard.latest = heard.band.decodes heard.slots = 4 display.update(heard, 1_700_000_000.0) with console.capture() as caught: console.print(display.render(now=1_700_000_005.0)) shown = caught.get() assert "last slot" in shown and "heard so far" in shown assert "DL1UDO" in shown and "JO31" in shown assert "4 slots" in shown def test_the_report_says_nothing_rather_than_an_empty_table(): from rich.console import Console console = Console(width=100, force_terminal=False) band = ft8.Band() band.slots = 12 with console.capture() as caught: ft8.report(console, band, ft8.defaults()) shown = caught.get() assert "nothing decoded" in shown assert "clock" in shown, "the usual cause is worth naming" def test_the_report_names_the_furthest_station_when_it_can(): from rich.console import Console console = Console(width=120, force_terminal=False) band = ft8.Band() band.slots = 4 band.add(_decode("CQ VK4BLE QG62", calls=("VK4BLE",), grid="QG62")) band.add(_decode("CQ DL1UDO JO31", calls=("DL1UDO",), grid="JO31")) options = ft8.defaults() options.grid = "IO91" with console.capture() as caught: ft8.report(console, band, options) shown = caught.get() assert "furthest heard" in shown and "VK4BLE" in shown def test_the_report_leaves_distances_out_when_it_has_nowhere_to_measure_from(): from rich.console import Console console = Console(width=120, force_terminal=False) band = ft8.Band() band.add(_decode("CQ DL1UDO JO31", calls=("DL1UDO",), grid="JO31")) with console.capture() as caught: ft8.report(console, band, ft8.defaults()) shown = caught.get() assert "DL1UDO" in shown and "away" not in shown # --------------------------------------------------------------------------- # Opening a receiver # --------------------------------------------------------------------------- def test_every_name_these_modules_import_actually_exists(): """Imports inside a function are not checked until that function runs. Which is how `open_rtlsdr` -- a name nothing in this program has ever exported -- sat in the one line that opens the receiver, past every test, until somebody chose Listen and was told the section was broken. The tests reached that line only through the simulator, which takes the other branch. """ import ast import importlib missing = [] for name in ("ft8", "ft8code", "ft8wave", "ft8log", "ft8sim", "ft8tables"): module = importlib.import_module(f"bandsaunter.{name}") tree = ast.parse(Path(module.__file__).read_text()) for node in ast.walk(tree): if not isinstance(node, ast.ImportFrom) or node.level != 1: continue if not node.module: continue target = importlib.import_module(f"bandsaunter.{node.module}") for alias in node.names: if not hasattr(target, alias.name): missing.append(f"{name}: {node.module}.{alias.name}") assert not missing, missing def test_the_made_up_band_needs_no_receiver_and_says_it_is_made_up(): from rich.console import Console console = Console(width=90, force_terminal=False) options = ft8.defaults() options.simulate = True with console.capture() as caught: device = ft8.open_device(console, options) assert isinstance(device, ft8sim.SimulatedBand) assert "not there" in caught.get(), "a simulation must say so" device.close() def test_a_receiver_that_cannot_be_opened_is_reported_rather_than_raising(): """A menu has to survive it, and the person has to be told which of the two things went wrong: the receiver, or the listening.""" from rich.console import Console console = Console(width=90, force_terminal=False) options = ft8.defaults() options.device = 99 # no such dongle with console.capture() as caught: assert ft8.open_device(console, options) is None assert "cannot open the receiver" in caught.get() def test_nothing_claims_to_be_listening_before_there_is_a_receiver(tmp_path): """Announcing the frequency and then failing to open a dongle reads as though the listening went wrong, when what went wrong happened before any of it started.""" from rich.console import Console console = Console(width=90, force_terminal=False) options = ft8.defaults() options.device = 99 options.log = False with console.capture() as caught: assert ft8.listen(console, options, str(tmp_path)) == 0 shown = caught.get() assert "cannot open the receiver" in shown assert "listening on" not in shown # --------------------------------------------------------------------------- # Who each station is # --------------------------------------------------------------------------- def _register(tmp_path, entries=()): """A licence register that knows what it is told and asks nobody.""" from bandsaunter.callsign import CACHE_VERSION, Callsign, CallsignBook book = CallsignBook(online=False, cache=tmp_path / "callsigns.json") for call, name, where in entries: book._entries[call] = Callsign( call=call, name=name, location=where, status="found", version=CACHE_VERSION, fetched_at=time.time()) return book def _heard_from(band, text, calls, grid="", at=1_700_000_000.0): band.add(ft8wave.Decode(text=text, kind="standard", calls=calls, grid=grid, hertz=1200.0, offset=0.2, snr_db=-7.0, calling=text.startswith("CQ"), at=at)) def test_both_callsigns_in_an_exchange_are_looked_up(tmp_path): """The station being answered may never transmit within earshot, and is still a station this receiver knows about.""" register = _register(tmp_path) band = ft8.Band() band.register = register _heard_from(band, "W1AW KU0W DM42", ("W1AW", "KU0W"), "DM42") assert set(register._entries) == {"W1AW", "KU0W"} def test_a_station_is_asked_about_once_however_often_it_transmits(tmp_path): register = _register(tmp_path) band = ft8.Band() band.register = register for i in range(6): _heard_from(band, "CQ W1AW FN31", ("W1AW",), "FN31", at=1_700_000_000.0 + i * 15) assert set(register._entries) == {"W1AW"} def test_a_callsign_nobody_has_spelled_out_is_not_asked_about(tmp_path): """FT8 carries a compound callsign as a hash and expects the receiver to have heard it in full earlier. Until it has, there is nothing to ask.""" register = _register(tmp_path) band = ft8.Band() band.register = register _heard_from(band, "<...> ON7EE JO10", ("<...>", "ON7EE"), "JO10") assert set(register._entries) == {"ON7EE"} def test_a_rover_is_looked_up_under_its_licence(tmp_path): """ET3RFG/R is ET3RFG operating away from the licensed address, and no register has heard of the suffix.""" register = _register(tmp_path) band = ft8.Band() band.register = register _heard_from(band, "ET3RFG/R IN3ADG -23", ("ET3RFG/R", "IN3ADG")) assert set(register._entries) == {"ET3RFG", "IN3ADG"} def test_the_name_is_shown_when_it_has_arrived(tmp_path): register = _register(tmp_path, [("W1AW", "ARRL HQ Operators Club", "Newington, CT")]) band = ft8.Band() band.register = register _heard_from(band, "CQ W1AW FN31", ("W1AW",), "FN31") assert ft8.who_text(ft8.licensee(band, "W1AW")) == \ "ARRL HQ Operators Club — Newington, CT" def test_the_dx_the_register_cannot_know_still_says_which_country(): """Which on this band is the point: the databases are United States registers and the interesting stations are all somewhere else.""" from bandsaunter.callsign import Callsign who = Callsign(call="VK4BLE", country="Australia", status="unlisted") assert ft8.who_text(who) == "Australia" def test_nothing_is_looked_up_without_a_register(): band = ft8.Band() _heard_from(band, "CQ W1AW FN31", ("W1AW",), "FN31") assert ft8.licensee(band, "W1AW") is None assert ft8.open_register(ft8.Ft8Options(lookup=False)).online is False def test_the_report_gives_a_column_for_who_each_station_is(tmp_path): from rich.console import Console band = ft8.Band() band.register = _register(tmp_path, [("W1AW", "ARRL HQ Operators Club", "Newington, CT")]) band.slots = 2 _heard_from(band, "CQ W1AW FN31", ("W1AW",), "FN31") console = Console(width=150, force_terminal=False) with console.capture() as caught: ft8.report(console, band, ft8.Ft8Options(lookup=True)) shown = caught.get() assert "licensed to" in shown and "ARRL HQ Operators Club" in shown def test_the_report_leaves_the_column_out_when_lookups_are_off(tmp_path): from rich.console import Console band = ft8.Band() band.slots = 2 _heard_from(band, "CQ W1AW FN31", ("W1AW",), "FN31") console = Console(width=150, force_terminal=False) with console.capture() as caught: ft8.report(console, band, ft8.Ft8Options(lookup=False)) assert "licensed to" not in caught.get() def test_what_was_learned_is_written_to_the_disk(tmp_path): """Otherwise every evening asks the register about the same band again, which is the one thing caching it was for.""" from bandsaunter.callsign import CallsignBook where = tmp_path / "callsigns.json" band = ft8.Band() band.register = _register(tmp_path, [("W1AW", "ARRL HQ", "Newington, CT")]) band.register._dirty = True ft8.keep_lookups(band) assert where.exists() assert CallsignBook(online=False, cache=where).get("W1AW").name == "ARRL HQ" def test_listening_ends_by_writing_what_it_learned(tmp_path, monkeypatch): """Through the real path rather than by calling the helper directly.""" from rich.console import Console from bandsaunter.callsign import CACHE_VERSION, Callsign, CallsignBook where = tmp_path / "callsigns.json" monkeypatch.setattr(ft8, "open_register", lambda options: CallsignBook(online=False, cache=where)) monkeypatch.setattr(ft8, "open_device", lambda console, options: None) console = Console(width=100, force_terminal=False) options = ft8.defaults() options.log = False heard = ft8.Heard() heard.band.register = CallsignBook(online=False, cache=where) heard.band.register._entries["KU0W"] = Callsign( call="KU0W", name="Rod R Gowdy", status="found", version=CACHE_VERSION, fetched_at=time.time()) heard.band.register._dirty = True with console.capture(): ft8.finish(console, options, str(tmp_path), heard) assert CallsignBook(online=False, cache=where).get("KU0W").name == \ "Rod R Gowdy" def test_every_mode_that_hears_callsigns_looks_them_up_and_caches_them(): """The whole of the answer to "all possible features", as a test rather than as a promise. Written as a register of what is known to resolve callsigns, so that a module which starts doing it later fails here until somebody has decided whether it should. Checking only that the current ones work would say nothing about the next one. Aircraft are deliberately absent: a registration is not an amateur callsign and no amateur register has heard of one. They have their own cached book, which test_flights covers. """ import ast import importlib from bandsaunter import aprs, browse, scanner from bandsaunter.callsign import CallsignBook, _cache_path known = { "scanner": "speech transcripts and Morse idents", "browse": "the recording browser", "aprs": "APRS stations", "ft8": "FT8 exchanges", } found = {} for path in sorted(Path(aprs.__file__).parent.glob("*.py")): if path.stem in ("callsign", "__init__"): continue tree = ast.parse(path.read_text()) for node in ast.walk(tree): if isinstance(node, ast.ImportFrom) and node.module == "callsign": if any(a.name in ("CallsignBook", "find_callsigns", "licensed_call") for a in node.names): found[path.stem] = True assert set(found) == set(known), ( f"modules resolving callsigns changed: {sorted(found)} " f"vs {sorted(known)} -- add it here and give it a lookup, or " f"explain why it should not have one") # And every one of them resolves through the same file, so a callsign # heard on two bands is asked about once rather than twice. for book in (aprs.open_book(aprs.defaults()), ft8.open_register(ft8.defaults()), CallsignBook(online=False)): assert book.cache_path == _cache_path() assert browse is not None and scanner is not None def test_a_callsign_heard_on_two_bands_is_asked_about_once(tmp_path): """The point of one shared file: APRS and FT8 hear the same operators, and a register asked twice for the same answer is a register being asked one time too many.""" from bandsaunter import aprs from bandsaunter.callsign import CACHE_VERSION, Callsign, CallsignBook where = tmp_path / "callsigns.json" asked = [] class Counting(CallsignBook): def _request(self, url, call): asked.append(call) raise AssertionError("no network in tests") first = Counting(online=False, cache=where) first._entries["W1AW"] = Callsign(call="W1AW", name="ARRL HQ", status="found", version=CACHE_VERSION, fetched_at=time.time()) first._dirty = True first.save() # A different mode, a later evening, the same file. second = Counting(online=True, cache=where) assert second.get("W1AW").name == "ARRL HQ" assert asked == [], "it went back to the register for an answer it had"