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

Read the APRS channel: who is out there, and what they said

Browse source 
A third section, alongside the aircraft and the weather sensors, and for the
same reason as both: a scan stops on a signal, records it and moves on, while
APRS is a two-second transmission every few minutes from a hundred stations
sharing one frequency.  A sweep catches whichever happened to key up as it
passed.  `bandsaunter aprs` parks on the channel and catches all of them;
`bandsaunter packets` reads a log back.

Four layers, three of them new.

The link layer was already here, opportunistically, in the generic decoder --
a correlator, NRZI, HDLC and a checksum, run on whatever a scan happened to
record.  It is now a receiver.  What had to change is the state that survives
a block boundary: the tail of the audio so the correlators see no edge, the
phase of the sampling loop so a bit is not lost where one block meets the
next, the tone the line was last at, and the bits themselves.  A packet is
most of a second and a block is about one, so frames straddling the boundary
are not an edge case, they are most of them.

Above that, the APRS information field, which is not one format but about
twenty, chosen by the first character and accreted over thirty years.
Positions uncompressed and compressed; Mic-E, which every Kenwood and Yaesu
mobile sends and which hides the latitude inside the destination callsign
because in 1995 those six bytes were carrying the word "APRS" and nothing
else; weather with a position and without; messages, acknowledgements,
rejections and bulletins; objects and items; status; telemetry; third-party
traffic, credited to whoever originally sent it rather than to the gateway.
Course and speed, altitude, power and antenna height, range and the precision
extension, all of which ride in the comment.  Every one has a writer beside
its reader, so a packet goes in and the same packet comes out.

Above that the section: a registry of who is out there and what each last
said of each kind, distances and bearings from --at, a log keeping the whole
frame under whatever was made of it, a spreadsheet, a map, and a channel full
of stations that are not there for --simulate.

One rule is worth naming because it is the difference between a decoder and a
liar.  A packet whose format does not match what its first character promised
comes back as unparsed with its text intact.  Thirteen characters of a
*malformed* uncompressed position are perfectly good base-91, so trying one
format and falling back to the other does not fail on a bad packet -- it
succeeds, as a confident and completely different place, usually a thousand
miles away.  The specification makes the two unambiguous, a leading digit
always meaning uncompressed, and the rule is read rather than guessed at.

Two faults found by building it, both by measurement rather than by reading
the code again.  The framer handed back frames it had already reported,
because it trimmed its buffer to before them rather than after -- every packet
counted twice, for ever, which only shows up once the same signal is read
across more than one block.  And the invented channel truncated a
transmission at the end of the block it began in rather than carrying the
remainder over, which was invisible for as long as the simulated clock
advanced in exact seconds and put every transmission at a block boundary; the
moment it was paced against a real clock, nothing decoded at all.

204 new tests against seven deliberately broken builds, one of which survived
until a test was written for the case it actually breaks.  Full suite 2597
passed.  Built as 2026-09-20_02.

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-20 19:36:30 -07:00
parent 0f7e47e55e
commit 2b653c2c3e
16 changed files with 5543 additions and 17 deletions
Download patch file Download diff file Expand all files Collapse all files

629
tests/test_aprs.py Normal file
View file

