c3bacdce0c
* Migrate backend dataframe layer from pandas to polars Replace pandas with polars across the backend, dropping both pandas and its pyarrow parquet engine from the dependency set. numpy stays (the grading percentile math is unchanged). - climate.py: parquet IO, source→frame mappings, cache read/topup on polars. New _normalize_read casts the cached `date` column to pl.Date (older files were written by pandas as datetime64[ns]); frames now unify missing values as null so the grading boundary drops them consistently across sources. - grading.py: keep the numpy percentile core; swap the frame→numpy bridge to .to_numpy()/.drop_nulls(), day-of-year/year to polars dt expressions, and the per-row loop to iter_rows(named=True). - views.py: filter/anti-join/concat replace boolean-mask, isin and pd.concat; scalar dates are stdlib datetime.date; a local _months_before helper replaces DateOffset(months=) for the calendar-range default. - app.py, migrate.py: request-date parsing uses datetime.date, removing pandas from the endpoint and migrate layers entirely. - The date column is pl.Date end to end, eliminating the pandas normalize() calls and comparing cleanly against stdlib dates. Payloads are unchanged: calendar, day, grade and forecast responses are byte-for-byte identical to the pandas implementation on the same cached record. Tests ported to polars fixtures, with added coverage for the combined feels-like fallback, calendar month-offset (month-end/leap), and the concat/dedup "fresher source wins" rule. * Port notify.py to polars after merging dev's account system Merge origin/dev (accounts + notification subscriptions) and carry the pandas→ polars migration into the newly added notify.py, which the merge brought in still using pandas — with pandas removed from requirements this broke its import. - notify.py: _candidate_rows filters/sorts the recent bundle with polars expressions and returns iter_rows dicts; date scalars are datetime.date; history/recent emptiness via is_empty(). - test_notify.py: synthetic history/rows built with polars + datetime.
648 lines
28 KiB
Python
648 lines
28 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
|
|
|
|
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 _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")
|
|
except Exception as e: # noqa: BLE001
|
|
# Classify a forecast-API rate limit here 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
|
|
df = _to_frame(r.json()["daily"])
|
|
_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
|
|
]
|