df080f6571
The terminal board says what is overhead. This says where: a real map with the aircraft moving on it as the frames arrive, and beside each one a box carrying everything known about the flight -- type and registration, who operates it, where it came from and where it is going, height with a rate of climb, speed and heading, how far away and on what bearing, its position, how many frames it has sent and how long since the last one. Qt is asked for and not required. Four bindings are tried, the module imports on a machine with none of them, and asking for the window without one gets the instructions rather than a traceback -- before the receiver is opened, since nothing is gained by taking the dongle for a window that cannot be drawn. In the menu, "listen now" is now "passive capture" with a realtime display beside it. Closing the window leaves exactly the files pressing control-C leaves, because listen and watch share one read loop and one finishing step; the receiver runs on its own thread, so a slow repaint cannot cost a frame and a slow tile fetch cannot stall the picture. The animation's labels grew to match: flight level and speed, type and registration, and both ends of the route, each with a small flag of the country its airport is in. The flags are a table rather than a network -- twelve pixels by eight, where a flag is the arrangement that makes one recognisable rather than a rendering of the real thing -- and a country not in the table is named by its two letters, since a flag that is nearly another country's is worse than none. Where a route arrives as bare codes the country comes from the ICAO prefix. Four things found on the way. The window ignored --seconds, so "listen for ten minutes" meant something different with a window open; it closes itself now. The register was being asked twice per aircraft, once for labels and once for airport positions. Cached routes had no country in them, so the first real redraw drew no flags at all -- routes are versioned now. And past fourteen aircraft on one frame the labels go back to the callsign, the height and the speed, because five lines beside each of three hundred aircraft is a page of overlapping text with a map somewhere behind it. Long names are folded rather than allowed to stretch a box, breaking at the arrow of a route so the two ends stay whole; and the animation's label placement gained the same ring search the window uses, having only ever tried four spots. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016PsWPTweCT6pwxKngvVxcg
924 lines
37 KiB
Python
924 lines
37 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()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Framing the picture on the receiver
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def far_away(icao="BAD001", lat=-9.5, lon=-112.5) -> Track:
|
|
"""One aircraft in the wrong hemisphere: a decode that went wrong."""
|
|
return Track(icao=icao, callsign="GHOST",
|
|
fixes=[Fix(at=1_000_000.0, latitude=lat, longitude=lon,
|
|
altitude_ft=35_000, ground_speed_kt=400.0)],
|
|
first_seen=1_000_000.0, last_seen=1_000_000.0)
|
|
|
|
|
|
def test_the_middle_of_what_was_heard_is_where_the_receiver_is():
|
|
"""A median, so a handful of wrong positions cannot drag it anywhere."""
|
|
from bandsaunter.flightlog import centre_of
|
|
|
|
tracks = two_aircraft() + [far_away(), far_away("BAD002", 60.0, 120.0)]
|
|
lat, lon = centre_of(tracks)
|
|
assert lat == pytest.approx(51.0, abs=1.0)
|
|
assert lon == pytest.approx(-1.0, abs=1.5)
|
|
|
|
|
|
def test_a_radius_leaves_the_far_ones_off_the_map():
|
|
tracks = two_aircraft() + [far_away()]
|
|
view_all = fm.fit([t for t in tracks if t.located])
|
|
assert view_all.north - view_all.south > 50 # dragged across a globe
|
|
from bandsaunter.flightlog import centre_of, within
|
|
|
|
lat, lon = centre_of(two_aircraft())
|
|
near = [t for t in within(tracks, lat, lon, 100.0) if t.located]
|
|
assert {t.icao for t in near} == {t.icao for t in two_aircraft()}
|
|
|
|
|
|
def test_the_frame_is_the_radius_rather_than_whatever_turned_up():
|
|
"""The scale should not change with the traffic."""
|
|
from bandsaunter.flightlog import box_around, centre_of, distance_nm
|
|
|
|
lat, lon = centre_of(two_aircraft())
|
|
box = box_around(lat, lon, 100.0)
|
|
view = fm.fit(two_aircraft(), width=600, box=box)
|
|
assert view.south == pytest.approx(box[0]) and view.north == pytest.approx(box[2])
|
|
across = distance_nm(lat, box[1], lat, box[3])
|
|
assert across == pytest.approx(200.0, rel=0.02) # a hundred each way
|
|
|
|
|
|
def test_the_radius_is_kept_by_the_animation(tmp_path):
|
|
"""Without one the frame stretches to reach a bad position on the other
|
|
side of the equator, and the real aircraft come out a pixel wide."""
|
|
from bandsaunter.flightlog import box_around, centre_of
|
|
|
|
tracks = two_aircraft() + [far_away()]
|
|
wide = fm.animate(tracks, tmp_path / "wide.png", width=400, radius_nm=0)
|
|
tight = fm.animate(tracks, tmp_path / "tight.png", width=400, radius_nm=100)
|
|
assert wide.aircraft == 3 and tight.aircraft == 2
|
|
|
|
everything = fm.fit([t for t in tracks if t.located], width=400)
|
|
lat, lon = centre_of(two_aircraft())
|
|
framed = fm.fit(two_aircraft(), width=400,
|
|
box=box_around(lat, lon, 100.0))
|
|
assert everything.north - everything.south > 50 # most of a hemisphere
|
|
assert framed.north - framed.south < 4 # a hundred miles
|
|
|
|
|
|
def test_an_explicit_receiver_position_is_used_as_given(tmp_path):
|
|
"""Somewhere with no aircraft near it: everything falls outside."""
|
|
out = fm.animate(two_aircraft(), tmp_path / "elsewhere.png", width=400,
|
|
radius_nm=50, centre=(0.0, 0.0))
|
|
assert out is None
|
|
|
|
|
|
def test_a_track_with_one_bad_fix_keeps_the_rest_of_its_flight():
|
|
"""Per fix rather than per aircraft: one wrong position in the middle of
|
|
a real flight must not take the flight off the map with it."""
|
|
from bandsaunter.flightlog import within
|
|
|
|
track = straight()
|
|
good = len(track.fixes)
|
|
track.fixes.insert(len(track.fixes) // 2,
|
|
Fix(at=track.fixes[0].at + 1, latitude=-9.5,
|
|
longitude=-112.5, altitude_ft=35_000))
|
|
kept = within([track], 51.0, -1.0, 100.0)[0]
|
|
assert len(kept.fixes) == good
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Drawing off the edge
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.parametrize("x,y", [(-28, -89), (-4000, 400), (450, -9000),
|
|
(2000, 400), (450, 4000)])
|
|
def test_an_aircraft_off_the_picture_paints_nothing(x, y):
|
|
"""Clamping the near edge of the centre dot and leaving the far one alone
|
|
reads as a negative slice, which numpy counts back from the far side --
|
|
and fills most of the frame with one colour."""
|
|
img = np.zeros((300, 500), dtype=np.uint8)
|
|
fm._marker(img, x, y, 90.0, 20)
|
|
assert int((img != 0).sum()) == 0
|
|
|
|
|
|
@pytest.mark.parametrize("x,y", [(0, 0), (499, 299), (250, 150)])
|
|
def test_an_aircraft_on_the_picture_is_still_drawn(x, y):
|
|
img = np.zeros((300, 500), dtype=np.uint8)
|
|
fm._marker(img, x, y, 90.0, 20)
|
|
assert 0 < int((img != 0).sum()) < 60
|
|
|
|
|
|
def test_a_still_draws_where_they_were_not_where_they_might_have_got_to():
|
|
"""A still of a whole evening is drawn hours after most of the aircraft
|
|
stopped transmitting; flying them on for those hours would scatter them
|
|
across three states."""
|
|
tracks = [straight(seconds=600.0)]
|
|
view = fm.fit(tracks, width=500)
|
|
base = fm.background(view)
|
|
hours_later = tracks[0].last_seen + 6 * 3600
|
|
projected = fm.render_frame(base, view, tracks, hours_later,
|
|
stale=7 * 3600, project=True, labels=False)
|
|
still = fm.render_frame(base, view, tracks, hours_later,
|
|
stale=7 * 3600, project=False, labels=False)
|
|
|
|
def markers(frame):
|
|
"""Just the aircraft: what changed, in the aircraft colours, inside
|
|
the map itself. The altitude key along the bottom is drawn in those
|
|
same colours and belongs to the background."""
|
|
body = slice(view.top, view.top + view.height), \
|
|
slice(view.left, view.left + view.width)
|
|
drawn = (frame[body] != base[body]) & (frame[body] >= fm.RAMP) & \
|
|
(frame[body] < fm.TRAIL)
|
|
return int(drawn.sum())
|
|
|
|
assert markers(projected) == 0 # flown clean off the picture
|
|
assert markers(still) > 0 # drawn where it last was
|
|
last = tracks[0].fixes[-1]
|
|
x, y = view.xy(last.latitude, last.longitude)
|
|
assert (still[y - 6:y + 7, x - 6:x + 7] >= fm.RAMP).any()
|
|
|
|
|
|
def test_a_whole_evening_as_one_picture_keeps_every_aircraft_on_it(tmp_path):
|
|
tracks = two_aircraft()
|
|
for t in tracks: # heard hours ago
|
|
for f in t.fixes:
|
|
f.at -= 6 * 3600
|
|
t.first_seen -= 6 * 3600
|
|
t.last_seen -= 6 * 3600
|
|
tracks.append(straight(icao="C0FFEE", callsign="LATE", lat=51.2, lon=-0.9))
|
|
out = fm.animate(tracks, tmp_path / "evening.png", width=500)
|
|
assert out is not None and out.aircraft == 3
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Throwing out what could not have happened
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _with_bad_fix(track: Track, at_index: int, lat=-9.5, lon=-112.5) -> Track:
|
|
"""Drop one impossible position into the middle of a real flight."""
|
|
when = track.fixes[at_index].at + 0.001
|
|
track.fixes.insert(at_index + 1,
|
|
Fix(at=when, latitude=lat, longitude=lon,
|
|
altitude_ft=35_000, ground_speed_kt=400.0))
|
|
return track
|
|
|
|
|
|
def test_a_position_no_aircraft_could_reach_is_thrown_out():
|
|
from bandsaunter.flightlog import recheck
|
|
|
|
track = _with_bad_fix(straight(), 10)
|
|
good = len(track.fixes) - 1
|
|
clean, dropped = recheck([track])
|
|
assert dropped == 1
|
|
assert len(clean[0].fixes) == good
|
|
assert all(-90 <= f.latitude <= 90 for f in clean[0].fixes)
|
|
|
|
|
|
def test_a_flight_that_is_entirely_real_loses_nothing():
|
|
from bandsaunter.flightlog import recheck
|
|
|
|
clean, dropped = recheck(two_aircraft())
|
|
assert dropped == 0
|
|
assert [len(t.fixes) for t in clean] == [len(t.fixes) for t in two_aircraft()]
|
|
|
|
|
|
def test_a_bad_position_early_on_does_not_take_the_flight_with_it():
|
|
"""Keeping whatever is reachable from the last position kept lets one
|
|
bad fix become the reference, and then the truth is what gets thrown
|
|
away. On a real recording that discarded a fifth of everything."""
|
|
from bandsaunter.flightlog import recheck
|
|
|
|
track = _with_bad_fix(straight(), 0)
|
|
clean, dropped = recheck([track])
|
|
assert dropped == 1
|
|
assert len(clean[0].fixes) == len(straight().fixes)
|
|
|
|
|
|
def test_a_run_of_bad_positions_that_agree_with_each_other_still_goes():
|
|
"""Three bad decodes can land in the same wrong place and agree
|
|
perfectly; they are still wrong."""
|
|
from bandsaunter.flightlog import recheck
|
|
|
|
track = straight()
|
|
where = 10
|
|
for i in range(3):
|
|
track.fixes.insert(where + 1 + i,
|
|
Fix(at=track.fixes[where].at + 0.001 * (i + 1),
|
|
latitude=-9.5 + i * 0.001, longitude=-112.5,
|
|
altitude_ft=35_000))
|
|
clean, dropped = recheck([track])
|
|
assert dropped == 3
|
|
assert all(f.latitude > 0 for f in clean[0].fixes)
|
|
|
|
|
|
def test_a_position_that_is_not_on_earth_never_survives():
|
|
from bandsaunter.flightlog import recheck
|
|
|
|
track = _with_bad_fix(straight(), 5, lat=239.6, lon=-111.0)
|
|
clean, _ = recheck([track])
|
|
assert all(-90 <= f.latitude <= 90 for f in clean[0].fixes)
|
|
|
|
|
|
def test_an_aircraft_that_went_quiet_and_came_back_is_not_an_error():
|
|
"""Out of range for ten minutes, then heard again a long way off: that
|
|
is an aeroplane, not a bad decode, and there is no way to say otherwise."""
|
|
from bandsaunter.flightlog import recheck
|
|
|
|
track = straight(seconds=200.0)
|
|
last = track.fixes[-1]
|
|
track.fixes.append(Fix(at=last.at + 900, latitude=last.latitude + 1.5,
|
|
longitude=last.longitude + 2.0,
|
|
altitude_ft=35_000, ground_speed_kt=480.0))
|
|
clean, dropped = recheck([track])
|
|
assert dropped == 0
|
|
|
|
|
|
def test_two_positions_that_contradict_each_other_do_not_both_stand():
|
|
"""One of them is wrong and nothing says which; what comes out has at
|
|
least to be a story."""
|
|
from bandsaunter.flightlog import recheck
|
|
|
|
track = Track(icao="4CA1FA", fixes=[
|
|
Fix(at=0.0, latitude=51.5, longitude=-0.12),
|
|
Fix(at=0.5, latitude=-9.5, longitude=-112.5)])
|
|
clean, dropped = recheck([track])
|
|
assert dropped == 1
|
|
assert len(clean[0].fixes) == 1
|
|
|
|
|
|
def test_a_jitter_of_a_few_yards_is_never_called_an_error():
|
|
"""Positions arrive twice a second and are stamped to the millisecond,
|
|
so two of them a thousandth of a second apart imply thousands of knots
|
|
across a few yards."""
|
|
from bandsaunter.flightlog import recheck
|
|
|
|
track = Track(icao="4CA1FA", fixes=[
|
|
Fix(at=0.0, latitude=51.5000, longitude=-0.1200),
|
|
Fix(at=0.001, latitude=51.5001, longitude=-0.1201),
|
|
Fix(at=0.002, latitude=51.5002, longitude=-0.1202)])
|
|
clean, dropped = recheck([track])
|
|
assert dropped == 0
|
|
|
|
|
|
def test_nothing_survives_that_needed_an_impossible_speed():
|
|
"""The promise of the whole thing, said as one assertion."""
|
|
from bandsaunter.flightlog import (MAX_GROUND_SPEED_KT, distance_nm,
|
|
implied_speed_kt, recheck)
|
|
|
|
tracks = [_with_bad_fix(straight(), i) for i in (3, 12, 20)]
|
|
clean, dropped = recheck(tracks)
|
|
assert dropped == 3
|
|
for track in clean:
|
|
for a, b in zip(track.fixes, track.fixes[1:]):
|
|
if 0 < b.at - a.at <= 300 and distance_nm(
|
|
a.latitude, a.longitude, b.latitude, b.longitude) > 2:
|
|
assert implied_speed_kt(a, b) <= MAX_GROUND_SPEED_KT
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# What is written beside each aircraft
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class _Entry:
|
|
"""A register's answer, in the shape flightmap reads it."""
|
|
|
|
def __init__(self, **over):
|
|
self.type_code = over.get("type_code", "B739")
|
|
self.model = over.get("model", "737-932ER")
|
|
self.registration = over.get("registration", "N904DN")
|
|
self.origin_code = over.get("origin_code", "KATL")
|
|
self.origin = over.get("origin", "Atlanta")
|
|
self.origin_country = over.get("origin_country", "US")
|
|
self.destination_code = over.get("destination_code", "EGLL")
|
|
self.destination = over.get("destination", "London Heathrow")
|
|
self.destination_country = over.get("destination_country", "GB")
|
|
# The map draws an airport where a lookup gave it a position.
|
|
self.origin_lat = over.get("origin_lat", 33.6367)
|
|
self.origin_lon = over.get("origin_lon", -84.4281)
|
|
self.destination_lat = over.get("destination_lat", 51.4706)
|
|
self.destination_lon = over.get("destination_lon", -0.4619)
|
|
|
|
|
|
def test_the_label_says_height_speed_type_and_both_ends_of_the_route():
|
|
track = straight()
|
|
rows = fm.label_lines(track, track.fixes[0], "knots", _Entry())
|
|
text = [line for line, _flag in rows]
|
|
assert text[0].startswith("350") # flight level
|
|
assert "480KT" in text[0]
|
|
assert "B739 N904DN" in text
|
|
assert "KATL" in text and "EGLL" in text
|
|
|
|
|
|
def test_each_end_of_the_route_carries_its_own_country():
|
|
track = straight()
|
|
rows = dict((line, flag) for line, flag in
|
|
fm.label_lines(track, track.fixes[0], "knots", _Entry()))
|
|
assert rows["KATL"] == "US"
|
|
assert rows["EGLL"] == "GB"
|
|
|
|
|
|
def test_with_no_register_the_label_is_what_the_aircraft_itself_said():
|
|
track = straight()
|
|
rows = fm.label_lines(track, track.fixes[0], "knots", None)
|
|
assert len(rows) == 1 and rows[0][1] == ""
|
|
|
|
|
|
def test_an_airport_with_no_code_is_named_short_rather_than_in_full():
|
|
track = straight()
|
|
entry = _Entry(origin_code="", origin="Hartsfield Jackson Atlanta "
|
|
"International Airport")
|
|
said = [line for line, _ in fm.label_lines(track, track.fixes[0],
|
|
"knots", entry)]
|
|
assert "Hartsfield Jackson" in said
|
|
assert not any(len(line) > 20 for line in said), said
|
|
|
|
|
|
def test_the_flag_is_drawn_in_the_flag_colours():
|
|
from bandsaunter.flags import COLOUR_ORDER, FLAG_H, FLAG_W
|
|
|
|
img = np.full((40, 60), fm.BG, dtype=np.uint8)
|
|
fm.draw_flag(img, 5, 5, "JP")
|
|
patch = img[5:5 + FLAG_H, 5:5 + FLAG_W]
|
|
assert (patch >= fm.FLAG).all()
|
|
assert (patch < fm.FLAG + len(COLOUR_ORDER)).all()
|
|
# White at the corner and red in the middle: the flag of Japan.
|
|
assert patch[0, 0] == fm.flag_index("w")
|
|
assert patch[4, 6] == fm.flag_index("r")
|
|
|
|
|
|
def test_a_country_with_no_flag_is_named_in_letters_instead():
|
|
img = np.full((40, 60), fm.BG, dtype=np.uint8)
|
|
fm.draw_flag(img, 5, 5, "ZZ")
|
|
assert (img == fm.GRID).any() # the letters, in grey
|
|
assert not (img >= fm.FLAG).any() # and no flag pretending to be one
|
|
|
|
|
|
def test_a_flag_off_the_edge_of_the_picture_paints_nothing():
|
|
for x, y in ((-40, 5), (5, -40), (200, 5), (5, 200)):
|
|
img = np.full((40, 60), fm.BG, dtype=np.uint8)
|
|
fm.draw_flag(img, x, y, "US")
|
|
assert (img == fm.BG).all(), (x, y)
|
|
|
|
|
|
def test_the_flag_colours_fit_in_the_palette_beside_everything_else():
|
|
from bandsaunter.flags import COLOUR_ORDER
|
|
|
|
assert fm.FLAG + len(COLOUR_ORDER) < fm.TRANSPARENT
|
|
assert fm.PALETTE.shape == (256, 3)
|
|
# And each index really is the colour it claims.
|
|
from bandsaunter.flags import COLOURS
|
|
|
|
for letter in COLOUR_ORDER:
|
|
assert tuple(int(v) for v in fm.PALETTE[fm.flag_index(letter)]) == \
|
|
COLOURS[letter]
|
|
|
|
|
|
def test_the_register_is_asked_once_per_aircraft_not_once_per_frame(tmp_path):
|
|
"""A five-hundred-frame animation asking the same question five hundred
|
|
times would be five hundred times as rude."""
|
|
asked = []
|
|
|
|
class _Book:
|
|
def get(self, icao, callsign=""):
|
|
asked.append(icao)
|
|
return _Entry()
|
|
|
|
fm.animate(two_aircraft(), tmp_path / "counted.gif", fps=8, seconds=3,
|
|
width=400, book=_Book())
|
|
assert asked, "the register was never asked at all"
|
|
assert len(asked) == len(set(asked)), f"asked twice about the same: {asked}"
|
|
|
|
|
|
def test_the_route_reaches_the_drawn_picture(tmp_path):
|
|
"""End to end: a register with a route, and the flags on the frame."""
|
|
class _Book:
|
|
def get(self, icao, callsign=""):
|
|
return _Entry()
|
|
|
|
tracks = two_aircraft()
|
|
view = fm.fit(tracks, width=700)
|
|
base = fm.background(view)
|
|
known = {t.icao: _Entry() for t in tracks}
|
|
frame = fm.render_frame(base, view, tracks, tracks[0].first_seen + 60,
|
|
known=known)
|
|
assert (frame >= fm.FLAG).any(), "no flag was drawn"
|
|
assert _has_text(frame, "KATL")
|
|
assert _has_text(frame, "B739")
|
|
|
|
|
|
def test_a_crowded_frame_goes_back_to_the_short_label():
|
|
"""Five lines beside each of three hundred aircraft is not more
|
|
information, it is a page of overlapping text with a map behind it."""
|
|
many = [straight(icao=f"{i:06X}", callsign=f"FLT{i}",
|
|
lat=51.0 + i * 0.02, lon=-1.0 + i * 0.02)
|
|
for i in range(fm.CROWDED + 4)]
|
|
view = fm.fit(many, width=900)
|
|
base = fm.background(view)
|
|
known = {t.icao: _Entry() for t in many}
|
|
frame = fm.render_frame(base, view, many, many[0].first_seen + 60,
|
|
known=known)
|
|
assert not (frame >= fm.FLAG).any(), "still drawing flags when crowded"
|
|
assert not _has_text(frame, "B739")
|
|
|
|
|
|
def test_a_quiet_frame_keeps_every_detail():
|
|
tracks = two_aircraft()
|
|
view = fm.fit(tracks, width=700)
|
|
base = fm.background(view)
|
|
known = {t.icao: _Entry() for t in tracks}
|
|
frame = fm.render_frame(base, view, tracks, tracks[0].first_seen + 60,
|
|
known=known)
|
|
assert (frame >= fm.FLAG).any()
|
|
assert _has_text(frame, "B739")
|
|
|
|
|
|
def test_the_callsign_and_height_survive_a_crowd():
|
|
"""Whatever else goes, what the aircraft itself said stays."""
|
|
many = [straight(icao=f"{i:06X}", callsign=f"FLT{i}",
|
|
lat=51.0 + i * 0.02, lon=-1.0 + i * 0.02)
|
|
for i in range(fm.CROWDED + 4)]
|
|
view = fm.fit(many, width=900)
|
|
frame = fm.render_frame(fm.background(view), view, many,
|
|
many[0].first_seen + 60,
|
|
known={t.icao: _Entry() for t in many})
|
|
assert _has_text(frame, "FLT0")
|
|
|
|
|
|
def test_labels_step_aside_rather_than_landing_on_each_other():
|
|
"""Two aircraft passing close together is exactly the moment somebody is
|
|
looking at that part of the picture."""
|
|
close = [straight(icao=f"{i:06X}", callsign=f"FLT{i}",
|
|
lat=51.0 + i * 0.004, lon=-1.0 + i * 0.004, seconds=60)
|
|
for i in range(6)]
|
|
view = fm.fit(close, width=800, box=(50.8, -1.4, 51.3, -0.6))
|
|
taken = []
|
|
base = fm.background(view)
|
|
fm.render_frame(base, view, close, close[0].first_seen, labels=True)
|
|
# Place them by hand so the boxes can be compared.
|
|
for track in close:
|
|
now = track.at(track.first_seen)
|
|
x, y = view.xy(now.latitude, now.longitude)
|
|
fm._label(base, x, y, track, now, fm.RAMP, taken)
|
|
for i, one in enumerate(taken):
|
|
for two in taken[i + 1:]:
|
|
assert not fm._overlaps(one, two), f"{one} overlaps {two}"
|