@ -0,0 +1,629 @@
"""The APRS section: the registry, the log, the report and the two front ends.
The radio is in test_ax25.py and the formats are in test_packets.py. What is
here is everything between a parsed packet and a person.
"""
import json
import time
import numpy as np
import pytest
from rich.console import Console
from bandsaunter import aprs as ap
from bandsaunter import aprslog as al
from bandsaunter import ax25, packets
from bandsaunter.ui import AprsDisplay
RATE = 240_000.0
def packet(info="=4903.50N/07201.75W>088/036 mobile", source="W1AW-9",
at=1_000.0, snr=20.0, path=(), destination="APRS"):
got = packets.parse_info(info, destination)
got.source, got.at, got.snr = source, at, snr
got.path, got.destination = tuple(path), destination
return got
# ---------------------------------------------------------------------------
# The options
# ---------------------------------------------------------------------------
def test_every_option_names_a_field_that_exists():
fields = set(ap.AprsOptions().__dict__)
for option in ap.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(ap.AprsOptions().__dict__) == {o.key for o in ap.OPTIONS}
def test_every_option_is_in_a_group_and_says_what_it_does():
for option in ap.OPTIONS:
assert option.group in ap.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 ap.OPTION_GROUPS:
assert ap.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[
"aprs"]._actions:
known.update(action.option_strings)
for option in ap.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_the_options_survive_being_saved_and_read_back(tmp_path):
options = ap.AprsOptions(units="imperial", region="europe", hold=90.0,
location="51.5,-0.1", digipeated=False)
ap.save_options(options, tmp_path)
assert ap.load_options(tmp_path) == options
def test_a_settings_file_with_a_mistake_in_it_costs_the_defaults(tmp_path):
(tmp_path / "aprs.yaml").write_text("units: [not a unit\n")
assert ap.load_options(tmp_path) == ap.AprsOptions()
@pytest.mark.parametrize("options,broken", [
(ap.AprsOptions(rate=48_000.0), "sample rate"),
(ap.AprsOptions(seconds=-1.0), "negative"),
(ap.AprsOptions(hold=0.0), "display"),
(ap.AprsOptions(frequency=10.0), "frequency"),
])
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 ap.AprsOptions().validate() == []
@pytest.mark.parametrize("region,megahertz", [
("north-america", 144.390), ("europe", 144.800), ("australia", 145.175),
])
def test_each_region_has_its_own_channel(region, megahertz):
assert ap.channel_named(region) == pytest.approx(megahertz * 1e6)
def test_an_unknown_region_falls_back_rather_than_failing():
"""Refusing to listen over a misspelt region would be the wrong trade."""
assert ap.channel_named("narnia") == ax25.APRS_HZ
def test_the_region_sets_the_frequency_unless_one_was_given():
from bandsaunter.cli import build_parser
args = build_parser().parse_args(["aprs", "--region", "europe"])
assert args.region == "europe" and args.frequency is None
args = build_parser().parse_args(["aprs", "--region", "europe",
"--freq", "144900000"])
assert args.frequency == 144_900_000.0
# ---------------------------------------------------------------------------
# How far off, and which way
# ---------------------------------------------------------------------------
def test_a_distance_is_the_great_circle_one():
london, paris = (51.5074, -0.1278), (48.8566, 2.3522)
assert ap.distance_km(london, paris) == pytest.approx(343.0, abs=5.0)
assert ap.bearing_deg(london, paris) == pytest.approx(148.0, abs=3.0)
def test_a_station_at_the_aerial_is_no_distance_away():
here = (47.55, -122.30)
assert ap.distance_km(here, here) == pytest.approx(0.0, abs=0.001)
def test_distances_are_only_worked_out_when_there_is_somewhere_to_measure_from():
station = ap.Station(call="W1AW")
station.add(packet())
assert station.away(None) is None
assert station.away((47.55, -122.30))[0] > 0
@pytest.mark.parametrize("text,ok", [
("47.55,-122.30", True), ("47.55, -122.30", True), ("", False),
("nowhere", False), ("91,0", False), ("0,181", False),
])
def test_a_location_that_is_not_one_is_refused(text, ok):
assert (ap.coordinates(text) is not None) is ok
# ---------------------------------------------------------------------------
# Who is out there
# ---------------------------------------------------------------------------
def test_a_station_keeps_the_latest_of_each_kind_and_not_the_latest_packet():
"""A station sends a position, then a status, then weather; a display
showing only the newest packet loses two of the three every time."""
net = ap.Net()
net.add(packet(at=1_000.0))
net.add(packet(">Monitoring 146.52", at=1_016.0))
station = net.stations["W1AW-9"]
assert station.position is not None # kept from the first
assert station.status == "Monitoring 146.52"
assert station.packets == 2
def test_a_moving_station_keeps_a_track_and_a_still_one_does_not_repeat_itself():
net = ap.Net()
for i in range(5):
net.add(packet(packets.position_report(49.0 + i * 0.01, -72.0, "/>"),
at=1_000.0 + i * 20))
assert len(net.stations["W1AW-9"].track) == 5
for i in range(5):
net.add(packet(packets.position_report(49.0, -72.0, "/-"),
source="KU0W", at=2_000.0 + i * 20))
assert len(net.stations["KU0W"].track) == 1
def test_a_track_cannot_grow_without_limit():
# A step of a thousandth of a degree, which is coarser than the
# hundredth of a minute the format resolves to -- a finer one would be
# rounded to the same place and correctly not recorded twice.
net = ap.Net()
for i in range(ap.TRACK_POINTS + 50):
net.add(packet(packets.position_report(49.0 + i * 0.001, -72.0),
at=1_000.0 + i))
assert len(net.stations["W1AW-9"].track) == ap.TRACK_POINTS
def test_a_station_that_has_not_moved_is_not_recorded_as_having_moved():
"""Two packets from the same place are one point, not two."""
net = ap.Net()
for i in range(6):
net.add(packet(packets.position_report(49.0, -72.0),
at=1_000.0 + i * 60))
assert len(net.stations["W1AW-9"].track) == 1
def test_a_packet_that_arrived_through_a_digipeater_is_counted_apart():
"""Most of what a receiver hears is a hilltop repeating somebody, which
is the whole design; what was heard directly is the honest measure of
what the aerial can reach."""
net = ap.Net()
net.add(packet(at=1_000.0, path=("WIDE1-1*", "WIDE2-1")))
net.add(packet(at=1_060.0, path=()))
station = net.stations["W1AW-9"]
assert station.packets == 2 and station.direct == 1
def test_an_object_is_filed_under_its_own_name_not_its_senders():
net = ap.Net()
net.add(packet(packets.object_report("LEADER", 49.0, -72.0),
source="W1AW-9", at=1_000.0))
assert "LEADER" in net.stations and "W1AW-9" not in net.stations
assert net.stations["LEADER"].object_of == "W1AW-9"
def test_messages_are_kept_in_order_as_a_conversation():
net = ap.Net()
net.add(packet(packets.message_text("KU0W", "first"), at=1_000.0))
net.add(packet(packets.message_text("W1AW", "second"), source="KU0W",
at=1_060.0))
assert [m.message.text for m in net.messages] == ["first", "second"]
def test_a_station_that_has_gone_quiet_leaves_the_display_but_not_the_log():
net = ap.Net()
net.add(packet(at=1_000.0))
assert len(net.showing(300.0, now=1_200.0)) == 1
assert len(net.showing(300.0, now=1_400.0)) == 0
assert len(net.all()) == 1
def test_the_newest_station_is_at_the_top():
"""The opposite of the other displays, and deliberately: what somebody
watching a busy channel wants to know is what just came in."""
net = ap.Net()
net.add(packet(source="OLD", at=1_000.0))
net.add(packet(source="NEW", at=2_000.0))
assert [s.call for s in net.showing(1e6, now=2_000.0)] == ["NEW", "OLD"]
def test_a_net_can_be_rebuilt_from_a_log():
heard = [packet(at=1_000.0), packet(source="KU0W", at=1_001.0)]
assert {s.call for s in ap.Net.of(heard).all()} == {"W1AW-9", "KU0W"}
# ---------------------------------------------------------------------------
# The log
# ---------------------------------------------------------------------------
def test_a_packet_survives_being_written_and_read_back(tmp_path):
log = al.AprsLog(tmp_path / "aprs_x.jsonl", frequency=ax25.APRS_HZ)
sent = packet(path=("WIDE1-1*",))
log.append(sent, ax25.frame_from(ax25.frame_bytes("W1AW-9", "APRS",
sent.info)))
log.close()
back = al.read_logs([log.path])
assert len(back) == 1
got = back[0]
assert got.source == "W1AW-9" and got.kind == "position"
assert got.position.latitude == pytest.approx(49.0583, abs=0.0002)
assert got.course == 88 and got.path == ("WIDE1-1*",)
assert got.snr == pytest.approx(20.0)
def test_the_whole_frame_goes_down_underneath_whatever_was_understood(tmp_path):
"""APRS has a long tail of formats, so a packet this version cannot read
is kept in full for a version that can."""
log = al.AprsLog(tmp_path / "aprs_x.jsonl")
raw = ax25.frame_bytes("W1AW", "APRS", "!!! not a format !!!")
frame = ax25.frame_from(raw)
log.append(packets.parse(frame), frame)
log.close()
body = json.loads(log.path.read_text().splitlines()[1])
assert bytes.fromhex(body["hex"]) == raw
assert body["kind"] == "unparsed"
assert body["info"] == "!!! not a format !!!"
def test_weather_and_messages_survive_the_log(tmp_path):
log = al.AprsLog(tmp_path / "aprs_x.jsonl")
log.append(packet("@092345z4903.50N/07201.75W_220/004g005t077h50b09900",
source="KB1XYZ"))
log.append(packet(packets.message_text("KU0W", "on my way", "7")))
log.close()
back = al.read_logs([log.path])
assert back[0].weather["temperature"].value == pytest.approx(25.0, abs=0.1)
assert back[1].message.to == "KU0W" and back[1].message.number == "7"
def test_a_half_written_last_line_costs_that_line_and_not_the_evening(tmp_path):
log = al.AprsLog(tmp_path / "aprs_x.jsonl")
for i in range(5):
log.append(packet(at=1_000.0 + i))
log.close()
with open(log.path, "a") as fh:
fh.write('{"t": 9, "src": "W1A')
assert len(al.read_logs([log.path])) == 5
def test_the_header_line_is_not_read_back_as_a_packet(tmp_path):
log = al.AprsLog(tmp_path / "aprs_x.jsonl", receiver="device 0")
log.close()
head = json.loads(log.path.read_text().splitlines()[0])
assert head["log"] == "bandsaunter-aprs"
assert al.read_logs([log.path]) == []
def test_a_log_is_written_as_it_goes_and_not_when_it_closes(tmp_path):
log = al.AprsLog(tmp_path / "aprs_x.jsonl")
log.append(packet())
assert len(log.path.read_text().splitlines()) == 2
log.close()
# ---------------------------------------------------------------------------
# Out to a spreadsheet and a globe
# ---------------------------------------------------------------------------
def test_the_spreadsheet_has_a_column_for_everything_anything_reported(tmp_path):
heard = [packet(at=1_000.0),
packet("@092345z4903.50N/07201.75W_220/004g005t077h50b09900",
source="KB1XYZ", at=1_001.0)]
where = al.write_csv(tmp_path / "a.csv", heard)
head = where.read_text().splitlines()[0].split(",")
assert "latitude" in head and "speed (km/h)" in head
assert any(h.startswith("temperature") for h in head)
def test_the_spreadsheet_can_be_written_the_other_way_round(tmp_path):
heard = [packet("@092345z4903.50N/07201.75W_220/004t032", source="KB1XYZ",
at=1_000.0)]
where = al.write_csv(tmp_path / "a.csv", heard, imperial=True)
head, row = where.read_text().splitlines()[:2]
columns = head.split(",")
assert "temperature (F)" in columns
assert row.split(",")[columns.index("temperature (F)")] == "32.0"
def test_a_map_has_a_pin_for_every_station_and_a_line_for_what_moved(tmp_path):
heard = [packet(packets.position_report(49.0 + i * 0.01, -72.0, "/>"),
at=1_000.0 + i * 20) for i in range(4)]
heard.append(packet(packets.position_report(50.0, -73.0), source="KU0W",
at=1_100.0))
where = al.write_kml(tmp_path / "a.kml", heard)
text = where.read_text()
assert text.count("<Placemark>") == 3 # a track and two pins
assert "<LineString>" in text and "W1AW-9" in text and "KU0W" in text
def test_a_map_of_stations_that_never_said_where_they_were_is_not_drawn(tmp_path):
"""A pin at nowhere puts a station off the west coast of Africa."""
heard = [packet(">Monitoring 146.52", at=1_000.0)]
assert al.write_kml(tmp_path / "a.kml", heard) is None
# ---------------------------------------------------------------------------
# Listening
# ---------------------------------------------------------------------------
class Silence:
"""A receiver that hears nothing, for testing the loop rather than the air."""
def __init__(self, blocks=2):
self.blocks = blocks
self.tuned = 0.0
def tune(self, hz, settle=True):
self.tuned = hz
return int(hz)
def read_samples(self, count, flush=False):
if self.blocks <= 0:
return np.zeros(0, dtype=np.complex64)
self.blocks -= 1
return np.zeros(count, dtype=np.complex64)
def close(self):
return None
class Channel:
"""The invented channel, wrapped so a test can drive it a block at a time."""
def __init__(self, blocks, rate=RATE, seed=3):
self.sky = ap.SimulatedChannel(sample_rate=rate, seed=seed)
self.left = blocks
def tune(self, hz, settle=True):
return int(hz)
def read_samples(self, count, flush=False):
if self.left <= 0:
return np.zeros(0, dtype=np.complex64)
self.left -= 1
return self.sky.read_samples(count)
def close(self):
return None
def test_the_receiver_is_tuned_to_the_channel():
device = Silence()
ap.pump(device, ap.AprsOptions(rate=RATE, region="europe",
frequency=ap.channel_named("europe")),
ap.Net(), None, time.time())
assert device.tuned == pytest.approx(144_800_000.0)
def test_listening_stops_when_the_receiver_does():
device = Silence(blocks=3)
assert ap.pump(device, ap.AprsOptions(rate=RATE), ap.Net(), None,
time.time()) == 0
assert device.blocks == 0
def test_a_whole_session_hears_the_invented_channel(tmp_path, monkeypatch):
monkeypatch.setattr(ap, "open_device",
lambda console, opts: Channel(120))
options = ap.AprsOptions(rate=RATE, simulate=True, packets_seen=True,
csv=True, kml=True, report=False)
console = Console(width=120, force_terminal=False)
with console.capture():
heard = ap.listen(console, options, str(tmp_path))
assert heard.stations >= 4
assert heard.log_path.exists() and heard.csv_path.exists()
assert heard.kml_path.exists()
assert len(al.read_logs([heard.log_path])) == heard.packets
kinds = {s.kind for s in heard.net.all()}
assert {"position", "weather", "object"} <= kinds
assert heard.net.messages
def test_a_transmission_that_straddles_a_block_is_still_heard():
"""A packet is most of a second and a block is about one, so this is not
an edge case -- it is most of them. A simulated clock advancing in exact
seconds hides it by putting every transmission at a block boundary.
"""
sky = ap.SimulatedChannel(sample_rate=RATE, seed=3)
demod, receiver = ap.make_receiver(ap.AprsOptions(rate=RATE))
found = 0
for _ in range(50):
# Blocks of an awkward length, so nothing lines up.
audio, _iq = demod.step(sky.read_samples(int(RATE * 0.77)))
found += len(receiver.feed(audio))
assert found >= 3, "transmissions are being cut off at the block boundary"
def test_relayed_packets_can_be_left_out(tmp_path, monkeypatch):
monkeypatch.setattr(ap, "open_device", lambda console, opts: Channel(120))
console = Console(width=120, force_terminal=False)
options = ap.AprsOptions(rate=RATE, packets_seen=True, log=False,
report=False, digipeated=False)
with console.capture():
heard = ap.listen(console, options, str(tmp_path))
for station in heard.net.all():
assert station.direct == station.packets
def test_a_receiver_that_will_not_open_is_a_message_and_not_a_traceback(
tmp_path, monkeypatch):
monkeypatch.setattr(ap, "open_device", lambda console, opts: None)
console = Console(width=120, force_terminal=False)
with console.capture():
heard = ap.listen(console, ap.AprsOptions(), str(tmp_path))
assert heard.stations == 0 and heard.log_path is None
def test_hearing_nothing_says_what_to_check(tmp_path, monkeypatch):
monkeypatch.setattr(ap, "open_device", lambda console, opts: Silence(2))
console = Console(width=120, force_terminal=False)
with console.capture() as cap:
ap.listen(console, ap.AprsOptions(rate=RATE, packets_seen=True,
log=False), str(tmp_path))
assert "region" in cap.get()
# ---------------------------------------------------------------------------
# The report and the display
# ---------------------------------------------------------------------------
def rendered(net, options=None):
console = Console(width=130, force_terminal=False)
with console.capture() as cap:
ap.report(console, net, options or ap.AprsOptions())
return cap.get()
def test_the_report_says_who_was_heard_and_what_they_said():
net = ap.Net()
net.add(packet(at=1_000.0))
net.add(packet("@092345z4903.50N/07201.75W_220/004g005t077h50b09900",
source="KB1XYZ", at=1_001.0))
net.add(packet(packets.message_text("KU0W", "on my way"), at=1_002.0))
out = rendered(net)
assert "W1AW-9" in out and "KB1XYZ" in out
assert "stations heard" in out and "what they said" in out
assert "what passed between them" in out
assert "on my way" in out
def test_the_report_shows_distances_only_when_it_knows_where_the_aerial_is():
net = ap.Net()
net.add(packet(at=1_000.0))
assert "away" not in rendered(net)
assert "away" in rendered(net, ap.AprsOptions(location="47.55,-122.30"))
def test_the_report_counts_what_it_could_not_read():
net = ap.Net()
net.add(packet("!!! not a format !!!", at=1_000.0))
assert "cannot read" in rendered(net)
@pytest.mark.parametrize("level,colour", [
(4.0, "red"), (12.0, "yellow"), (24.0, "green"),
])
def test_the_signal_is_coloured_so_an_aerial_can_be_aimed_by_it(level, colour):
assert colour in ap.signal_text(level)
def test_no_signal_is_shown_as_nothing_rather_than_as_zero():
assert ap.signal_text(0.0) == ""
def shown(net, width=120, now=None, **kw):
console = Console(width=width, force_terminal=False)
display = AprsDisplay(console, **kw)
display.started = (now or time.time()) - 60
display.update(net, net.packets)
with console.capture() as cap:
console.print(display.render(now=now))
return cap.get()
def test_the_display_shows_a_station_and_what_it_last_said():
net = ap.Net()
now = time.time()
net.add(packet(at=now))
out = shown(net, now=now)
assert "W1AW-9" in out and "49.0583N" in out and "144.39 MHz" in out
def test_the_display_still_fits_a_narrow_terminal():
net = ap.Net()
now = time.time()
net.add(packet(at=now))
for width in (60, 70, 86, 100, 140):
out = shown(net, width=width, now=now, home=(47.55, -122.30))
assert max(len(line) for line in out.splitlines()) <= width
assert "W1AW-9" in out
def test_an_empty_display_says_what_to_check():
assert "region" in shown(ap.Net())
# ---------------------------------------------------------------------------
# The command line and the menu
# ---------------------------------------------------------------------------
def test_reading_a_log_back_prints_it_and_writes_the_files(tmp_path,
monkeypatch):
from bandsaunter.cli import build_parser, cmd_packets
import bandsaunter.cli as cli
log = tmp_path / "aprs_x.jsonl"
with al.AprsLog(log) as fh:
for i in range(4):
fh.append(packet(packets.position_report(49.0 + i * 0.01, -72.0),
at=1_000.0 + i * 20))
console = Console(width=130, force_terminal=False)
monkeypatch.setattr(cli, "console", console)
args = build_parser().parse_args(["packets", str(log), "--csv", "--kml"])
with console.capture() as cap:
assert cmd_packets(args) == 0
assert "W1AW-9" in cap.get()
assert log.with_suffix(".csv").exists() and log.with_suffix(".kml").exists()
def test_reading_a_log_back_can_be_narrowed_to_one_station(tmp_path,
monkeypatch):
from bandsaunter.cli import build_parser, cmd_packets
import bandsaunter.cli as cli
log = tmp_path / "aprs_x.jsonl"
with al.AprsLog(log) as fh:
fh.append(packet(at=1_000.0))
fh.append(packet(source="KU0W", at=1_001.0))
console = Console(width=130, force_terminal=False)
monkeypatch.setattr(cli, "console", console)
args = build_parser().parse_args(["packets", str(log), "--station", "KU0W"])
with console.capture() as cap:
assert cmd_packets(args) == 0
out = cap.get()
assert "KU0W" in out and "W1AW-9" not in out
def test_asking_for_a_log_that_is_not_there_is_a_message(tmp_path,
monkeypatch):
from bandsaunter.cli import build_parser, cmd_packets
import bandsaunter.cli as cli
console = Console(width=120, force_terminal=False)
monkeypatch.setattr(cli, "console", console)
args = build_parser().parse_args(["packets", str(tmp_path / "no.jsonl")])
with console.capture() as cap:
assert cmd_packets(args) == 1
assert "no such file" in cap.get()
def test_the_menu_can_be_opened_and_left(monkeypatch):
import bandsaunter.tui as tui
from bandsaunter.config import ScanConfig
answers = iter(["b"])
monkeypatch.setattr(tui, "_ask",
lambda console, prompt, default="": next(answers))
console = Console(width=120, force_terminal=False)
with console.capture() as cap:
tui.aprs_menu(console, ScanConfig())
out = cap.get()
assert "144.390 MHz" in out
for group in ap.OPTION_GROUPS:
assert group.lower() in out
def test_one_set_of_option_screens_drives_all_three_sections():
import bandsaunter.tui as tui
from bandsaunter import aircraft, weather
for module in (aircraft, weather, ap):
assert tui._section(module) is module
assert callable(module.defaults)
assert module.OPTION_GROUPS and module.OPTIONS

