espresso_frame/server/app/weather/ec.py

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