diff --git a/alembic/versions/0002_climate_hypertables.py b/alembic/versions/0002_climate_hypertables.py new file mode 100644 index 0000000..8993cc8 --- /dev/null +++ b/alembic/versions/0002_climate_hypertables.py @@ -0,0 +1,112 @@ +"""climate record: TimescaleDB hypertables + +Moves the raw daily climate record out of per-cell parquet files and into +Postgres, so the app queries the database instead of the filesystem (see +data/climate_store.py). Three objects: + +* ``climate_history`` — a TimescaleDB hypertable of the full daily archive, + durable/LOGGED (a 45-year, rate-limited refetch is expensive). Range-partitioned + on ``date`` with a large 5-year chunk interval (the access pattern is + all-rows-for-one-cell, so time chunking is incidental — a big interval keeps the + chunk count tiny). Old chunks compress; the recent chunk stays writable for the + hourly tail top-up. +* ``climate_recent`` — a plain table (NOT a hypertable) for the recent+forecast + bundle. It holds *future* dates and is fully rewritten hourly, both of which + fight time-partitioning. +* ``climate_sync`` — per-cell freshness (epoch seconds), replacing the parquet + file mtimes that drove the topup cadence, the forecast TTL, and the + ``recent_stamp`` token. + +Postgres-only: guarded to no-op on the SQLite bind (tests / CI / offline Alembic), +where climate.py uses the parquet backend and these tables are never needed. + +Revision ID: 0002_climate_hypertables +Revises: 0001_initial_schema +Create Date: 2026-07-20 +""" +from alembic import op + +revision = "0002_climate_hypertables" +down_revision = "0001_initial_schema" +branch_labels = None +depends_on = None + + +# The daily measurement columns shared by both tables (see data/climate_store.COLS). +_MEASURES = """ + tmax DOUBLE PRECISION, + tmin DOUBLE PRECISION, + precip DOUBLE PRECISION, + wind DOUBLE PRECISION, + gust DOUBLE PRECISION, + humid DOUBLE PRECISION, + fmax DOUBLE PRECISION, + fmin DOUBLE PRECISION, + feels DOUBLE PRECISION +""" + + +def upgrade() -> None: + if op.get_bind().dialect.name != "postgresql": + return # SQLite fallback (tests/CI): climate uses the parquet backend. + + op.execute("CREATE EXTENSION IF NOT EXISTS timescaledb") + + # History: durable hypertable. PK (cell_id, date) includes the range-partition + # column (date), which TimescaleDB requires of any unique index, and gives the + # single-cell chronological scan the read path needs for free. + op.execute(f""" + CREATE TABLE climate_history ( + cell_id TEXT NOT NULL, + date DATE NOT NULL, + {_MEASURES}, + PRIMARY KEY (cell_id, date) + ) + """) + op.execute( + "SELECT create_hypertable('climate_history', " + "by_range('date', INTERVAL '5 years'))") + + # Columnar compression, grouped per cell. Only chunks older than a year are + # compressed — the hourly tail top-up only ever writes dates within the last + # week, i.e. the current uncompressed chunk, so compression never collides with + # a write. + op.execute(""" + ALTER TABLE climate_history SET ( + timescaledb.compress, + timescaledb.compress_segmentby = 'cell_id', + timescaledb.compress_orderby = 'date' + ) + """) + op.execute("SELECT add_compression_policy('climate_history', INTERVAL '365 days')") + + # Recent + forecast bundle: plain table (holds future dates, rewritten hourly). + op.execute(f""" + CREATE TABLE climate_recent ( + cell_id TEXT NOT NULL, + date DATE NOT NULL, + {_MEASURES}, + PRIMARY KEY (cell_id, date) + ) + """) + + # Per-cell freshness — epoch seconds (DOUBLE mirrors store.py's updated_at and + # makes recent_stamp = int(recent_synced_at) trivial). + op.execute(""" + CREATE TABLE climate_sync ( + cell_id TEXT PRIMARY KEY, + history_synced_at DOUBLE PRECISION, + recent_synced_at DOUBLE PRECISION, + cols_version SMALLINT NOT NULL DEFAULT 1 + ) + """) + + +def downgrade() -> None: + if op.get_bind().dialect.name != "postgresql": + return + op.execute("DROP TABLE IF EXISTS climate_sync") + op.execute("DROP TABLE IF EXISTS climate_recent") + op.execute("DROP TABLE IF EXISTS climate_history") + # Leave the timescaledb extension installed — other objects may rely on it and + # dropping an extension is a heavier, rarely-wanted operation. diff --git a/data/climate.py b/data/climate.py index b52a90c..3507ad4 100644 --- a/data/climate.py +++ b/data/climate.py @@ -17,6 +17,7 @@ import polars as pl from core import audit from core import metrics import paths +from data import climate_store from data import store CACHE_DIR = os.path.join(paths.DATA_DIR, "cache") @@ -277,6 +278,85 @@ def _write_cache(df: pl.DataFrame, path: str) -> None: df.drop("doy", strict=False).write_parquet(path, compression="zstd") +# --- cache backend dispatch -------------------------------------------------- +# The raw daily record lives in TimescaleDB hypertables on Postgres (prod) and in +# per-cell parquet files otherwise (dev / tests / offline). These helpers hide the +# choice so the loaders below stay backend-agnostic: they return the same shapes a +# parquet read produced — a frame WITHOUT the derived `doy` column, plus an age in +# seconds since the last write (the mtime age the topup/TTL logic keyed on). The +# active backend is climate_store.is_postgres() (same THERMOGRAPH_DATABASE_URL +# switch as accounts/db.py and data/store.py). + + +def _read_history_backed(cell_id: str): + """(schema-complete history frame without doy, age_s) or None.""" + if climate_store.is_postgres(): + hit = climate_store.read_history(cell_id) + if hit is None: + return None + df, synced_at = hit + return df, time.time() - synced_at + return _read_history_cache(_cache_path(cell_id)) + + +def _stale_history_backed(cell_id: str): + """Any cached history rows regardless of schema completeness (for the + all-fetches-failed stale serve), or None.""" + if climate_store.is_postgres(): + return climate_store.read_history_raw(cell_id) + path = _cache_path(cell_id) + return _normalize_read(pl.read_parquet(path)) if os.path.exists(path) else None + + +def _write_history_backed(cell_id: str, full_df: pl.DataFrame, + delta_df: "pl.DataFrame | None" = None) -> None: + """Persist a cell's history. Parquet rewrites the whole file from ``full_df``; + Postgres upserts ``delta_df`` when given (just the topped-up tail) else the full + frame — the upsert makes a delta write equivalent to a full rewrite.""" + if climate_store.is_postgres(): + climate_store.write_history( + cell_id, delta_df if delta_df is not None else full_df, time.time()) + else: + _write_cache(full_df, _cache_path(cell_id)) + + +def _touch_history_backed(cell_id: str) -> None: + """Reset the hourly topup timer when the tail is already current, without a + rewrite — bump history_synced_at on Postgres, or the file mtime on parquet.""" + if climate_store.is_postgres(): + climate_store.touch_history(cell_id, time.time()) + else: + try: + os.utime(_cache_path(cell_id), None) + except OSError: + pass + + +def _read_recent_backed(cell_id: str): + """(recent+forecast frame without doy, age_s) or None.""" + if climate_store.is_postgres(): + hit = climate_store.read_recent(cell_id) + if hit is None: + return None + df, synced_at = hit + return df, time.time() - synced_at + path = _rf_cache_path(cell_id) + if not os.path.exists(path): + return None + try: + return _normalize_read(pl.read_parquet(path)), time.time() - os.path.getmtime(path) + except Exception: # noqa: BLE001 - a truncated/corrupt cache file is a miss + return None + + +def _write_recent_backed(cell_id: str, df: pl.DataFrame) -> None: + """Persist a cell's recent+forecast bundle (full replace on both backends).""" + if climate_store.is_postgres(): + climate_store.write_recent(cell_id, df, time.time()) + else: + _write_cache(df, _rf_cache_path(cell_id)) + + 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 @@ -503,12 +583,13 @@ def _read_history_cache(path): return df, time.time() - os.path.getmtime(path) -def _topup_tail(cell: dict, df: pl.DataFrame, path: str) -> pl.DataFrame: +def _topup_tail(cell: dict, df: pl.DataFrame) -> 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 + cell_id = cell["id"] + with _cell_lock(cell_id): + fresh = _read_history_backed(cell_id) # another thread may have just refreshed it if fresh is not None: df, age_s = fresh if age_s < HISTORY_TOPUP_INTERVAL: @@ -518,8 +599,7 @@ def _topup_tail(cell: dict, df: pl.DataFrame, path: str) -> pl.DataFrame: 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 + _touch_history_backed(cell_id) # tail already current — reset the hourly timer return df try: recent = _fetch_history_range( @@ -531,10 +611,13 @@ def _topup_tail(cell: dict, df: pl.DataFrame, path: str) -> pl.DataFrame: # 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") + tail = recent.drop("doy", strict=False) + merged = (pl.concat([df, tail], how="diagonal_relaxed") .unique(subset="date", keep="last", maintain_order=True) .sort("date")) - _write_cache(merged, path) + # Postgres only needs the new/refetched tail rows upserted (ON CONFLICT gives + # the same keep="last" precedence); parquet rewrites the whole file. + _write_history_backed(cell_id, merged, delta_df=tail) return merged @@ -546,11 +629,11 @@ def get_history(cell: dict) -> tuple[pl.DataFrame, dict]: 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"])) + """History from the 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_backed(cell["id"]) if hit is None: return None return _derive_metrics(_with_doy(hit[0])) @@ -562,24 +645,24 @@ def _load_history(cell: dict) -> tuple[pl.DataFrame, dict]: 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"]) + cell_id = cell["id"] - hit = _read_history_cache(path) + hit = _read_history_backed(cell_id) 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 = _topup_tail(cell, df) # 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 + with _cell_lock(cell_id): + hit = _read_history_backed(cell_id) # 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 + stale = _stale_history_backed(cell_id) def _serve_stale(): return _with_doy(stale), {"cached": True, "stale": True, "cache_age_days": None} @@ -615,7 +698,7 @@ def _load_history(cell: dict) -> tuple[pl.DataFrame, dict]: daily=_archive_limit_daily) # Store the raw record; percentiles are derived at request time. - _write_cache(df, path) + _write_history_backed(cell_id, df) return df, {"cached": False, "cache_age_days": 0, "source": source} @@ -628,9 +711,13 @@ def _rf_cache_path(cell_id: str) -> str: 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.""" + """Identity stamp (whole seconds) of the cell's cached recent+forecast bundle; + 0 when absent. On Postgres it's int(recent_synced_at) from climate_sync; on + parquet it's the file mtime. Either way it changes exactly when the + recent/forecast data is rewritten (and NOT on a stale serve), so it's the + freshness token for every payload that grades recent or future days.""" + if climate_store.is_postgres(): + return int(climate_store.recent_synced_at(cell_id)) try: return int(os.path.getmtime(_rf_cache_path(cell_id))) except OSError: @@ -644,17 +731,17 @@ def get_recent_forecast(cell: dict) -> pl.DataFrame: def load_cached_recent_forecast(cell: dict) -> "pl.DataFrame | None": - """Recent+forecast from the parquet cache ONLY — never fetches upstream, and - unlike the loader below it does not care how stale the file is. The sibling of + """Recent+forecast from the cache ONLY — never fetches upstream, and unlike the + loader below it does not care how stale the record is. The sibling of load_cached_history, for the same reason: the homepage precompute sweeps every cached city and must not spend a single upstream request doing it. None when the cell has no cached recent/forecast record.""" - path = _rf_cache_path(cell["id"]) - if not os.path.exists(path): + hit = _read_recent_backed(cell["id"]) + if hit is None: return None try: - return _derive_metrics(_with_doy(_normalize_read(pl.read_parquet(path)))) - except Exception: # noqa: BLE001 - a truncated/corrupt cache file is a miss, not a crash + return _derive_metrics(_with_doy(hit[0])) + except Exception: # noqa: BLE001 - a truncated/corrupt cache record is a miss, not a crash return None @@ -665,11 +752,12 @@ def _load_recent_forecast(cell: dict) -> pl.DataFrame: 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))) + 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) params = { "latitude": cell["center_lat"], @@ -696,10 +784,10 @@ def _load_recent_forecast(cell: dict) -> pl.DataFrame: # Last resort: serve the stale cache if we have one. An hours-old bundle # (which still carries the recent observed days MET Norway lacks) beats a # hard failure, mirroring _load_history's stale-serve. Return WITHOUT - # rewriting it, so its mtime stays old and the next request still retries - # upstream first rather than serving this as if it were fresh. - if os.path.exists(path): - return _with_doy(_normalize_read(pl.read_parquet(path))) + # rewriting it, so recent_stamp stays old and the next request still + # retries upstream first rather than serving this as if it were fresh. + if hit is not None: + return _with_doy(hit[0]) # No cache either: surface the original error, classifying an Open-Meteo # rate limit as the typed, daily-aware WeatherUnavailable (the archive # path classifies its own inside _load_history). @@ -707,7 +795,7 @@ def _load_recent_forecast(cell: dict) -> pl.DataFrame: daily = "daily" in _rate_limit_reason(e).lower() raise WeatherUnavailable(limit_message(daily), daily=daily) from e raise - _write_cache(df, path) + _write_recent_backed(cell_id, df) return df diff --git a/data/climate_store.py b/data/climate_store.py new file mode 100644 index 0000000..d54bdcd --- /dev/null +++ b/data/climate_store.py @@ -0,0 +1,289 @@ +"""Raw climate record store — TimescaleDB hypertables (Postgres) behind polars. + +The per-cell daily record (the multi-decade archive history and the hourly +recent+forecast bundle) is the *source of truth* for grading. On Postgres it +lives in two tables managed by Alembic (``backend/alembic/versions``): + +* ``climate_history`` — a TimescaleDB **hypertable** (durable/LOGGED) of the full + daily archive, keyed ``(cell_id, date)``. Expensive to refetch (45 years, + rate-limited upstream), so it is durable, not a throwaway cache. +* ``climate_recent`` — a plain table holding the recent-observations + forward + forecast bundle (includes *future* dates), fully rewritten each refresh. +* ``climate_sync`` — per-cell freshness (epoch seconds), replacing the parquet + file mtimes that used to drive the topup cadence, the forecast TTL, and the + ``recent_stamp`` token embedded in derived-payload validity (see web/views.py). + +This module is the Postgres backend only. ``climate.py`` keeps the parquet cache +as the backend when ``THERMOGRAPH_DATABASE_URL`` is *not* a Postgres URL (dev, +tests, offline tooling) — the dialect switch mirrors ``data/store.py`` and +``accounts/db.py`` exactly, which is what keeps the test suite Postgres-free. + +Every helper is **fail-soft**: a connection or query that fails is swallowed and +returns "absent" (``None`` / ``0.0``). Unlike ``store.py`` (which degrades to +recompute-from-parquet), a miss here degrades to *fetch-from-upstream* — prod has +no parquet fallback, and ``climate.py``'s loaders already treat "no cache" as +"go fetch". +""" +import os +import threading + +import polars as pl + +# Bump when a new persisted metric column is added to the daily record. A cell +# whose stored rows predate the bump reads back as schema-incomplete (see +# read_history), so climate.py refetches to fill the new column — the Postgres +# analogue of the parquet NEW_COLS presence check. +COLS_VERSION = 1 + +# The persisted daily columns, in a stable order. ``humid`` is the RAW relative +# humidity the archive returns; absolute humidity + wet-bulb are derived at the +# read boundary in climate._derive_metrics, never stored (so the store matches +# what the parquet write path persisted). ``feels`` IS stored. ``doy`` is not — +# climate._with_doy recomputes it at read time. +COLS = ("date", "tmax", "tmin", "precip", "wind", "gust", "humid", "fmax", "fmin", "feels") + +# Explicit polars schema so a frame read back from Postgres has the exact dtypes +# the parquet path produced (pl.Date + Float64), even when the result is empty — +# so grading.py / scoring.py / web/views.py are untouched by the backend switch. +_SCHEMA: dict[str, pl.DataType] = { + "date": pl.Date, + **{c: pl.Float64 for c in COLS if c != "date"}, +} + +DATABASE_URL = os.environ.get("THERMOGRAPH_DATABASE_URL", "").strip() +IS_POSTGRES = DATABASE_URL.startswith("postgresql") + + +def is_postgres() -> bool: + """True when the climate record is served from Postgres (prod), False when it + falls back to the parquet cache (dev / tests / offline).""" + return IS_POSTGRES + + +def _libpq_dsn(url: str) -> str: + """A plain libpq URL psycopg accepts: drop the SQLAlchemy driver suffix + (``postgresql+asyncpg://`` / ``+psycopg://`` -> ``postgresql://``).""" + return url.replace("+asyncpg", "").replace("+psycopg", "") + + +_PG_DSN = _libpq_dsn(DATABASE_URL) if IS_POSTGRES else "" + +# Per-thread connection: uvicorn serves on a thread pool and psycopg connections +# aren't thread-safe, so each thread lazily opens its own (mirrors store.py). +_local = threading.local() + + +def _conn(): + """Thread-local autocommit psycopg connection, or None when Postgres is off or + unreachable. A dropped connection is retired so the next call reconnects.""" + if not IS_POSTGRES: + return None + conn = getattr(_local, "conn", None) + if conn is not None: + if getattr(conn, "closed", False): + _local.conn = None + else: + return conn + try: + import psycopg # local import: only the Postgres path needs it + conn = psycopg.connect(_PG_DSN, autocommit=True) + _local.conn = conn + return conn + except Exception: # noqa: BLE001 - the store is optional; callers fetch on miss + return None + + +# --- freshness (climate_sync) ----------------------------------------------- + +def _get_sync(conn, cell_id): + """(history_synced_at, recent_synced_at, cols_version) for a cell, or None.""" + row = conn.execute( + "SELECT history_synced_at, recent_synced_at, cols_version " + "FROM climate_sync WHERE cell_id = %s", + (cell_id,), + ).fetchone() + return row + + +def _bump_sync(conn, cell_id, *, history=None, recent=None, cols_version=None) -> None: + """Upsert one cell's freshness. A None field leaves the stored value unchanged + on update (COALESCE), so a recent-bundle write never disturbs history_synced_at + and vice-versa. On insert, cols_version defaults to 1.""" + conn.execute( + """ + INSERT INTO climate_sync (cell_id, history_synced_at, recent_synced_at, cols_version) + VALUES (%(c)s, %(h)s, %(r)s, COALESCE(%(v)s, 1)) + ON CONFLICT (cell_id) DO UPDATE SET + history_synced_at = COALESCE(%(h)s, climate_sync.history_synced_at), + recent_synced_at = COALESCE(%(r)s, climate_sync.recent_synced_at), + cols_version = COALESCE(%(v)s, climate_sync.cols_version) + """, + {"c": cell_id, "h": history, "r": recent, "v": cols_version}, + ) + + +def history_synced_at(cell_id: str) -> float: + """Epoch seconds of the cell's last history write/topup; 0.0 if absent.""" + conn = _conn() + if conn is None: + return 0.0 + try: + row = _get_sync(conn, cell_id) + return float(row[0]) if row and row[0] is not None else 0.0 + except Exception: # noqa: BLE001 + return 0.0 + + +def recent_synced_at(cell_id: str) -> float: + """Epoch seconds of the cell's last recent-bundle write; 0.0 if absent. Backs + climate.recent_stamp, so it must advance only on a real rewrite (see write_recent).""" + conn = _conn() + if conn is None: + return 0.0 + try: + row = _get_sync(conn, cell_id) + return float(row[1]) if row and row[1] is not None else 0.0 + except Exception: # noqa: BLE001 + return 0.0 + + +def cols_version(cell_id: str) -> int: + """The stored schema version for a cell's history; 0 if absent.""" + conn = _conn() + if conn is None: + return 0 + try: + row = _get_sync(conn, cell_id) + return int(row[2]) if row and row[2] is not None else 0 + except Exception: # noqa: BLE001 + return 0 + + +# --- reads ------------------------------------------------------------------ + +def _read_frame(conn, table: str, cell_id: str) -> pl.DataFrame: + """A cell's rows from ``table`` as a polars frame with the canonical schema. + ``table`` is a module constant, never user input.""" + rows = conn.execute( + f"SELECT {', '.join(COLS)} FROM {table} WHERE cell_id = %s ORDER BY date", + (cell_id,), + ).fetchall() + return pl.DataFrame(rows, schema=_SCHEMA, orient="row") + + +def read_history(cell_id: str) -> "tuple[pl.DataFrame, float] | None": + """(history frame, history_synced_at) for a cell, or None when there is no + schema-complete cached record. A stored cols_version below COLS_VERSION reads + as absent, so climate.py refetches to add the new column(s).""" + conn = _conn() + if conn is None: + return None + try: + sync = _get_sync(conn, cell_id) + if sync is None or sync[0] is None: + return None + if (sync[2] or 0) < COLS_VERSION: + return None + df = _read_frame(conn, "climate_history", cell_id) + if df.is_empty(): + return None + return df, float(sync[0]) + except Exception: # noqa: BLE001 + return None + + +def read_history_raw(cell_id: str) -> "pl.DataFrame | None": + """A cell's history rows regardless of schema version — for the stale-serve + fallback when every upstream fetch fails (the Postgres analogue of reading the + parquet file directly, bypassing the NEW_COLS completeness check). None when + there are no rows.""" + conn = _conn() + if conn is None: + return None + try: + df = _read_frame(conn, "climate_history", cell_id) + return None if df.is_empty() else df + except Exception: # noqa: BLE001 + return None + + +def read_recent(cell_id: str) -> "tuple[pl.DataFrame, float] | None": + """(recent+forecast frame, recent_synced_at) for a cell, or None when absent.""" + conn = _conn() + if conn is None: + return None + try: + sync = _get_sync(conn, cell_id) + if sync is None or sync[1] is None: + return None + df = _read_frame(conn, "climate_recent", cell_id) + if df.is_empty(): + return None + return df, float(sync[1]) + except Exception: # noqa: BLE001 + return None + + +# --- writes ----------------------------------------------------------------- + +def _copy_rows(cur, table: str, cell_id: str, df: pl.DataFrame) -> None: + """COPY a cell's frame into ``table`` (text format). NaN was already folded to + null upstream in climate._finalize_frame, so nulls copy cleanly.""" + with cur.copy(f"COPY {table} (cell_id, {', '.join(COLS)}) FROM STDIN") as copy: + for row in df.select(COLS).iter_rows(): + copy.write_row((cell_id, *row)) + + +def write_history(cell_id: str, df: pl.DataFrame, synced_at: float) -> None: + """Upsert a cell's history rows and stamp history_synced_at. Idempotent: a + re-fetched day supersedes its stored duplicate (ON CONFLICT DO UPDATE), which + mirrors the parquet unique(subset="date", keep="last") merge. Best-effort.""" + conn = _conn() + if conn is None: + return + try: + with conn.transaction(): + conn.execute( + "CREATE TEMP TABLE _hist_stage (LIKE climate_history) ON COMMIT DROP") + with conn.cursor() as cur: + _copy_rows(cur, "_hist_stage", cell_id, df) + set_cols = ", ".join(f"{c} = EXCLUDED.{c}" for c in COLS if c != "date") + conn.execute( + f"INSERT INTO climate_history (cell_id, {', '.join(COLS)}) " + f"SELECT cell_id, {', '.join(COLS)} FROM _hist_stage " + f"ON CONFLICT (cell_id, date) DO UPDATE SET {set_cols}") + _bump_sync(conn, cell_id, history=synced_at, cols_version=COLS_VERSION) + except Exception: # noqa: BLE001 - failing to cache must not fail the request + pass + + +def touch_history(cell_id: str, synced_at: float) -> None: + """Bump only history_synced_at — the analogue of ``os.utime`` on the parquet + file when the archive tail is already current, so the hourly topup timer resets + without a rewrite. Best-effort.""" + conn = _conn() + if conn is None: + return + try: + _bump_sync(conn, cell_id, history=synced_at) + except Exception: # noqa: BLE001 + pass + + +def write_recent(cell_id: str, df: pl.DataFrame, synced_at: float) -> None: + """Replace a cell's recent+forecast bundle (delete + COPY in one transaction) + and stamp recent_synced_at. The window slides and forecast values change each + refresh, so a full replace is the simplest correct write. Call ONLY on a real + rewrite — recent_synced_at feeds recent_stamp, which must not advance when a + stale bundle is served without a refetch. Best-effort.""" + conn = _conn() + if conn is None: + return + try: + with conn.transaction(): + conn.execute("DELETE FROM climate_recent WHERE cell_id = %s", (cell_id,)) + with conn.cursor() as cur: + _copy_rows(cur, "climate_recent", cell_id, df) + _bump_sync(conn, cell_id, recent=synced_at) + except Exception: # noqa: BLE001 + pass diff --git a/migrate_cache_to_pg.py b/migrate_cache_to_pg.py new file mode 100644 index 0000000..393bb62 --- /dev/null +++ b/migrate_cache_to_pg.py @@ -0,0 +1,96 @@ +"""One-shot backfill of the parquet climate cache into the TimescaleDB tables. + +Run ONCE during the cutover, after ``alembic upgrade head`` has created the +``climate_history`` / ``climate_recent`` / ``climate_sync`` tables. It loads every +cached cell's parquet record into Postgres so a freshly-cut-over server serves +history/recent from the database immediately instead of refetching decades of +archive (which the upstream rate limit would throttle hard). + +Idempotent and resumable: a cell whose stored sync timestamp already matches its +parquet mtime is skipped, and history rows upsert (``ON CONFLICT``), so re-running +after an interruption only does the missing work. + +Key detail: rows are read RAW (``pl.read_parquet`` + ``_normalize_read``), NOT via +``climate.load_cached_*`` — the loaders apply ``_derive_metrics``, which replaces +raw relative humidity with absolute humidity. The DB must store the raw record the +live write path persists, so absolute humidity / wet-bulb stay read-time +derivations. The per-cell ``synced_at`` is set to the parquet FILE MTIME (not now), +so ``recent_stamp`` — hence every derived-payload validity token — is unchanged +across the cutover. + +Usage: + THERMOGRAPH_DATABASE_URL=postgresql+psycopg://thermograph:PW@127.0.0.1:5432/thermograph \\ + python migrate_cache_to_pg.py # reads data/cache/*.parquet +""" +import os +import sys + +import polars as pl + +from data import climate +from data import climate_store +from data import grid + + +def _is_schema_complete(df: pl.DataFrame) -> bool: + """The same NEW_COLS guard climate._read_history_cache applies: a record missing + any post-original column predates the wind/humidity features and is skipped so + the server refetches it lazily (with the current schema) on first touch.""" + return all(c in df.columns for c in climate.NEW_COLS) + + +def migrate() -> int: + if not climate_store.is_postgres(): + sys.exit("THERMOGRAPH_DATABASE_URL must be a postgresql URL (the migration target)") + if not os.path.isdir(climate.CACHE_DIR): + print("No parquet cache directory yet — nothing to migrate.") + return 0 + + hist = recent = current = skipped = 0 + for fname in sorted(os.listdir(climate.CACHE_DIR)): + if not fname.endswith(".parquet"): + continue + path = os.path.join(climate.CACHE_DIR, fname) + is_recent = fname.endswith("_rf.parquet") + cell_id = fname[:-len("_rf.parquet")] if is_recent else fname[:-len(".parquet")] + try: + grid.from_id(cell_id) # validate the id is a real cell (skip stray files) + except ValueError: + print(f" {fname}: unrecognized cache filename — skipped") + skipped += 1 + continue + + mtime = os.path.getmtime(path) + synced = (climate_store.recent_synced_at(cell_id) if is_recent + else climate_store.history_synced_at(cell_id)) + if synced >= mtime: + current += 1 + continue + + try: + df = climate._normalize_read(pl.read_parquet(path)) + except Exception as e: # noqa: BLE001 - a truncated/corrupt file is a skip, not a crash + print(f" {fname}: unreadable ({e}) — skipped") + skipped += 1 + continue + + if is_recent: + climate_store.write_recent(cell_id, df, mtime) + recent += 1 + print(f" {cell_id}: recent bundle ({df.height} rows)") + else: + if not _is_schema_complete(df): + print(f" {cell_id}: pre-schema history — skipped (refetches lazily)") + skipped += 1 + continue + climate_store.write_history(cell_id, df, mtime) + hist += 1 + print(f" {cell_id}: history ({df.height} rows)") + + print(f"{hist} history + {recent} recent loaded, {current} already current, " + f"{skipped} skipped.") + return 0 + + +if __name__ == "__main__": + sys.exit(migrate()) diff --git a/tests/data/test_climate_store.py b/tests/data/test_climate_store.py new file mode 100644 index 0000000..d3edcc1 --- /dev/null +++ b/tests/data/test_climate_store.py @@ -0,0 +1,152 @@ +"""The climate cache backend switch (data/climate_store.py + climate.py dispatch). + +Two layers: + +* **Hermetic** (always run): with no THERMOGRAPH_DATABASE_URL the store falls back + to the parquet cache, so ``is_postgres()`` is False, ``recent_stamp`` reads the + file mtime, and the backend helpers round-trip through parquet. This is the path + CI exercises — no Postgres required. +* **Gated** (skipped unless ``THERMOGRAPH_TEST_DATABASE_URL`` points at a reachable + Postgres): round-trips history/recent frames through the real psycopg + COPY + bridge and asserts dtype fidelity, upsert keep-last, and the recent_stamp token + semantics. +""" +import datetime +import os + +import polars as pl +import pytest + +from data import climate +from data import climate_store + + +def _hist_frame(n=5, tmax0=70.0): + """A raw daily frame with the persisted columns (no doy — the write path drops it).""" + d0 = datetime.date(1995, 1, 1) + cols = {"date": [d0 + datetime.timedelta(days=i) for i in range(n)]} + for c in ("tmax", "tmin", "precip", "wind", "gust", "humid", "fmax", "fmin", "feels"): + cols[c] = [float(tmax0 + i) for i in range(n)] + return pl.DataFrame(cols) + + +# --- hermetic (parquet fallback) -------------------------------------------- + +def test_selects_parquet_when_no_database_url(): + # The test env sets no THERMOGRAPH_DATABASE_URL (see conftest), so the climate + # record is served from parquet, not Postgres. + assert climate_store.is_postgres() is False + + +def test_recent_stamp_reads_mtime_on_parquet(monkeypatch, tmp_path): + monkeypatch.setattr(climate, "CACHE_DIR", str(tmp_path)) + cell_id = "12_-34" + assert climate.recent_stamp(cell_id) == 0 # absent -> 0 + climate._write_recent_backed(cell_id, _hist_frame()) + stamp = climate.recent_stamp(cell_id) + assert stamp == int(os.path.getmtime(climate._rf_cache_path(cell_id))) + assert stamp > 0 + + +def test_history_backend_roundtrips_through_parquet(monkeypatch, tmp_path): + monkeypatch.setattr(climate, "CACHE_DIR", str(tmp_path)) + cell_id = "5_6" + assert climate._read_history_backed(cell_id) is None # nothing cached yet + climate._write_history_backed(cell_id, _hist_frame(n=4)) + hit = climate._read_history_backed(cell_id) + assert hit is not None + df, age_s = hit + assert df.height == 4 and "doy" not in df.columns + assert age_s >= 0.0 + + +def test_recent_backend_roundtrips_through_parquet(monkeypatch, tmp_path): + monkeypatch.setattr(climate, "CACHE_DIR", str(tmp_path)) + cell_id = "7_8" + assert climate._read_recent_backed(cell_id) is None + climate._write_recent_backed(cell_id, _hist_frame(n=3)) + hit = climate._read_recent_backed(cell_id) + assert hit is not None and hit[0].height == 3 + + +# --- gated Postgres integration --------------------------------------------- + +@pytest.fixture +def pg_store(): + """Point climate_store at THERMOGRAPH_TEST_DATABASE_URL for one test, creating + the tables (plain — the bridge logic doesn't need the hypertable) and wiping + them. Restores the module's dialect state afterwards.""" + url = os.environ.get("THERMOGRAPH_TEST_DATABASE_URL", "").strip() + if not url: + pytest.skip("set THERMOGRAPH_TEST_DATABASE_URL to run the Postgres bridge test") + import psycopg + + saved = (climate_store.IS_POSTGRES, climate_store._PG_DSN) + climate_store.IS_POSTGRES = True + climate_store._PG_DSN = climate_store._libpq_dsn(url) + if hasattr(climate_store._local, "conn"): + del climate_store._local.conn + + ddl_cols = ", ".join(f"{c} DOUBLE PRECISION" for c in climate_store.COLS if c != "date") + with psycopg.connect(climate_store._PG_DSN, autocommit=True) as conn: + for tbl in ("climate_history", "climate_recent"): + conn.execute(f"CREATE TABLE IF NOT EXISTS {tbl} " + f"(cell_id TEXT NOT NULL, date DATE NOT NULL, {ddl_cols}, " + f"PRIMARY KEY (cell_id, date))") + conn.execute("CREATE TABLE IF NOT EXISTS climate_sync " + "(cell_id TEXT PRIMARY KEY, history_synced_at DOUBLE PRECISION, " + "recent_synced_at DOUBLE PRECISION, cols_version SMALLINT NOT NULL DEFAULT 1)") + conn.execute("TRUNCATE climate_history, climate_recent, climate_sync") + + yield climate_store + + climate_store.IS_POSTGRES, climate_store._PG_DSN = saved + if hasattr(climate_store._local, "conn"): + del climate_store._local.conn + + +def test_pg_history_roundtrip_and_dtypes(pg_store): + pg_store.write_history("1_2", _hist_frame(n=5, tmax0=70.0), 1000.0) + hit = pg_store.read_history("1_2") + assert hit is not None + df, synced = hit + assert synced == 1000.0 + assert df.schema["date"] == pl.Date and df.schema["tmax"] == pl.Float64 + assert "doy" not in df.columns + assert df.height == 5 and df["tmax"].to_list() == [70, 71, 72, 73, 74] + + +def test_pg_history_upsert_keeps_last(pg_store): + pg_store.write_history("1_2", _hist_frame(n=3, tmax0=10.0), 1000.0) + # Re-write day 2 with a new value + append a 4th day. + d0 = datetime.date(1995, 1, 1) + tail = pl.DataFrame({ + "date": [d0 + datetime.timedelta(days=1), d0 + datetime.timedelta(days=3)], + **{c: [999.0, 40.0] for c in + ("tmax", "tmin", "precip", "wind", "gust", "humid", "fmax", "fmin", "feels")}, + }) + pg_store.write_history("1_2", tail, 2000.0) + df, synced = pg_store.read_history("1_2") + assert synced == 2000.0 and df.height == 4 + assert df.filter(pl.col("date") == d0 + datetime.timedelta(days=1))["tmax"].item() == 999.0 + + +def test_pg_recent_stamp_advances_only_on_write(pg_store): + assert pg_store.recent_synced_at("1_2") == 0.0 + pg_store.write_recent("1_2", _hist_frame(n=4), 5000.0) + assert pg_store.recent_synced_at("1_2") == 5000.0 + # A history write must not disturb the recent stamp, and vice-versa. + pg_store.write_history("1_2", _hist_frame(n=2), 6000.0) + assert pg_store.recent_synced_at("1_2") == 5000.0 + assert pg_store.history_synced_at("1_2") == 6000.0 + + +def test_pg_cols_version_gate(pg_store): + pg_store.write_history("1_2", _hist_frame(n=2), 1000.0) + assert pg_store.read_history("1_2") is not None + pg_store.COLS_VERSION = pg_store.COLS_VERSION + 1 + try: + assert pg_store.read_history("1_2") is None # stale schema -> refetch + assert pg_store.read_history_raw("1_2") is not None # raw ignores the gate + finally: + pg_store.COLS_VERSION -= 1