METAR

"""METAR response and summary types - matches weather-metar field tables."""from __future__ import annotationsfrom dataclasses import dataclassfrom typing import LiteralFlightRules = Literal["VFR", "MVFR", "IFR", "LIFR"]@dataclassclass MetarRequest:    """Path/query input for GET /weather/metar/{icao}."""    icao: str    parsed: bool = False@dataclassclass MetarWind:    direction: int    speed: int    gust: int | None    variable: list[int]@dataclassclass MetarVisibility:    value: float | None    repr: str@dataclassclass MetarCloudLayer:    type: str    base: int    repr: str@dataclassclass MetarWxCode:    repr: str    value: str@dataclassclass MetarParsed:    time: str    wind: MetarWind    visibility: MetarVisibility    clouds: list[MetarCloudLayer]    temperature: float    dewpoint: float    altimeter: float    flight_rules: FlightRules    wx_codes: list[MetarWxCode]    remarks: str    density_altitude: int    pressure_altitude: int    relative_humidity: float@dataclassclass MetarResponse:    raw: str    icao: str    airport_name: str    timestamp: str    parsed: MetarParsed | None = None@dataclassclass MetarSummary:    """Shape returned by metar_summary() in the fetch example."""    icao: str    category: FlightRules    wind_kt: int    gust_kt: int | None    visibility_display: str    age_min: float    stale: bool

import osimport timefrom dataclasses import dataclassfrom datetime import datetime, timezonefrom typing import Literalimport requests# MetarSummary - helper return type. Full wire types (MetarResponse, MetarParsed, …)# are in the Client types section on the METAR docs page.FlightRules = Literal["VFR", "MVFR", "IFR", "LIFR"]@dataclassclass MetarSummary:    icao: str    category: FlightRules    wind_kt: int    gust_kt: int | None    visibility_display: str    age_min: float    stale: boolHEADERS = {    "x-api-key": os.getenv("SKYLINK_API_KEY", "YOUR_API_KEY")}BASE = "https://data.skylinkapi.com/v3.1"CACHE: dict[str, tuple[dict, float]] = {}METAR_TTL = 300  # 5 min - safe inside the 30–60 min observation cycledef parse_observation_time(iso: str) -> datetime:    """Parse parsed.time - API returns ISO 8601 with timezone offset."""    return datetime.fromisoformat(iso.replace("Z", "+00:00"))def fetch_metar(icao: str) -> dict | None:    cache_key = f"metar:{icao}:parsed"    cached = CACHE.get(cache_key)    if cached and time.time() < cached[1]:        return cached[0]    try:        r = requests.get(            f"{BASE}/weather/metar/{icao}",            headers=HEADERS,            params={"parsed": "true"},            timeout=(10, 15),        )    except requests.Timeout as exc:        raise RuntimeError(f"SkyLink timeout fetching METAR for {icao}") from exc    if r.status_code == 404:        return None    r.raise_for_status()    data = r.json()    CACHE[cache_key] = (data, time.time() + METAR_TTL)    return datadef metar_summary(icao: str) -> MetarSummary | None:    metar = fetch_metar(icao)    if not metar:        return None    p = metar["parsed"]    obs = parse_observation_time(p["time"])    age_min = (datetime.now(timezone.utc) - obs).total_seconds() / 60    vis = p["visibility"]["value"]    return MetarSummary(        icao=icao,        category=p["flight_rules"],        wind_kt=p["wind"]["speed"],        gust_kt=p["wind"]["gust"],        visibility_display=f"{vis} SM" if vis is not None else "> 6 SM",        age_min=round(age_min, 1),        stale=age_min > 90,    )if __name__ == "__main__":    import json    from dataclasses import asdict    summary = metar_summary("KJFK")    print(json.dumps(asdict(summary), indent=2) if summary else "METAR not available")