6927a53ea0
The forecast path had no backup — unlike history, which falls back to NASA POWER. When Open-Meteo's forecast API was rate-limited or down, the recent/forecast bundle (and every endpoint that grades future days) failed with a 503. Add MET Norway (yr.no) Locationforecast as a keyless, global forecast backup, mirroring the NASA POWER role for history: - _metno_to_frame: aggregates MET Norway's sub-daily timeseries into the daily schema, converting units (°C→°F, mm→in, m/s→mph) and deriving feels-like from the NWS heat index / wind chill (MET has no gusts or apparent temperature). Precip prefers the 1-hour block and falls back to the 6-hour block so the hourly→6-hourly resolution switch never double-counts. - _fetch_forecast_metno: the backup fetch, with the ToS-required identifying User-Agent and coordinates rounded to 4 decimals. - _load_recent_forecast: on any Open-Meteo forecast failure, try MET Norway before surfacing the error; a shared rate limit still raises the typed, daily-aware WeatherUnavailable. MET Norway is forecast-only (no recent past days), so it's a degraded-but-working fallback: the forecast / day-ahead views keep serving during an Open-Meteo outage. Tests cover the daily aggregation + unit conversion (incl. the no-double-count precip rule) and the fallback wiring.
734 lines
32 KiB
Python
734 lines
32 KiB
Python
"""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 polars as pl
|
|
|
|
import audit
|
|
import store
|
|
|
|
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
|
|
|
|
# Backup forward forecast when Open-Meteo's forecast API is unavailable. MET Norway
|
|
# (yr.no) is free + keyless + global, mirroring NASA POWER's role for history. It
|
|
# returns a sub-daily timeseries in metric units with no gusts or apparent temp, so
|
|
# we aggregate to daily, convert units, and derive feels-like like the NASA path.
|
|
# It is forecast-only — no recent past days — so it's a degraded-but-working fallback.
|
|
METNO_URL = "https://api.met.no/weatherapi/locationforecast/2.0/complete"
|
|
# 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.
|
|
METNO_UA = "Thermograph/0.2 (+https://thermograph.org)"
|
|
|
|
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?
|
|
|
|
|
|
class WeatherUnavailable(RuntimeError):
|
|
"""Weather data can't be fetched right now (rate limit, upstream outage with
|
|
no cached fallback). Carries user-facing text; ``daily`` marks a daily-quota
|
|
exhaustion (resets after UTC midnight) rather than a transient burst limit.
|
|
The API layer maps this to a retryable 503 — it never needs to parse the
|
|
message, and new upstream sources classify their own failures here."""
|
|
|
|
def __init__(self, message: str, daily: bool = False):
|
|
super().__init__(message)
|
|
self.daily = daily
|
|
|
|
|
|
def limit_message(daily: bool) -> str:
|
|
"""The user-facing rate-limit copy, in one place."""
|
|
if 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 service is rate-limited right now — please try again in a minute."
|
|
|
|
|
|
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 _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 _normalize_read(df: pl.DataFrame) -> pl.DataFrame:
|
|
"""Normalize a frame read from the parquet cache: the daily `date` column is a
|
|
calendar date, so coerce it to ``pl.Date`` (older files were written by pandas
|
|
as a nanosecond ``Datetime``). Downstream date math and comparisons all key on
|
|
``pl.Date`` / stdlib ``datetime.date``."""
|
|
if df.schema["date"] != pl.Date:
|
|
df = df.with_columns(pl.col("date").cast(pl.Date))
|
|
return df
|
|
|
|
|
|
def _combined_feels_expr(hi: str = "fmax", lo: str = "fmin") -> pl.Expr:
|
|
"""Expression for 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 (the coalesce picks the non-null side when a comparison is null)."""
|
|
amax, amin = pl.col(hi), pl.col(lo)
|
|
hot, cold = amax - COMFORT_F, COMFORT_F - amin
|
|
chosen = pl.when(hot >= cold).then(amax).otherwise(amin)
|
|
return pl.coalesce([chosen, amax, amin])
|
|
|
|
|
|
def _derive_humidity(df: pl.DataFrame) -> pl.DataFrame:
|
|
"""Replace the raw mean *relative* humidity column with *absolute* humidity
|
|
(grams of water vapor per m³).
|
|
|
|
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 = ((pl.col("tmax") + pl.col("tmin")) / 2.0 - 32.0) * 5.0 / 9.0
|
|
rh = pl.col("humid").cast(pl.Float64, strict=False)
|
|
es = 6.112 * (17.67 * tmean_c / (tmean_c + 243.5)).exp() # sat. vapor pressure, hPa
|
|
return df.with_columns(
|
|
(es * rh * 2.1674 / (273.15 + tmean_c)).round(1).alias("humid")) # abs. humidity, g/m³
|
|
|
|
|
|
def _with_doy(df: pl.DataFrame) -> pl.DataFrame:
|
|
"""(Re)attach the int16 day-of-year column the grading windows key on."""
|
|
return df.with_columns(pl.col("date").dt.ordinal_day().cast(pl.Int16).alias("doy"))
|
|
|
|
|
|
def _write_cache(df: pl.DataFrame, path: str) -> None:
|
|
"""Persist a daily record to the parquet cache — the raw record only, never
|
|
the derived doy column (it's recomputed at read time)."""
|
|
os.makedirs(CACHE_DIR, exist_ok=True)
|
|
df.drop("doy", strict=False).write_parquet(path, compression="zstd")
|
|
|
|
|
|
def _om_daily_params(cell: dict, **window) -> dict:
|
|
"""The Open-Meteo daily-request params every fetch shares — location, the
|
|
variable set, imperial units. The date/window selectors (start_date/end_date
|
|
or past_days/forecast_days) come in as kwargs."""
|
|
return {
|
|
"latitude": cell["center_lat"],
|
|
"longitude": cell["center_lon"],
|
|
"daily": DAILY_VARS,
|
|
"timezone": "auto",
|
|
"temperature_unit": "fahrenheit",
|
|
"precipitation_unit": "inch",
|
|
"wind_speed_unit": "mph",
|
|
**window,
|
|
}
|
|
|
|
|
|
def _finalize_frame(df: pl.DataFrame) -> pl.DataFrame:
|
|
"""Shared tail of every source→frame mapping: unify missing values as null, add
|
|
the combined feels-like, apply the valid-day filter, and attach day-of-year. A
|
|
usable climate day needs a real high/low; other columns may be missing and
|
|
simply grade as None. Float NaN (from numpy-derived columns) is folded into null
|
|
so the whole pipeline models "missing" one way — the grading boundary drops it."""
|
|
df = df.with_columns(pl.col(pl.Float32, pl.Float64).fill_nan(None))
|
|
df = df.with_columns(_combined_feels_expr().alias("feels"))
|
|
df = df.drop_nulls(subset=["tmax", "tmin"])
|
|
return _with_doy(df)
|
|
|
|
|
|
def _to_frame(daily: dict) -> pl.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 = pl.DataFrame(
|
|
{
|
|
"date": 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.
|
|
"fmax": col("apparent_temperature_max"),
|
|
"fmin": col("apparent_temperature_min"),
|
|
}
|
|
)
|
|
df = df.with_columns(
|
|
pl.col("date").str.to_date(),
|
|
*[pl.col(c).cast(pl.Float64, strict=False)
|
|
for c in ("tmax", "tmin", "precip", "wind", "gust", "fmax", "fmin")],
|
|
)
|
|
return _finalize_frame(df)
|
|
|
|
|
|
def _fetch_history(cell: dict) -> pl.DataFrame:
|
|
end = (datetime.date.today() - datetime.timedelta(days=ARCHIVE_LATENCY_DAYS)).isoformat()
|
|
params = _om_daily_params(cell, start_date=START_DATE, end_date=end)
|
|
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) -> pl.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 [None 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 = pl.DataFrame({
|
|
"date": dates,
|
|
"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": [None] * len(dates), # POWER has no gusts
|
|
"humid": col("RH2M", lambda v: v),
|
|
})
|
|
df = df.with_columns(
|
|
pl.col("date").str.to_date("%Y%m%d"),
|
|
*[pl.col(c).cast(pl.Float64, strict=False)
|
|
for c in ("tmax", "tmin", "precip", "wind", "gust")],
|
|
)
|
|
# 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. numpy NaN
|
|
# from these lands in float columns and is folded to null in _finalize_frame.
|
|
df = df.with_columns(
|
|
pl.Series("fmax", _heat_index(df["tmax"].to_numpy(), df["humid"].to_numpy())),
|
|
pl.Series("fmin", _wind_chill(df["tmin"].to_numpy(), df["wind"].to_numpy())),
|
|
)
|
|
return _finalize_frame(df)
|
|
|
|
|
|
def _fetch_history_nasa(cell: dict) -> pl.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 _metno_to_frame(props: dict) -> pl.DataFrame:
|
|
"""Map a MET Norway Locationforecast (yr.no) response to our daily schema.
|
|
|
|
MET Norway returns a per-timestep timeseries (hourly near-term, then 6-hourly)
|
|
in metric units, with no gusts or apparent temperature, so we aggregate to daily
|
|
extremes/means, convert units (°C→°F, mm→in, m/s→mph), and approximate feels-like
|
|
from the NWS heat index / wind chill — the same treatment as the NASA POWER path."""
|
|
# Aggregate the sub-daily steps into per-day values, keyed by (UTC) calendar day.
|
|
agg: dict[str, dict] = {}
|
|
for step in props.get("timeseries", []):
|
|
day = step["time"][:10]
|
|
data = step.get("data", {})
|
|
inst = data.get("instant", {}).get("details", {})
|
|
a = agg.setdefault(day, {"t": [], "rh": [], "wind": [], "precip": None})
|
|
if (t := inst.get("air_temperature")) is not None:
|
|
a["t"].append(t)
|
|
if (rh := inst.get("relative_humidity")) is not None:
|
|
a["rh"].append(rh)
|
|
if (w := inst.get("wind_speed")) is not None:
|
|
a["wind"].append(w)
|
|
# Precip: prefer the 1-hour block (hourly near-term), else the 6-hour block
|
|
# (6-hourly far-term). Never add both — they overlap — so summing each step's
|
|
# chosen block avoids double counting across the resolution switch.
|
|
p = (data.get("next_1_hours") or {}).get("details", {}).get("precipitation_amount")
|
|
if p is None:
|
|
p = (data.get("next_6_hours") or {}).get("details", {}).get("precipitation_amount")
|
|
if p is not None:
|
|
a["precip"] = (a["precip"] or 0.0) + p
|
|
|
|
dates = sorted(agg)
|
|
c2f = lambda c: c * 9.0 / 5.0 + 32.0
|
|
df = pl.DataFrame({
|
|
"date": dates,
|
|
"tmax": [c2f(max(agg[d]["t"])) if agg[d]["t"] else None for d in dates],
|
|
"tmin": [c2f(min(agg[d]["t"])) if agg[d]["t"] else None for d in dates],
|
|
"precip": [agg[d]["precip"] / 25.4 if agg[d]["precip"] is not None else None
|
|
for d in dates],
|
|
"wind": [max(agg[d]["wind"]) * 2.2369362920544 if agg[d]["wind"] else None
|
|
for d in dates],
|
|
"gust": [None] * len(dates), # MET Norway has no gusts
|
|
"humid": [sum(agg[d]["rh"]) / len(agg[d]["rh"]) if agg[d]["rh"] else None
|
|
for d in dates],
|
|
})
|
|
df = df.with_columns(
|
|
pl.col("date").str.to_date(),
|
|
*[pl.col(c).cast(pl.Float64, strict=False)
|
|
for c in ("tmax", "tmin", "precip", "wind", "gust")],
|
|
)
|
|
df = df.with_columns(
|
|
pl.Series("fmax", _heat_index(df["tmax"].to_numpy(), df["humid"].to_numpy())),
|
|
pl.Series("fmin", _wind_chill(df["tmin"].to_numpy(), df["wind"].to_numpy())),
|
|
)
|
|
return _finalize_frame(df)
|
|
|
|
|
|
def _fetch_forecast_metno(cell: dict) -> pl.DataFrame:
|
|
"""Backup forward-forecast fetch from MET Norway / yr.no (used when Open-Meteo's
|
|
forecast API is unavailable). Coordinates are rounded to 4 decimals per MET's
|
|
ToS (improves their cache hit rate)."""
|
|
r = _request(
|
|
METNO_URL,
|
|
{"lat": round(cell["center_lat"], 4), "lon": round(cell["center_lon"], 4)},
|
|
60,
|
|
phase="forecast_metno",
|
|
headers={"User-Agent": METNO_UA},
|
|
)
|
|
return _metno_to_frame(r.json().get("properties", {}))
|
|
|
|
|
|
def _fetch_history_range(cell: dict, start_date: str, end_date: str) -> pl.DataFrame:
|
|
"""Fetch just a date range of archive history (used to top up the recent tail)."""
|
|
params = _om_daily_params(cell, start_date=start_date, end_date=end_date)
|
|
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 = _normalize_read(pl.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: pl.DataFrame, path: str) -> pl.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 = df["date"].max()
|
|
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
|
|
# Concatenate archive-first, recent-last so `keep="last"` prefers a freshly
|
|
# fetched day over its cached duplicate; maintain_order keeps that precedence
|
|
# before the final chronological sort.
|
|
merged = (pl.concat([df, recent.drop("doy", strict=False)], how="diagonal_relaxed")
|
|
.unique(subset="date", keep="last", maintain_order=True)
|
|
.sort("date"))
|
|
_write_cache(merged, path)
|
|
return merged
|
|
|
|
|
|
def get_history(cell: dict) -> tuple[pl.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)
|
|
return _derive_humidity(df), meta
|
|
|
|
|
|
def load_cached_history(cell: dict) -> pl.DataFrame | None:
|
|
"""History from the parquet cache ONLY — never fetches upstream and never tops
|
|
up the tail. Powers the warm-only prefetch path (which must not spend upstream
|
|
quota) and the offline migrate script. None when the cell has no
|
|
(schema-complete) cached record."""
|
|
hit = _read_history_cache(_cache_path(cell["id"]))
|
|
if hit is None:
|
|
return None
|
|
return _derive_humidity(_with_doy(hit[0]))
|
|
|
|
|
|
def _load_history(cell: dict) -> tuple[pl.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
|
|
return _with_doy(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
|
|
return _with_doy(df), {"cached": True, "cache_age_days": round(age_s / 86400.0, 1)}
|
|
|
|
stale = _normalize_read(pl.read_parquet(path)) if os.path.exists(path) else None
|
|
|
|
def _serve_stale():
|
|
return _with_doy(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 WeatherUnavailable(limit_message(_archive_limit_daily),
|
|
daily=_archive_limit_daily)
|
|
|
|
# Store the raw record; percentiles are derived at request time.
|
|
_write_cache(df, path)
|
|
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 recent_stamp(cell_id: str) -> int:
|
|
"""Identity stamp (mtime, whole seconds) of the cell's cached recent+forecast
|
|
parquet; 0 when absent. Changes exactly when the recent/forecast data does, so
|
|
it's the freshness token for every payload that grades recent or future days."""
|
|
try:
|
|
return int(os.path.getmtime(_rf_cache_path(cell_id)))
|
|
except OSError:
|
|
return 0
|
|
|
|
|
|
def get_recent_forecast(cell: dict) -> pl.DataFrame:
|
|
"""Recent observations + forward forecast, with humidity as absolute humidity
|
|
(g/m³). Thin wrapper over the raw loader (see below)."""
|
|
return _derive_humidity(_load_recent_forecast(cell))
|
|
|
|
|
|
def _load_recent_forecast(cell: dict) -> pl.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:
|
|
return _with_doy(_normalize_read(pl.read_parquet(path)))
|
|
|
|
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,
|
|
}
|
|
try:
|
|
r = _request(FORECAST_URL, params, 60, phase="recent_forecast_fetch")
|
|
df = _to_frame(r.json()["daily"])
|
|
except Exception as e: # noqa: BLE001
|
|
# Open-Meteo forecast unavailable (rate limit or outage). Fall back to MET
|
|
# Norway (yr.no) — global + keyless — mirroring the archive's NASA POWER
|
|
# backup. MET Norway is forecast-only (no recent past days), so this keeps
|
|
# the forecast / day-ahead views working in a degraded form.
|
|
try:
|
|
df = _fetch_forecast_metno(cell)
|
|
except Exception: # noqa: BLE001 - backup unavailable too; surface the original
|
|
# Classify the original Open-Meteo rate limit so callers get the typed
|
|
# error (the archive path classifies its own inside _load_history).
|
|
if is_rate_limit(e):
|
|
daily = "daily" in _rate_limit_reason(e).lower()
|
|
raise WeatherUnavailable(limit_message(daily), daily=daily) from e
|
|
raise
|
|
_write_cache(df, path)
|
|
return df
|
|
|
|
|
|
_REVGEO_CACHE: dict[str, str | None] = {}
|
|
|
|
# Nominatim's usage policy allows ~1 request/second and rejects bursts. The compare
|
|
# page loads several locations at once, so serialize the reverse lookups here and
|
|
# space them out — otherwise the burst is rate-limited, a null label gets cached, and
|
|
# those locations are left showing bare coordinates.
|
|
_REVGEO_LOCK = threading.Lock()
|
|
_REVGEO_MIN_INTERVAL = 1.1 # seconds between successive Nominatim reverse calls
|
|
_revgeo_last = 0.0
|
|
|
|
|
|
def reverse_geocode_cached(lat: float, lon: float) -> tuple[bool, str | None]:
|
|
"""(found, label) from the in-memory + SQLite revgeo caches only — never calls
|
|
Nominatim. Used by paths that must not add upstream traffic (prefetch)."""
|
|
key = store.revgeo_key(lat, lon)
|
|
if key in _REVGEO_CACHE:
|
|
return (True, _REVGEO_CACHE[key])
|
|
found, label = store.get_revgeo(key)
|
|
if found:
|
|
_REVGEO_CACHE[key] = label
|
|
return (found, label)
|
|
|
|
|
|
def reverse_geocode(lat: float, lon: float) -> str | None:
|
|
"""Best-effort neighbourhood/city label for a point (OpenStreetMap Nominatim).
|
|
|
|
Cached per ~cell — in memory for the process and in SQLite across restarts —
|
|
so panning around (or redeploying) doesn't re-hit the service, and failures
|
|
return None so the caller can fall back to bare coordinates.
|
|
"""
|
|
key = store.revgeo_key(lat, lon)
|
|
found, label = reverse_geocode_cached(lat, lon)
|
|
if found:
|
|
return label
|
|
# Serialize + rate-limit the upstream call. Concurrent callers (the compare page
|
|
# loads several locations at once) would otherwise burst past Nominatim's ~1/sec
|
|
# limit and get rate-limited, caching a null label. Under the lock we re-check the
|
|
# cache (a peer may have just resolved this cell) and space successive calls out.
|
|
global _revgeo_last
|
|
with _REVGEO_LOCK:
|
|
found, label = reverse_geocode_cached(lat, lon)
|
|
if found:
|
|
return label
|
|
wait = _REVGEO_MIN_INTERVAL - (time.monotonic() - _revgeo_last)
|
|
if wait > 0:
|
|
time.sleep(wait)
|
|
label = None
|
|
try:
|
|
r = _request(
|
|
"https://nominatim.openstreetmap.org/reverse",
|
|
# zoom 14 resolves to the suburb/neighbourhood level so we can lead
|
|
# with it when OSM has one (zoom 10 only ever returns the city).
|
|
# accept-language=en keeps labels in one script worldwide (matches
|
|
# the forward geocoder's language=en).
|
|
{"lat": lat, "lon": lon, "format": "jsonv2", "zoom": 14,
|
|
"addressdetails": 1, "accept-language": "en"},
|
|
15,
|
|
phase="reverse_geocode",
|
|
headers={"User-Agent": "Thermograph/0.1 (local weather grading app)"},
|
|
)
|
|
a = r.json().get("address", {}) or {}
|
|
# Lead with the neighbourhood when available, then the city, then region.
|
|
neighborhood = (a.get("neighbourhood") or a.get("suburb")
|
|
or a.get("quarter") or a.get("city_district")
|
|
or a.get("borough"))
|
|
city = (a.get("city") or a.get("town") or a.get("village")
|
|
or a.get("hamlet") or a.get("county"))
|
|
region = a.get("state") or a.get("province") or a.get("region")
|
|
country = a.get("country")
|
|
# dict.fromkeys drops duplicates (e.g. neighbourhood == city) in order.
|
|
# Country trails the neighbourhood/city/region so the label reads as a full
|
|
# hierarchy ("West Seattle, Seattle, Washington, United States"); the
|
|
# frontend leads with the first part and mutes the rest.
|
|
parts = dict.fromkeys(p for p in (neighborhood, city, region, country) if p)
|
|
label = ", ".join(parts) or None
|
|
except Exception: # noqa: BLE001 - reverse geocoding is a nicety, never fatal
|
|
label = None
|
|
_revgeo_last = time.monotonic()
|
|
_REVGEO_CACHE[key] = label
|
|
store.put_revgeo(key, label) # survive restarts (a None label retries after its TTL)
|
|
return label
|
|
|
|
|
|
def geocode(name: str, count: int = 5) -> list[dict]:
|
|
"""Look up places by name worldwide 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"),
|
|
"population": g.get("population"),
|
|
}
|
|
for g in results
|
|
]
|