"""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)}"