dustcouncil/bandsaunter
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Listen to FT8: fifteen seconds of everybody at once

Browse source 
A new section, alongside the aircraft, the weather sensors and APRS.

FT8 is the odd one out among the things this program listens to, and the
reason is worth stating because it shapes everything below.  Every station on
the band transmits in the same quarter-minute slots, on the same dial
frequency, fifty hertz wide each, stacked across three kilohertz of audio.
One receiver parked on one frequency therefore hears the whole band's worth of
stations at once -- and hears most of them well below the noise, because half
of what is sent is error-correcting code.  That is the entire trick: a rate of
about one half buys a mode that decodes twenty-odd decibels under what an
operator can hear.  A receiver that took the loudest tone of each symbol and
hoped would decode almost nothing, which is why the tone detector reports how
confident it is bit by bit rather than what it thinks it heard.

Written from first principles except for two tables.  The checksum, the
belief propagation over the sparse graph, the Costas sync search, the
waterfall, the soft-bit metric, and the seventy-seven bits that hold two
callsigns and a grid square are all here.  The generator and the parity-check
matrix are not: they cannot be derived, being the code itself rather than
consequences of anything, so they are taken from ft8_lib under its MIT licence
with the attribution it asks for, and said so in the readme, the manual and
the file.  No decoding logic came with them.  That the two agree -- and they
are not derivable from one another, the generator's parity half running to
fifty-odd bits a row against the sparse matrix's six or seven -- is a test
rather than an assumption.

Tested against the air, not against itself.  Eleven off-air recordings with
published decodes: ninety-seven of a hundred and fifty messages, no false
decodes, timing within a hundredth of a second, frequency within a hertz,
signal reports within half a decibel on average.  The third not decoded are
the weakest in each slot; a mature decoder subtracts what it has decoded and
looks again in the remainder, and does ordered-statistics decoding where
belief propagation fails, and neither is built here.  What is here decodes
nothing that other receivers did not also hear, which is the property that
matters in a log.  Ten whole codewords lifted off the air are in the tests as
a permanent fixture, so the recordings can go missing and the regression
cannot.

Three things that looked like bugs and were not, and three that were.  The
half-second timing discrepancy was the convention: a transmission is 12.64
seconds in a slot of fifteen and everybody starts half a second in, so
lateness is reported against that.  Synthetic signals at known offsets proved
the clock self-consistent before anything was changed.  The signal reports
were twenty-one decibels optimistic because those recordings have a receiver
passband above three kilohertz, putting a whole-band median twelve to sixteen
decibels below the real noise floor -- so noise is now measured beside the
signal, and in the tone that was actually sent rather than the loudest of
eight, the largest of eight noisy numbers being well above their mean even
with no signal at all.  And the test transmitter was thirteen decibels
pessimistic, scaling its noise into a fifty-hertz reference instead of the
sampled bandwidth, which made the decoder look deaf when it was the test
signal that had been quietly attenuated.

The real bug the simulator caught was a one-block timestamp error: samples
were dated a block earlier than they were taken, which slid every slot slice a
second late and cut the first half-second -- three symbols, part of the
opening Costas array -- off every transmission on the band.  One decode a slot
became six.

Reachable both ways, as everything here is.  Twenty-one options, every one of
them a command-line flag and a line in the menu, both built from one table so
they cannot disagree -- and a test that says so, since an option in no group
would be settable from the command line and invisible in the menu.  The band
list says which channels a plain dongle can reach and which need an
upconverter, because almost all the activity is on shortwave and finding that
out by listening to silence for ten minutes is the wrong way to learn it.  The
default is two metres, which a plain dongle can hear.

--grid turns decodes into geography: every CQ says where it is, so each gets a
distance and a bearing and the furthest is named.  --adif writes the log in
the form every amateur logging program imports, marked as heard rather than
worked, because nothing here transmits and an ADIF that let a logging program
treat these as contacts would put claims into somebody's log that they cannot
make.

One bug shipped and found by being used rather than by being tested: the
line that opens the receiver called a function this program has never had.
Every test reached it through the simulator, which takes the other branch, so
the one line that matters to somebody with an aerial was the one line never
run.  There is now a check that every name these modules import actually
exists -- it names the missing one rather than failing somewhere downstream --
and two that say a receiver which cannot be opened is reported rather than
raised, and that nothing claims to be listening before there is one.  It had
been announcing the frequency first, so a dongle that would not open read as
listening that had gone wrong.

Ninety-six new tests.  Full suite 2781 passed.  Built as 2026-09-21_04.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016PsWPTweCT6pwxKngvVxcg
This commit is contained in:
The Dust Council 2026-09-21 16:09:57 -07:00
parent e203b3e581
commit 93120b80a6
14 changed files with 4247 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

839
tests/test_ft8.py Normal file
View file

@ -0,0 +1,839 @@
"""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 "<CALL:6>DL1UDO" in body
assert "<MODE:3>FT8" in body
assert "<MY_GRIDSQUARE:4>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("<EOR>") == 1
# And the one kept is the best report, that being the one worth passing on.
assert "<RST_RCVD:2>-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