TAF

"""TAF response and departure-summary types - matches weather-taf field tables."""from __future__ import annotationsfrom dataclasses import dataclassfrom typing import LiteralFlightRules = Literal["VFR", "MVFR", "IFR", "LIFR"]TafPeriodType = Literal["FROM", "TEMPO", "BECMG"]@dataclassclass TafRequest:    """Path/query input for GET /weather/taf/{icao}."""    icao: str    parsed: bool = False@dataclassclass TafWind:    direction: int    speed: int    gust: int | None    variable: list[int]@dataclassclass TafVisibility:    value: float | None    repr: str@dataclassclass TafCloudLayer:    type: str    base: int    repr: str@dataclassclass TafWxCode:    repr: str    value: str@dataclassclass TafProbability:    repr: str    value: Literal[30, 40]    spoken: str@dataclassclass TafPeriod:    type: TafPeriodType    start_time: str    end_time: str    wind: TafWind    visibility: TafVisibility    clouds: list[TafCloudLayer]    wx_codes: list[TafWxCode]    flight_rules: FlightRules    probability: TafProbability | None    turbulence: list[object]    icing: list[object]@dataclassclass TafParsed:    start_time: str    end_time: str    forecast: list[TafPeriod]@dataclassclass TafResponse:    raw: str    icao: str    airport_name: str    timestamp: str    parsed: TafParsed | None = None@dataclassclass TafModifierSummary:    type: TafPeriodType    category: FlightRules    probability: int | None    wx: list[str]@dataclassclass TafDepartureSummary:    """Shape returned by taf_at_departure() in the fetch example."""    icao: str    valid_to: str    baseline_category: FlightRules | None    modifiers: list[TafModifierSummary]    worst_from_category: FlightRules | None

import osimport timefrom dataclasses import dataclassfrom datetime import datetime, timezonefrom typing import Literalimport requests# TafDepartureSummary - helper return type. Full wire types (TafResponse, TafPeriod, …)# are in the Client types section on the TAF docs page.FlightRules = Literal["VFR", "MVFR", "IFR", "LIFR"]TafPeriodType = Literal["FROM", "TEMPO", "BECMG"]@dataclassclass TafModifierSummary:    type: TafPeriodType    category: FlightRules    probability: int | None    wx: list[str]@dataclassclass TafDepartureSummary:    icao: str    valid_to: str    baseline_category: FlightRules | None    modifiers: list[TafModifierSummary]    worst_from_category: FlightRules | NoneHEADERS = {    "x-api-key": os.getenv("SKYLINK_API_KEY", "YOUR_API_KEY")}BASE = "https://data.skylinkapi.com/v3.1"CACHE: dict[str, tuple[dict, float]] = {}TAF_TTL = 900CATEGORY_RANK = {"VFR": 0, "MVFR": 1, "IFR": 2, "LIFR": 3}def parse_time(iso: str) -> datetime:    return datetime.fromisoformat(iso.replace("Z", "+00:00"))def fetch_taf(icao: str) -> dict | None:    cache_key = f"taf:{icao}:parsed"    cached = CACHE.get(cache_key)    if cached and time.time() < cached[1]:        return cached[0]    try:        r = requests.get(            f"{BASE}/weather/taf/{icao}",            headers=HEADERS,            params={"parsed": "true"},            timeout=(10, 15),        )    except requests.Timeout as exc:        raise RuntimeError(f"SkyLink timeout fetching TAF for {icao}") from exc    if r.status_code == 404:        return None    r.raise_for_status()    data = r.json()    CACHE[cache_key] = (data, time.time() + TAF_TTL)    return datadef taf_at_departure(icao: str, departure: datetime | None = None) -> TafDepartureSummary | None:    taf = fetch_taf(icao)    if not taf:        return None    at = departure or datetime.now(timezone.utc)    forecast = taf["parsed"]["forecast"]    baseline = next(        (            p            for p in forecast            if p["type"] == "FROM"            and parse_time(p["start_time"]) <= at <= parse_time(p["end_time"])        ),        None,    )    modifiers = [        TafModifierSummary(            type=p["type"],            category=p["flight_rules"],            probability=(p["probability"] or {}).get("value"),            wx=[w["repr"] for w in p["wx_codes"]],        )        for p in forecast        if p is not baseline and parse_time(p["start_time"]) <= at <= parse_time(p["end_time"])    ]    from_groups = [p for p in forecast if p["type"] == "FROM"]    worst = (        max(from_groups, key=lambda p: CATEGORY_RANK.get(p["flight_rules"], -1))["flight_rules"]        if from_groups        else None    )    return TafDepartureSummary(        icao=icao,        valid_to=taf["parsed"]["end_time"],        baseline_category=baseline["flight_rules"] if baseline else None,        modifiers=modifiers,        worst_from_category=worst,    )if __name__ == "__main__":    import json    from dataclasses import asdict    summary = taf_at_departure("KJFK")    print(json.dumps(asdict(summary), indent=2) if summary else "TAF not available")