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

Read the pictures, the aircraft, the meters and the sensors

Browse source 
Hexadecimal is a true answer to "what did that say" and not a useful one.
This is the work of turning the rest of what a receiver hears into
something a person can read, and most of it is pictures.

PICTURES

Three of the things on the air are images rather than sounds, and all three
arrive as the audio a scan already records:

  SSTV     14.230 and 144.5 MHz   Martin M1/M2, Scottie S1/S2/DX, Robot 36/72
  APT      137-138 MHz            the NOAA weather satellites
  HF fax   2-20 MHz, sideband     the marine weather charts

Each is written from its published specification, and the generators used to
test them are written from the same specification without reference to the
decoders -- so a picture that comes back matching the one that went in is
evidence about the format.  Every SSTV mode reproduces its published line
time exactly, which is worth failing a test over: a line a few milliseconds
long walks the picture off the screen inside ten lines.  Against synthetic
transmissions at 30 dB SNR, SSTV is 96-98% of pixels exact, APT correlates
at 0.97 and fax at 0.998; all three still read at 6-12 dB.

None of the three is guessed at, and that is what makes it safe to try them
on every recording.  SSTV needs its VIS header, APT needs both line syncs at
the right distance from each other, fax needs the phasing signal.  No false
pictures in 295 attempts over noise, tones, speech and swept whistles.

Two things had to be got right beyond the arithmetic.  A band-pass does not
switch between two tones, it slides between them, so every edge is measured
at the midpoint of the slide rather than at the first sample past a
threshold -- the earlier version was reading the coarse search stride back
as the edge and shifting Martin M1 sideways by a whole colour bar.  And a
picture now keeps its capture whatever the content check made of it: a
satellite is a steady tone with a wobble on it and SSTV is a whistle, so
both were being discarded as "no signal content" having already been
recognised.

PNG is written here rather than pulled in from Pillow.  A scanner that
cannot start because an imaging library is missing is worse than one that
cannot draw.

saunterbrowse marks a picture in the list, gives its path in full -- wrapped
rather than cut off, because half a path opens nothing -- and moves or
deletes the PNGs with the recording.  o prints the picture's path, not the
audio's.

GRIB is not a modulation and is not pretended to be one.  It is the format
weather models are published in and it travels by satellite link and by
e-mail; where a decoded byte stream begins with its magic number it is
named, and that is all.

AIRCRAFT

`bandsaunter adsb` parks the receiver on 1090 MHz and reads Mode S extended
squitter: address, callsign, altitude, position, speed.  A command of its
own because a megabit a second will not go through a channel twelve and a
half kilohertz wide.  Every frame carries a 24-bit checksum so there is no
threshold anywhere in it -- with one trap, which is that a frame of all
zeros satisfies that checksum and silence is exactly that.  Positions round
trip exactly through compact position reporting, and a pair straddling a
longitude-zone boundary is refused rather than resolved against two grids.

METERS AND SENSORS

Itron ERT utility meters on 900 MHz and AcuRite weather sensors on 433 MHz
are named rather than reported as hex, and neither is believed without its
own checksum -- BCH(255,239) for the meter, a checksum and four parity bits
for the sensor.  Both are implemented from published descriptions and
checked against frames built from the same descriptions, which proves the
framing and the arithmetic and is not the same as having held a meter.

HEX INTO WORDS

Everything else that decodes to bits now gets its fields named where the
shape is standard, its text read out where there is text, and its bytes laid
out in groups with the printable characters beside them.

The text search is where the care went, because printability is not
evidence.  Forty framings of each packet, and seven-bit values printable
three in four, meant a bar set on printability called 64% of random payloads
text.  Real text is nearly all one case where random letters are half and
half, two fifths vowels where random is a fifth, and mostly alphanumeric
where random draws punctuation one time in four.  Together: under 0.5%,
measured in the suite.

CALLSIGNS

The licensed address is recorded in full -- the street, not merely the town
-- and goes into the KML with everything else.  US amateur records are
public by law and carry it; holding it and not saying so is worse than
either showing it or not asking, and --no-lookup asks for none of it.

Also here: Morse is decoded again from the whole recording where the capture
was made in cw mode.  The first pass works from the classifier's buffer,
which holds a few seconds -- enough to say "this is Morse", not enough to
catch a callsign whole between two word gaps, so a beacon repeating every
eight seconds through an eight-second window was never identified.

And classify._psk_order took the logarithm of zero on a silent block.

1318 tests, up from 1161.

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-08-29 19:38:37 -07:00
parent b4718aa425
commit 3d7f76118e
32 changed files with 4426 additions and 39 deletions
Download patch file Download diff file Expand all files Collapse all files

118
tests/adsb_gen.py Normal file
View file

