espresso_frame/server/app/weather/nws.py

"""NWS (National Weather Service, api.weather.gov) provider -- US
locations only, free, no API key, but requires a `User-Agent` header
identifying the calling app (NWS API policy: unidentified traffic gets
throttled/blocked). No geocoder of its own; app/weather/__init__.py's
geocode_city (Open-Meteo's) resolves a place name to lat/lon regardless
of which provider is then chosen to fetch with it.

Unlike Open-Meteo's numeric WMO codes, NWS periods carry a `shortForecast`
text description and an `icon` URL encoding a condition code (e.g.
".../icons/land/day/tsra,40?size=medium") -- _category_from_period below
normalizes either into the same shared category set
(clear/partly_cloudy/cloudy/fog/rain/snow/thunderstorm) Open-Meteo's
weather_category() already produces, so app/weather_render.py's build_*
functions never need to know which provider supplied an entry.

Pure functions -- no ORM, no FastAPI Depends -- same testability
philosophy as calendar_feed.py/caldav_client.py/app/weather/open_meteo.py.
"""

from __future__ import annotations

from datetime import datetime

import httpx

from . import WeatherFetchError

HTTP_TIMEOUT_S = 15.0
BASE_URL = "https://api.weather.gov"
# NWS blocks/deprioritizes traffic with no identifying User-Agent -- a
# contact-ish string is their documented convention, not a real account.
HEADERS = {"User-Agent": "espresso_frame-weather-widget (self-hosted photo frame project)"}

# NWS's icon condition codes (https://api.weather.gov/icons), grouped into
# the same handful of categories Open-Meteo's _CODE_CATEGORIES maps WMO
# codes to. "wind_"-prefixed variants (e.g. wind_skc) are just the same
# sky condition plus wind -- stripped before lookup, since this project's
# vendored icon set (app/weather_render.py, app/weather_icons/) doesn't
# have a separate windy glyph.
_ICON_CODE_CATEGORIES = {
    "skc": "clear", "clear": "clear",
    "few": "partly_cloudy", "sct": "partly_cloudy",
    "bkn": "cloudy", "ovc": "cloudy",
    "fog": "fog", "haze": "fog", "smoke": "fog", "dust": "fog",
    "rain": "rain", "rain_showers": "rain", "rain_showers_hi": "rain",
    "showers": "rain", "drizzle": "rain",
    "snow": "snow", "rain_snow": "snow", "sleet": "snow",
    "fzra": "snow", "rain_fzra": "snow", "snow_fzra": "snow",
    "tsra": "thunderstorm", "tsra_sct": "thunderstorm", "tsra_hi": "thunderstorm",
}

# Fallback keyword match against shortForecast text, in priority order --
# used when the icon URL can't be parsed at all (unexpected shape) or its
# condition code isn't in the table above (NWS occasionally adds new
# icon variants).
_TEXT_CATEGORY_KEYWORDS = [
    ("thunderstorm", "thunderstorm"), ("tstm", "thunderstorm"),
    ("snow", "snow"), ("sleet", "snow"), ("ice", "snow"),
    ("rain", "rain"), ("shower", "rain"), ("drizzle", "rain"),
    ("fog", "fog"), ("haze", "fog"), ("mist", "fog"), ("smoke", "fog"),
    ("overcast", "cloudy"), ("cloudy", "cloudy"),
    ("clear", "clear"), ("sunny", "clear"), ("fair", "clear"),
]


def _category_from_icon(icon_url: str) -> str | None:
    """The first condition code segment out of an icon URL path like
    ".../icons/land/day/tsra,40/skc,20?size=medium" -- only the first
    (the dominant/nearest-term condition) is used, same "one glyph per
    entry" simplicity as Open-Meteo's single-WMO-code-per-day shape."""
    path = icon_url.split("?")[0]
    segments = [s for s in path.split("/") if s]
    for i, seg in enumerate(segments):
        if seg in ("day", "night") and i + 1 < len(segments):
            code = segments[i + 1].split(",")[0]
            code = code.removeprefix("wind_")
            return _ICON_CODE_CATEGORIES.get(code)
    return None


def _category_from_text(text: str) -> str:
    lowered = text.lower()
    for keyword, category in _TEXT_CATEGORY_KEYWORDS:
        if keyword in lowered:
            return category
    return "cloudy"  # same generic-icon fallback Open-Meteo's weather_category uses


