"""Fetch historical + recent daily weather from Open-Meteo and cache to parquet. The full multi-decade daily record for a grid cell is fetched once and stored as a zstd-compressed parquet file keyed by cell id. Percentiles are computed on the fly from this raw record (see grading.py), which keeps the cache small and lets us handle ties (e.g. many zero-precip days) correctly. """ import datetime import os import threading import time import httpx import numpy as np import pandas as pd import audit CACHE_DIR = os.path.join(os.path.dirname(__file__), "..", "data", "cache") MAX_ATTEMPTS = 3 # per upstream call, before giving up START_DATE = "1980-01-01" # ERA5 reaches back to 1940; 1980 = 45 yrs, fast + robust ARCHIVE_LATENCY_DAYS = 6 # ERA5 archive lags real time by a few days # History rarely changes, so the full multi-decade archive is fetched once and # cached indefinitely (refetched only to add new metric columns). Only the recent # tail is refreshed — a small incremental fetch, at most hourly. HISTORY_TOPUP_INTERVAL = 3600 # seconds between recent-tail refresh attempts FORECAST_TTL_HOURS = 1 # refetch the forward forecast hourly to track updates ARCHIVE_URL = "https://archive-api.open-meteo.com/v1/archive" FORECAST_URL = "https://api.open-meteo.com/v1/forecast" # Backup archive when Open-Meteo is unavailable (e.g. its daily rate limit). NASA # POWER is free + keyless, global, daily from 1981. It lacks gusts + apparent temp, # so gusts read as unavailable and "feels like" is computed from heat index/chill. NASA_POWER_URL = "https://power.larc.nasa.gov/api/temporal/daily/point" NASA_START = "19810101" NASA_FILL = -900.0 # POWER's missing-value sentinel is ~-999 DAILY_VARS = ( "temperature_2m_max,temperature_2m_min,precipitation_sum," "wind_speed_10m_max,wind_gusts_10m_max," "apparent_temperature_max,apparent_temperature_min," "relative_humidity_2m_mean" ) # Columns added after the original tmax/tmin/precip schema. A cached parquet # missing any of these predates the wind/feels/humidity features (or the split # apparent high/low) and is refetched so the new metrics show up immediately # instead of after the 30-day cache TTL. NEW_COLS = ("wind", "gust", "feels", "humid", "fmax", "fmin") # Thermoneutral baseline (°F). The combined "feels like" metric reports whichever # apparent-temperature extreme — the heat-index-driven daily max or the # wind-chill-driven daily min — sits further from this comfort point, so a single # daily value captures both heat index (hot side) and wind chill (cold side). COMFORT_F = 65.0 def _cache_path(cell_id: str) -> str: return os.path.join(CACHE_DIR, f"{cell_id}.parquet") # One lock per cell so concurrent requests/refreshes don't each pull the full # multi-decade archive (which quickly trips the upstream rate limit). _LOCKS_GUARD = threading.Lock() _CELL_LOCKS: dict[str, threading.Lock] = {} # When the archive rate-limits us (429), back off globally for a bit rather than # re-hitting it on every refresh — that only prolongs the limit. ARCHIVE_COOLDOWN = 120 # seconds _archive_cooldown_until = 0.0 _archive_limit_daily = False # is the current cooldown a daily-quota exhaustion? def _is_rate_limit(e) -> bool: return getattr(getattr(e, "response", None), "status_code", None) == 429 def _rate_limit_reason(e) -> str: """Upstream's human-readable 429 reason, if present in the response body.""" resp = getattr(e, "response", None) if resp is not None: try: j = resp.json() if isinstance(j, dict) and j.get("reason"): return str(j["reason"]) except Exception: # noqa: BLE001 pass return "" def _seconds_to_utc_reset() -> float: """Seconds until just after the next UTC midnight (when daily quotas reset).""" now = datetime.datetime.now(datetime.timezone.utc) reset = (now + datetime.timedelta(days=1)).replace(hour=0, minute=10, second=0, microsecond=0) return max((reset - now).total_seconds(), 600) def _note_rate_limit(e) -> None: """Record a rate limit: back off ~2 min, or until tomorrow for a daily quota.""" global _archive_cooldown_until, _archive_limit_daily reason = _rate_limit_reason(e).lower() _archive_limit_daily = "daily" in reason or "tomorrow" in reason _archive_cooldown_until = time.time() + ( _seconds_to_utc_reset() if _archive_limit_daily else ARCHIVE_COOLDOWN) def _cooldown_message() -> str: if _archive_limit_daily: return ("Open-Meteo's daily request limit is exhausted — new locations will work again " "tomorrow. Places you've already viewed still work.") return "The weather archive is rate-limited right now — please try again in a minute." def _cell_lock(cell_id: str) -> threading.Lock: with _LOCKS_GUARD: lk = _CELL_LOCKS.get(cell_id) if lk is None: lk = _CELL_LOCKS[cell_id] = threading.Lock() return lk def _request(url, params, timeout, *, phase, headers=None, attempts=MAX_ATTEMPTS): """GET with bounded retries; every retry and the final failure are logged to the errors folder (tagged ``retry`` / ``error``). A 429 (rate limit) fails fast without retrying, so we don't hammer the limit and make it worse.""" last = None for attempt in range(1, attempts + 1): try: r = httpx.get(url, params=params, timeout=timeout, headers=headers) r.raise_for_status() return r except Exception as e: # noqa: BLE001 - upstream/network failures are expected last = e status = getattr(getattr(e, "response", None), "status_code", None) rate_limited = status == 429 final = rate_limited or attempt == attempts audit.log_event( "error" if final else "retry", {"phase": phase, "attempt": attempt, "max_attempts": attempts, "url": url, "status": status, "error": repr(e)}, ) if final: break time.sleep(min(0.5 * 2 ** (attempt - 1), 4.0)) raise last def _derive_humidity(df: pd.DataFrame) -> pd.DataFrame: """Replace the raw mean *relative* humidity column with *absolute* humidity (grams of water vapor per m³), in place. Absolute humidity is derived from the day's mean RH and its mean temperature ((tmax+tmin)/2) via the Magnus saturation-vapor-pressure formula. It's a far more informative "how muggy was it" signal than relative humidity, which mostly tracks the day/night temperature swing (cold nights read ~100% RH regardless of actual moisture). Applied at the read boundary so the parquet cache keeps the raw RH the archive returns — no refetch needed for existing cells.""" if "humid" not in df.columns: return df tmean_c = ((df["tmax"] + df["tmin"]) / 2.0 - 32.0) * 5.0 / 9.0 rh = pd.to_numeric(df["humid"], errors="coerce") es = 6.112 * np.exp(17.67 * tmean_c / (tmean_c + 243.5)) # sat. vapor pressure, hPa df["humid"] = (es * rh * 2.1674 / (273.15 + tmean_c)).round(1) # absolute humidity, g/m³ return df def _combined_feels(amax: pd.Series, amin: pd.Series) -> pd.Series: """One daily "feels like" value: the apparent-temperature extreme furthest from the comfort baseline — the heat-index high on warm days, the wind-chill low on cold ones. Falls back to whichever side is present if one is missing.""" hot = amax - COMFORT_F cold = COMFORT_F - amin feels = amax.where(hot >= cold, amin) # NaN comparisons pick the min side return feels.fillna(amax).fillna(amin) def _to_frame(daily: dict) -> pd.DataFrame: n = len(daily["time"]) # Older upstream responses (or a narrowed variable set) may omit a series; fall # back to an all-null column of the right length so the frame shape is stable. def col(key): vals = daily.get(key) return vals if vals else [None] * n df = pd.DataFrame( { "date": pd.to_datetime(daily["time"]), "tmax": daily["temperature_2m_max"], "tmin": daily["temperature_2m_min"], "precip": daily["precipitation_sum"], "wind": col("wind_speed_10m_max"), "gust": col("wind_gusts_10m_max"), "humid": col("relative_humidity_2m_mean"), } ) # The apparent (felt) high and low are kept as their own columns — graded # independently — alongside the combined `feels` (whichever side is further # from the fixed comfort baseline), which the weekly/day views still use. amax = pd.to_numeric(pd.Series(col("apparent_temperature_max")), errors="coerce") amin = pd.to_numeric(pd.Series(col("apparent_temperature_min")), errors="coerce") df["fmax"] = amax df["fmin"] = amin df["feels"] = _combined_feels(amax, amin) # Need a valid high/low to be a usable climate day; the rest may be NaN and are # simply ungraded for that day (the percentile helpers return None). df = df.dropna(subset=["tmax", "tmin"]).reset_index(drop=True) df["doy"] = df["date"].dt.dayofyear.astype("int16") return df def _fetch_history(cell: dict) -> pd.DataFrame: end = (datetime.date.today() - datetime.timedelta(days=ARCHIVE_LATENCY_DAYS)).isoformat() params = { "latitude": cell["center_lat"], "longitude": cell["center_lon"], "start_date": START_DATE, "end_date": end, "daily": DAILY_VARS, "timezone": "auto", "temperature_unit": "fahrenheit", "precipitation_unit": "inch", "wind_speed_unit": "mph", } r = _request(ARCHIVE_URL, params, 180, phase="history_fetch") return _to_frame(r.json()["daily"]) def _heat_index(t_f, rh): """NWS heat index (°F); below ~80°F it's just the air temperature.""" T, R = t_f, rh hi = (-42.379 + 2.04901523 * T + 10.14333127 * R - 0.22475541 * T * R - 0.00683783 * T * T - 0.05481717 * R * R + 0.00122874 * T * T * R + 0.00085282 * T * R * R - 0.00000199 * T * T * R * R) return np.where(np.asarray(T, dtype="float64") >= 80, hi, T) def _wind_chill(t_f, v_mph): """NWS wind chill (°F); applies only when cold + breezy, else the air temp.""" T = np.asarray(t_f, dtype="float64") V = np.clip(np.asarray(v_mph, dtype="float64"), 0, None) Vp = np.power(V, 0.16) wc = 35.74 + 0.6215 * T - 35.75 * Vp + 0.4275 * T * Vp return np.where((T <= 50) & (V >= 3), wc, T) def _nasa_to_frame(param: dict) -> pd.DataFrame: """Map a NASA POWER daily response to our (tmax/tmin/precip/wind/gust/humid/feels) schema, converting units (°C→°F, mm→in, m/s→mph) and computing feels-like.""" dates = sorted(param.get("T2M_MAX", {}).keys()) def col(key, transform): d = param.get(key, {}) return [np.nan if (v is None or v <= NASA_FILL) else transform(v) for v in (d.get(k) for k in dates)] c2f = lambda c: c * 9.0 / 5.0 + 32.0 df = pd.DataFrame({ "date": pd.to_datetime(dates, format="%Y%m%d"), "tmax": col("T2M_MAX", c2f), "tmin": col("T2M_MIN", c2f), "precip": col("PRECTOTCORR", lambda mm: mm / 25.4), "wind": col("WS10M_MAX", lambda ms: ms * 2.2369362920544), "gust": [np.nan] * len(dates), # POWER has no gusts "humid": col("RH2M", lambda v: v), }) # POWER has no apparent temperature; approximate the felt high with the NWS # heat index and the felt low with NWS wind chill (each falls back to the air # temperature outside its regime), mirroring the Open-Meteo columns. df["fmax"] = pd.Series(_heat_index(df["tmax"], df["humid"]), index=df.index) df["fmin"] = pd.Series(_wind_chill(df["tmin"], df["wind"]), index=df.index) df["feels"] = _combined_feels(df["fmax"], df["fmin"]) df = df.dropna(subset=["tmax", "tmin"]).reset_index(drop=True) df["doy"] = df["date"].dt.dayofyear.astype("int16") return df def _fetch_history_nasa(cell: dict) -> pd.DataFrame: """Backup history fetch from NASA POWER (used when Open-Meteo is unavailable).""" end = (datetime.date.today() - datetime.timedelta(days=ARCHIVE_LATENCY_DAYS)).strftime("%Y%m%d") params = { "parameters": "T2M_MAX,T2M_MIN,PRECTOTCORR,WS10M_MAX,RH2M", "community": "RE", "latitude": cell["center_lat"], "longitude": cell["center_lon"], "start": NASA_START, "end": end, "format": "JSON", } r = _request(NASA_POWER_URL, params, 180, phase="history_nasa") return _nasa_to_frame(r.json()["properties"]["parameter"]) def _fetch_history_range(cell: dict, start_date: str, end_date: str) -> pd.DataFrame: """Fetch just a date range of archive history (used to top up the recent tail).""" params = { "latitude": cell["center_lat"], "longitude": cell["center_lon"], "start_date": start_date, "end_date": end_date, "daily": DAILY_VARS, "timezone": "auto", "temperature_unit": "fahrenheit", "precipitation_unit": "inch", "wind_speed_unit": "mph", } r = _request(ARCHIVE_URL, params, 60, phase="history_topup") return _to_frame(r.json()["daily"]) def _read_history_cache(path): """Schema-complete cached frame (no doy) + its file age in seconds, or None. No age expiry — history is cached indefinitely; a pre-wind/humidity schema is the only reason to refetch (to add the new metric columns).""" if not os.path.exists(path): return None df = pd.read_parquet(path) if not all(c in df.columns for c in NEW_COLS): return None return df, time.time() - os.path.getmtime(path) def _topup_tail(cell: dict, df: pd.DataFrame, path: str) -> pd.DataFrame: """Append newly-available archive days to a cached record. Best-effort and serialized per cell; a small incremental fetch, not the full multi-decade pull.""" global _archive_cooldown_until with _cell_lock(cell["id"]): fresh = _read_history_cache(path) # another thread may have just refreshed it if fresh is not None: df, age_s = fresh if age_s < HISTORY_TOPUP_INTERVAL: return df if time.time() < _archive_cooldown_until: return df expected = datetime.date.today() - datetime.timedelta(days=ARCHIVE_LATENCY_DAYS) cached_max = pd.Timestamp(df["date"].max()).date() if cached_max >= expected: try: os.utime(path, None) # tail already current — reset the hourly timer except OSError: pass return df try: recent = _fetch_history_range( cell, (cached_max + datetime.timedelta(days=1)).isoformat(), expected.isoformat()) except Exception as e: # noqa: BLE001 - keep the cached record on failure if _is_rate_limit(e): _note_rate_limit(e) return df merged = (pd.concat([df, recent.drop(columns=["doy"], errors="ignore")]) .drop_duplicates(subset="date", keep="last") .sort_values("date").reset_index(drop=True)) merged.to_parquet(path, compression="zstd", index=False) return merged def get_history(cell: dict) -> tuple[pd.DataFrame, dict]: """Return (daily history frame, cache metadata) for a cell, with humidity as absolute humidity (g/m³). Thin wrapper over the raw loader (see below).""" df, meta = _load_history(cell) _derive_humidity(df) return df, meta def _load_history(cell: dict) -> tuple[pd.DataFrame, dict]: """Return (daily history frame, cache metadata) for a cell. The full archive is cached indefinitely (fetched once); only the recent tail is topped up, at most hourly. Concurrent callers are serialized so only one archive fetch happens; on an upstream failure (e.g. 429) any existing cache is served.""" path = _cache_path(cell["id"]) hit = _read_history_cache(path) if hit is not None: df, age_s = hit if age_s > HISTORY_TOPUP_INTERVAL: df = _topup_tail(cell, df, path) # refresh just the recent days df = df.copy() df["doy"] = df["date"].dt.dayofyear.astype("int16") return df, {"cached": True, "cache_age_days": round(age_s / 86400.0, 1)} global _archive_cooldown_until with _cell_lock(cell["id"]): hit = _read_history_cache(path) # another thread may have populated it while we waited if hit is not None: df, age_s = hit df["doy"] = df["date"].dt.dayofyear.astype("int16") return df, {"cached": True, "cache_age_days": round(age_s / 86400.0, 1)} stale = pd.read_parquet(path) if os.path.exists(path) else None def _serve_stale(): stale["doy"] = stale["date"].dt.dayofyear.astype("int16") return stale, {"cached": True, "stale": True, "cache_age_days": None} # Fetch. Open-Meteo is primary (richer: gusts + apparent temp); NASA POWER is # the backup when Open-Meteo is unavailable (network error or its daily limit). # Skip Open-Meteo entirely while it's in a rate-limit cooldown. df = None source = None if time.time() >= _archive_cooldown_until: try: df = _fetch_history(cell) source = "open-meteo" except Exception as e: # noqa: BLE001 if _is_rate_limit(e): _note_rate_limit(e) if df is None: try: df = _fetch_history_nasa(cell) source = "nasa-power" except Exception: # noqa: BLE001 - backup unavailable too df = None if df is None: if stale is not None: return _serve_stale() raise RuntimeError(_cooldown_message()) os.makedirs(CACHE_DIR, exist_ok=True) # Store the raw record; percentiles are derived at request time. df.drop(columns=["doy"]).to_parquet(path, compression="zstd", index=False) return df, {"cached": False, "cache_age_days": 0, "source": source} RECENT_PAST_DAYS = 25 # recent observations window (covers the ~2-week graded view) FORECAST_DAYS = 8 # today + 7 days ahead def _rf_cache_path(cell_id: str) -> str: return os.path.join(CACHE_DIR, f"{cell_id}_rf.parquet") def get_recent_forecast(cell: dict) -> pd.DataFrame: """Recent observations + forward forecast, with humidity as absolute humidity (g/m³). Thin wrapper over the raw loader (see below).""" df = _load_recent_forecast(cell) _derive_humidity(df) return df def _load_recent_forecast(cell: dict) -> pd.DataFrame: """Recent observations AND the forward forecast in ONE forecast-API call. Both the recent (past) view and the forecast (future) view slice from this, so a cell needs just one upstream forecast request per hour (plus the ~monthly archive fetch). Cached per cell for one hour so it still follows model updates. """ path = _rf_cache_path(cell["id"]) if os.path.exists(path): age_h = (time.time() - os.path.getmtime(path)) / 3600.0 if age_h < FORECAST_TTL_HOURS: df = pd.read_parquet(path) df["doy"] = df["date"].dt.dayofyear.astype("int16") return df params = { "latitude": cell["center_lat"], "longitude": cell["center_lon"], "daily": DAILY_VARS, "timezone": "auto", "temperature_unit": "fahrenheit", "precipitation_unit": "inch", "wind_speed_unit": "mph", "past_days": RECENT_PAST_DAYS, "forecast_days": FORECAST_DAYS, } r = _request(FORECAST_URL, params, 60, phase="recent_forecast_fetch") df = _to_frame(r.json()["daily"]) os.makedirs(CACHE_DIR, exist_ok=True) df.drop(columns=["doy"]).to_parquet(path, compression="zstd", index=False) return df _REVGEO_CACHE: dict[tuple[float, float], str | None] = {} def reverse_geocode(lat: float, lon: float) -> str | None: """Best-effort "City, State" label for a point (OpenStreetMap Nominatim). Cached per ~cell so panning around doesn't hammer the service, and failures return None so the caller can fall back to bare coordinates. """ key = (round(lat, 3), round(lon, 3)) if key in _REVGEO_CACHE: return _REVGEO_CACHE[key] label = None try: r = _request( "https://nominatim.openstreetmap.org/reverse", {"lat": lat, "lon": lon, "format": "jsonv2", "zoom": 10, "addressdetails": 1}, 15, phase="reverse_geocode", headers={"User-Agent": "Thermograph/0.1 (local weather grading app)"}, ) a = r.json().get("address", {}) or {} city = (a.get("city") or a.get("town") or a.get("village") or a.get("hamlet") or a.get("suburb") or a.get("county")) region = a.get("state") or a.get("province") or a.get("region") label = ", ".join(p for p in (city, region) if p) or None except Exception: # noqa: BLE001 - reverse geocoding is a nicety, never fatal label = None _REVGEO_CACHE[key] = label return label def geocode(name: str, count: int = 5) -> list[dict]: """Look up places by name (US/Canada biased) via Open-Meteo's geocoder.""" r = _request( "https://geocoding-api.open-meteo.com/v1/search", {"name": name, "count": count, "language": "en", "format": "json"}, 30, phase="geocode", ) results = r.json().get("results", []) or [] return [ { "name": g.get("name"), "admin1": g.get("admin1"), "country": g.get("country"), "country_code": g.get("country_code"), "lat": g.get("latitude"), "lon": g.get("longitude"), } for g in results ]