"""Weather strip for calendar frame mode (agenda/today & tomorrow/week views only -- never month, there's no room, see calendar_render.py's _BUILDERS). A frame can list multiple cities; each is geocoded once via Open-Meteo's free geocoding API (no API key, no signup, no per-request quota to manage) when added from the Calendar tab, then its daily forecast is refreshed on its own throttle -- same shape idiom as calendar_feed.py's merge-fetch cache. Pure functions -- no ORM, no FastAPI Depends -- same testability philosophy as calendar_feed.py/caldav_client.py. """ from __future__ import annotations import httpx HTTP_TIMEOUT_S = 15.0 GEOCODE_URL = "https://geocoding-api.open-meteo.com/v1/search" FORECAST_URL = "https://api.open-meteo.com/v1/forecast" CHECK_INTERVAL_S = 3 * 60 * 60 # weather doesn't need calendar_feed's 20-minute cadence FORECAST_DAYS = 14 # comfortably covers the week view's furthest browse-forward # Open-Meteo's WMO weather codes (https://open-meteo.com/en/docs), grouped # into the handful of icon categories calendar_render.py actually draws. _CODE_CATEGORIES = { 0: "clear", 1: "partly_cloudy", 2: "partly_cloudy", 3: "cloudy", 45: "fog", 48: "fog", 51: "rain", 53: "rain", 55: "rain", 56: "rain", 57: "rain", 61: "rain", 63: "rain", 65: "rain", 66: "rain", 67: "rain", 80: "rain", 81: "rain", 82: "rain", 71: "snow", 73: "snow", 75: "snow", 77: "snow", 85: "snow", 86: "snow", 95: "thunderstorm", 96: "thunderstorm", 99: "thunderstorm", } class WeatherFetchError(Exception): """Geocoding or forecast fetch failed -- network, no match, or an unexpected response shape. Raised loudly; callers (the Calendar tab's add-city endpoint, get_or_refresh_weather) decide what to do.""" def weather_category(code: int) -> str: """Falls back to "cloudy" for any WMO code Open-Meteo might add later that isn't in the table above -- an unrecognized code shouldn't drop a day's weather entirely, just render with a generic icon.""" return _CODE_CATEGORIES.get(code, "cloudy") # Open-Meteo's geocoder matches on the bare place name only -- "Portland, # OR" returns zero results even though "Portland" alone returns three # (OR/ME/IN, disambiguated by population-ranked order). So a ", " # or ", " qualifier is split off client-side and used to filter # among the candidates instead of being sent as part of the search term. _US_STATE_ABBREVIATIONS = { "al": "alabama", "ak": "alaska", "az": "arizona", "ar": "arkansas", "ca": "california", "co": "colorado", "ct": "connecticut", "de": "delaware", "fl": "florida", "ga": "georgia", "hi": "hawaii", "id": "idaho", "il": "illinois", "in": "indiana", "ia": "iowa", "ks": "kansas", "ky": "kentucky", "la": "louisiana", "me": "maine", "md": "maryland", "ma": "massachusetts", "mi": "michigan", "mn": "minnesota", "ms": "mississippi", "mo": "missouri", "mt": "montana", "ne": "nebraska", "nv": "nevada", "nh": "new hampshire", "nj": "new jersey", "nm": "new mexico", "ny": "new york", "nc": "north carolina", "nd": "north dakota", "oh": "ohio", "ok": "oklahoma", "or": "oregon", "pa": "pennsylvania", "ri": "rhode island", "sc": "south carolina", "sd": "south dakota", "tn": "tennessee", "tx": "texas", "ut": "utah", "vt": "vermont", "va": "virginia", "wa": "washington", "wv": "west virginia", "wi": "wisconsin", "wy": "wyoming", "dc": "district of columbia", } def _matches_qualifier(result: dict, qualifier: str) -> bool: q = qualifier.strip().lower() expanded = _US_STATE_ABBREVIATIONS.get(q, q) admin1 = (result.get("admin1") or "").lower() country = (result.get("country") or "").lower() country_code = (result.get("country_code") or "").lower() return expanded in admin1 or expanded in country or q == country_code def geocode_city(name: str) -> dict: """Best-match {"label", "latitude", "longitude"} for a free-text city name, optionally qualified with a state/country (e.g. "Portland, OR") via Open-Meteo's geocoder. label is the resolved place name (city + admin1/country when available), not necessarily what the user typed -- shown back so they can confirm it found the right place before it's saved.""" query, _, qualifier = name.partition(",") query, qualifier = query.strip(), qualifier.strip() try: resp = httpx.get(GEOCODE_URL, params={"name": query, "count": 10}, timeout=HTTP_TIMEOUT_S) resp.raise_for_status() data = resp.json() except httpx.HTTPError as e: raise WeatherFetchError(str(e)) from e results = data.get("results") or [] if not results: raise WeatherFetchError(f"No location found matching {name!r}") if qualifier: qualified = [r for r in results if _matches_qualifier(r, qualifier)] if not qualified: raise WeatherFetchError(f"No location found matching {name!r}") results = qualified r = results[0] parts = [r["name"]] if r.get("admin1"): parts.append(r["admin1"]) if r.get("country"): parts.append(r["country"]) return {"label": ", ".join(parts), "latitude": r["latitude"], "longitude": r["longitude"]} def fetch_daily_forecast(latitude: float, longitude: float, units: str) -> dict[str, dict]: """{"YYYY-MM-DD": {"code": int, "high": float, "low": float}, ...} for the next FORECAST_DAYS days, already in `units` ("fahrenheit"/"celsius") -- Open-Meteo converts server-side, so there's no client-side unit math to get wrong.""" try: resp = httpx.get(FORECAST_URL, params={ "latitude": latitude, "longitude": longitude, "daily": "weathercode,temperature_2m_max,temperature_2m_min", "temperature_unit": units, "timezone": "auto", "forecast_days": FORECAST_DAYS, }, timeout=HTTP_TIMEOUT_S) resp.raise_for_status() daily = resp.json()["daily"] return { day: {"code": code, "high": high, "low": low} for day, code, high, low in zip( daily["time"], daily["weathercode"], daily["temperature_2m_max"], daily["temperature_2m_min"] ) } except (httpx.HTTPError, KeyError) as e: raise WeatherFetchError(str(e)) from e