89 lines
3.7 KiB
Python
89 lines
3.7 KiB
Python
"""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
|
|
|
|
from data import climate
|
|
from data import grid
|
|
from data import store
|
|
from api import payloads
|
|
|
|
|
|
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.is_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 = payloads.history_token(history)
|
|
start_ts, end_ts = payloads.cal_span(history, None, None, 24)
|
|
cal_key = payloads.calendar_key(start_ts, end_ts, 24)
|
|
last = history["date"].max()
|
|
day_key = payloads.day_key(last)
|
|
|
|
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,
|
|
payloads.build_calendar(cell, history, start_ts, end_ts, 24, place))
|
|
if not have_day:
|
|
store.put_payload("day", cell_id, day_key, token,
|
|
payloads.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())
|