bandsaunter/tests/test_flights.py
The Dust Council eae60cb04d Write the aircraft down, and draw where they went
ADS-B was a live table and nothing else: an aircraft was overhead for four
minutes and then gone, with nothing kept.  Now everything heard goes into
adsb_<time>.jsonl as it arrives -- one object per frame, the raw hex beside
what was read out of it, flushed per line because a listening session ends
with control-C -- with a readable report beside it.

flights.py asks who the aircraft are: adsbdb for the airframe and the
route, hexdb behind it, cached for a month.  What needs no website is
answered without one, because the ICAO address block says which country
registered the aircraft and the first three letters of an airline callsign
are its designator.  Nothing but the address and the callsign heard on the
air is ever sent.

  bandsaunter flights [LOG...] --out sky.gif

reads a log back and draws the evening as a map with the clock running.
Every frame is a moment: each aircraft is where it actually was then,
interpolated between the position reports either side of it and
dead-reckoned from its last speed and heading between them, and dropped
rather than guessed at once it has not been heard for --stale seconds.
The GIF is written here -- palette, LZW, frame differencing against a
transparent index -- so nothing but numpy is needed; ffmpeg writes an MP4
where it happens to be installed, and .png draws the whole evening at once.

The decoder needed 6.3 s to read a second of sky, so a live capture was
losing six frames in seven.  Reading the bits off a running total instead
of summing each window takes that to 0.6 s, with identical output.

--simulate flies six aircraft that are not there past a receiver that is
not there, through the real encoder, the real checksum and the real
decoder, so all of this can be tried without an aerial.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016PsWPTweCT6pwxKngvVxcg
2026-09-03 22:22:49 -07:00

488 lines
19 KiB
Python

