thermograph/migrate.py

"""Backfill the SQLite derived store from the existing parquet cell caches.

For every cell with a cached archive record, materialize the payloads that are
derivable offline — the default 2-year calendar and the latest-day detail — plus
its reverse-geocode label, so a server restarted (or freshly deployed) onto an
existing parquet cache serves those views from the database immediately instead
of regrading each cell on first touch.

Idempotent and resumable: cells whose derived rows already match the current
validity token are skipped, so re-running after an interruption (or after new
archive days arrive) only does the missing work. Never fetches weather data —
history comes strictly from the parquet cache (cells without one are skipped and
will materialize lazily on first request, as always). Reverse geocoding makes at
most one Nominatim call per unlabeled cell, throttled to ~1/s per the service's
policy. Safe to run alongside a live server: writes go through the same WAL
store the server reads, and readers fall back to recomputing on any miss.

Run via `make migrate`.
"""
import os
import sys
import time

import pandas as pd

import climate
import grid
import store
from app import PAYLOAD_VER, _build_calendar, _build_day, _cal_span, _hist_end


def migrate() -> int:
    if not os.path.isdir(climate.CACHE_DIR):
        print("No parquet cache directory yet — nothing to migrate.")
        return 0
    cells = sorted(
        f[: -len(".parquet")]
        for f in os.listdir(climate.CACHE_DIR)
        if f.endswith(".parquet") and not f.endswith("_rf.parquet")
    )
    built = current = skipped = 0
    for cell_id in cells:
        try:
            cell = grid.from_id(cell_id)
        except ValueError:
            print(f"  {cell_id}: unrecognized cache filename — skipped")
            skipped += 1
            continue
        history = climate.load_cached_history(cell)
        if history is None or history.empty:
            # Pre-current-schema record; the server refetches it lazily on first
            # touch (see climate.NEW_COLS) — nothing to materialize offline.
            print(f"  {cell_id}: no schema-complete record — skipped")
            skipped += 1
            continue

        token = f"{PAYLOAD_VER}:{_hist_end(history)}"
        start_ts, end_ts = _cal_span(history, None, None, 24)
        cal_key = f"{start_ts.date().isoformat()}:{end_ts.date().isoformat()}:24"
        last = pd.Timestamp(history["date"].max()).normalize()
        day_key = last.date().isoformat()

        have_cal = store.get_payload("calendar", cell_id, cal_key, token) is not None
        have_day = store.get_payload("day", cell_id, day_key, token) is not None
        if have_cal and have_day:
            current += 1
            continue

        found, place = climate.reverse_geocode_cached(cell["center_lat"], cell["center_lon"])
        if not found:
            place = climate.reverse_geocode(cell["center_lat"], cell["center_lon"])
            time.sleep(1.1)   # Nominatim usage policy: at most ~1 request/second

        if not have_cal:
            store.put_payload("calendar", cell_id, cal_key, token,
                              _build_calendar(cell, history, start_ts, end_ts, 24, place))
        if not have_day:
            store.put_payload("day", cell_id, day_key, token,
                              _build_day(cell, history, last, place))
        built += 1
        print(f"  {cell_id}: materialized ({place or 'no label'})")

    s = store.stats()
    print(f"{built} cell(s) materialized, {current} already current, {skipped} skipped"
          f" · derived rows: {s['derived']} · revgeo: {s['revgeo']}"
          f" · db: {s['bytes'] / 1e6:.1f} MB → {s['db_path']}")
    return 0


if __name__ == "__main__":
    sys.exit(migrate())