Phase 3: flip recent/forecast leg off Open-Meteo (NASA past + MET Norway forward) #1

Closed
admin_emi wants to merge 1 commit from feature/metno-recent-flip into feature/era5-seed
2 changed files with 129 additions and 92 deletions

View file

@ -60,7 +60,8 @@ MIN_ARCHIVE_DAYS = 3650 # ~10 yrs: far above any sync window, far bel
# cached indefinitely (refetched only to add new metric columns). Only the recent # cached indefinitely (refetched only to add new metric columns). Only the recent
# tail is refreshed — a small incremental fetch, at most hourly. # tail is refreshed — a small incremental fetch, at most hourly.
HISTORY_TOPUP_INTERVAL = 3600 # seconds between recent-tail refresh attempts HISTORY_TOPUP_INTERVAL = 3600 # seconds between recent-tail refresh attempts
FORECAST_TTL_HOURS = 1 # refetch the forward forecast hourly to track updates FORECAST_TTL_HOURS = 4 # refetch the recent+forecast bundle every ~4h (lower IO;
# daily-resolution forecast normals move slowly)
# The archive (historical) endpoint. Overridable so it can point at a self-hosted # The archive (historical) endpoint. Overridable so it can point at a self-hosted
# Open-Meteo instance (ERA5 in object storage) instead of the rate-limited public # Open-Meteo instance (ERA5 in object storage) instead of the rate-limited public
@ -80,12 +81,13 @@ NASA_POWER_URL = "https://power.larc.nasa.gov/api/temporal/daily/point"
NASA_START = "19810101" NASA_START = "19810101"
NASA_FILL = -900.0 # POWER's missing-value sentinel is ~-999 NASA_FILL = -900.0 # POWER's missing-value sentinel is ~-999
# Backup forward forecast when Open-Meteo's forecast API is unavailable. MET Norway # Forward-forecast source (primary for the forward days). MET Norway (yr.no) is free
# (yr.no) is free + keyless + global, mirroring NASA POWER's role for history. It # + keyless + global. It returns a sub-daily timeseries in metric units with no gusts
# returns a sub-daily timeseries in metric units with no gusts or apparent temp, so # or apparent temp, so we aggregate to daily, convert units, derive feels-like like
# we aggregate to daily, convert units, and derive feels-like like the NASA path. # the NASA path, and fill gusts from Meteostat. It is forecast-only (no recent past),
# It is forecast-only — no recent past days — so it's a degraded-but-working fallback. # so the recent observed days come from a NASA POWER range instead. The /compact
METNO_URL = "https://api.met.no/weatherapi/locationforecast/2.0/complete" # endpoint carries every field we use at a smaller payload than /complete.
METNO_URL = "https://api.met.no/weatherapi/locationforecast/2.0/compact"
# MET Norway's ToS requires an identifying User-Agent (a missing/generic one is # MET Norway's ToS requires an identifying User-Agent (a missing/generic one is
# 403'd); include the app and a contact URL so they can reach us if usage misbehaves. # 403'd); include the app and a contact URL so they can reach us if usage misbehaves.
METNO_UA = "Thermograph/0.2 (+https://thermograph.org)" METNO_UA = "Thermograph/0.2 (+https://thermograph.org)"
@ -824,6 +826,10 @@ def _load_history(cell: dict) -> tuple[pl.DataFrame, dict]:
RECENT_PAST_DAYS = 25 # recent observations window (covers the ~2-week graded view) RECENT_PAST_DAYS = 25 # recent observations window (covers the ~2-week graded view)
FORECAST_DAYS = 8 # today + 7 days ahead FORECAST_DAYS = 8 # today + 7 days ahead
# NASA POWER near-real-time lags a couple of days, so the recent observed window ends
# here; the small today-1/today-2 gap before the MET Norway forecast begins grades as
# missing (drop_nulls), bracketed by history behind and forecast ahead.
RECENT_END_LAG_DAYS = 2
def _rf_cache_path(cell_id: str) -> str: def _rf_cache_path(cell_id: str) -> str:
@ -865,20 +871,27 @@ def load_cached_recent_forecast(cell: dict) -> "pl.DataFrame | None":
return None return None
def _load_recent_forecast(cell: dict) -> pl.DataFrame: def _fetch_recent_forecast(cell: dict) -> pl.DataFrame:
"""Recent observations AND the forward forecast in ONE forecast-API call. """Recent observations + forward forecast WITHOUT Open-Meteo: the recent observed
window from a NASA POWER range (measured), the forward days from MET Norway,
merged by date. Gusts which neither source carries are filled from Meteostat:
measured for the observed days, estimated from wind for the forecast days."""
today = datetime.date.today()
start = (today - datetime.timedelta(days=RECENT_PAST_DAYS)).isoformat()
end = (today - datetime.timedelta(days=RECENT_END_LAG_DAYS)).isoformat()
past = _fetch_history_nasa(cell, start=start, end=end) # measured + Meteostat gusts
fwd = meteostat.fill_gusts(
cell["center_lat"], cell["center_lon"], _fetch_forecast_metno(cell))
# Forecast wins on any overlapping day (freshest model value for today); the NASA
# observed days fill the recent past MET Norway lacks.
return (pl.concat([past, fwd], how="diagonal_relaxed")
.unique(subset="date", keep="last", maintain_order=True)
.sort("date"))
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.
"""
cell_id = cell["id"]
hit = _read_recent_backed(cell_id)
if hit is not None:
cached, age_s = hit
if age_s / 3600.0 < FORECAST_TTL_HOURS:
return _with_doy(cached)
def _fetch_recent_forecast_om(cell: dict) -> pl.DataFrame:
"""Fallback recent+forecast bundle from the Open-Meteo forecast API (the former
primary): recent past + forward days in one call."""
params = { params = {
"latitude": cell["center_lat"], "latitude": cell["center_lat"],
"longitude": cell["center_lon"], "longitude": cell["center_lon"],
@ -890,49 +903,59 @@ def _load_recent_forecast(cell: dict) -> pl.DataFrame:
"past_days": RECENT_PAST_DAYS, "past_days": RECENT_PAST_DAYS,
"forecast_days": FORECAST_DAYS, "forecast_days": FORECAST_DAYS,
} }
# Skip Open-Meteo's forecast endpoint entirely while it's in its own r = _request(FORECAST_URL, params, 60, phase="recent_forecast_fetch")
# rate-limit cooldown (see _note_forecast_rate_limit) -- mirrors return _to_frame(r.json()["daily"])
# _load_history's archive-cooldown check, so a forecast brownout doesn't
# retry-storm the endpoint from every subscribed cell and every live request
# independently; it falls straight through to the MET Norway backup instead. def _load_recent_forecast(cell: dict) -> pl.DataFrame:
"""Recent observations AND the forward forecast for a cell.
The primary source is keyless and Open-Meteo-free: a NASA POWER recent-past range
plus the MET Norway forward forecast (see _fetch_recent_forecast). The Open-Meteo
forecast API is the fallback (skipped while it's in its own rate-limit cooldown,
see _note_forecast_rate_limit), then a stale cache. Cached per cell for
FORECAST_TTL_HOURS so it follows model updates without per-hour upstream IO.
"""
cell_id = cell["id"]
hit = _read_recent_backed(cell_id)
if hit is not None:
cached, age_s = hit
if age_s / 3600.0 < FORECAST_TTL_HOURS:
return _with_doy(cached)
df = None df = None
primary_error = None try:
if time.time() >= _forecast_cooldown_until: df = _fetch_recent_forecast(cell) # NASA recent + MET forward (primary)
except Exception: # noqa: BLE001 - keyless primary unavailable; try Open-Meteo
pass
forecast_error = None
if df is None and time.time() >= _forecast_cooldown_until:
# Fall back to the Open-Meteo forecast API, skipped while it's in its own
# rate-limit cooldown so a brownout doesn't retry-storm the endpoint.
try: try:
r = _request(FORECAST_URL, params, 60, phase="recent_forecast_fetch") df = _fetch_recent_forecast_om(cell)
df = _to_frame(r.json()["daily"])
except Exception as e: # noqa: BLE001 except Exception as e: # noqa: BLE001
if is_rate_limit(e): if is_rate_limit(e):
_note_forecast_rate_limit(e) _note_forecast_rate_limit(e)
primary_error = e forecast_error = e
if df is None: if df is None:
# Open-Meteo forecast unavailable (rate limit, cooldown, or outage). Fall # Last resort: serve the stale cache if we have one, WITHOUT rewriting it, so
# back to MET Norway (yr.no) — global + keyless — mirroring the archive's # recent_stamp stays old and the next request retries upstream first (mirrors
# NASA POWER backup. MET Norway is forecast-only (no recent past days), # _load_history's stale-serve).
# so this keeps the forecast / day-ahead views working in a degraded form. if hit is not None:
try: return _with_doy(hit[0])
df = _fetch_forecast_metno(cell) # No cache either: surface the error, classifying an Open-Meteo rate limit as
except Exception: # noqa: BLE001 - backup unavailable too # the typed, daily-aware WeatherUnavailable. forecast_error is None when the
# Last resort: serve the stale cache if we have one. An hours-old bundle # fallback was skipped outright (cooldown active) — report the cooldown then.
# (which still carries the recent observed days MET Norway lacks) beats a if forecast_error is not None and is_rate_limit(forecast_error):
# hard failure, mirroring _load_history's stale-serve. Return WITHOUT daily = "daily" in _rate_limit_reason(forecast_error).lower()
# rewriting it, so recent_stamp stays old and the next request still raise WeatherUnavailable(limit_message(daily), daily=daily) from forecast_error
# retries upstream first rather than serving this as if it were fresh. if forecast_error is not None:
if hit is not None: raise forecast_error
return _with_doy(hit[0]) raise WeatherUnavailable(limit_message(_forecast_limit_daily),
# No cache either: surface the original error, classifying an Open-Meteo daily=_forecast_limit_daily)
# rate limit as the typed, daily-aware WeatherUnavailable (the archive
# path classifies its own inside _load_history). primary_error is None
# when the primary fetch was skipped outright (cooldown active) --
# report the cooldown itself in that case.
if primary_error is not None and is_rate_limit(primary_error):
daily = "daily" in _rate_limit_reason(primary_error).lower()
raise WeatherUnavailable(limit_message(daily), daily=daily) from primary_error
if primary_error is not None:
raise primary_error
raise WeatherUnavailable(limit_message(_forecast_limit_daily),
daily=_forecast_limit_daily)
_write_recent_backed(cell_id, df) _write_recent_backed(cell_id, df)
return df return df

View file

@ -137,34 +137,53 @@ def test_metno_to_frame_aggregates_daily_and_converts_units():
assert round(d2["precip"], 4) == round(6.0 / 25.4, 4) # only the 6-hour block assert round(d2["precip"], 4) == round(6.0 / 25.4, 4) # only the 6-hour block
def test_recent_forecast_falls_back_to_metno(monkeypatch, tmp_path): def test_recent_forecast_merges_nasa_past_and_metno_forward(monkeypatch, tmp_path):
"""When Open-Meteo's forecast API fails, the MET Norway backup serves the frame.""" """The recent/forecast bundle is built from a NASA POWER recent-past range plus the
MET Norway forward forecast, merged by date (Open-Meteo not consulted)."""
monkeypatch.setattr(climate, "CACHE_DIR", str(tmp_path)) monkeypatch.setattr(climate, "CACHE_DIR", str(tmp_path))
cell = {"id": "fallback_cell", "center_lat": 47.6062, "center_lon": -122.3321} cell = {"id": "rf_primary", "center_lat": 47.6, "center_lon": -122.3}
met = {"timeseries": [
_metno_step("2026-07-16T00:00:00Z", 18.0, 55.0, 3.0, p1=0.0),
_metno_step("2026-07-17T00:00:00Z", 22.0, 45.0, 5.0, p6=1.0),
]}
class Resp: past = climate._finalize_approximated(pl.DataFrame({
def json(self): return {"properties": met} "date": [datetime.date(2026, 7, 10), datetime.date(2026, 7, 11)],
"tmax": [70.0, 72.0], "tmin": [50.0, 51.0], "precip": [0.0, 0.1],
"wind": [5.0, 6.0], "humid": [60.0, 55.0],
}))
fwd = climate._finalize_approximated(pl.DataFrame({
"date": [datetime.date(2026, 7, 16), datetime.date(2026, 7, 17)],
"tmax": [80.0, 82.0], "tmin": [60.0, 61.0], "precip": [0.0, 0.0],
"wind": [3.0, 4.0], "humid": [40.0, 45.0],
}))
monkeypatch.setattr(climate, "_fetch_history_nasa", lambda c, start, end: past)
monkeypatch.setattr(climate, "_fetch_forecast_metno", lambda c: fwd)
monkeypatch.setattr(climate.meteostat, "fill_gusts", lambda lat, lon, df: df)
def _om_should_not_run(c): raise AssertionError("Open-Meteo forecast should not run")
monkeypatch.setattr(climate, "_fetch_recent_forecast_om", _om_should_not_run)
def fake_request(url, params, timeout, *, phase, headers=None, attempts=climate.MAX_ATTEMPTS):
if url == climate.FORECAST_URL:
raise RuntimeError("open-meteo forecast outage")
assert url == climate.METNO_URL
assert headers and headers.get("User-Agent"), "MET Norway needs a User-Agent"
return Resp()
monkeypatch.setattr(climate, "_request", fake_request)
df = climate._load_recent_forecast(cell) df = climate._load_recent_forecast(cell)
assert len(df) == 2 and df["date"].max() == datetime.date(2026, 7, 17) assert df["date"].to_list() == [
datetime.date(2026, 7, 10), datetime.date(2026, 7, 11),
datetime.date(2026, 7, 16), datetime.date(2026, 7, 17)]
def test_recent_forecast_falls_back_to_open_meteo(monkeypatch, tmp_path):
"""When the NASA + MET Norway primary fails, the Open-Meteo forecast API serves."""
monkeypatch.setattr(climate, "CACHE_DIR", str(tmp_path))
monkeypatch.setattr(climate, "_forecast_cooldown_until", 0.0)
cell = {"id": "rf_fallback", "center_lat": 47.6, "center_lon": -122.3}
def _primary_down(c): raise RuntimeError("NASA + MET Norway down")
monkeypatch.setattr(climate, "_fetch_recent_forecast", _primary_down)
om = climate._to_frame(_om_daily(3))
monkeypatch.setattr(climate, "_fetch_recent_forecast_om", lambda c: om)
df = climate._load_recent_forecast(cell)
assert df.height == om.height
def test_recent_forecast_serves_stale_cache_when_all_sources_fail(monkeypatch, tmp_path): def test_recent_forecast_serves_stale_cache_when_all_sources_fail(monkeypatch, tmp_path):
"""With Open-Meteo AND MET Norway both down, an existing (stale) cache is served """With every source down (the NASA + MET Norway primary and the Open-Meteo
rather than failing and it is NOT rewritten, so its mtime stays old and the fallback), an existing (stale) cache is served rather than failing and it is NOT
next request still retries upstream first.""" rewritten, so its mtime stays old and the next request still retries upstream first."""
import os import os
import time import time
monkeypatch.setattr(climate, "CACHE_DIR", str(tmp_path)) monkeypatch.setattr(climate, "CACHE_DIR", str(tmp_path))
@ -370,28 +389,23 @@ def test_history_tail_falls_back_to_open_meteo(monkeypatch):
# --- forecast cooldown (mirrors the archive path's, tracked separately) ----- # --- forecast cooldown (mirrors the archive path's, tracked separately) -----
def test_forecast_cooldown_skips_open_meteo_and_falls_to_metno(monkeypatch, tmp_path): def test_forecast_cooldown_skips_the_open_meteo_fallback(monkeypatch, tmp_path):
"""While the forecast cooldown is active, the forecast fetch must not call """While the forecast cooldown is active, the Open-Meteo FALLBACK is skipped: with
Open-Meteo at all -- straight to the MET Norway backup, mirroring the keyless primary (NASA + MET Norway) also down and no cache, the load surfaces
_load_history's archive-cooldown skip.""" WeatherUnavailable without ever calling Open-Meteo."""
import time as time_mod import time as time_mod
monkeypatch.setattr(climate, "CACHE_DIR", str(tmp_path)) monkeypatch.setattr(climate, "CACHE_DIR", str(tmp_path))
monkeypatch.setattr(climate, "_forecast_cooldown_until", time_mod.time() + 60) monkeypatch.setattr(climate, "_forecast_cooldown_until", time_mod.time() + 60)
cell = {"id": "fc_cooldown_cell", "center_lat": 47.6, "center_lon": -122.3} cell = {"id": "fc_cooldown_cell", "center_lat": 47.6, "center_lon": -122.3}
met = {"timeseries": [_metno_step("2026-07-16T00:00:00Z", 18.0, 55.0, 3.0, p1=0.0)]} def _primary_down(c): raise RuntimeError("NASA + MET Norway down")
monkeypatch.setattr(climate, "_fetch_recent_forecast", _primary_down)
def _om_should_not_run(c):
raise AssertionError("Open-Meteo fallback must not run during cooldown")
monkeypatch.setattr(climate, "_fetch_recent_forecast_om", _om_should_not_run)
class Resp: with pytest.raises(climate.WeatherUnavailable):
def json(self): return {"properties": met} climate._load_recent_forecast(cell)
def fake_request(url, params, timeout, *, phase, headers=None, attempts=climate.MAX_ATTEMPTS):
assert url != climate.FORECAST_URL, "must not call Open-Meteo during cooldown"
assert url == climate.METNO_URL
return Resp()
monkeypatch.setattr(climate, "_request", fake_request)
df = climate._load_recent_forecast(cell)
assert len(df) == 1
def test_forecast_429_sets_its_own_cooldown(monkeypatch, tmp_path): def test_forecast_429_sets_its_own_cooldown(monkeypatch, tmp_path):