96fc21ac7d
Four things the ADS-B mode was missing, and one it was actively getting
wrong.
The band plan lists 1090 MHz because that is where ADS-B is, so choosing
it from the band plan is the obvious thing to do -- and it records the
bursts as clicks in a WAV file and decodes nothing, silently. Both the
scanner and the menus now say so, before the sweep starts, and name the
mode that does decode it. It is not refused: looking at the raw spectrum
is a fair thing to want.
Menu 5, Aircraft (ADS-B), is the whole mode without a command line. Every
option on one screen with a line saying what it does, ?N for the long
version and the flag it corresponds to, l to listen, m to draw a map from
any log, s to keep the options. The listening and the drawing moved into
bandsaunter/aircraft.py so the menus and the command line run the same
code.
While it listens the screen is a live board: one line per aircraft in the
order first heard, the counter climbing as frames arrive, height coloured
low warm to high cold with an arrow for climb or descent, the age of the
last report going green to red, and the line removed once nothing has been
heard for --hold seconds, everything below moving up. The registers are
asked while it runs, so registration, type, operator and route fill
themselves in as the answers arrive.
--speed-unit knots|mph|kph changes the heading of that board, the speed
beside every aircraft on the map and the speeds in the report, and moves
the distances with it so that one picture never carries two different
miles. The log stays in knots, which is what the aircraft broadcast.
And there is a real map under the flight paths: {z}/{x}/{y} tiles fetched
once, cached in ~/.cache/bandsaunter/tiles, reprojected from Web Mercator
pixel by pixel, inverted and dimmed so the aircraft stay the brightest
thing on the picture. The PNGs are decoded here -- zlib and the five row
filters from the specification, checked byte for byte against Pillow on
real tiles -- so nothing new is depended on. Tiles are cached and never
re-fetched, every request says who is asking, and the attribution is drawn
onto the picture, because a GIF travels without its readme.
conftest now fails any test that reaches for a tile server or a register.
It caught four of these on the way in.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016PsWPTweCT6pwxKngvVxcg
462 lines
18 KiB
Python
462 lines
18 KiB
Python
"""The moving map: what is drawn, and whether the file is really a GIF.
|
|
|
|
A picture is checked here the way a picture has to be -- by reading the pixels
|
|
back -- and the file by taking it apart with a reader written from the GIF
|
|
specification rather than from the encoder in the program. An encoder checked
|
|
against itself is not checked.
|
|
"""
|
|
import struct
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from bandsaunter import flightmap as fm
|
|
from bandsaunter.flightlog import Fix, Track
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# A GIF reader, built from the specification
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def lzw_decode(data: bytes, code_bits: int) -> bytes:
|
|
"""GIF's LZW, the reading side, written from the standard."""
|
|
clear, end = 1 << code_bits, (1 << code_bits) + 1
|
|
table = {i: bytes([i]) for i in range(clear)}
|
|
width = code_bits + 1
|
|
counter = end + 1 # counts codes read, not entries added
|
|
out = bytearray()
|
|
previous = None
|
|
value = held = at = 0
|
|
while True:
|
|
while held < width and at < len(data):
|
|
value |= data[at] << held
|
|
held += 8
|
|
at += 1
|
|
if held < width:
|
|
break
|
|
code = value & ((1 << width) - 1)
|
|
value >>= width
|
|
held -= width
|
|
if code == clear:
|
|
table = {i: bytes([i]) for i in range(clear)}
|
|
width, counter, previous = code_bits + 1, end + 1, None
|
|
continue
|
|
if code == end:
|
|
break
|
|
# The width goes up as codes are read rather than as the table is
|
|
# filled: the decoder builds its table one code behind the encoder,
|
|
# and counting entries instead would widen the codes one too late.
|
|
read_at = counter
|
|
if counter < 4096:
|
|
counter += 1
|
|
if counter > (1 << width) and width < 12:
|
|
width += 1
|
|
if code in table:
|
|
entry = table[code]
|
|
elif previous is not None:
|
|
entry = previous + previous[:1]
|
|
else:
|
|
raise ValueError(f"code {code} before anything defined it")
|
|
out += entry
|
|
if previous is not None and read_at - 1 < 4096:
|
|
table[read_at - 1] = previous + entry[:1]
|
|
previous = entry
|
|
return bytes(out)
|
|
|
|
|
|
def read_gif(path: Path) -> dict:
|
|
"""Take a GIF apart: the screen, the palette and every frame in it."""
|
|
raw = Path(path).read_bytes()
|
|
assert raw[:6] == b"GIF89a", "not a GIF89a"
|
|
width, height, packed, _bg, _aspect = struct.unpack("<HHBBB", raw[6:13])
|
|
at = 13
|
|
table_size = 2 ** ((packed & 0x07) + 1)
|
|
assert packed & 0x80, "no global colour table"
|
|
palette = np.frombuffer(raw[at:at + table_size * 3],
|
|
dtype=np.uint8).reshape(-1, 3)
|
|
at += table_size * 3
|
|
|
|
frames, loops, control = [], False, {}
|
|
while at < len(raw):
|
|
block = raw[at]
|
|
if block == 0x3B: # trailer
|
|
break
|
|
if block == 0x21: # extension
|
|
label = raw[at + 1]
|
|
at += 2
|
|
body = b""
|
|
while raw[at]:
|
|
size = raw[at]
|
|
body += raw[at + 1:at + 1 + size]
|
|
at += size + 1
|
|
at += 1
|
|
if label == 0xF9:
|
|
control = {"disposal": (body[0] >> 2) & 0x07,
|
|
"transparent": body[3] if body[0] & 1 else None,
|
|
"delay": struct.unpack("<H", body[1:3])[0]}
|
|
elif label == 0xFF and body.startswith(b"NETSCAPE2.0"):
|
|
loops = True
|
|
continue
|
|
if block == 0x2C: # image descriptor
|
|
left, top, w, h, flags = struct.unpack("<HHHHB", raw[at + 1:at + 10])
|
|
at += 10
|
|
assert not flags & 0x80, "a local colour table was not expected"
|
|
code_bits = raw[at]
|
|
at += 1
|
|
data = b""
|
|
while raw[at]:
|
|
size = raw[at]
|
|
data += raw[at + 1:at + 1 + size]
|
|
at += size + 1
|
|
at += 1
|
|
pixels = np.frombuffer(lzw_decode(data, code_bits),
|
|
dtype=np.uint8)
|
|
assert pixels.size == w * h, f"{pixels.size} pixels for {w}x{h}"
|
|
frames.append({"left": left, "top": top, "width": w, "height": h,
|
|
"pixels": pixels.reshape(h, w), **control})
|
|
continue
|
|
raise ValueError(f"unknown block {block:#x} at {at}")
|
|
return {"width": width, "height": height, "palette": palette,
|
|
"frames": frames, "loops": loops}
|
|
|
|
|
|
def played(gif: dict) -> list[np.ndarray]:
|
|
"""Every frame as it appears on the screen, patches laid over each other."""
|
|
canvas = np.zeros((gif["height"], gif["width"]), dtype=np.uint8)
|
|
out = []
|
|
for frame in gif["frames"]:
|
|
patch = frame["pixels"]
|
|
top, left = frame["top"], frame["left"]
|
|
area = canvas[top:top + frame["height"], left:left + frame["width"]]
|
|
if frame.get("transparent") is None:
|
|
area[:, :] = patch
|
|
else:
|
|
keep = patch != frame["transparent"]
|
|
area[keep] = patch[keep]
|
|
out.append(canvas.copy())
|
|
return out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tracks to draw
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def straight(icao="4CA1FA", callsign="RYR1234", lat=51.0, lon=-1.0,
|
|
heading=90.0, speed=480.0, altitude=35_000, seconds=600.0,
|
|
every=20.0, start=1_000_000.0) -> Track:
|
|
"""One aircraft flying a straight line at a steady speed."""
|
|
from bandsaunter.flightlog import move
|
|
|
|
fixes = []
|
|
when = 0.0
|
|
while when <= seconds:
|
|
step = speed * when / 3600.0
|
|
here = move(lat, lon, heading, step)
|
|
fixes.append(Fix(at=start + when, latitude=here[0], longitude=here[1],
|
|
altitude_ft=altitude, ground_speed_kt=speed,
|
|
track_deg=heading))
|
|
when += every
|
|
return Track(icao=icao, callsign=callsign, fixes=fixes,
|
|
frames=len(fixes) * 3, first_seen=start,
|
|
last_seen=start + seconds)
|
|
|
|
|
|
def two_aircraft() -> list[Track]:
|
|
return [straight(),
|
|
straight(icao="A835AF", callsign="UAL1902", lat=51.4, lon=-0.6,
|
|
heading=250.0, speed=300.0, altitude=9_000)]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Colours and geometry
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_the_palette_is_a_full_table_with_room_for_transparency():
|
|
assert fm.PALETTE.shape == (256, 3)
|
|
assert fm.TRANSPARENT == 255
|
|
|
|
|
|
def test_altitude_becomes_a_colour_that_climbs_with_it():
|
|
steps = [fm.altitude_step(ft) for ft in (0, 5_000, 20_000, 35_000, 45_000)]
|
|
assert steps == sorted(steps)
|
|
assert steps[0] == 0 and steps[-1] == fm.RAMP_STEPS - 1
|
|
assert fm.altitude_step(90_000) == fm.RAMP_STEPS - 1 # clamped
|
|
|
|
|
|
def test_the_map_holds_every_position_that_was_reported():
|
|
tracks = two_aircraft()
|
|
view = fm.fit(tracks, width=800)
|
|
for track in tracks:
|
|
for fix in track.fixes:
|
|
assert view.inside(fix.latitude, fix.longitude)
|
|
x, y = view.xy(fix.latitude, fix.longitude)
|
|
assert view.left <= x < view.left + view.width
|
|
assert view.top <= y < view.top + view.height
|
|
|
|
|
|
def test_a_mile_across_is_a_mile_up_the_picture():
|
|
"""Longitude is squeezed by the cosine, or everything at fifty degrees
|
|
north comes out stretched half as wide again."""
|
|
view = fm.fit(two_aircraft(), width=800)
|
|
per_nm_x = view.width / view.width_nm
|
|
tall_nm = (view.north - view.south) * 60.0
|
|
per_nm_y = view.height / tall_nm
|
|
assert per_nm_x == pytest.approx(per_nm_y, rel=0.02)
|
|
|
|
|
|
def test_one_aircraft_heard_once_still_gets_a_map():
|
|
track = Track(icao="4CA1FA", fixes=[Fix(at=1.0, latitude=51.0,
|
|
longitude=-1.0)])
|
|
view = fm.fit([track], width=400)
|
|
assert view is not None and view.north > view.south
|
|
|
|
|
|
def test_nothing_to_draw_draws_nothing(tmp_path):
|
|
assert fm.fit([Track(icao="4CA1FA")]) is None
|
|
assert fm.animate([Track(icao="4CA1FA")], tmp_path / "x.gif") is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# What ends up on the picture
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_the_background_has_a_grid_a_scale_and_a_title():
|
|
view = fm.fit(two_aircraft(), width=800)
|
|
base = fm.background(view, title="6 AIRCRAFT 2026-09-03")
|
|
assert (base == fm.GRID).sum() > 200 # the graticule and the border
|
|
assert (base == fm.INK).sum() > 40 # the title
|
|
assert (base == fm.DIM).sum() > 40 # scale bar and axis labels
|
|
assert (base == fm.RAMP + fm.RAMP_STEPS - 1).any() # the key
|
|
|
|
|
|
def test_an_aircraft_is_drawn_where_it_was_at_that_moment():
|
|
tracks = [straight()]
|
|
view = fm.fit(tracks, width=800)
|
|
base = fm.background(view)
|
|
when = tracks[0].first_seen + 300.0
|
|
frame = fm.render_frame(base, view, tracks, when)
|
|
fix = tracks[0].at(when)
|
|
x, y = view.xy(fix.latitude, fix.longitude)
|
|
patch = frame[y - 6:y + 7, x - 6:x + 7]
|
|
assert (patch >= fm.RAMP).any() and (patch < fm.TRAIL).any()
|
|
|
|
|
|
def test_the_aircraft_moves_between_frames_and_leaves_a_trail():
|
|
tracks = [straight()]
|
|
view = fm.fit(tracks, width=800)
|
|
base = fm.background(view)
|
|
early = fm.render_frame(base, view, tracks, tracks[0].first_seen + 60)
|
|
late = fm.render_frame(base, view, tracks, tracks[0].first_seen + 540)
|
|
assert not np.array_equal(early, late)
|
|
trail_early = (early >= fm.TRAIL) & (early < fm.TRANSPARENT)
|
|
trail_late = (late >= fm.TRAIL) & (late < fm.TRANSPARENT)
|
|
assert trail_late.sum() > trail_early.sum()
|
|
|
|
|
|
def test_an_aircraft_is_not_drawn_before_it_was_ever_heard():
|
|
tracks = [straight()]
|
|
view = fm.fit(tracks, width=800)
|
|
base = fm.background(view)
|
|
frame = fm.render_frame(base, view, tracks, tracks[0].first_seen - 10)
|
|
assert np.array_equal(frame, base)
|
|
|
|
|
|
def test_an_aircraft_long_gone_is_not_drawn_at_a_guessed_position():
|
|
tracks = [straight()]
|
|
view = fm.fit(tracks, width=800)
|
|
base = fm.background(view)
|
|
frame = fm.render_frame(base, view, tracks,
|
|
tracks[0].last_seen + 900, stale=300)
|
|
assert np.array_equal(frame, base)
|
|
|
|
|
|
def test_the_clock_and_the_count_are_drawn_over_the_map():
|
|
tracks = two_aircraft()
|
|
view = fm.fit(tracks, width=800)
|
|
base = fm.background(view)
|
|
frame = fm.render_frame(base, view, tracks, tracks[0].first_seen + 60,
|
|
clock="19:45:02")
|
|
assert (frame == fm.PANEL).any()
|
|
assert (frame == fm.INK).sum() > (base == fm.INK).sum()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# The file
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_the_lzw_stream_reads_back_as_what_went_in():
|
|
for body in (b"\x00" * 300, bytes(range(256)) * 3,
|
|
bytes([7, 7, 7, 8, 9, 7, 7, 8]) * 40):
|
|
assert lzw_decode(fm._lzw(body, 8), 8) == body
|
|
|
|
|
|
def test_a_long_stream_survives_the_table_filling_up():
|
|
rng = np.random.default_rng(4)
|
|
body = rng.integers(0, 60, size=200_000, dtype=np.uint8).tobytes()
|
|
assert lzw_decode(fm._lzw(body, 8), 8) == body
|
|
|
|
|
|
def test_the_animation_is_a_gif_that_loops(tmp_path):
|
|
out = fm.animate(two_aircraft(), tmp_path / "flights.gif", fps=10,
|
|
seconds=4, width=480)
|
|
assert out is not None and out.path.exists()
|
|
gif = read_gif(out.path)
|
|
assert gif["loops"], "a map that plays once and stops"
|
|
assert len(gif["frames"]) == out.frames
|
|
assert (gif["width"], gif["height"]) == (out.width, out.height)
|
|
assert gif["frames"][0]["delay"] == 10 # hundredths, so 10 a second
|
|
|
|
|
|
def test_only_what_changed_is_written_after_the_first_frame(tmp_path):
|
|
out = fm.animate(two_aircraft(), tmp_path / "flights.gif", fps=8,
|
|
seconds=4, width=480)
|
|
gif = read_gif(out.path)
|
|
first = gif["frames"][0]
|
|
assert (first["width"], first["height"]) == (out.width, out.height)
|
|
later = gif["frames"][1:]
|
|
assert later, "one frame is not an animation"
|
|
assert all(f["width"] * f["height"] < out.width * out.height for f in later)
|
|
assert all(f["transparent"] is not None for f in later)
|
|
|
|
|
|
def test_the_frames_played_back_show_the_aircraft_moving(tmp_path):
|
|
tracks = [straight()]
|
|
out = fm.animate(tracks, tmp_path / "one.gif", fps=8, seconds=4, width=480)
|
|
screens = played(read_gif(out.path))
|
|
assert len(screens) == out.frames
|
|
|
|
def where(frame):
|
|
lit = np.argwhere((frame >= fm.RAMP) & (frame < fm.TRAIL))
|
|
return lit.mean(axis=0)
|
|
|
|
start, end = where(screens[1]), where(screens[-1])
|
|
assert abs(end[1] - start[1]) > 20 # it went east across the picture
|
|
for frame in screens:
|
|
assert frame.shape == (out.height, out.width)
|
|
|
|
|
|
def test_the_animation_says_how_much_flying_it_covers(tmp_path):
|
|
tracks = two_aircraft()
|
|
out = fm.animate(tracks, tmp_path / "flights.gif", fps=10, seconds=5,
|
|
width=400)
|
|
assert out.covers == pytest.approx(600.0, abs=1.0)
|
|
played_for = out.frames / out.fps
|
|
assert played_for == pytest.approx(5.0, rel=0.3)
|
|
assert out.speed == pytest.approx(out.covers / played_for, rel=0.05)
|
|
assert "aircraft" in out.summary()
|
|
|
|
|
|
def test_asking_for_a_speed_gives_that_speed(tmp_path):
|
|
out = fm.animate(two_aircraft(), tmp_path / "flights.gif", fps=10,
|
|
speed=60.0, width=400)
|
|
assert out.speed == pytest.approx(60.0, rel=0.1)
|
|
assert out.frames == pytest.approx(600 / 60 * 10, abs=2)
|
|
|
|
|
|
def test_a_still_picture_is_a_png_of_the_whole_evening(tmp_path):
|
|
from bandsaunter.images import PNG_SIGNATURE
|
|
|
|
out = fm.animate(two_aircraft(), tmp_path / "flights.png", width=400)
|
|
assert out.kind == "png"
|
|
assert out.path.read_bytes()[:8] == PNG_SIGNATURE
|
|
|
|
|
|
def test_pillow_agrees_that_it_is_an_animation(tmp_path):
|
|
"""Not a dependency; when it happens to be installed it is a second
|
|
opinion from a decoder nobody here wrote."""
|
|
Image = pytest.importorskip("PIL.Image")
|
|
out = fm.animate(two_aircraft(), tmp_path / "flights.gif", fps=10,
|
|
seconds=3, width=400)
|
|
with Image.open(out.path) as picture:
|
|
assert picture.n_frames == out.frames
|
|
assert picture.size == (out.width, out.height)
|
|
picture.seek(0)
|
|
first = np.array(picture.convert("RGB"))
|
|
picture.seek(picture.n_frames - 1)
|
|
last = np.array(picture.convert("RGB"))
|
|
assert not np.array_equal(first, last)
|
|
|
|
|
|
@pytest.mark.skipif(not fm.ffmpeg_available(), reason="ffmpeg is not installed")
|
|
def test_a_video_can_be_written_where_ffmpeg_exists(tmp_path):
|
|
out = fm.animate(two_aircraft(), tmp_path / "flights.mp4", fps=10,
|
|
seconds=3, width=400)
|
|
assert out.kind == "mp4" and out.path.stat().st_size > 1000
|
|
probe = subprocess.run(["ffprobe", "-v", "error", "-select_streams", "v:0",
|
|
"-show_entries", "stream=width,height",
|
|
"-of", "csv=p=0", str(out.path)],
|
|
capture_output=True, text=True)
|
|
if probe.returncode == 0:
|
|
assert probe.stdout.strip() == f"{out.width},{out.height}"
|
|
|
|
|
|
def test_the_picture_is_an_even_number_of_pixels_across(tmp_path):
|
|
"""Video encoders refuse an odd width, and it costs nothing to be even."""
|
|
for width in (401, 402, 555):
|
|
view = fm.fit(two_aircraft(), width=width)
|
|
w, h = fm.canvas_size(view)
|
|
assert w % 2 == 0 and h % 2 == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Speeds in whatever the user reads
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _label_text(unit: str, width: int = 700) -> np.ndarray:
|
|
"""Draw one aircraft and hand back the pixels its label was written in."""
|
|
tracks = [straight(speed=480.0)]
|
|
view = fm.fit(tracks, width=width)
|
|
base = fm.background(view, unit=unit)
|
|
return fm.render_frame(base, view, tracks, tracks[0].first_seen + 120,
|
|
unit=unit)
|
|
|
|
|
|
def _has_text(frame: np.ndarray, text: str) -> bool:
|
|
"""Whether a string was stamped anywhere in the frame, found by drawing
|
|
it again and looking for the same pattern."""
|
|
from bandsaunter.images import GLYPH_H, draw_text, text_width
|
|
|
|
stamp = np.zeros((GLYPH_H, max(1, text_width(text))), dtype=np.uint8)
|
|
draw_text(stamp, 0, 0, text, 1)
|
|
rows, cols = stamp.shape
|
|
wanted = stamp.astype(bool)
|
|
if not wanted.any():
|
|
return False
|
|
for y in range(frame.shape[0] - rows):
|
|
for x in range(frame.shape[1] - cols):
|
|
patch = frame[y:y + rows, x:x + cols]
|
|
if np.all(patch[wanted] != fm.BG) and \
|
|
np.all(patch[~wanted] == patch[~wanted][0]):
|
|
return True
|
|
return False
|
|
|
|
|
|
@pytest.mark.parametrize("unit,label", [("knots", "KT"), ("mph", "MPH"),
|
|
("kph", "KM/H")])
|
|
def test_the_speed_on_the_map_carries_its_unit(unit, label):
|
|
"""A bare 480 beside an aircraft is three different speeds depending on
|
|
who is reading it."""
|
|
assert _has_text(_label_text(unit), label)
|
|
|
|
|
|
def test_the_number_beside_it_is_converted():
|
|
assert _has_text(_label_text("knots"), "480KT")
|
|
assert _has_text(_label_text("mph"), "552MPH")
|
|
assert _has_text(_label_text("kph"), "889KM/H")
|
|
|
|
|
|
@pytest.mark.parametrize("unit,label", [("knots", "NM"), ("mph", "MI"),
|
|
("kph", "KM")])
|
|
def test_the_scale_bar_uses_the_same_kind_of_mile(unit, label):
|
|
"""Miles an hour beside a scale in nautical miles is two different miles
|
|
on one picture."""
|
|
view = fm.fit(two_aircraft(), width=700)
|
|
assert _has_text(fm.background(view, unit=unit), label)
|
|
|
|
|
|
def test_the_animation_takes_the_unit_through_to_the_file(tmp_path):
|
|
out = fm.animate(two_aircraft(), tmp_path / "mph.png", width=500,
|
|
unit="mph")
|
|
assert out is not None and out.path.is_file()
|