"""Environment Canada (ECCC MSC GeoMet OGC API, api.weather.gc.ca) provider -- Canada-only, free, no API key. Unlike NWS's grid-point lookup or Open-Meteo's plain lat/lon REST, EC's `citypageweather-realtime` collection (an OGC API - Features collection, the modern replacement for the old dd.weatheroffice.gc.ca XML feed -- that host no longer resolves) is only queryable by bounding box, not a direct by-coordinate endpoint -- this widens the box progressively until it finds at least one site, then picks the nearest by straight-line distance. This station/bbox-lookup shape (not simple lat/lon REST) is exactly why EC was documented as a follow-up rather than shipped alongside Open-Meteo/NWS in the first pass -- see docs/widgets.md. EC's numeric icon codes are its own set, distinct from WMO's (Open- Meteo) or NWS's icon-URL condition codes. _category_from_code_and_text below only maps the codes actually confirmed against live data (see this module's own tests, captured from real api.weather.gc.ca responses), falling back to the same keyword-match-on-condition-text safety net app/weather/nws.py uses for anything unmapped -- correctness comes from the text fallback, the numeric table is just a fast path. Pure functions -- no ORM, no FastAPI Depends -- same testability philosophy as calendar_feed.py/caldav_client.py/app/weather/open_meteo.py/nws.py. """ from __future__ import annotations import math from datetime import datetime, timedelta import httpx from . import WeatherFetchError HTTP_TIMEOUT_S = 15.0 BASE_URL = "https://api.weather.gc.ca" COLLECTION = "citypageweather-realtime" HEADERS = {"User-Agent": "espresso_frame-weather-widget (self-hosted photo frame project)"} # Progressively wider bounding boxes (degrees) around the target point -- # EC's ~844 sites are dense near cities but sparse in the north, so a # small box can come back empty even for a real Canadian location. 25 # degrees (~2000-2700 km depending on latitude) is already far beyond # _MAX_DISTANCE_KM, so there's no point widening past it. _BBOX_PADDINGS_DEG = (1.0, 3.0, 8.0, 25.0) # See _nearest_site's own docstring for why this exists and how it was # calibrated (a real Miami query matched 1824 km away without it). _MAX_DISTANCE_KM = 300 _ICON_CODE_CATEGORIES = { 0: "clear", 1: "clear", 30: "clear", 2: "partly_cloudy", 5: "partly_cloudy", 31: "partly_cloudy", 32: "partly_cloudy", 3: "cloudy", 4: "cloudy", 10: "cloudy", 33: "cloudy", 6: "rain", 12: "rain", 28: "rain", 36: "rain", 9: "thunderstorm", 19: "thunderstorm", 39: "thunderstorm", 24: "fog", } _TEXT_CATEGORY_KEYWORDS = [ ("thunderstorm", "thunderstorm"), ("tstm", "thunderstorm"), ("tornado", "thunderstorm"), ("flurr", "snow"), ("snow", "snow"), ("sleet", "snow"), ("ice pellet", "snow"), ("hail", "snow"), ("freezing", "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_code_and_text(code: int | None, text: str) -> str: if code is not None and code in _ICON_CODE_CATEGORIES: return _ICON_CODE_CATEGORIES[code] lowered = text.lower() for keyword, category in _TEXT_CATEGORY_KEYWORDS: if keyword in lowered: return category return "cloudy" # same generic-icon fallback Open-Meteo/NWS both use def _convert_temp(celsius: float, units: str) -> float: """EC's citypage feed reports temperatures in Celsius only (its unitType is always "metric" in this feed) -- convert to the widget's requested units, no-op if celsius was actually asked for.""" return celsius * 9 / 5 + 32 if units == "fahrenheit" else celsius def _haversine_km(lat1: float, lon1: float, lat2: float, lon2: float) -> float: r = 6371.0 p1, p2 = math.radians(lat1), math.radians(lat2) dphi = math.radians(lat2 - lat1) dlambda = math.radians(lon2 - lon1) a = math.sin(dphi / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dlambda / 2) ** 2 return 2 * r * math.asin(math.sqrt(a)) def _items(bbox: str) -> list[dict]: try: resp = httpx.get( f"{BASE_URL}/collections/{COLLECTION}/items", params={"f": "json", "bbox": bbox, "limit": 50}, headers=HEADERS, timeout=HTTP_TIMEOUT_S, ) resp.raise_for_status() return resp.json()["features"] except (httpx.HTTPError, KeyError) as e: raise WeatherFetchError(str(e)) from e def _nearest_site(latitude: float, longitude: float) -> dict: """This location's nearest Environment Canada citypage site's `properties` dict -- widens the bounding box until it finds one within _MAX_DISTANCE_KM, trying every padding rather than stopping at the first non-empty box: since each padding's box is a superset of the previous one's, a wider box's nearest match can only be the same distance or closer, never farther, so an early non-empty box whose nearest site is still too far away doesn't mean a closer one isn't waiting just outside it. Without the distance cutoff, an unconditional "just take whatever's nearest" happily matches a US or overseas location to some real EC site thousands of km away (confirmed live: Miami matched to Leamington, Ontario, 1824 km off) instead of reporting that EC simply doesn't cover this location. 300 km is generous enough for a legitimate rural-Canada query against EC's sparse northern coverage (~844 sites total) while still correctly rejecting a non-Canadian one -- a US border city like Seattle, genuinely ~100 km from the nearest EC site in Victoria, BC, still passes. Deliberately NOT a single whole-country query instead of progressive widening: that collection response is ~29 MB for cheap, in-city lookups fetching a handful of nearby sites.""" best_distance_km = None for pad in _BBOX_PADDINGS_DEG: bbox = f"{longitude - pad},{latitude - pad},{longitude + pad},{latitude + pad}" features = _items(bbox) if not features: continue nearest = min( features, key=lambda f: _haversine_km( latitude, longitude, f["geometry"]["coordinates"][1], f["geometry"]["coordinates"][0] ), ) best_distance_km = _haversine_km( latitude, longitude, nearest["geometry"]["coordinates"][1], nearest["geometry"]["coordinates"][0] ) if best_distance_km <= _MAX_DISTANCE_KM: return nearest["properties"] raise WeatherFetchError("No Environment Canada site found near this location (is it in Canada?)") def fetch_current(latitude: float, longitude: float, units: str) -> dict: props = _nearest_site(latitude, longitude) cc = props["currentConditions"] temp_c = cc["temperature"]["value"]["en"] code = (cc.get("iconCode") or {}).get("value") text = (cc.get("condition") or {}).get("en") or "" return {"temp": _convert_temp(temp_c, units), "category": _category_from_code_and_text(code, text)} def fetch_hourly(latitude: float, longitude: float, units: str, hours: int = 48) -> list[dict]: """EC's hourlyForecastGroup is a fixed 24-hour window (unlike Open- Meteo/NWS's own longer hourly ranges) -- `hours` just caps how much of it gets returned, same as the other providers.""" props = _nearest_site(latitude, longitude) entries = props["hourlyForecastGroup"]["hourlyForecasts"] result = [] for h in entries[:hours]: temp_c = h["temperature"]["value"]["en"] code = (h.get("iconCode") or {}).get("value") text = (h.get("condition") or {}).get("en") or "" result.append({ "time": h["timestamp"], "temp": _convert_temp(temp_c, units), "category": _category_from_code_and_text(code, text), }) return result def fetch_daily(latitude: float, longitude: float, units: str, days: int) -> dict[str, dict]: """Pairs EC's named day/night periods ("Today"/"Tonight"/"Tuesday"/ "Tuesday night"/...) into calendar dates by walking them in issued order -- unlike NWS's periods (which carry a real startTime), EC's forecast periods are named relative to "today", not dated, so the date is inferred: a "night" period shares its preceding day period's date, any other period starts the calendar day after the previous one (the forecastGroup's own issued-at timestamp anchors day 0).""" props = _nearest_site(latitude, longitude) group = props["forecastGroup"] issued = datetime.fromisoformat(group["timestamp"]["en"].replace("Z", "+00:00")).date() by_date: dict[str, dict] = {} order: list[str] = [] current_day = None for period in group["forecasts"]: name = period["period"]["textForecastName"]["en"].strip().lower() if "night" in name: day_date = current_day or issued else: day_date = issued if current_day is None else current_day + timedelta(days=1) current_day = day_date key = day_date.isoformat() entry = by_date.setdefault(key, {"category": None}) if key not in order: order.append(key) temps = period.get("temperatures", {}).get("temperature") or [] if temps: temp = _convert_temp(temps[0]["value"]["en"], units) if temps[0]["class"]["en"] == "high": entry["high"] = temp else: entry["low"] = temp if entry["category"] is None: icon = (period.get("abbreviatedForecast") or {}).get("icon") or {} text = (period.get("abbreviatedForecast") or {}).get("textSummary", {}).get("en") or "" entry["category"] = _category_from_code_and_text(icon.get("value"), text) result = {} for key in order[:max(1, days)]: entry = by_date[key] if "high" not in entry and "low" not in entry: continue result[key] = { "high": entry.get("high", entry.get("low")), "low": entry.get("low", entry.get("high")), "category": entry["category"] or "cloudy", } return result