@ -0,0 +1,118 @@
"""Generate Mode S extended squitter frames, so the ADS-B decoder can be tested.
Built from the standard rather than from the decoder: the parity is computed
here with the same polynomial but the frames are assembled independently, and
a decoder that reads back what was put in has read the format rather than
agreed with itself.
"""
import math
import numpy as np
from bandsaunter.adsb import CALLSIGN_CHARS, PREAMBLE_US, crc24
def with_parity(payload: bytes) -> bytes:
"""A frame with its 24 parity bits appended, as a transmitter sends it."""
return payload + crc24(payload).to_bytes(3, "big")
def identification(icao: int, callsign: str, category: int = 0) -> bytes:
"""A DF17 type-4 frame: the aircraft saying what it is called."""
me = bytearray(7)
me[0] = (4 << 3) | (category & 0x07)
text = callsign.upper().ljust(8)[:8]
bits = ""
for ch in text:
index = CALLSIGN_CHARS.find(ch)
bits += format(index if index >= 0 else 32, "06b")
packed = int(bits, 2).to_bytes(6, "big")
me[1:7] = packed
return with_parity(bytes([17 << 3 | 5]) + icao.to_bytes(3, "big")
+ bytes(me))
def _cpr(lat: float, lon: float, odd: bool) -> tuple[int, int]:
"""Compact position reporting, the encoding side."""
def nl(latitude):
if abs(latitude) >= 87.0:
return 1
if latitude == 0:
return 59
inner = 1 - (1 - math.cos(math.pi / 30)) / \
math.cos(math.radians(abs(latitude))) ** 2
return int(math.floor(2 * math.pi / math.acos(max(-1.0, min(1.0, inner)))))
i = 1 if odd else 0
d_lat = 360.0 / (60 - i)
j = math.floor(lat / d_lat) + math.floor(
0.5 + (lat % d_lat) / d_lat)
y = int(round(131072 * ((lat % d_lat) / d_lat)))
zones = nl(lat) - i
d_lon = 360.0 / zones if zones > 0 else 360.0
x = int(round(131072 * ((lon % d_lon) / d_lon)))
del j
return y & 0x1FFFF, x & 0x1FFFF
def airborne_position(icao: int, lat: float, lon: float, altitude_ft: int,
odd: bool) -> bytes:
"""A DF17 type-11 frame: where the aircraft is and how high."""
encoded = int(round((altitude_ft + 1000) / 25.0))
field = format(encoded, "011b")
alt = field[:7] + "1" + field[7:] # the Q bit, 25-foot steps
y, x = _cpr(lat, lon, odd)
me_bits = (format(11, "05b") + "000" + alt + "0"
+ ("1" if odd else "0") + format(y, "017b")
+ format(x, "017b"))
me = int(me_bits, 2).to_bytes(7, "big")
return with_parity(bytes([17 << 3 | 5]) + icao.to_bytes(3, "big") + me)
def velocity(icao: int, east_kt: int, north_kt: int,
vertical_fpm: int = 0) -> bytes:
"""A DF17 type-19 frame: ground speed and climb rate."""
ew_sign = "1" if east_kt < 0 else "0"
ns_sign = "1" if north_kt < 0 else "0"
ew = min(1023, abs(east_kt) + 1)
ns = min(1023, abs(north_kt) + 1)
rate = min(511, abs(vertical_fpm) // 64 + 1) if vertical_fpm else 0
me_bits = (format(19, "05b") + "001" + "00000"
+ ew_sign + format(ew, "010b")
+ ns_sign + format(ns, "010b")
+ "0" + ("1" if vertical_fpm < 0 else "0")
+ format(rate, "09b") + "0" * 10)
me = int(me_bits[:56].ljust(56, "0"), 2).to_bytes(7, "big")
return with_parity(bytes([17 << 3 | 5]) + icao.to_bytes(3, "big") + me)
def modulate(frames, sample_rate: float = 2_000_000.0, gap_us: float = 60.0,
amplitude: float = 1.0, noise: float = 0.0,
seed: int = 0) -> np.ndarray:
"""Turn frames into the magnitude a receiver would see at 1090 MHz.
Pulse-position: a preamble of four pulses, then one microsecond per bit
with the energy in the first half for a one and the second half for a
zero.
"""
rng = np.random.default_rng(seed)
per_us = sample_rate / 1e6
out = [np.zeros(int(round(gap_us * per_us)))]
for frame in frames:
bits = "".join(format(b, "08b") for b in frame)
span = np.zeros(int(round((8 + len(bits)) * per_us)))
for at in PREAMBLE_US:
lo = int(round(at * per_us))
span[lo:lo + int(round(0.5 * per_us))] = amplitude
for i, bit in enumerate(bits):
base = (8 + i) * per_us
lo = int(round(base if bit == "1" else base + 0.5 * per_us))
span[lo:lo + int(round(0.5 * per_us))] = amplitude
out.append(span)
out.append(np.zeros(int(round(gap_us * per_us))))
signal = np.concatenate(out)
if noise:
signal = signal + noise * np.abs(
rng.standard_normal(signal.size)
+ 1j * rng.standard_normal(signal.size)) / math.sqrt(2)
return signal.astype(np.complex64)

198
tests/image_gen.py Normal file
View file

@ -0,0 +1,198 @@
"""Generate SSTV, APT and HF fax audio, so the decoders can be tested.
Written from the mode specifications rather than from the decoders, so that a
decoder agreeing with one of these is evidence rather than a tautology.
"""
import numpy as np
from bandsaunter.sstv import (BIT_ONE_HZ, BIT_ZERO_HZ, BLACK_HZ, BREAK_HZ,
LEADER_HZ, SYNC_HZ, WHITE_HZ, SSTVMode)
def fm(segments, rate: float) -> np.ndarray:
"""Turn ``(hertz, seconds)`` pairs into one continuous-phase tone."""
phase = 0.0
out = []
for hz, seconds in segments:
n = max(1, int(round(seconds * rate)))
freq = np.full(n, float(hz)) if np.isscalar(hz) else np.asarray(hz)
if freq.size != n:
freq = np.interp(np.linspace(0, 1, n),
np.linspace(0, 1, freq.size), freq)
step = 2.0 * np.pi * freq / rate
angles = phase + np.cumsum(step)
phase = float(angles[-1] % (2.0 * np.pi))
out.append(np.sin(angles))
return np.concatenate(out) if out else np.zeros(0)
def _levels(row: np.ndarray) -> np.ndarray:
"""A row of 0-255 as the frequencies that carry it."""
return BLACK_HZ + np.asarray(row, dtype=np.float64) / 255.0 * (
WHITE_HZ - BLACK_HZ)
def vis_header(code: int) -> list:
bits = [(code >> i) & 1 for i in range(7)]
bits.append(sum(bits) % 2) # even parity
out = [(LEADER_HZ, 0.300), (BREAK_HZ, 0.010), (LEADER_HZ, 0.300),
(SYNC_HZ, 0.030)]
out += [(BIT_ONE_HZ if b else BIT_ZERO_HZ, 0.030) for b in bits]
out.append((SYNC_HZ, 0.030)) # stop bit
return out
def sstv_audio(mode: SSTVMode, image: np.ndarray, rate: float = 16000.0,
lines: int | None = None, lead: float = 0.2,
snr_db: float = 40.0, seed: int = 0) -> np.ndarray:
"""A whole SSTV transmission: header, then the picture, line by line.
Built from the mode's own segment list, so that a decoder agreeing with
this is agreeing about the published line structure rather than about a
second copy of the same guess.
"""
rng = np.random.default_rng(seed)
height = lines if lines is not None else min(mode.height, image.shape[0])
segments = [(0.0, lead)] + vis_header(mode.vis)
if not mode.sync_first:
# Scottie sends one sync pulse before the first line, because its
# per-line pulse comes two thirds of the way through.
segments.append((SYNC_HZ, mode.sync))
for y in range(height):
row = image[y % image.shape[0]].astype(np.float64)
red, green, blue = row[:, 0], row[:, 1], row[:, 2]
luma = 0.299 * red + 0.587 * green + 0.114 * blue
chroma_u = 0.564 * (blue - luma) + 128.0
chroma_v = 0.713 * (red - luma) + 128.0
scans = {"R": red, "G": green, "B": blue, "Y": luma,
"U": chroma_u, "V": chroma_v,
"C": chroma_v if y % 2 == 0 else chroma_u}
for what, seconds in mode.segments():
if seconds <= 0:
continue
if what == "sync":
segments.append((SYNC_HZ, seconds))
elif len(what) == 1:
segments.append((_levels(scans[what]), seconds))
else:
segments.append((BLACK_HZ, seconds))
audio = fm(segments, rate)
if snr_db < 60:
noise = np.sqrt(np.mean(audio ** 2) / (10 ** (snr_db / 10.0)))
audio = audio + noise * rng.standard_normal(audio.size)
return audio
def colour_card(width: int = 320, height: int = 256) -> np.ndarray:
"""A picture with structure a decoder can be checked against.
Colour bars across the top, a grey wedge down the middle and a border, so
that a picture decoded with the colours swapped, the lines out of order or
the geometry wrong looks wrong in a way a test can measure.
"""
image = np.zeros((height, width, 3), dtype=np.uint8)
bars = [(255, 255, 255), (255, 255, 0), (0, 255, 255), (0, 255, 0),
(255, 0, 255), (255, 0, 0), (0, 0, 255), (0, 0, 0)]
band = height // 2
for i, colour in enumerate(bars):
lo = i * width // len(bars)
hi = (i + 1) * width // len(bars)
image[:band, lo:hi] = colour
wedge = np.linspace(0, 255, width).astype(np.uint8)
image[band:, :] = wedge[None, :, None]
image[:, :2] = 255
image[:, -2:] = 255
return image
# ---------------------------------------------------------------------------
# APT
# ---------------------------------------------------------------------------
APT_SYNC_A = "0" + "1100" * 7 + "0" * 8 # 1040 Hz square, 39 words
APT_SYNC_B = "0" + "11100" * 7 + "0" * 3
def apt_audio(image: np.ndarray, rate: float = 16000.0,
snr_db: float = 40.0, seed: int = 0,
subcarrier: float = 2400.0) -> np.ndarray:
"""A NOAA APT transmission: two lines a second, 2080 words each.
``image`` is greyscale and becomes channel A; channel B is the same
picture inverted, which is what a real pass looks like when one channel is
visible and the other infrared.
"""
rng = np.random.default_rng(seed)
words_per_line = 2080
lines = image.shape[0]
frame = np.zeros((lines, words_per_line), dtype=np.float64)
for y in range(lines):
row = np.interp(np.linspace(0, 1, 909),
np.linspace(0, 1, image.shape[1]),
image[y].astype(np.float64))
line = np.zeros(words_per_line)
line[0:39] = np.array([255.0 if c == "1" else 11.0
for c in APT_SYNC_A.ljust(39, "0")])
line[39:86] = 11.0 # space A
line[86:995] = row
line[995:1040] = 128.0 # telemetry A
line[1040:1079] = np.array([255.0 if c == "1" else 11.0
for c in APT_SYNC_B.ljust(39, "0")])
line[1079:1126] = 11.0
line[1126:2035] = 255.0 - row
line[2035:2080] = 128.0
frame[y] = line
words = frame.reshape(-1)
word_rate = 4160.0
n = int(round(words.size / word_rate * rate))
envelope = np.interp(np.linspace(0, 1, n), np.linspace(0, 1, words.size),
words) / 255.0
t = np.arange(n) / rate
audio = (0.1 + 0.9 * envelope) * np.sin(2 * np.pi * subcarrier * t)
if snr_db < 60:
noise = np.sqrt(np.mean(audio ** 2) / (10 ** (snr_db / 10.0)))
audio = audio + noise * rng.standard_normal(audio.size)
return audio
def grey_card(width: int = 909, height: int = 40) -> np.ndarray:
"""A greyscale picture with a hard edge and a ramp, for APT and fax."""
image = np.zeros((height, width), dtype=np.uint8)
image[:, :] = np.linspace(0, 255, width).astype(np.uint8)[None, :]
image[height // 3:2 * height // 3, width // 4:width // 2] = 255
image[:, ::128] = 0
return image
# ---------------------------------------------------------------------------
# HF fax
# ---------------------------------------------------------------------------
def fax_audio(image: np.ndarray, rate: float = 16000.0, lpm: float = 120.0,
ioc: int = 576, black: float = 1500.0, white: float = 2300.0,
start_seconds: float = 5.0, phasing_lines: int = 20,
snr_db: float = 40.0, seed: int = 0) -> np.ndarray:
"""A weather fax transmission: start tone, phasing, then the chart."""
rng = np.random.default_rng(seed)
pixels = int(round(np.pi * ioc))
line_seconds = 60.0 / lpm
segments = [(black, 0.5)]
# The start tone: 300 Hz for 120 lpm, sent as black/white alternation.
tone_period = 1.0 / 300.0
for _ in range(int(start_seconds / tone_period)):
segments += [(white, tone_period / 2), (black, tone_period / 2)]
# Phasing: a white pulse at the start of each line, black for the rest.
for _ in range(phasing_lines):
segments += [(white, line_seconds * 0.05),
(black, line_seconds * 0.95)]
for y in range(image.shape[0]):
row = np.interp(np.linspace(0, 1, pixels),
np.linspace(0, 1, image.shape[1]),
image[y].astype(np.float64))
segments.append((black + row / 255.0 * (white - black), line_seconds))
audio = fm(segments, rate)
if snr_db < 60:
noise = np.sqrt(np.mean(audio ** 2) / (10 ** (snr_db / 10.0)))
audio = audio + noise * rng.standard_normal(audio.size)
return audio

56
tests/test_callsign.py
View file

@ -515,3 +515,59 @@ def test_an_unlisted_callsign_is_cached_too(tmp_path, monkeypatch):
CallsignBook._apply(entry, {"status": "INVALID"})
assert entry.status == "unlisted"
assert entry.version == CACHE_VERSION
# -- the whole address -------------------------------------------------------
#
# US amateur licence records are public by law and carry the street the
# licence was issued to. It is recorded because holding it and not saying so
# would be worse than either showing it or not asking for it, and --no-lookup
# asks for none of it.
def test_the_street_is_read_off_the_licence(tmp_path):
b = book(tmp_path, {"W1AW": {
"status": "VALID", "current": {"callsign": "W1AW"},
"name": "ARRL HQ OPERATORS CLUB",
"address": {"line1": "225 MAIN ST", "line2": "NEWINGTON, CT 06111"},
"location": {"latitude": "41.714", "longitude": "-72.727"}}})
entry = b.get("W1AW")
b.wait(5.0)
assert entry.street == "225 Main St"
assert entry.location == "Newington, CT"
assert entry.postcode == "06111"
assert entry.address == "225 Main St, Newington, CT, 06111"
def test_the_street_is_shown_under_the_callsign(tmp_path):
b = book(tmp_path, {"W1AW": {
"status": "VALID", "current": {"callsign": "W1AW", "operClass": "CLUB"},
"name": "ARRL", "address": {"line1": "225 MAIN ST",
"line2": "NEWINGTON, CT 06111"},
"location": {"gridsquare": "FN31pr"}}})
entry = b.get("W1AW")
b.wait(5.0)
assert "225 Main St" in entry.details()
def test_an_address_with_nothing_in_it_is_an_empty_string(tmp_path):
b = book(tmp_path, {"W1AW": {"status": "INVALID"}})
entry = b.get("W1AW")
b.wait(5.0)
assert entry.address == ""
def test_a_record_cached_before_the_street_is_asked_about_again(tmp_path):
"""The cache is versioned so an old entry is not kept with a field empty."""
import json
import time as _time
cache = tmp_path / "calls.json"
cache.write_text(json.dumps({"W1AW": {
"call": "W1AW", "name": "ARRL", "status": "found",
"fetched_at": _time.time(), "version": 2}}))
b = StubBook({"W1AW": {
"status": "VALID", "current": {"callsign": "W1AW"}, "name": "ARRL",
"address": {"line1": "225 MAIN ST", "line2": "NEWINGTON, CT 06111"},
"location": {}}}, cache=cache)
entry = b.get("W1AW")
b.wait(5.0)
assert entry.street == "225 Main St"

15
tests/test_ident.py
View file

@ -55,7 +55,8 @@ class StubBook(CallsignBook):
def _scanner(tmp_path, mhz: str, **over) -> Scanner:
cfg = ScanConfig(ranges=parse_range_list(mhz),
output_dir=str(tmp_path), transcribe=False,
record_seconds=10, hang_seconds=1.5,
record_seconds=over.pop("record_seconds", 10),
hang_seconds=1.5,
max_runtime_seconds=over.pop("seconds", 45),
**over)
scanner = Scanner(cfg, device=SimulatedDevice(realtime=False).open())
@ -101,9 +102,14 @@ def test_a_beacon_is_identified_from_the_words_that_survived(tmp_path):
"""A continuous beacon is always caught partway through.
Every capture of one begins and ends in the middle of the message, so
what can be said about it is whatever lies between two word gaps.
what can be said about it is whatever lies between two word gaps -- which
means the capture has to be longer than one repeat before any word is
certain to be bounded on both sides. This beacon repeats every eight
seconds; a ten-second capture of it is identifiable only by luck, and a
twenty-second one always.
"""
scanner = _scanner(tmp_path, "144.05M-144.15M", seconds=30)
scanner = _scanner(tmp_path, "144.05M-144.15M", seconds=50,
record_seconds=20)
scanner.run()
assert "W1AW" in scanner.heard
assert set(scanner.heard) == {"W1AW"}, \
@ -112,7 +118,8 @@ def test_a_beacon_is_identified_from_the_words_that_survived(tmp_path):
def test_the_hit_keeps_both_the_text_and_the_part_it_can_be_identified_from(
tmp_path):
scanner = _scanner(tmp_path, "144.05M-144.15M", seconds=30)
scanner = _scanner(tmp_path, "144.05M-144.15M", seconds=30,
record_seconds=20)
scanner.run()
cw = [h for h in scanner.hits if h.morse_text]
assert cw

493
tests/test_images.py Normal file
View file

@ -0,0 +1,493 @@
"""Pictures off the air: SSTV, weather satellites and shortwave fax.
Every one of these decoders is written against a published specification, and
the generators in ``image_gen.py`` are written against the same one without
reference to the decoders. So a picture that comes back matching the one
that went in is evidence about the format rather than a decoder agreeing with
itself.
Two things are measured throughout. How close the picture is -- as the share
of pixels within a sixth of full scale, because a filter softens every edge
and an exact match is not a thing analogue television does. And how often a
decoder draws a picture from something that is not one, which has to be never:
a scanner that fills a directory with beautifully rendered static is worse
than one that draws nothing.
"""
import struct
import zlib
import numpy as np
import pytest
from bandsaunter.apt import decode_apt, CHANNEL_A, WORDS_PER_LINE
from bandsaunter.fax import decode_fax, find_phasing, _levels
from bandsaunter.images import (ImageDecode, PNG_SIGNATURE, instantaneous_frequency,
resample_to, write_png)
from bandsaunter.pictures import find_image
from bandsaunter.sstv import MODES, VIS_CODES, decode_sstv, find_vis
from image_gen import (apt_audio, fax_audio, grey_card, sstv_audio, colour_card)
FS = 16000.0
def close(got: np.ndarray, want: np.ndarray, within: int = 40) -> float:
"""The share of pixels that agree to within ``within`` of 255."""
got = np.asarray(got, dtype=float)
want = np.asarray(want, dtype=float)[:got.shape[0]]
return float((np.abs(got - want[:got.shape[0]]) < within).mean())
def correlation(got: np.ndarray, want: np.ndarray) -> float:
a = np.asarray(got, dtype=float)
b = np.asarray(want, dtype=float)[:a.shape[0], :a.shape[1]]
a = (a - a.mean()) / (a.std() or 1.0)
b = (b - b.mean()) / (b.std() or 1.0)
return float((a * b).mean())
# ---------------------------------------------------------------------------
# PNG
# ---------------------------------------------------------------------------
def _read_png(path):
"""Read back a PNG this module wrote, so a test can check the pixels.
Only what write_png produces: eight bits, no interlacing, filter type
zero on every row. A general reader would be a second implementation to
get wrong.
"""
raw = path.read_bytes()
assert raw[:8] == PNG_SIGNATURE
at, chunks = 8, {}
idat = b""
while at < len(raw):
length = struct.unpack(">I", raw[at:at + 4])[0]
tag = raw[at + 4:at + 8]
body = raw[at + 8:at + 8 + length]
if tag == b"IDAT":
idat += body
else:
chunks[tag] = body
at += 12 + length
width, height, depth, colour = struct.unpack(">IIBB", chunks[b"IHDR"][:10])
assert depth == 8
per = 3 if colour == 2 else 1
data = zlib.decompress(idat)
stride = width * per + 1
rows = []
for y in range(height):
line = data[y * stride:(y + 1) * stride]
assert line[0] == 0, "only the unfiltered form is written"
rows.append(np.frombuffer(line[1:], dtype=np.uint8))
out = np.stack(rows)
return out.reshape(height, width, 3) if per == 3 else out
def test_a_grey_png_round_trips(tmp_path):
want = (np.arange(64 * 40).reshape(40, 64) % 256).astype(np.uint8)
write_png(tmp_path / "g.png", want)
assert np.array_equal(_read_png(tmp_path / "g.png"), want)
def test_a_colour_png_round_trips(tmp_path):
want = colour_card(64, 32)
write_png(tmp_path / "c.png", want)
assert np.array_equal(_read_png(tmp_path / "c.png"), want)
def test_a_png_is_written_whole_or_not_at_all(tmp_path):
"""Atomic, so an interrupted scan cannot leave half a picture."""
path = tmp_path / "p.png"
write_png(path, np.zeros((4, 4), dtype=np.uint8))
assert path.exists()
assert not list(tmp_path.glob("*.tmp"))
def test_an_array_that_is_not_a_picture_is_refused(tmp_path):
with pytest.raises(ValueError):
write_png(tmp_path / "x.png", np.zeros((4, 4, 2)))
# ---------------------------------------------------------------------------
# SSTV
# ---------------------------------------------------------------------------
def test_every_mode_adds_up_to_its_published_line_time():
"""The number the whole decode hangs on, checked rather than believed.
A line time a few milliseconds out walks the picture off the bottom of
the screen; these are the published figures for each mode.
"""
for mode in MODES:
assert mode.line_seconds * 1000 == pytest.approx(mode.line_ms,
abs=0.001), mode.name
def test_the_vis_codes_are_the_assigned_ones():
assert VIS_CODES[44].name == "Martin M1"
assert VIS_CODES[60].name == "Scottie S1"
assert VIS_CODES[8].name == "Robot 36"
@pytest.mark.parametrize("mode", MODES, ids=lambda m: m.name)
def test_a_transmission_comes_back_as_the_picture_that_was_sent(mode):
card = colour_card()
got = decode_sstv(sstv_audio(mode, card, FS, lines=20, snr_db=30), FS)
assert got is not None and got.ok, got and got.note
assert got.mode == mode.name
assert got.width == mode.width
assert close(got.pixels, card) > 0.88, close(got.pixels, card)
@pytest.mark.parametrize("snr", [30, 20, 12, 9])
def test_it_still_reads_through_noise(snr):
mode = MODES[0]
card = colour_card()
got = decode_sstv(sstv_audio(mode, card, FS, lines=16, snr_db=snr), FS)
assert got is not None and got.ok, f"{snr} dB: {got and got.note}"
assert close(got.pixels, card) > 0.85
def test_a_capture_that_ends_partway_keeps_what_arrived():
"""Two minutes is longer than most captures, so partial is the normal case."""
mode = MODES[0]
got = decode_sstv(sstv_audio(mode, colour_card(), FS, lines=20), FS)
assert got.ok and not got.complete
assert got.height == 20
assert "of 256 lines" in got.note
def test_the_mode_is_read_from_the_header_not_guessed():
"""A picture decoded as the wrong mode is a picture and is wrong."""
card = colour_card()
for mode in (MODES[0], MODES[3], MODES[6]):
got = decode_sstv(sstv_audio(mode, card, FS, lines=12), FS)
assert got.mode == mode.name
@pytest.mark.parametrize("seed", range(8))
def test_nothing_that_is_not_sstv_becomes_a_picture(seed):
rng = np.random.default_rng(seed)
t = np.arange(int(FS * 4)) / FS
for signal in (rng.standard_normal(t.size),
np.sin(2 * np.pi * 1900 * t),
np.sin(2 * np.pi * 1500 * t) + 0.4 * rng.standard_normal(t.size),
np.sin(2 * np.pi * (1700 + 400 * np.sin(2 * np.pi * 3 * t)) * t)):
got = decode_sstv(signal, FS)
assert got is None or not got.ok
def test_a_header_for_a_mode_this_does_not_know_says_so():
from image_gen import fm, vis_header
audio = fm([(0.0, 0.2)] + vis_header(2) + [(1500.0, 2.0)], FS)
got = decode_sstv(audio, FS)
assert got is not None and not got.ok
assert "not decoded here" in got.note
def test_the_header_is_found_where_it_actually_is():
mode = MODES[0]
audio = sstv_audio(mode, colour_card(), FS, lines=4, lead=0.5)
freq = instantaneous_frequency(audio, FS, low=900.0, high=2600.0)
found = find_vis(freq, FS)
assert found is not None
code, start = found
assert code == mode.vis
# lead + leader + break + leader + start bit + 8 bits + stop bit
want = 0.5 + 0.300 + 0.010 + 0.300 + 0.030 * 10
assert start / FS == pytest.approx(want, abs=0.004)
# ---------------------------------------------------------------------------
# APT
# ---------------------------------------------------------------------------
def test_a_satellite_pass_comes_back_as_the_picture():
card = grey_card(909, 40)
got = decode_apt(apt_audio(card, FS, snr_db=25), FS, frequency=137.1e6)
assert got is not None and got.ok, got and got.note
assert got.width == WORDS_PER_LINE
channel = np.asarray(got.channels["A"], dtype=float)
assert correlation(channel, card) > 0.9
@pytest.mark.parametrize("snr", [30, 20, 12, 6])
def test_the_pass_survives_a_weak_signal(snr):
card = grey_card(909, 40)
got = decode_apt(apt_audio(card, FS, snr_db=snr), FS, frequency=137.1e6)
assert got is not None and got.ok
assert correlation(np.asarray(got.channels["A"], float), card) > 0.75
def test_both_sensors_are_cut_out_separately():
card = grey_card(909, 30)
got = decode_apt(apt_audio(card, FS), FS, frequency=137.1e6)
assert set(got.channels) == {"A", "B"}
assert got.channels["A"].shape[1] == CHANNEL_A[1]
def test_it_is_only_tried_in_the_satellite_band():
"""Nothing outside 137 MHz is APT, and looking anyway finds sync in static."""
card = grey_card(909, 30)
audio = apt_audio(card, FS)
assert decode_apt(audio, FS, frequency=146.52e6) is None
assert decode_apt(audio, FS, frequency=137.62e6).ok
@pytest.mark.parametrize("seed", range(6))
def test_static_does_not_become_a_satellite_pass(seed):
rng = np.random.default_rng(seed)
t = np.arange(int(FS * 12)) / FS
for signal in (rng.standard_normal(t.size),
np.sin(2 * np.pi * 2400 * t),
np.sin(2 * np.pi * 2400 * t) * (1 + 0.5 * rng.standard_normal(t.size)),
np.sin(2 * np.pi * 2400 * t) * (1 + 0.9 * np.sin(2 * np.pi * 7 * t))):
got = decode_apt(signal, FS, frequency=137.1e6)
assert got is None or not got.ok
# ---------------------------------------------------------------------------
# HF fax
# ---------------------------------------------------------------------------
def test_a_chart_comes_back_as_the_chart():
card = grey_card(800, 30)
got = decode_fax(fax_audio(card, FS, snr_db=25), FS)
assert got is not None and got.ok, got and got.note
assert got.mode.startswith("120 lpm")
assert correlation(got.pixels, _stretched(card, got.width)) > 0.95
def _stretched(card, width):
return np.stack([resample_to(row.astype(float), width) for row in card])
@pytest.mark.parametrize("snr", [30, 20, 12, 6])
def test_the_chart_survives_a_weak_signal(snr):
card = grey_card(800, 30)
got = decode_fax(fax_audio(card, FS, snr_db=snr), FS)
assert got is not None and got.ok
assert correlation(got.pixels, _stretched(card, got.width)) > 0.9
def test_the_phasing_signal_gives_the_line_rate():
card = grey_card(800, 20)
for lpm in (60.0, 120.0, 180.0):
audio = fax_audio(card, FS, lpm=lpm, snr_db=30)
freq = instantaneous_frequency(audio, FS, low=1200.0, high=2600.0)
found = find_phasing(_levels(freq), FS)
assert found is not None and found.lpm == lpm
def test_the_start_tone_is_not_mistaken_for_the_chart():
"""It is a black-and-white alternation, so its average is mid-grey.
Taking "the first line that is not black" as the start of the picture
made every chart thirty seconds of tone.
"""
card = grey_card(800, 20)
got = decode_fax(fax_audio(card, FS, start_seconds=6.0), FS)
assert got.ok
assert got.height <= card.shape[0] + 2
@pytest.mark.parametrize("seed", range(6))
def test_a_band_with_nothing_on_it_produces_no_chart(seed):
rng = np.random.default_rng(seed)
t = np.arange(int(FS * 25)) / FS
speech = (np.sin(2 * np.pi * (1800 + 300 * np.sin(2 * np.pi * 4 * t)) * t)
* (0.5 + 0.5 * np.sin(2 * np.pi * 2.5 * t)))
for signal in (rng.standard_normal(t.size), np.sin(2 * np.pi * 1900 * t),
speech, speech + 0.3 * rng.standard_normal(t.size)):
got = decode_fax(signal, FS)
assert got is None or not got.ok
# ---------------------------------------------------------------------------
# Which decoder gets offered what
# ---------------------------------------------------------------------------
def test_sstv_is_recognised_wherever_it_is_heard():
audio = sstv_audio(MODES[0], colour_card(), FS, lines=12)
for frequency in (14.230e6, 144.5e6, 0.0):
got = find_image(audio, FS, frequency=frequency)
assert got is not None and got.ok and got.kind == "SSTV"
def test_a_satellite_is_only_looked_for_in_its_own_band():
audio = apt_audio(grey_card(909, 30), FS)
assert find_image(audio, FS, frequency=137.1e6).kind == "APT"
assert find_image(audio, FS, frequency=433.92e6) is None
def test_fax_is_looked_for_on_shortwave_and_in_sideband():
audio = fax_audio(grey_card(800, 20), FS)
assert find_image(audio, FS, frequency=8.040e6).kind == "HF fax"
assert find_image(audio, FS, frequency=14.2e6, mode="usb").kind == "HF fax"
assert find_image(audio, FS, frequency=146.52e6, mode="nfm") is None
def test_an_ordinary_recording_is_not_a_picture():
rng = np.random.default_rng(0)
t = np.arange(int(FS * 6)) / FS
speech = np.sin(2 * np.pi * 300 * t) * (0.5 + 0.5 * np.sin(2 * np.pi * 3 * t))
assert find_image(speech + 0.1 * rng.standard_normal(t.size), FS,
frequency=146.52e6) is None
def test_a_decode_can_be_saved_and_read_back(tmp_path):
got = find_image(sstv_audio(MODES[0], colour_card(), FS, lines=12), FS)
path = got.save(tmp_path / "picture.png")
assert path and (tmp_path / "picture.png").exists()
assert _read_png(tmp_path / "picture.png").shape == (got.height, got.width, 3)
def test_an_empty_decode_saves_nothing(tmp_path):
assert ImageDecode().save(tmp_path / "nothing.png") == ""
assert not (tmp_path / "nothing.png").exists()
# ---------------------------------------------------------------------------
# Reading one back
# ---------------------------------------------------------------------------
#
# The point of decoding a picture is that somebody looks at it, and the
# browser cannot draw a PNG in a terminal. So what it owes them is to say
# plainly that this recording is a picture and exactly where the file is.
import json
import wave
from rich.console import Console
from bandsaunter.browse import Browser, Player, scan_directory
def _picture_capture(directory, **over):
stem = "0144.500000MHz--2026-08-29_10_00_00-nfm"
with wave.open(str(directory / f"{stem}.wav"), "wb") as w:
w.setnchannels(1)
w.setsampwidth(2)
w.setframerate(16000)
w.writeframes(b"\0\0" * 16000)
write_png(directory / f"{stem}.png", colour_card(320, 24))
hit = {"frequency": 144.5e6, "category": "image",
"classification": "SSTV (Martin M1)", "image_kind": "SSTV",
"image_mode": "Martin M1", "image_width": 320, "image_height": 24,
"image_complete": False,
"image_path": str(directory / f"{stem}.png"), "confidence": 0.9}
hit.update(over)
(directory / f"{stem}.json").write_text(json.dumps({"hit": hit}))
return directory / f"{stem}.wav"
def _browser(directory, width=100, height=30):
return Browser(directory, player=Player([]),
console=Console(width=width, height=height,
force_terminal=False, no_color=True))
def _frame(browser):
with browser.console.capture() as cap:
browser.console.print(browser.render())
return cap.get()
def test_the_browser_says_a_recording_is_a_picture(tmp_path):
_picture_capture(tmp_path)
shown = _frame(_browser(tmp_path))
assert "picture" in shown
assert "SSTV Martin M1 320x24 partial" in shown
def test_it_gives_the_path_in_full(tmp_path):
"""In full, wrapped where it has to be: half a path opens nothing."""
path = _picture_capture(tmp_path).with_suffix(".png")
shown = _frame(_browser(tmp_path))
flat = "".join(shown.split()).replace("│", "")
assert str(path) in flat
def test_a_short_path_is_on_one_line(tmp_path):
path = _picture_capture(tmp_path).with_suffix(".png")
assert str(path) in _frame(_browser(tmp_path, width=len(str(path)) + 12))
def test_the_list_row_says_so_too(tmp_path):
_picture_capture(tmp_path)
browser = _browser(tmp_path)
assert "SSTV" in _frame(browser).split("recordings in")[1]
def test_a_picture_can_be_searched_for_by_kind(tmp_path):
_picture_capture(tmp_path)
browser = _browser(tmp_path)
browser.query = "sstv"
browser.apply()
assert len(browser.view) == 1
def test_o_prints_the_picture_rather_than_the_audio(tmp_path):
path = _picture_capture(tmp_path).with_suffix(".png")
browser = _browser(tmp_path)
assert not browser.handle("o")
assert browser.message == str(path)
def test_the_picture_is_found_beside_the_recording_without_a_sidecar_path(
tmp_path):
"""A directory copied somewhere else keeps the sidecar, not the path in it."""
_picture_capture(tmp_path, image_path="/gone/nowhere.png")
cap = scan_directory(tmp_path)[0]
assert cap.image_path.endswith("-nfm.png")
def test_the_picture_moves_with_the_recording_when_it_is_filed(tmp_path):
_picture_capture(tmp_path)
browser = _browser(tmp_path)
browser.handle("S")
moved = {p.suffix for p in (tmp_path / "saved").iterdir()}
assert moved == {".wav", ".json", ".png"}
def test_every_channel_of_a_satellite_pass_belongs_to_the_recording(tmp_path):
wav = _picture_capture(tmp_path)
for name in ("A", "B"):
write_png(wav.with_name(wav.stem + f"_{name}.png"),
grey_card(64, 8))
cap = scan_directory(tmp_path)[0]
assert len([p for p in cap.files() if p.suffix == ".png"]) == 3
assert len(cap.images) == 3
def test_a_simulated_transmission_becomes_a_file_on_disk(tmp_path):
"""The whole path, from a signal on the air to a PNG beside the recording.
The demo band carries a real Martin M1 transmission, so this exercises
detection, the content check that used to throw pictures away, the
decode, and the write.
"""
from bandsaunter.config import ScanConfig
from bandsaunter.ranges import parse_range_list
from bandsaunter.scanner import Scanner
from bandsaunter.simulator import SimulatedDevice
cfg = ScanConfig(ranges=parse_range_list("144.45M-144.55M"),
output_dir=str(tmp_path), transcribe=False,
record_seconds=20, hang_seconds=2.0,
min_record_seconds=1.0, max_runtime_seconds=60,
dwell_seconds=0.05)
scanner = Scanner(cfg, device=SimulatedDevice(realtime=False).open())
scanner.prepare()
scanner.run()
pictures = [h for h in scanner.hits if h.image_kind]
assert pictures, "the SSTV transmission produced no picture"
hit = pictures[0]
assert hit.image_kind == "SSTV" and hit.image_mode == "Martin M1"
assert hit.category == "image"
assert hit.kept, "a picture was decoded and then discarded"
written = list(tmp_path.glob("*.png"))
assert written
assert _read_png(written[0]).shape[1] == 320

368
tests/test_signals_named.py Normal file
View file

@ -0,0 +1,368 @@
"""The formats that say what they are: ADS-B, utility meters, weather sensors.
What these three have in common is that none of them needs to be believed.
Every ADS-B frame carries a 24-bit checksum, every meter message a 16-bit BCH,
every AcuRite message a checksum and four parity bits -- so the only question
a test can usefully ask is whether the fields come back holding what was put
in, and whether anything that is not one of these is ever mistaken for one.
Each generator is written from the published format rather than from the
decoder beside it, so a round trip is evidence about the format.
"""
import numpy as np
import pytest
import adsb_gen as gen
from bandsaunter.adsb import (AircraftRegistry, SAMPLE_RATE, crc24,
decode_adsb, decode_frames, global_position)
from bandsaunter.ism import (SCM_PREAMBLE, acurite_frame, decode_acurite,
decode_ism, decode_scm, scm_frame)
from bandsaunter.payload import (MIN_TEXT, hexdump, interpret, read_text,
readable)
RATE = 2_000_000.0
# ---------------------------------------------------------------------------
# ADS-B
# ---------------------------------------------------------------------------
def test_the_parity_is_the_one_mode_s_uses():
"""A good frame leaves nothing behind; a corrupted one does."""
frame = gen.identification(0x4CA1FA, "RYR1234")
assert crc24(frame) == 0
broken = bytearray(frame)
broken[5] ^= 0x01
assert crc24(bytes(broken)) != 0
def test_an_aircraft_saying_its_callsign_is_read_back():
frames, registry = decode_adsb(
gen.modulate([gen.identification(0x4CA1FA, "RYR1234")]), RATE)
assert len(frames) == 1
assert registry.aircraft["4CA1FA"].callsign == "RYR1234"
@pytest.mark.parametrize("lat,lon", [(51.5, -0.12), (40.7, -74.0),
(-33.9, 151.2), (0.5, 0.5),
(60.2, 24.9)])
def test_a_position_needs_two_frames_and_comes_back_exactly(lat, lon):
"""One frame is ambiguous by hundreds of miles; the pair is not."""
frames = [gen.airborne_position(0xA0B1C2, lat, lon, 30000, odd=False),
gen.airborne_position(0xA0B1C2, lat, lon, 30000, odd=True)]
_, registry = decode_adsb(gen.modulate(frames), RATE)
craft = registry.aircraft["A0B1C2"]
assert craft.latitude == pytest.approx(lat, abs=0.01)
assert craft.longitude == pytest.approx(lon, abs=0.01)
def test_one_position_frame_alone_places_nothing():
_, registry = decode_adsb(gen.modulate(
[gen.airborne_position(0xA0B1C2, 51.5, -0.12, 30000, odd=False)]), RATE)
assert not registry.aircraft["A0B1C2"].located
@pytest.mark.parametrize("feet", [0, 1000, 12000, 35000, 43000])
def test_altitude_comes_back_in_feet(feet):
_, registry = decode_adsb(gen.modulate(
[gen.airborne_position(0xABCDEF, 51.5, -0.12, feet, odd=False)]), RATE)
assert registry.aircraft["ABCDEF"].altitude_ft == pytest.approx(feet,
abs=25)
def test_speed_and_climb_rate_come_back():
_, registry = decode_adsb(gen.modulate(
[gen.velocity(0x4CA1FA, 250, -180, 1216)]), RATE)
craft = registry.aircraft["4CA1FA"]
assert craft.ground_speed_kt == pytest.approx(308, abs=3)
assert craft.track_deg == pytest.approx(126, abs=2)
assert craft.vertical_rate_fpm == pytest.approx(1216, abs=64)
@pytest.mark.parametrize("noise", [0.0, 0.05, 0.15, 0.30])
def test_every_frame_survives_a_noisy_receiver(noise):
frames = [gen.identification(0x4CA1FA, "RYR1234"),
gen.airborne_position(0x4CA1FA, 51.5, -0.12, 35000, odd=False),
gen.airborne_position(0x4CA1FA, 51.5, -0.12, 35000, odd=True),
gen.velocity(0x4CA1FA, 250, -180, 1216)]
got, registry = decode_adsb(gen.modulate(frames, noise=noise, seed=3),
RATE)
assert len(got) == len(frames)
assert registry.aircraft["4CA1FA"].located
def test_many_aircraft_are_kept_apart():
frames = []
for i, icao in enumerate((0x4CA1FA, 0xA0B1C2, 0x3C6444, 0x780102)):
frames.append(gen.identification(icao, f"FLT{i}"))
frames.append(gen.airborne_position(icao, 50 + i, -1 - i, 30000, False))
frames.append(gen.airborne_position(icao, 50 + i, -1 - i, 30000, True))
_, registry = decode_adsb(gen.modulate(frames), RATE)
assert len(registry) == 4
assert all(craft.located for craft in registry.aircraft.values())
@pytest.mark.parametrize("seed", range(6))
def test_silence_does_not_become_aircraft(seed):
"""A frame of all zeros satisfies the checksum, and silence is one."""
rng = np.random.default_rng(seed)
for samples in (np.zeros(int(RATE // 4), dtype=np.complex64),
(rng.standard_normal(int(RATE // 4))
+ 1j * rng.standard_normal(int(RATE // 4))
).astype(np.complex64) * 0.1):
assert decode_frames(samples, RATE) == []
def test_a_rate_too_low_to_see_a_bit_is_refused():
"""One megabit a second cannot be read at one megasample a second."""
iq = gen.modulate([gen.identification(0x4CA1FA, "TEST")])
assert decode_frames(iq, SAMPLE_RATE / 2) == []
def test_a_corrupted_frame_is_dropped_rather_than_reported():
frame = bytearray(gen.identification(0x4CA1FA, "RYR1234"))
frame[6] ^= 0xFF
assert decode_frames(gen.modulate([bytes(frame)]), RATE) == []
def test_a_position_straddling_a_zone_boundary_is_refused():
"""Two frames from different latitude bands cannot be combined.
The longitude zones get wider towards the poles, so a pair that came from
either side of a boundary would be resolved against two different grids
and land somewhere neither of them was.
"""
from bandsaunter.adsb import Frame
# These two resolve to latitudes with a different number of longitude
# zones, which is exactly the case the standard says cannot be combined.
even = Frame(cpr_lat=2048, cpr_lon=0, cpr_odd=False)
odd = Frame(cpr_lat=12288, cpr_lon=0, cpr_odd=True)
assert global_position(even, odd) is None
# And a pair that does not straddle one still resolves.
lat, lon = gen._cpr(51.5, -0.12, odd=False)
olat, olon = gen._cpr(51.5, -0.12, odd=True)
assert global_position(Frame(cpr_lat=lat, cpr_lon=lon),
Frame(cpr_lat=olat, cpr_lon=olon,
cpr_odd=True)) is not None
# ---------------------------------------------------------------------------
# Utility meters
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("meter,reading,ert", [
(12345678, 987654, 4), (0x3FFFFFF, 16777215, 5), (1, 0, 11),
(555555, 42, 12), (67108863, 1, 13),
])
def test_a_meter_reading_comes_back_as_it_was_sent(meter, reading, ert):
got = decode_scm(scm_frame(meter, reading, ert))
assert got is not None
assert got.identifier == str(meter)
assert ("reading", str(reading)) in got.fields
@pytest.mark.parametrize("ert,name", [(4, "electricity"), (5, "gas"),
(11, "water"), (12, "gas")])
def test_the_kind_of_meter_is_named(ert, name):
got = decode_scm(scm_frame(7, 7, ert))
assert got.device == f"{name} meter"
def test_a_meter_type_that_is_not_known_is_numbered_not_guessed():
got = decode_scm(scm_frame(7, 7, ert_type=2))
assert "type 2" in got.device
def test_the_tamper_flags_are_reported():
got = decode_scm(scm_frame(7, 7, tamper_physical=2, tamper_encoder=1))
names = dict(got.fields)
assert names["physical tamper"] == "2"
assert names["encoder tamper"] == "1"
def test_a_message_is_found_after_whatever_came_before_it():
bits = "0101101" + scm_frame(4242, 999) + "1101"
assert decode_scm(bits).identifier == "4242"
def test_a_meter_message_with_a_bit_wrong_is_refused():
bits = list(scm_frame(12345678, 987654))
bits[40] = "1" if bits[40] == "0" else "0"
assert decode_scm("".join(bits)) is None
def test_a_meter_number_too_large_is_an_error_not_a_wrong_reading():
with pytest.raises(ValueError):
scm_frame(1 << 27, 5)
with pytest.raises(ValueError):
scm_frame(5, 1 << 25)
def test_random_bits_behind_a_real_preamble_are_refused():
"""The preamble is 21 bits and turns up; the checksum is what matters."""
rng = np.random.default_rng(0)
accepted = 0
for _ in range(3000):
body = "".join(rng.integers(0, 2, 75).astype(str))
if decode_scm(SCM_PREAMBLE + body) is not None:
accepted += 1
assert accepted == 0
# ---------------------------------------------------------------------------
# AcuRite
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("sensor,celsius,humidity,channel", [
(0x1234, 21.5, 48, "A"), (0x0001, -20.0, 5, "C"), (0x3FFF, 45.3, 100, "B"),
(0x2AAA, 0.0, 50, "D"), (0x0555, -39.9, 1, "A"),
])
def test_a_sensor_reading_comes_back_as_it_was_sent(sensor, celsius,
humidity, channel):
got = decode_acurite(acurite_frame(sensor, celsius, humidity, channel))
assert got is not None
assert got.identifier == f"{sensor:04X}"
fields = dict(got.fields)
assert fields["temperature"] == f"{celsius:.1f} C"
assert fields["humidity"] == f"{humidity}%"
assert fields["channel"] == channel
def test_a_flat_battery_is_reported_and_a_good_one_is_not():
good = decode_acurite(acurite_frame(0x1234, 20.0, 50, "A", False))
flat = decode_acurite(acurite_frame(0x1234, 20.0, 50, "A", True))
assert "battery" not in dict(good.fields)
assert dict(flat.fields)["battery"] == "low"
def test_a_message_is_found_wherever_in_the_burst_it_starts():
bits = "10110" + acurite_frame(0x0ABC, 12.3, 77, "B") + "0011"
assert decode_acurite(bits).identifier == "0ABC"
def test_a_sensor_message_with_a_bit_wrong_is_refused():
bits = list(acurite_frame(0x1234, 21.5, 48, "A"))
bits[20] = "1" if bits[20] == "0" else "0"
assert decode_acurite("".join(bits)) is None
def test_a_reading_outside_what_the_sensor_can_report_is_refused():
"""The checksum can be satisfied by a message the hardware cannot send."""
from bandsaunter.ism import _parity
data = [0x00, 0x01, 0x04, 0x7F, 0x0F, 0x7F]
for i in range(2, 6):
if _parity(data[i]) != 1:
data[i] |= 0x80
data.append(sum(data[:6]) & 0xFF)
bits = "".join(format(b, "08b") for b in data)
assert decode_acurite(bits) is None
def test_the_band_decides_which_is_tried_first_and_nothing_else():
meter = scm_frame(4242, 999)
assert decode_ism(meter, frequency=915e6).kind == "SCM"
assert decode_ism(meter, frequency=433.92e6).kind == "SCM"
def test_neither_reads_a_message_out_of_nothing():
rng = np.random.default_rng(1)
accepted = sum(1 for _ in range(5000)
if decode_ism("".join(rng.integers(0, 2, 96).astype(str)))
is not None)
# The seven-byte sensor message is checked at every offset, so the bar is
# a rate rather than zero. What reaches this in the scanner has already
# had to arrive identically several times over.
assert accepted / 5000 < 0.01
# ---------------------------------------------------------------------------
# Reading a payload
# ---------------------------------------------------------------------------
def _encode(text: str, width: int = 8, msb: bool = True, before: int = 0,
after: int = 0, seed: int = 0) -> str:
rng = np.random.default_rng(seed)
bits = "".join(rng.integers(0, 2, before).astype(str))
for ch in text:
chunk = format(ord(ch), f"0{width}b")
bits += chunk if msb else chunk[::-1]
return bits + "".join(rng.integers(0, 2, after).astype(str))
@pytest.mark.parametrize("message", [
"HELLO WORLD", "ENGINE 4 RESPOND", "the quick brown fox",
"Meeting at seven", "BATTERY LOW", "unit twelve en route",
])
@pytest.mark.parametrize("width,msb", [(8, True), (8, False), (7, True)])
def test_text_in_a_packet_is_read_out(message, width, msb):
got = read_text(_encode(message, width, msb))
assert got is not None, message
assert got.text.lower() in message.lower()
def test_text_is_found_behind_a_preamble_and_an_address():
got = read_text(_encode("STATION OPEN", before=13, after=7, seed=4))
assert got is not None and "STATION OPEN" in got.text
def test_random_payloads_are_almost_never_read_as_text():
"""Printability is not evidence, and this is the measurement that says so.
Every framing at every offset in both bit orders is about forty readings
of each packet, and seven-bit values are printable three times in four, so
a bar set on printability alone called 64% of random payloads text.
"""
rng = np.random.default_rng(5)
hits = 0
total = 0
for n_bits in (24, 32, 48, 64, 96, 128, 192, 256, 512):
for _ in range(250):
total += 1
if read_text("".join(rng.integers(0, 2, n_bits).astype(str))):
hits += 1
assert hits / total < 0.01, f"{100 * hits / total:.1f}% read as text"
def test_a_run_of_one_case_and_no_vowels_is_not_text():
assert read_text(_encode("XKCDZQRT")) is None
def test_the_bytes_are_laid_out_with_their_characters_beside_them():
lines = hexdump(_encode("ABCDEFGH"))
assert lines[0].startswith("0000")
assert "|ABCDEFGH|" in lines[0]
def test_a_trailing_part_byte_is_shown_as_bits_not_padded():
"""Padding four bits to a byte invents four zeroes nobody sent."""
lines = hexdump("1" * 12)
assert "4 bit(s): 1111" in lines[-1]
def test_a_remote_control_gets_its_fields_named():
readings = interpret("1" * 20 + "0100", encoding="PWM")
fields = [r for r in readings if r.kind == "fields"]
assert fields and fields[0].how.startswith("EV1527")
assert dict(fields[0].fields)["button"] == "B"
def test_every_payload_gets_at_least_its_bytes_back():
lines = readable("10110010" * 4)
assert lines and lines[0].startswith("0000")
def test_a_payload_that_says_what_it_is_is_named():
"""GRIB gets asked about as though it were a modulation.
It is not: it is the format weather models are published in, and it
arrives by satellite data link and by e-mail. Where a byte stream begins
with its magic number that is worth saying; nothing here renders one.
"""
bits = "".join(format(b, "08b") for b in b"GRIB\x00\x00\x00\x02payload!")
named = [r for r in interpret(bits) if r.kind == "fields"]
assert named and "GRIB" in named[0].line()
def test_an_ordinary_payload_claims_no_format():
bits = "".join(format(b, "08b") for b in b"\x01\x02\x03\x04\x05\x06\x07\x08")
assert not [r for r in interpret(bits) if r.kind == "fields"]