260
tests/test_ax25.py Normal file
View file

@ -0,0 +1,260 @@
"""AX.25: the frames APRS rides in, and getting them off the air.
Every frame here is built by the encoder that sits beside the decoder, keyed
out as real Bell 202 audio, and read back. That proves the framing, the bit
stuffing, the NRZI, the checksum and the clock recovery; it does not prove
anything a description and an implementation of it might both get wrong, and
the module says so.
The other half is about what must not be read. This runs for hours with the
squelch open, so a frame either satisfies sixteen bits of CRC or it never
existed.
"""
import numpy as np
import pytest
from bandsaunter import ax25
RATE = 22_050.0
def heard(frames, rate=RATE, chunk=None, noise=0.02, amplitude=0.5):
"""Frames keyed out and read back, optionally a block at a time."""
if isinstance(frames, (bytes, bytearray)):
frames = [frames]
audio = ax25.modulate(frames, rate, amplitude=amplitude, noise=noise)
receiver = ax25.Receiver(rate)
if chunk is None:
return receiver.feed(audio, when=1_000.0)
out = []
for i in range(0, audio.size, chunk):
out += receiver.feed(audio[i:i + chunk], when=1_000.0)
return out
# ---------------------------------------------------------------------------
# What a frame is made of
# ---------------------------------------------------------------------------
def test_a_frame_comes_back_with_everything_it_was_sent_with():
raw = ax25.frame_bytes("W1AW-5", "APRS", "=4123.45N/07203.12W-hi",
path=("WIDE1-1*", "WIDE2-1"))
frame = ax25.frame_from(raw)
assert frame is not None
assert frame.source.plain == "W1AW-5" and frame.source.ssid == 5
assert frame.destination.plain == "APRS"
assert [str(h) for h in frame.path] == ["WIDE1-1*", "WIDE2-1"]
assert frame.text() == "=4123.45N/07203.12W-hi"
assert frame.route() == "W1AW-5>APRS,WIDE1-1*,WIDE2-1"
def test_a_station_with_no_ssid_is_written_without_one():
frame = ax25.frame_from(ax25.frame_bytes("W1AW", "APRS", "x"))
assert frame.source.plain == "W1AW"
assert str(frame.source) == "W1AW"
@pytest.mark.parametrize("ssid", range(16))
def test_every_ssid_survives(ssid):
call = f"KU0W-{ssid}" if ssid else "KU0W"
frame = ax25.frame_from(ax25.frame_bytes(call, "APRS", "x"))
assert frame.source.ssid == ssid
assert frame.source.plain == call
def test_a_digipeater_that_has_repeated_a_frame_says_so():
"""The H bit, which is how the path records where a frame has been.
The hops with it set are where the frame went; the rest are where it was
asked to go and has not been yet.
"""
frame = ax25.frame_from(ax25.frame_bytes(
"W1AW", "APRS", "x", path=("WIDE1-1*", "WIDE2-1")))
assert [str(h) for h in frame.heard_through] == ["WIDE1-1*"]
assert frame.path[0].repeated and not frame.path[1].repeated
def test_the_frame_type_apris_uses_is_recognised_and_others_are_named():
ui = ax25.frame_from(ax25.frame_bytes("W1AW", "APRS", "x"))
assert ui.unnumbered_information and ui.kind == "unnumbered information"
other = ax25.frame_from(ax25.frame_bytes("W1AW", "APRS", "x",
control=0x2F, pid=0xF0))
assert not other.unnumbered_information
assert other.kind == "set async balanced"
def test_a_frame_with_a_broken_check_is_refused():
raw = bytearray(ax25.frame_bytes("W1AW", "APRS", "hello"))
for i in range(len(raw)):
broken = bytearray(raw)
broken[i] ^= 0x01
assert ax25.frame_from(bytes(broken)) is None, f"byte {i} got through"
def test_a_frame_whose_addresses_are_not_callsigns_is_refused():
"""Sixteen bits of CRC is strong and this band is busy.
A frame whose addresses are unprintable passed the check by accident,
and there is no reason to put it on a display.
"""
raw = bytearray(ax25.frame_bytes("W1AW", "APRS", "x"))
raw[0] = 0x02 # an unprintable callsign
body = bytes(raw[:-2])
check = ax25.fcs(body)
assert ax25.frame_from(body + bytes([check & 0xFF, check >> 8])) is None
def test_a_frame_too_short_to_be_one_is_refused():
assert ax25.frame_from(b"") is None
assert ax25.frame_from(b"\x00" * 10) is None
@pytest.mark.parametrize("info", ["", "x", "!" * 256, "\x00\xff binary"])
def test_an_information_field_of_any_shape_survives(info):
frame = ax25.frame_from(ax25.frame_bytes("W1AW", "APRS", info))
assert frame is not None and frame.text() == info
# ---------------------------------------------------------------------------
# HDLC
# ---------------------------------------------------------------------------
def test_a_zero_is_stuffed_after_five_ones_and_taken_back_out():
assert ax25.stuff("11111" + "1") == "111110" + "1"
assert ax25.unstuff(ax25.stuff("1" * 40)) == "1" * 40
assert ax25.stuff("0" * 40) == "0" * 40
def test_stuffing_is_what_stops_a_flag_appearing_inside_a_frame():
assert ax25.FLAG not in ax25.stuff("0" + ax25.FLAG + "0")
def test_nrzi_is_the_same_data_read_from_either_polarity():
"""A zero is a change of tone and a one is no change, so inverting the
whole stream -- swapping mark for space, or wiring a discriminator up
backwards -- decodes to exactly the same bits. Nothing here ever has to
guess at polarity, and that is why."""
bits = "110100111000101"
sent = ax25.nrzi(bits)
upside_down = "".join("1" if b == "0" else "0" for b in sent)
assert ax25.un_nrzi("1" + sent) == bits
assert ax25.un_nrzi("0" + upside_down) == bits
def test_a_stream_is_consumed_up_to_the_last_flag_and_no_further():
"""What a caller reading a continuous signal may forget.
Trimming to before a frame that has already been reported hands it back
on the next block, and every packet is counted twice for ever.
"""
raw = ax25.frame_bytes("W1AW", "APRS", "hello")
bits = ax25.bits_of(raw, flags=2)
frames, used = ax25.hdlc_frames(bits)
assert frames == [raw]
assert used > 0
again, _ = ax25.hdlc_frames(bits[used:])
assert again == []
def test_a_frame_still_arriving_is_kept_rather_than_thrown_away():
raw = ax25.frame_bytes("W1AW", "APRS", "hello")
bits = ax25.bits_of(raw, flags=2)
half = bits[:len(bits) // 2]
frames, used = ax25.hdlc_frames(half)
assert frames == []
rest, _ = ax25.hdlc_frames(half[used:] + bits[len(bits) // 2:])
assert rest == [raw]
# ---------------------------------------------------------------------------
# Off the air
# ---------------------------------------------------------------------------
def test_a_packet_survives_being_keyed_out_and_read_back():
raw = ax25.frame_bytes("W1AW-5", "APRS", "=4123.45N/07203.12W-hi",
path=("WIDE1-1*",))
got = heard(raw)
assert [f.raw for f in got] == [raw]
assert got[0].at == 1_000.0
@pytest.mark.parametrize("chunk", [220, 1_102, 2_205, 11_025, 44_100])
def test_a_packet_is_read_once_however_the_audio_is_cut_into_blocks(chunk):
"""A packet is most of a second and a block is about one, so a frame
straddling the boundary is not an edge case -- it is most of them."""
raw = ax25.frame_bytes("W1AW-5", "APRS", "=4123.45N/07203.12W-hi")
got = heard(raw, chunk=chunk)
assert [f.raw for f in got] == [raw], f"{len(got)} frames at {chunk}"
def test_several_packets_back_to_back_all_come_through():
raws = [ax25.frame_bytes("W1AW", "APRS", f">beacon {i}") for i in range(4)]
got = heard(raws, chunk=2_205)
assert [f.text() for f in got] == [f">beacon {i}" for i in range(4)]
@pytest.mark.parametrize("rate", [11_025.0, 22_050.0, 24_000.0, 48_000.0])
def test_it_works_at_the_audio_rates_a_demodulator_might_hand_over(rate):
raw = ax25.frame_bytes("W1AW", "APRS", "=4123.45N/07203.12W-")
assert [f.raw for f in heard(raw, rate=rate, chunk=int(rate // 10))] == [raw]
def test_a_weak_packet_is_still_read():
raw = ax25.frame_bytes("W1AW", "APRS", "=4123.45N/07203.12W-")
got = heard(raw, noise=0.2, amplitude=0.5) # about 8 dB
assert [f.raw for f in got] == [raw]
def test_nothing_is_read_out_of_noise():
rng = np.random.default_rng(1)
for _ in range(80):
receiver = ax25.Receiver(RATE)
assert receiver.feed(rng.normal(0.0, 0.3, int(RATE))) == []
def test_a_receiver_can_be_reset_and_carries_nothing_over():
raw = ax25.frame_bytes("W1AW", "APRS", "hello")
audio = ax25.modulate(raw, RATE, noise=0.02)
receiver = ax25.Receiver(RATE)
receiver.feed(audio[:audio.size // 2])
receiver.reset()
assert receiver.feed(audio[audio.size // 2:]) == []
def test_the_receiver_counts_what_it_has_seen():
raw = ax25.frame_bytes("W1AW", "APRS", "hello")
receiver = ax25.Receiver(RATE)
receiver.feed(ax25.modulate(raw, RATE, noise=0.02))
assert receiver.frames == 1
assert receiver.bytes_seen >= len(raw)
# ---------------------------------------------------------------------------
# Where APRS is
# ---------------------------------------------------------------------------
def test_every_region_has_a_channel_and_they_are_all_different():
frequencies = [hz for _key, hz, _where in ax25.APRS_CHANNELS]
assert len(set(frequencies)) == len(frequencies)
assert all(144e6 < hz < 146e6 for hz in frequencies)
assert ax25.APRS_CHANNELS[0][1] == ax25.APRS_HZ == 144_390_000.0
def test_a_block_holding_more_frames_than_one_pass_takes_repeats_none():
"""The framer returns at most so many at a time, and says how far it got.
If it stopped at the limit without saying it had consumed the frames it
just returned, the caller would keep them and hand them back on the next
pass -- every packet in a busy second counted twice.
"""
raws = [ax25.frame_bytes("W1AW", "APRS", f">beacon {i}") for i in range(9)]
bits = "".join(ax25.bits_of(raw, flags=2) for raw in raws)
seen, at = [], 0
for _ in range(6):
frames, used = ax25.hdlc_frames(bits[at:], most=3)
if not frames:
break
seen += frames
at += used
assert seen == raws, f"{len(seen)} frames out of {len(raws)}"

18
tests/test_manpage.py
View file

@ -42,11 +42,13 @@ def test_every_setting_explains_itself_in_plain_words(page):
def test_the_commands_and_the_keys_are_documented(page):
for word in ("scan", "bands", "config", "transcribe", "devices",
"profiles", "analyze", "weather", "readings", "sensors"):
"profiles", "analyze", "weather", "readings", "sensors",
"aprs", "packets"):
assert f".B {word}\n" in page, f"command {word} undocumented"
for section in ("SYNOPSIS", "DESCRIPTION", "COMMANDS", "OPTIONS",
"SETTINGS", "FILES", "ENVIRONMENT", "EXAMPLES",
"AIRCRAFT OPTIONS", "WEATHER SENSORS", "WEATHER OPTIONS"):
"AIRCRAFT OPTIONS", "WEATHER SENSORS", "WEATHER OPTIONS",
"APRS", "APRS OPTIONS"):
assert f".SH {section}" in page
@ -161,8 +163,20 @@ def test_the_manual_lists_every_weather_option(page):
assert option.key in page, f"{option.key} is not named in the manual"
def test_the_manual_lists_every_aprs_option(page):
"""The same again, for the third section, from the third table."""
from bandsaunter import aprs as ap
for option in ap.OPTIONS:
flags = tuple(option.flags) + tuple(option.off_flags)
assert any(flag in page for flag in flags), \
f"{option.key} ({', '.join(flags)}) is not in the manual"
assert option.key in page, f"{option.key} is not named in the manual"
def test_the_manual_says_where_the_sensor_names_are_kept(page):
"""They are the only thing this program stores that somebody typed."""
assert "sensors.yaml" in page
assert "weather.yaml" in page
assert "weather_" in page
assert "aprs.yaml" in page and "aprs_" in page

365
tests/test_packets.py Normal file
View file

@ -0,0 +1,365 @@
"""The APRS information field: about twenty formats, and what must not happen.
Every format here has a writer beside its reader, so a position goes in and
the same position comes out. That proves the arithmetic and the framing and
proves nothing about anything the specification and this both get wrong --
which is worth saying, because the last section of this program shipped
unable to decode anything at all for exactly that reason.
The other half is about refusing to guess. A packet whose format does not
match what its first character promised comes back as unparsed, with its text
intact, rather than as a confident position a thousand miles from where the
station is.
"""
import math
import pytest
from bandsaunter import packets as p
# Every quadrant, the prime meridian, the equator, and the longitudes either
# side of the boundaries Mic-E re-maps to keep its bytes printable.
PLACES = [
(49.0583, -72.0292), (-33.8688, 151.2093), (51.5074, -0.1278),
(35.6762, 139.6503), (5.0, -5.0), (0.0, 0.0), (-45.0, -179.5),
(64.1466, -21.9426), (-1.2921, 36.8219), (47.55, -122.30),
]
# ---------------------------------------------------------------------------
# Positions
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("latitude,longitude", PLACES)
def test_an_uncompressed_position_comes_back_where_it_was_sent(latitude,
longitude):
info = p.position_report(latitude, longitude, "/>", "hello")
packet = p.parse_info(info)
assert packet.kind == "position"
assert packet.position.latitude == pytest.approx(latitude, abs=0.0002)
assert packet.position.longitude == pytest.approx(longitude, abs=0.0002)
assert packet.position.symbol == "car"
assert packet.comment == "hello"
@pytest.mark.parametrize("latitude,longitude", PLACES)
def test_a_compressed_position_comes_back_where_it_was_sent(latitude,
longitude):
packet = p.parse_info(p.compressed_report(latitude, longitude, "/>"))
assert packet.kind == "position"
assert packet.position.compressed
assert packet.position.latitude == pytest.approx(latitude, abs=0.0002)
assert packet.position.longitude == pytest.approx(longitude, abs=0.0002)
@pytest.mark.parametrize("latitude,longitude", PLACES)
def test_a_mic_e_position_comes_back_where_it_was_sent(latitude, longitude):
"""The awkward one: half of it lives in the destination callsign, and
the longitude is re-mapped in three ranges to keep the bytes printable."""
destination, info = p.mic_e_report(latitude, longitude, course=251,
speed=37.0, symbol="/j",
status="returning")
packet = p.parse_info(info, destination)
assert packet.kind == "position"
assert packet.position.latitude == pytest.approx(latitude, abs=0.0002)
assert packet.position.longitude == pytest.approx(longitude, abs=0.0002)
assert packet.course == 251
assert packet.speed == pytest.approx(37.0, abs=1.0)
assert packet.status == "returning"
assert packet.position.symbol == "jeep"
def test_the_compressed_position_from_the_specification_reads_correctly():
"""The example in the specification itself: 49 30.00 N, 72 45.00 W, with
a pre-computed radio range of 20.13 miles."""
packet = p.parse_info("!/5L!!<*e7>{?!")
assert packet.position.latitude == pytest.approx(49.5, abs=0.0001)
assert packet.position.longitude == pytest.approx(-72.75, abs=0.0001)
assert packet.range == pytest.approx(20.13 * 1.609344, rel=0.01)
@pytest.mark.parametrize("blanked,within_km", [(0, 0.0), (1, 0.34), (2, 3.4),
(3, 34.0), (4, 340.0)])
def test_a_station_that_blanks_its_minutes_is_not_drawn_as_a_point(blanked,
within_km):
"""Blanking the minute digits is a deliberate act by the operator.
Drawing a fuzzy position as a sharp one is a lie they specifically asked
not to be told, so the count is kept and turned into a distance.
"""
info = p.position_report(49.0583, -72.0292, "/-", ambiguity=blanked)
packet = p.parse_info(info)
assert packet.position.ambiguity == blanked
assert packet.position.uncertainty_km == within_km
@pytest.mark.parametrize("info", [
"=4760.37N/07201.75W-", # sixty minutes is not a minute
"=4903.50X/07201.75W-", # not a hemisphere
"=49 3.50N/07201.75W-", # a gap in the middle, not ambiguity
"=9903.50N/07201.75W-", # past the pole
"=4903.50N/19201.75W-", # past the antimeridian
])
def test_a_position_that_cannot_be_one_is_not_reported_as_a_position(info):
"""And in particular is not quietly re-read as a compressed position.
Thirteen characters of a malformed uncompressed position are perfectly
good base-91, so falling back does not fail -- it succeeds, as a
confident and completely different place.
"""
packet = p.parse_info(info)
assert packet.kind == "unparsed"
assert packet.position is None
assert packet.info == info # the text is kept regardless
def test_a_leading_digit_always_means_an_uncompressed_position():
"""Which is why the compressed format writes a numeric overlay as a
letter: so the two can never be confused by anything that reads the rule."""
overlaid = p.parse_info("!a5L!!<*e7> sT")
assert overlaid.position.table == "0" # the overlay, as a digit
assert overlaid.position.compressed
@pytest.mark.parametrize("timestamp", ["092345z", "092345/", "234500h"])
def test_a_position_with_a_timestamp_keeps_both(timestamp):
info = p.position_report(49.0583, -72.0292, "/-", timestamp=timestamp)
packet = p.parse_info(info, now=1_600_000_000.0)
assert packet.position is not None
assert packet.reported > 0.0
def test_a_timestamp_with_no_year_lands_on_the_nearest_side_of_today():
"""None of the four formats carries a year, so a packet stamped the
thirty-first heard on the first is from yesterday, not four weeks on."""
import calendar
now = calendar.timegm((2026, 3, 1, 0, 30, 0, 0, 0, 0))
when, _rest = p.timestamp_from("282345z", now)
assert now - when < 3 * 24 * 3600 # late February, not next year
assert when < now
# ---------------------------------------------------------------------------
# What rides in the comment
# ---------------------------------------------------------------------------
def test_a_course_and_speed_are_read_in_the_units_they_are_sent_in():
"""Knots here, which is not the same as the miles an hour a weather
report uses for wind, and the two differ by fifteen per cent."""
packet = p.parse_info(p.position_report(49.0, -72.0, "/>", course=88,
speed=66.672))
assert packet.course == 88
assert packet.speed == pytest.approx(66.672, abs=1.0)
def test_an_antenna_description_is_read_rather_than_shown_as_letters():
extra, rest = p.extensions_from("PHG5132Hi")
assert extra["power"] == 25 # watts, from a single digit
assert extra["height"] == pytest.approx(10.0 * 2 ** 1 * 0.3048, rel=0.01)
assert extra["gain"] == 3 and extra["beam"] == "E"
assert rest == "Hi"
def test_a_pre_computed_range_is_read():
extra, rest = p.extensions_from("RNG0050here")
assert extra["range"] == pytest.approx(50 * 1.609344, rel=0.01)
assert rest == "here"
def test_a_comment_that_is_only_a_comment_is_left_alone():
extra, rest = p.extensions_from("Hello there")
assert extra == {} and rest == "Hello there"
def test_an_altitude_is_pulled_out_of_wherever_in_the_comment_it_sits():
extra, rest = p.comment_from("climbing /A=012345 steadily")
assert extra["altitude"] == pytest.approx(12345 * 0.3048, rel=0.001)
assert "A=" not in rest and "climbing" in rest and "steadily" in rest
def test_an_altitude_is_shown_in_metres_and_not_rounded_to_a_kilometre():
"""Four hundred metres and the ground both read as "0 km", which is the
sort of rounding that makes a number worse than no number."""
assert p.height_text(376.0) == "376 m"
assert p.height_text(376.0, imperial=True) == "1,234 ft"
# ---------------------------------------------------------------------------
# Weather
# ---------------------------------------------------------------------------
def test_a_weather_report_with_a_position_reads_both():
info = "@092345z4903.50N/07201.75W_220/004g005t077r000p000P000h50b09900"
packet = p.parse_info(info)
assert packet.kind == "weather"
assert packet.position is not None
wx = packet.weather
assert wx["temperature"].value == pytest.approx(25.0, abs=0.1)
assert wx["humidity"].value == 50
assert wx["pressure"].value == pytest.approx(990.0, abs=0.1)
assert wx["wind from"].value == 220
assert wx["wind"].value == pytest.approx(4 * 1.852, abs=0.1)
def test_a_positionless_weather_report_reads_the_numbers():
packet = p.parse_info("_10090556c220s004g005t077r000p000P000h50b09900")
assert packet.kind == "weather" and packet.position is None
assert packet.weather["wind"].value == pytest.approx(4 * 1.609344, abs=0.1)
def test_the_letter_s_means_wind_speed_or_snow_depending_on_the_form():
"""The same letter, and the two differ by a factor of forty.
In a positionless report the wind arrives as "c" then "s"; in a position
report it arrives in the course-and-speed field, so a later "s" is
snowfall. Guessing is not an option.
"""
positionless = p.parse_info("_10090556c220s004g005t077")
assert "wind" in positionless.weather and "snow" not in positionless.weather
with_position = p.parse_info(
"@092345z4903.50N/07201.75W_220/004g005t077s010")
assert with_position.weather["snow"].value == pytest.approx(254.0, abs=1)
assert with_position.weather["wind"].value == pytest.approx(4 * 1.852,
abs=0.1)
def test_a_humidity_of_zero_means_a_hundred_per_cent():
packet = p.parse_info("_10090556c220s004g005t077h00")
assert packet.weather["humidity"].value == 100
def test_a_weather_report_comes_back_as_it_was_written():
info = p.weather_report(49.0583, -72.0292, timestamp="092345z",
wind_from=220, wind=7.4, gust=18.0,
temperature=25.0, humidity=50, pressure=990.0)
packet = p.parse_info(info)
assert packet.weather["temperature"].value == pytest.approx(25.0, abs=0.3)
assert packet.weather["humidity"].value == 50
assert packet.weather["pressure"].value == pytest.approx(990.0, abs=0.1)
# ---------------------------------------------------------------------------
# Messages, objects, items, status, telemetry
# ---------------------------------------------------------------------------
def test_a_message_reads_with_its_addressee_and_its_number():
packet = p.parse_info(p.message_text("KU0W", "on my way", number="42"))
assert packet.kind == "message"
assert packet.message.to == "KU0W"
assert packet.message.text == "on my way"
assert packet.message.number == "42"
@pytest.mark.parametrize("body,field,value", [
("ack42", "acknowledges", "42"),
("rej42", "rejects", "42"),
])
def test_an_acknowledgement_is_not_read_as_a_message_saying_ack(body, field,
value):
packet = p.parse_info(":W1AW :" + body)
assert getattr(packet.message, field) == value
assert packet.message.text == ""
def test_a_bulletin_is_addressed_to_everybody_and_says_so():
packet = p.parse_info(":BLN1 :Net tonight at eight")
assert packet.message.is_bulletin and packet.message.bulletin == "1"
assert "Net tonight" in packet.message.describe()
def test_an_object_carries_a_name_that_is_not_the_station_that_placed_it():
info = p.object_report("LEADER", 49.0583, -72.0292, "/>",
timestamp="092345z")
packet = p.parse_info(info)
assert packet.kind == "object"
assert packet.name == "LEADER"
assert packet.station == "LEADER" # the object, not the sender
assert packet.position.latitude == pytest.approx(49.0583, abs=0.0002)
assert packet.live
def test_an_object_can_be_killed():
info = p.object_report("LEADER", 49.0583, -72.0292, live=False)
assert p.parse_info(info).live is False
def test_an_item_is_an_object_without_a_timestamp():
packet = p.parse_info(p.item_report("AID", 49.0583, -72.0292))
assert packet.kind == "item" and packet.name == "AID"
assert packet.position is not None
def test_a_status_report_is_kept_as_what_the_operator_typed():
packet = p.parse_info(">Monitoring 146.52")
assert packet.kind == "status" and packet.status == "Monitoring 146.52"
def test_telemetry_reads_as_five_channels_and_eight_bits():
packet = p.parse_info(p.telemetry_report("005", [199, 0, 255, 73, 123],
"01101001"))
assert packet.kind == "telemetry"
assert packet.telemetry.analogue == (199.0, 0.0, 255.0, 73.0, 123.0)
assert packet.telemetry.digital == "01101001"
def test_a_relayed_packet_is_credited_to_whoever_originally_sent_it():
inner = p.position_report(49.0583, -72.0292, "/>")
packet = p.parse_info("}W1AW-5>APRS,TCPIP*:" + inner)
assert packet.source == "W1AW-5"
assert packet.position is not None
assert "relayed" in packet.comment
# ---------------------------------------------------------------------------
# Refusing to guess
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("info", [
"", "x", "!", "=short", ":notamessage", ";tooshort", "T#", ">",
"@notatimestamp", "}", "_", "`", "'",
])
def test_a_packet_that_cannot_be_read_keeps_its_text_and_says_so(info):
packet = p.parse_info(info)
assert packet.position is None
assert packet.info == info
assert packet.kind in p.KINDS
def test_nothing_is_read_out_of_random_text():
import random
rng = random.Random(4)
alphabet = "".join(chr(c) for c in range(32, 127))
placed = 0
for _ in range(4000):
info = "".join(rng.choice(alphabet) for _ in range(rng.randint(5, 60)))
packet = p.parse_info(info)
if packet.position is not None:
placed += 1
# Some random text really is a valid compressed position -- thirteen
# printable characters is all one takes -- so the bar is a rate. What
# reaches this off the air has also had to satisfy a sixteen-bit CRC.
assert placed <= 120, f"{placed} of 4000 random strings became a place"
# ---------------------------------------------------------------------------
# Symbols
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("table,code,name", [
("/", ">", "car"), ("/", "_", "weather station"), ("/", "#", "digipeater"),
("/", "O", "balloon"), ("\\", "s", "boat"), ("/", "[", "person"),
])
def test_a_station_is_described_by_what_it_draws_itself_as(table, code, name):
assert p.symbol_name(table, code) == name
def test_an_overlaid_symbol_says_which_character_is_over_it():
assert "(T)" in p.symbol_name("T", "#")
def test_a_symbol_nobody_named_is_reported_by_its_characters():
assert p.symbol_name("/", "\x01") == "symbol /\x01"