Winds Aloft

"""Winds aloft response types - matches winds-aloft field tables."""from dataclasses import dataclassfrom typing import Literal@dataclassclass WindLevel:    altitude_ft: int    wind_direction: int | None    wind_speed_kt: int | None    temperature_c: int | None    light_and_variable: bool    raw: str@dataclassclass WindsAloftStation:    station: str    latitude: float    longitude: float    winds: list[WindLevel]    raw_text: str | None@dataclassclass WindsAloftResponse:    bbox: str    forecast_hour: int    level: str    valid_time: str | None    stations: list[WindsAloftStation]    total: int@dataclassclass CruiseWindSummary:    station: str    altitude_ft: int    direction_deg: int | None    speed_kt: int | None    temperature_c: int | None

import osimport timefrom typing import TypedDictimport requests# CruiseWindSummary - helper return type. Full wire types are in the Client types section.class WindLevel(TypedDict):    altitude_ft: int    wind_direction: int | None    wind_speed_kt: int | None    temperature_c: int | None    light_and_variable: bool    raw: strclass Station(TypedDict):    station: str    latitude: float    longitude: float    winds: list[WindLevel]    raw_text: str | Noneclass WindsAloftResponse(TypedDict):    bbox: str    forecast_hour: int    level: str    valid_time: str | None    stations: list[Station]    total: intclass CruiseWindSummary(TypedDict):    station: str    altitude_ft: int    direction_deg: int | None    speed_kt: int | None    temperature_c: int | NoneHEADERS = {    "x-api-key": os.getenv("SKYLINK_API_KEY", "YOUR_API_KEY")}BASE = "https://data.skylinkapi.com/v3"CACHE: dict[str, tuple[WindsAloftResponse, float]] = {}WINDS_TTL = 3600  # 1 h - FB products update on a fixed scheduledef fetch_winds_aloft(bbox: str, forecast: int = 12, level: str = "low") -> WindsAloftResponse | None:    cache_key = f"winds:{bbox}:{forecast}:{level}"    cached = CACHE.get(cache_key)    if cached and time.time() < cached[1]:        return cached[0]    try:        r = requests.get(            f"{BASE}/weather/winds-aloft",            headers=HEADERS,            params={"bbox": bbox, "forecast": forecast, "level": level},            timeout=(10, 20),        )    except requests.Timeout as exc:        raise RuntimeError(f"SkyLink timeout fetching winds aloft for {bbox}") from exc    if r.status_code == 404:        return None    r.raise_for_status()    data: WindsAloftResponse = r.json()    CACHE[cache_key] = (data, time.time() + WINDS_TTL)    return datadef nearest_cruise_wind(    bbox: str,    target_altitude_ft: int = 18_000,    forecast: int = 12,) -> CruiseWindSummary | None:    """Pick the wind level closest to target altitude from the first station in the bbox."""    payload = fetch_winds_aloft(bbox, forecast=forecast, level="low")    if not payload or not payload["stations"]:        return None    station = payload["stations"][0]    levels = station["winds"]    if not levels:        return None    best = min(levels, key=lambda w: abs(w["altitude_ft"] - target_altitude_ft))    return {        "station": station["station"],        "altitude_ft": best["altitude_ft"],        "direction_deg": best["wind_direction"],        "speed_kt": best["wind_speed_kt"],        "temperature_c": best["temperature_c"],    }if __name__ == "__main__":    import json    from dataclasses import asdict    bbox = "-74.5,40.3,-73.5,41.0"    cruise = nearest_cruise_wind(bbox, target_altitude_ft=18_000)    print(json.dumps(asdict(cruise), indent=2) if cruise else "No winds data")