FAA Delays

"""FAA NAS delay response models for /delays/faa endpoints."""from __future__ import annotationsfrom dataclasses import dataclassfrom typing import Literal@dataclassclass GroundDelay:    airport: str    reason: str    airport_name: str | None = None    avg_delay: str | None = None    max_delay: str | None = None@dataclassclass GroundStop:    airport: str    reason: str    airport_name: str | None = None    end_time: str | None = None@dataclassclass Closure:    airport: str    reason: str    airport_name: str | None = None    begin: str | None = None    reopen: str | None = None@dataclassclass AirspaceFlowProgram:    facility: str    reason: str    fca_start: str | None = None    fca_end: str | None = None@dataclassclass FaaDelayResponse:    ground_delays: list[GroundDelay]    ground_stops: list[GroundStop]    closures: list[Closure]    airspace_flow_programs: list[AirspaceFlowProgram]    total_alerts: int = 0    message: str | None = NoneDelayKind = Literal["GDP", "GS", "Closure", "AFP"]@dataclassclass DelayAlert:    kind: DelayKind    airport: str    reason: str    avg_delay: str | None    max_delay: str | None

import osimport timefrom dataclasses import dataclassimport requestsHEADERS = {    "x-api-key": os.getenv("SKYLINK_API_KEY", "YOUR_API_KEY")}BASE = "https://data.skylinkapi.com/v3"CACHE: dict[str, tuple[dict, float]] = {}DELAY_TTL = 120  # 2 min - FAA feed is real-time@dataclassclass DelayAlert:    kind: str    airport: str    reason: str    avg_delay: str | None    max_delay: str | Nonedef fetch_faa_delays(icao: str | None = None) -> dict:    key = f"delays:{icao or 'all'}"    cached = CACHE.get(key)    if cached and time.time() < cached[1]:        return cached[0]    path = f"{BASE}/delays/faa/{icao.upper()}" if icao else f"{BASE}/delays/faa"    try:        r = requests.get(path, headers=HEADERS, timeout=(10, 20))    except requests.Timeout as exc:        raise RuntimeError("SkyLink timeout fetching FAA delays") from exc    r.raise_for_status()    data = r.json()    CACHE[key] = (data, time.time() + DELAY_TTL)    return datadef delay_alerts(icao: str | None = None) -> list[DelayAlert]:    data = fetch_faa_delays(icao)    alerts: list[DelayAlert] = []    for item in data.get("ground_delays", []):        alerts.append(            DelayAlert(                kind="GDP",                airport=item["airport"],                reason=item["reason"],                avg_delay=item.get("avg_delay"),                max_delay=item.get("max_delay"),            )        )    for item in data.get("ground_stops", []):        alerts.append(            DelayAlert(kind="GS", airport=item["airport"], reason=item["reason"], avg_delay=None, max_delay=None)        )    for item in data.get("closures", []):        alerts.append(            DelayAlert(kind="Closure", airport=item["airport"], reason=item["reason"], avg_delay=None, max_delay=None)        )    for item in data.get("airspace_flow_programs", []):        alerts.append(            DelayAlert(                kind="AFP",                airport=item["facility"],                reason=item["reason"],                avg_delay=None,                max_delay=None,            )        )    return alertsif __name__ == "__main__":    import json    delays = fetch_faa_delays("KJFK")    print(json.dumps(delays, indent=2, default=str))