"""Aircraft: who they are, what was written down, and what it reads back as.
Nothing here goes near a network. The registers are stubbed with the shapes
they really answer with -- checked against the live services when this was
written -- so a test failing means the reader changed, not that a website is
down.
"""
import json
import time
import pytest
from bandsaunter.flights import (Flight, FlightBook, airline_of,
describe_address)
from bandsaunter.flightlog import (Fix, FlightLog, Track, bearing_deg,
distance_nm, move, read_logs, report,
write_kml)
# ---------------------------------------------------------------------------
# What the address alone says
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("icao,country", [
("A835AF", "United States"), # N628TS
("406B3B", "United Kingdom"), # G-DGEF
("3C6444", "Germany"), # D-AIBD
("4CA1FA", "Ireland"), # EI-DDH
("7C4A1B", "Australia"),
("C01234", "Canada"),
("484200", "Netherlands"),
])
def test_the_address_says_which_country_registered_it(icao, country):
"""Fixed by treaty, so no website is needed and none is asked."""
assert describe_address(icao) == country
def test_an_address_in_no_block_is_not_guessed_at():
assert describe_address("F00000") == ""
assert describe_address("") == ""
assert describe_address("nonsense") == ""
def test_a_military_block_is_named_before_the_country_it_sits_inside():
assert describe_address("ADFFFF") == "United States military"
@pytest.mark.parametrize("callsign,airline", [
("RYR1234", "Ryanair"),
("BAW49", "British Airways"),
("UAL1902", "United Airlines"),
("RCH445", "United States Air Mobility Command (Reach)"),
])
def test_the_callsign_says_which_airline_is_flying(callsign, airline):
assert airline_of(callsign) == airline
@pytest.mark.parametrize("callsign", ["N517HP", "G-ABCD", "", "XX"])
def test_a_registration_flown_as_a_callsign_is_not_an_airline(callsign):
"""A private aircraft uses its registration, and reading three letters of
that as an airline designator would invent one."""
assert airline_of(callsign) == ""
# ---------------------------------------------------------------------------
# The registers
# ---------------------------------------------------------------------------
ADSBDB_AIRCRAFT = {"response": {"aircraft": {
"type": "737 8AS", "icao_type": "B738", "manufacturer": "Boeing",
"mode_s": "4CA1FA", "registration": "EI-DYP",
"registered_owner_country_iso_name": "IE",
"registered_owner_country_name": "Ireland",
"registered_owner": "Ryanair"}}}
ADSBDB_ROUTE = {"response": {"flightroute": {
"callsign": "RYR1234",
"airline": {"name": "Ryanair", "icao": "RYR"},
"origin": {"icao_code": "EGSS", "iata_code": "STN",
"name": "London Stansted Airport", "municipality": "London",
"latitude": 51.885, "longitude": 0.235},
"destination": {"icao_code": "EGNX", "iata_code": "EMA",
"name": "East Midlands Airport", "municipality": "Nottingham",
"latitude": 52.8311, "longitude": -1.32806}}}}
HEXDB_AIRCRAFT = {"ModeS": "3C6444", "Registration": "D-AIBD",
"Manufacturer": "Airbus", "ICAOTypeCode": "A319",
"Type": "A319 112", "RegisteredOwners": "Lufthansa",
"OperatorFlagCode": "DLH"}
@pytest.fixture
def register(monkeypatch, tmp_path):
"""A book that answers from a table instead of the internet."""
asked: list[str] = []
answers: dict[str, object] = {}
def request(self, url):
asked.append(url)
for fragment, body in answers.items():
if fragment in url:
return body
raise OSError("not found")
monkeypatch.setattr(FlightBook, "_request", request)
book = FlightBook(cache=tmp_path / "flights.json")
return book, asked, answers
def test_a_register_answers_with_the_airframe_and_the_route(register):
book, asked, answers = register
answers["adsbdb.com/v0/aircraft"] = ADSBDB_AIRCRAFT
answers["adsbdb.com/v0/callsign"] = ADSBDB_ROUTE
entry = book.get("4CA1FA", "RYR1234")
book.wait(5.0)
assert entry.status == "found"
assert entry.registration == "EI-DYP"
assert entry.type_code == "B738"
assert entry.operator == "Ryanair"
assert entry.origin_code == "EGSS" and entry.destination_code == "EGNX"
assert "Stansted" in entry.route and "East Midlands" in entry.route
assert entry.origin_lat == pytest.approx(51.885)
def test_the_second_register_is_asked_when_the_first_has_nothing(register):
book, asked, answers = register
answers["hexdb.io/api/v1/aircraft"] = HEXDB_AIRCRAFT
entry = book.get("3C6444")
book.wait(5.0)
assert entry.status == "found"
assert entry.registration == "D-AIBD"
assert entry.manufacturer == "Airbus"
assert any("adsbdb" in url for url in asked), "the first was skipped"
def test_a_route_written_as_one_string_is_read_as_two_airports(register):
book, asked, answers = register
answers["hexdb.io/api/v1/route"] = {"flight": "BAW123",
"route": "EGLL-OTHH"}
entry = book.get("400001", "BAW123")
book.wait(5.0)
assert (entry.origin_code, entry.destination_code) == ("EGLL", "OTHH")
def test_an_aircraft_no_register_holds_is_remembered_as_unlisted(register):
book, asked, answers = register
answers["adsbdb.com/v0/aircraft"] = {"response": {"aircraft": None}}
entry = book.get("ABCDEF")
book.wait(5.0)
assert entry.status == "unlisted"
assert entry.country == "United States" # the address still says this
def test_nothing_reachable_leaves_what_the_address_said(register):
book, asked, answers = register # no answers at all
entry = book.get("4CA1FA", "RYR1234")
book.wait(5.0)
assert entry.status == "offline"
assert entry.country == "Ireland"
assert entry.airline == "Ryanair" # from the callsign, not a website
def test_offline_asks_nobody(tmp_path, monkeypatch):
def refuse(self, url):
raise AssertionError(f"asked {url} while offline")
monkeypatch.setattr(FlightBook, "_request", refuse)
book = FlightBook(online=False, cache=tmp_path / "c.json")
entry = book.get("A835AF", "UAL1902")
book.wait(1.0)
assert entry.status == "local"
assert entry.country == "United States"
assert entry.airline == "United Airlines"
def test_only_the_address_and_the_callsign_are_ever_sent(register):
"""A register is told what was heard on the air and nothing else."""
book, asked, answers = register
answers["adsbdb.com"] = ADSBDB_AIRCRAFT
book.get("4CA1FA", "RYR1234")
book.wait(5.0)
assert asked
for url in asked:
tail = url.split("://", 1)[1]
for piece in tail.split("/")[1:]:
assert piece in ("v0", "aircraft", "callsign", "api", "v1",
"route", "icao", "4CA1FA", "RYR1234"), url
def test_an_answer_is_kept_and_the_register_is_not_asked_twice(register,
tmp_path):
book, asked, answers = register
answers["adsbdb.com/v0/aircraft"] = ADSBDB_AIRCRAFT
book.get("4CA1FA")
book.wait(5.0)
book.save()
again = FlightBook(cache=tmp_path / "flights.json")
entry = again.get("4CA1FA")
assert entry.registration == "EI-DYP"
assert entry.status == "found"
def test_a_cache_from_another_version_is_ignored(tmp_path):
body = {"aircraft": {"4CA1FA": {"icao": "4CA1FA", "registration": "EI-DYP",
"status": "found", "version": 0,
"fetched_at": time.time()}}}
(tmp_path / "c.json").write_text(json.dumps(body))
book = FlightBook(online=False, cache=tmp_path / "c.json")
assert book.get("4CA1FA").registration == ""
def test_what_is_printed_says_the_aircraft_the_operator_and_the_route():
flight = Flight(icao="4CA1FA", callsign="RYR1234", registration="EI-DYP",
manufacturer="Boeing", model="737-8AS", operator="Ryanair",
origin="Stansted", destination="East Midlands")
assert flight.aircraft == "Boeing 737-8AS (EI-DYP)"
assert flight.route == "Stansted → East Midlands"
assert "Ryanair" in flight.summary()
assert "registration: EI-DYP" in flight.details()
# ---------------------------------------------------------------------------
# The log
# ---------------------------------------------------------------------------
class _Frame:
"""Just enough of a decoded frame for the log to write one down."""
def __init__(self, icao="4CA1FA", df=17, tc=11, data=b"\x8d\x4c\xa1\xfa",
callsign="", altitude_ft=0, ground_speed_kt=0.0,
track_deg=0.0, vertical_rate_fpm=0):
self.icao, self.df, self.type_code, self.data = icao, df, tc, data
self.callsign = callsign
self.altitude_ft = altitude_ft
self.ground_speed_kt = ground_speed_kt
self.track_deg = track_deg
self.vertical_rate_fpm = vertical_rate_fpm
class _Craft:
def __init__(self, lat=0.0, lon=0.0, callsign=""):
self.latitude, self.longitude, self.callsign = lat, lon, callsign
@property
def located(self):
return bool(self.latitude or self.longitude)
def _log(tmp_path, name="adsb_test.jsonl"):
return FlightLog(tmp_path / name, receiver="test", frequency=1090e6,
sample_rate=2e6, started=1_000_000.0)
def test_the_log_keeps_the_raw_frame_next_to_what_was_read_out_of_it(tmp_path):
log = _log(tmp_path)
log.append(_Frame(data=bytes.fromhex("8D4CA1FA9905A11E202C00D3450D"),
altitude_ft=35000),
_Craft(51.5, -0.12), when=1_000_001.0)
log.close()
lines = [json.loads(x) for x in
log.path.read_text().splitlines() if x.strip()]
assert lines[0]["log"] == "bandsaunter-adsb"
assert lines[1]["hex"] == "8D4CA1FA9905A11E202C00D3450D"
assert lines[1]["lat"] == 51.5 and lines[1]["alt_ft"] == 35000
def test_the_log_is_on_the_disk_before_the_session_ends(tmp_path):
"""A listening session ends with control-C, so nothing may wait for a
clean shutdown."""
log = _log(tmp_path)
log.append(_Frame(), _Craft(51.5, -0.12), when=1_000_001.0)
assert "4CA1FA" in log.path.read_text() # not closed, already written
log.close()
def test_what_was_written_reads_back_as_a_track(tmp_path):
log = _log(tmp_path)
for i in range(5):
log.append(_Frame(altitude_ft=30000 + i * 100, ground_speed_kt=420.0,
track_deg=90.0, callsign="RYR1234"),
_Craft(51.5 + i * 0.01, -0.12 + i * 0.02, "RYR1234"),
when=1_000_000.0 + i * 10)
log.close()
tracks = read_logs(log.path)
assert len(tracks) == 1
track = tracks[0]
assert track.icao == "4CA1FA" and track.callsign == "RYR1234"
assert len(track.fixes) == 5
assert track.frames == 5
assert track.distance_nm > 0
assert track.altitude_range == (30000, 30400)
def test_an_aircraft_that_never_moved_is_one_point_not_seven_thousand(tmp_path):
"""A transponder on a stand reports the same place twice a second."""
log = _log(tmp_path)
for i in range(50):
log.append(_Frame(altitude_ft=0), _Craft(51.5, -0.12),
when=1_000_000.0 + i)
log.close()
track = read_logs(log.path)[0]
assert len(track.fixes) == 1
assert track.frames == 50
def test_two_logs_read_as_one_evening(tmp_path):
first = _log(tmp_path, "a.jsonl")
first.append(_Frame(), _Craft(51.5, -0.12), when=1_000_000.0)
first.close()
second = _log(tmp_path, "b.jsonl")
second.append(_Frame(icao="3C6444"), _Craft(50.0, 8.0), when=1_000_100.0)
second.close()
tracks = read_logs([first.path, second.path])
assert [t.icao for t in tracks] == ["4CA1FA", "3C6444"]
def test_a_half_written_last_line_does_not_lose_the_rest(tmp_path):
"""Control-C during a write, or a full disk: the evening still reads."""
log = _log(tmp_path)
log.append(_Frame(), _Craft(51.5, -0.12), when=1_000_000.0)
log.close()
with log.path.open("a") as handle:
handle.write('{"t": 1000001.0, "icao": "3C64')
assert len(read_logs(log.path)) == 1
def test_a_position_fix_is_given_the_speed_the_aircraft_last_reported(tmp_path):
"""Position and velocity arrive in different frames; the aeroplane is the
same aeroplane a second later."""
log = _log(tmp_path)
log.append(_Frame(ground_speed_kt=420.0, track_deg=90.0), None,
when=1_000_000.0)
log.append(_Frame(), _Craft(51.5, -0.12), when=1_000_001.0)
log.append(_Frame(), _Craft(51.6, -0.10), when=1_000_011.0)
log.close()
track = read_logs(log.path)[0]
assert track.fixes[0].ground_speed_kt == 0.0 or track.fixes[0].track_deg
assert track.fixes[-1].ground_speed_kt == 420.0
# ---------------------------------------------------------------------------
# Time and speed: where an aircraft was between two reports
# ---------------------------------------------------------------------------
def _track(**over) -> Track:
fixes = [Fix(at=0.0, latitude=51.0, longitude=0.0, altitude_ft=10_000,
ground_speed_kt=600.0, track_deg=90.0),
Fix(at=60.0, latitude=51.0, longitude=0.2648, altitude_ft=12_000,
ground_speed_kt=600.0, track_deg=90.0)]
track = Track(icao="4CA1FA", callsign="RYR1234", fixes=fixes,
first_seen=0.0, last_seen=60.0)
for key, value in over.items():
setattr(track, key, value)
return track
def test_halfway_between_two_reports_is_halfway_along():
fix = _track().at(30.0)
assert fix.longitude == pytest.approx(0.1324, abs=1e-3)
assert fix.altitude_ft == pytest.approx(11_000, abs=20)
def test_before_the_first_report_the_aircraft_is_not_drawn():
assert _track().at(-1.0) is None
def test_after_the_last_report_it_carries_on_at_the_speed_it_said():
"""Ten knots-minutes on: dead reckoning, not a jump."""
fix = _track().at(120.0, stale=300.0)
assert fix is not None
flown = distance_nm(51.0, 0.2648, fix.latitude, fix.longitude)
assert flown == pytest.approx(10.0, rel=0.05) # 600 kt for a minute
def test_an_aircraft_not_heard_for_a_long_time_stops_being_drawn():
"""Five minutes on it has flown fifty miles and is a guess."""
assert _track().at(60.0 + 400.0, stale=300.0) is None
def test_the_trail_is_everywhere_it_had_been_by_then():
track = _track()
assert len(track.trail(30.0)) == 2 # one fix and where it is
assert track.trail(30.0)[-1].longitude == pytest.approx(0.1324, abs=1e-3)
def test_a_bearing_and_a_distance_agree_with_the_move_they_describe():
lat, lon = move(51.0, 0.0, 90.0, 60.0)
assert distance_nm(51.0, 0.0, lat, lon) == pytest.approx(60.0, rel=1e-3)
assert bearing_deg(51.0, 0.0, lat, lon) == pytest.approx(90.0, abs=0.5)
def test_a_track_across_the_date_line_does_not_go_the_long_way_round():
track = Track(fixes=[Fix(at=0.0, latitude=0.0, longitude=179.9),
Fix(at=10.0, latitude=0.0, longitude=-179.9)])
fix = track.at(5.0)
assert abs(fix.longitude) > 179.0
# ---------------------------------------------------------------------------
# The readable report, and the map for Google Earth
# ---------------------------------------------------------------------------
def test_the_report_says_who_where_and_how_far(tmp_path):
log = _log(tmp_path)
for i in range(4):
log.append(_Frame(callsign="RYR1234", altitude_ft=30_000,
ground_speed_kt=420.0, track_deg=90.0),
_Craft(51.5, -0.12 + i * 0.05, "RYR1234"),
when=1_000_000.0 + i * 30)
log.close()
lines = report(read_logs(log.path), title="test")
text = "\n".join(lines)
assert "1 aircraft" in text
assert "4CA1FA" in text and "RYR1234" in text
assert "flew:" in text and "altitude: 30,000 ft" in text
def test_the_report_keeps_the_register_apart_from_the_air(register, tmp_path):
book, asked, answers = register
answers["adsbdb.com/v0/aircraft"] = ADSBDB_AIRCRAFT
book.get("4CA1FA")
book.wait(5.0)
log = _log(tmp_path)
log.append(_Frame(), _Craft(51.5, -0.12), when=1_000_000.0)
log.close()
text = "\n".join(report(read_logs(log.path), book))
assert "registration: EI-DYP" in text
assert "heard:" in text
def test_nothing_heard_is_said_rather_than_drawn_as_an_empty_table():
assert report([]) == ["nothing heard."]
def test_google_earth_gets_the_whole_path_not_just_the_last_place(tmp_path):
log = _log(tmp_path)
for i in range(3):
log.append(_Frame(altitude_ft=10_000), _Craft(51.5 + i * 0.1, -0.12),
when=1_000_000.0 + i * 20)
log.close()
out = write_kml(tmp_path / "flights.kml", read_logs(log.path))
body = out.read_text()
assert "<LineString>" in body
assert body.count(",") > 3
assert "3048" in body or "3048" in body.replace(" ", "") # feet to metres
def test_a_log_with_no_positions_makes_no_map(tmp_path):
log = _log(tmp_path)
log.append(_Frame(), None, when=1_000_000.0)
log.close()
assert write_kml(tmp_path / "flights.kml", read_logs(log.path)) is None
# ---------------------------------------------------------------------------
# End to end, through the real decoder
# ---------------------------------------------------------------------------
def test_a_simulated_sky_is_heard_recorded_and_read_back(tmp_path):
"""The whole path: frames encoded, modulated, decoded, logged, re-read."""
from bandsaunter.adsb import AircraftRegistry, SAMPLE_RATE, decode_frames
from bandsaunter.adsb import SimulatedSky, default_sky
sky = SimulatedSky(default_sky(seed=3), noise=0.02, seed=1)
registry = AircraftRegistry()
log = _log(tmp_path)
when = 1_000_000.0
for _ in range(8):
block = sky.read_samples(int(SAMPLE_RATE))
for frame in decode_frames(block, SAMPLE_RATE):
craft = registry.add(frame, when=when + frame.at_sample / SAMPLE_RATE)
log.append(frame, craft, when=when + frame.at_sample / SAMPLE_RATE)
when += 1.0
log.close()
tracks = read_logs(log.path)
assert len(tracks) == len(default_sky())
assert all(t.callsign for t in tracks)
assert all(t.located for t in tracks)
for track in tracks:
low, high = track.altitude_range
assert 0 < high < 50_000
assert track.top_speed_kt > 50
# It flew at the speed it said it was flying, give or take the
# quarter of a knot the encoding rounds to.
flown = track.distance_nm
expected = track.top_speed_kt * track.seconds / 3600.0
assert flown == pytest.approx(expected, rel=0.35, abs=0.2)