def _category_from_period(period: dict) -> str:
    icon = period.get("icon") or ""
    category = _category_from_icon(icon) if icon else None
    return category or _category_from_text(period.get("shortForecast") or "")


def _convert_temp(value: float, from_unit: str, to_units: str) -> float:
    """NWS periods report temperatureUnit per-period (almost always "F"
    for US points) -- converts to the widget's own requested `units`
    ("fahrenheit"/"celsius") only if they actually differ, so this is a
    no-op in the common case."""
    to_unit = "F" if to_units == "fahrenheit" else "C"
    if from_unit == to_unit:
        return value
    if from_unit == "F" and to_unit == "C":
        return (value - 32) * 5 / 9
    return value * 9 / 5 + 32


def _points(latitude: float, longitude: float) -> dict:
    try:
        resp = httpx.get(
            f"{BASE_URL}/points/{latitude:.4f},{longitude:.4f}", headers=HEADERS, timeout=HTTP_TIMEOUT_S
        )
        resp.raise_for_status()
        return resp.json()["properties"]
    except (httpx.HTTPError, KeyError) as e:
        raise WeatherFetchError(f"NWS points lookup failed (is this location in the US?): {e}") from e


def _periods(url: str) -> list[dict]:
    try:
        resp = httpx.get(url, headers=HEADERS, timeout=HTTP_TIMEOUT_S)
        resp.raise_for_status()
        return resp.json()["properties"]["periods"]
    except (httpx.HTTPError, KeyError) as e:
        raise WeatherFetchError(str(e)) from e


def fetch_current(latitude: float, longitude: float, units: str) -> dict:
    """{"temp": float, "category": str} -- NWS has no simple by-
    coordinate "current conditions" endpoint (that needs a second
    stations-list + latest-observation lookup); this approximates
    "current" with the first hourly forecast period instead, which is
    plenty for a frame that only refreshes every few hours."""
    points = _points(latitude, longitude)
    periods = _periods(points["forecastHourly"])
    if not periods:
        raise WeatherFetchError("NWS returned no hourly forecast periods")
    period = periods[0]
    temp = _convert_temp(period["temperature"], period.get("temperatureUnit", "F"), units)
    return {"temp": temp, "category": _category_from_period(period)}


def fetch_hourly(latitude: float, longitude: float, units: str, hours: int = 48) -> list[dict]:
    """[{"time": ISO 8601 string, "temp": float, "category": str}, ...],
    one entry per hour -- NWS's forecastHourly is already 1-hour
    resolution, same as Open-Meteo's."""
    points = _points(latitude, longitude)
    periods = _periods(points["forecastHourly"])
    return [
        {
            "time": p["startTime"],
            "temp": _convert_temp(p["temperature"], p.get("temperatureUnit", "F"), units),
            "category": _category_from_period(p),
        }
        for p in periods[:hours]
    ]


def fetch_daily(latitude: float, longitude: float, units: str, days: int) -> dict[str, dict]:
    """{"YYYY-MM-DD": {"high": float, "low": float, "category": str}, ...}
    -- NWS's /forecast returns day/night period pairs (12h each, ~7 days
    / 14 periods), not one row per day; this pairs them by calendar date
    (a daytime period's high, the following night's low) and clamps to
    however many full dates actually came back once `days` is asked for
    more than that -- no error, just fewer days, same graceful-
    degradation idiom as app/weather_render.py's draw_weather_row."""
    points = _points(latitude, longitude)
    periods = _periods(points["forecast"])

    by_date: dict[str, dict] = {}
    order: list[str] = []
    for p in periods:
        day = datetime.fromisoformat(p["startTime"]).date().isoformat()
        entry = by_date.setdefault(day, {"category": None})
        if day not in order:
            order.append(day)
        temp = _convert_temp(p["temperature"], p.get("temperatureUnit", "F"), units)
        if p["isDaytime"]:
            entry["high"] = temp
            entry["category"] = _category_from_period(p)
        else:
            entry["low"] = temp
            if entry["category"] is None:
                entry["category"] = _category_from_period(p)

    result = {}
    for day in order[:max(1, days)]:
        entry = by_date[day]
        if "high" not in entry and "low" not in entry:
            continue
        result[day] = {
            "high": entry.get("high", entry.get("low")),
            "low": entry.get("low", entry.get("high")),
            "category": entry["category"] or "cloudy",
        }
    return result