bsncubed
408fde440d
Photograph/screenshot a boarding pass, decode its BCBP barcode (PDF417/ Aztec/QR/DataMatrix), fetch the historical flight track from FlightAware the day after the flight, stage it for review, and write it into AirTrail via its REST API on approval. ntfy notifications carry signed Approve/Deny/Investigate actions.
71 lines
2.1 KiB
Python
71 lines
2.1 KiB
Python
"""IATA <-> ICAO lookups for airports and airlines, loaded once from the
|
|
trimmed CSVs in data/ (see scripts/build_reference_data.py). No network
|
|
access at runtime.
|
|
"""
|
|
|
|
import csv
|
|
from pathlib import Path
|
|
|
|
DATA_DIR = Path(__file__).resolve().parent / "data"
|
|
|
|
# Known-correct overrides, checked before the bundled CSV. The CSV (OpenFlights)
|
|
# can lag real-world rebrands - e.g. it still maps VA to the ICAO code of the
|
|
# defunct "V Australia Airlines" (VAU) rather than the current Virgin
|
|
# Australia code (VOZ). These are the examples called out in the brief.
|
|
CARRIER_ICAO_OVERRIDES = {
|
|
"QF": "QFA",
|
|
"VA": "VOZ",
|
|
"JQ": "JST",
|
|
"EK": "UAE",
|
|
}
|
|
|
|
|
|
def _load_csv(path: Path) -> dict[str, tuple[str, str]]:
|
|
"""Returns iata -> (icao, name)."""
|
|
result = {}
|
|
if not path.exists():
|
|
return result
|
|
with path.open(newline="", encoding="utf-8") as f:
|
|
for row in csv.DictReader(f):
|
|
iata = (row.get("iata") or "").strip().upper()
|
|
icao = (row.get("icao") or "").strip().upper()
|
|
name = (row.get("name") or "").strip()
|
|
if iata and icao:
|
|
result[iata] = (icao, name)
|
|
return result
|
|
|
|
|
|
_AIRPORTS = _load_csv(DATA_DIR / "airports.csv")
|
|
_AIRLINES = _load_csv(DATA_DIR / "airlines.csv")
|
|
|
|
|
|
def iata_to_icao_airport(iata_code: str) -> str | None:
|
|
if not iata_code:
|
|
return None
|
|
entry = _AIRPORTS.get(iata_code.strip().upper())
|
|
return entry[0] if entry else None
|
|
|
|
|
|
def airport_name(iata_code: str) -> str | None:
|
|
if not iata_code:
|
|
return None
|
|
entry = _AIRPORTS.get(iata_code.strip().upper())
|
|
return entry[1] if entry else None
|
|
|
|
|
|
def iata_to_icao_airline(iata_code: str) -> str | None:
|
|
if not iata_code:
|
|
return None
|
|
code = iata_code.strip().upper()
|
|
if code in CARRIER_ICAO_OVERRIDES:
|
|
return CARRIER_ICAO_OVERRIDES[code]
|
|
entry = _AIRLINES.get(code)
|
|
return entry[0] if entry else None
|
|
|
|
|
|
def airline_name(iata_code: str) -> str | None:
|
|
if not iata_code:
|
|
return None
|
|
entry = _AIRLINES.get(iata_code.strip().upper())
|
|
return entry[1] if entry else None
|