125 lines
4.4 KiB
Python
125 lines
4.4 KiB
Python
"""FAA NASSTATUS API client — live airport delay/status data, no API key required.
|
|
|
|
Endpoint: https://nasstatus.faa.gov/api/airport-status-information
|
|
Returns XML with ground stops, ground delay programs, arrival/departure delays, closures.
|
|
"""
|
|
|
|
import xml.etree.ElementTree as ET
|
|
|
|
import httpx
|
|
|
|
BASE_URL = "https://nasstatus.faa.gov/api/airport-status-information"
|
|
|
|
# Cache parsed data to avoid hitting the API for every airport query
|
|
_cache: dict | None = None
|
|
_cache_raw: str = ""
|
|
|
|
|
|
async def _fetch_all_status() -> dict[str, list[dict]]:
|
|
"""Fetch and parse all airport status information from FAA."""
|
|
global _cache, _cache_raw
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=10.0, follow_redirects=True) as client:
|
|
resp = await client.get(BASE_URL)
|
|
resp.raise_for_status()
|
|
xml_text = resp.text
|
|
|
|
# Don't re-parse if unchanged
|
|
if xml_text == _cache_raw and _cache is not None:
|
|
return _cache
|
|
|
|
_cache_raw = xml_text
|
|
_cache = _parse_status_xml(xml_text)
|
|
return _cache
|
|
|
|
except Exception as e:
|
|
return {"_error": [{"error": str(e)}]}
|
|
|
|
|
|
def _parse_status_xml(xml_text: str) -> dict[str, list[dict]]:
|
|
"""Parse FAA airport status XML into a dict of airport → delays."""
|
|
delays_by_airport: dict[str, list[dict]] = {}
|
|
root = ET.fromstring(xml_text)
|
|
|
|
for delay_type in root.findall("Delay_type"):
|
|
name = delay_type.findtext("Name", "").strip()
|
|
|
|
# Ground Stop Programs
|
|
for program in delay_type.findall(".//Program"):
|
|
arpt = program.findtext("ARPT", "").strip()
|
|
if not arpt:
|
|
continue
|
|
delays_by_airport.setdefault(arpt, []).append({
|
|
"type": "ground_stop",
|
|
"reason": program.findtext("Reason", "unknown").strip(),
|
|
"end_time": program.findtext("End_Time", "").strip() or None,
|
|
})
|
|
|
|
# Ground Delay Programs
|
|
for gdp in delay_type.findall(".//Ground_Delay"):
|
|
arpt = gdp.findtext("ARPT", "").strip()
|
|
if not arpt:
|
|
continue
|
|
delays_by_airport.setdefault(arpt, []).append({
|
|
"type": "ground_delay_program",
|
|
"reason": gdp.findtext("Reason", "unknown").strip(),
|
|
"average_delay": gdp.findtext("Avg", "").strip() or None,
|
|
"max_delay": gdp.findtext("Max", "").strip() or None,
|
|
})
|
|
|
|
# Arrival/Departure Delays
|
|
for delay in delay_type.findall(".//Delay"):
|
|
arpt = delay.findtext("ARPT", "").strip()
|
|
if not arpt:
|
|
continue
|
|
delays_by_airport.setdefault(arpt, []).append({
|
|
"type": name.lower().replace(" ", "_") if name else "delay",
|
|
"reason": delay.findtext("Reason", "unknown").strip(),
|
|
"average_delay": delay.findtext("Avg", "").strip() or None,
|
|
"max_delay": delay.findtext("Max", "").strip() or None,
|
|
"trend": delay.findtext("Trend", "").strip() or None,
|
|
})
|
|
|
|
# Closures
|
|
for closure in delay_type.findall(".//Closure"):
|
|
arpt = closure.findtext("ARPT", "").strip()
|
|
if not arpt:
|
|
continue
|
|
delays_by_airport.setdefault(arpt, []).append({
|
|
"type": "closure",
|
|
"reason": closure.findtext("Reason", "unknown").strip(),
|
|
"begin": closure.findtext("Begin", "").strip() or None,
|
|
"end": closure.findtext("End", "").strip() or None,
|
|
})
|
|
|
|
return delays_by_airport
|
|
|
|
|
|
async def get_airport_status(airport_code: str) -> dict:
|
|
"""Fetch live FAA airport status for a single airport.
|
|
|
|
Returns ground delay programs, ground stops, closures.
|
|
Falls back to 'status_unavailable' on any error.
|
|
"""
|
|
all_status = await _fetch_all_status()
|
|
|
|
if "_error" in all_status:
|
|
return {
|
|
"airport": airport_code.upper(),
|
|
"status": "status_unavailable",
|
|
"error": all_status["_error"][0]["error"],
|
|
"source": "faa_nasstatus_live",
|
|
}
|
|
|
|
airport = airport_code.upper()
|
|
delays = all_status.get(airport, [])
|
|
|
|
return {
|
|
"airport": airport,
|
|
"has_delays": len(delays) > 0,
|
|
"status": "delays_active" if delays else "normal_operations",
|
|
"delays": delays,
|
|
"source": "faa_nasstatus_live",
|
|
}
|