"""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