dee262e130
Every capture that is not voice, or whose voice yields five characters or fewer of transcript, now gets a PNG of the waterfall it would have painted on screen: spectrogram from the IQ where it was kept, from the demodulated audio otherwise, captioned and labelled either way. The browser shows it in the picture panel, but only when there is no transcript, Morse or decoded data to show instead. bandsaunter waterfall [PATH...] [--all] [--redraw] [--min-chars N] draws them after the fact for recordings already on disk. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016PsWPTweCT6pwxKngvVxcg
497 lines
19 KiB
Python
497 lines
19 KiB
Python
"""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"
|
|
# Named, not merely globbed: every capture that produced no words has a
|
|
# waterfall drawn beside it, and an SSTV transmission produced a picture
|
|
# rather than words. Both are PNGs; only one of them is 320 across.
|
|
written = [p for p in tmp_path.glob("*.png")
|
|
if not p.name.endswith("_waterfall.png")]
|
|
assert written
|
|
assert _read_png(written[0]).shape[1] == 320
|