"""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 "" 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)