Spaces:
Running
Running
| """ | |
| Weather history and forecasts from Open-Meteo, at state centres (Phase 1's | |
| native level — LGA-level history is the same code with level='lga', paced over | |
| more days of quota). | |
| Quota is the real constraint, not code: Open-Meteo counts a request covering | |
| more than 2 weeks or more than 10 variables as several calls (weight ≈ days/14 × | |
| max(1, vars/10)), and the free tier allows 600/min, 5,000/hour, 10,000/day. 20 | |
| years × 37 states is ~19k weighted calls, so history is a *resumable backfill*: | |
| each run fetches whole missing years, oldest first, until it hits its budget, | |
| and the next run carries on where it stopped. A 429 ends the run cleanly as a | |
| partial success rather than an error. | |
| The free tier is for non-commercial use — fine for local Phase 1 development; a | |
| paid plan (or ECMWF open data direct) is needed before this runs for a live | |
| product (design doc, "Conflicts"). | |
| """ | |
| import math | |
| import sqlite3 | |
| import time | |
| from datetime import date, timedelta | |
| from typing import Callable | |
| import httpx | |
| from intel.db import now_iso | |
| from intel.freshness import record_attempt | |
| from intel.handoff import DATA_STREAM, INTERFACE, produce | |
| from intel.registry import WEATHER_VARIABLES | |
| ARCHIVE_URL = "https://archive-api.open-meteo.com/v1/archive" | |
| FORECAST_URL = "https://api.open-meteo.com/v1/forecast" | |
| VARIABLES = [v for v, _ in WEATHER_VARIABLES] | |
| DEFAULT_HISTORY_START = date(2006, 1, 1) | |
| # The reanalysis archive trails real time by about 5 days. | |
| ARCHIVE_LAG_DAYS = 6 | |
| PER_MINUTE_BUDGET = 500 # below Open-Meteo's 600/min, leaving headroom | |
| DEFAULT_RUN_BUDGET = 4500 # below the 5,000/hour limit | |
| Fetch = Callable[[str, dict], dict] | |
| class QuotaExhausted(Exception): | |
| pass | |
| def call_weight(days: int, n_vars: int = len(VARIABLES)) -> float: | |
| return max(1.0, days / 14) * max(1.0, n_vars / 10) | |
| def http_fetch(url: str, params: dict) -> dict: | |
| r = httpx.get(url, params=params, timeout=60.0) | |
| if r.status_code == 429: | |
| raise QuotaExhausted(r.text[:200]) | |
| r.raise_for_status() | |
| return r.json() | |
| def _points(con: sqlite3.Connection, level: str) -> list[sqlite3.Row]: | |
| return con.execute("select id, lat, lon from locations where level = ? order by id", (level,)).fetchall() | |
| def _store(con: sqlite3.Connection, location_id: str, payload: dict, is_forecast: bool, issued_at: str) -> int: | |
| daily = payload["daily"] | |
| now = now_iso() | |
| rows = [] | |
| for var in VARIABLES: | |
| for day, value in zip(daily["time"], daily.get(var, [])): | |
| if value is not None: | |
| rows.append((f"weather.{var}", location_id, day, int(is_forecast), issued_at, float(value), now)) | |
| con.executemany( | |
| """insert into observations (stream_id, location_id, period_start, is_forecast, issued_at, value, ingested_at) | |
| values (?, ?, ?, ?, ?, ?, ?) | |
| on conflict (stream_id, location_id, period_start, is_forecast, issued_at) | |
| do update set value = excluded.value, ingested_at = excluded.ingested_at""", | |
| rows, | |
| ) | |
| con.commit() | |
| return len(rows) | |
| def _missing_years(con: sqlite3.Connection, location_id: str, start: date, end: date) -> list[tuple[date, date]]: | |
| """Calendar-year spans (clipped to [start, end]) still missing rows for the | |
| first variable. A year that's complete from its first day up to some date | |
| (the current year, typically) only resumes from the day after; any other | |
| gap refetches the whole year — simpler than tracking holes, and rare.""" | |
| have = {r[0]: (r[1], r[2]) for r in con.execute( | |
| """select substr(period_start, 1, 4) y, count(*), max(period_start) from observations | |
| where stream_id = ? and location_id = ? and is_forecast = 0 group by y""", | |
| (f"weather.{VARIABLES[0]}", location_id), | |
| )} | |
| spans = [] | |
| for y in range(start.year, end.year + 1): | |
| a, b = max(start, date(y, 1, 1)), min(end, date(y, 12, 31)) | |
| count, last = have.get(str(y), (0, None)) | |
| if count >= (b - a).days + 1: | |
| continue | |
| if last and count == (date.fromisoformat(last) - a).days + 1: | |
| a = date.fromisoformat(last) + timedelta(days=1) | |
| spans.append((a, b)) | |
| return spans | |
| def backfill_history(con: sqlite3.Connection, level: str = "state", start: date = DEFAULT_HISTORY_START, | |
| budget: float = DEFAULT_RUN_BUDGET, fetch: Fetch = http_fetch, | |
| sleep: Callable[[float], None] = time.sleep, run_id: str = "manual") -> dict: | |
| end = date.today() - timedelta(days=ARCHIVE_LAG_DAYS) | |
| todo = [(p, a, b) for p in _points(con, level) for a, b in _missing_years(con, p["id"], start, end)] | |
| todo.sort(key=lambda t: (t[1], t[0]["id"])) # oldest years first, across all points | |
| spent = minute_spent = 0.0 | |
| minute_started = time.monotonic() | |
| fetched = rows = 0 | |
| stopped = None | |
| try: | |
| for p, a, b in todo: | |
| w = call_weight((b - a).days + 1) | |
| if spent + w > budget: | |
| stopped = "run budget reached" | |
| break | |
| if minute_spent + w > PER_MINUTE_BUDGET: | |
| sleep(max(0.0, 60 - (time.monotonic() - minute_started))) | |
| minute_spent, minute_started = 0.0, time.monotonic() | |
| payload = fetch(ARCHIVE_URL, { | |
| "latitude": p["lat"], "longitude": p["lon"], "start_date": a.isoformat(), "end_date": b.isoformat(), | |
| "daily": ",".join(VARIABLES), "timezone": "Africa/Lagos", | |
| }) | |
| rows += _store(con, p["id"], payload, is_forecast=False, issued_at="") | |
| spent += w | |
| minute_spent += w | |
| fetched += 1 | |
| except QuotaExhausted as e: | |
| stopped = f"Open-Meteo quota: {e}" | |
| except Exception as e: | |
| record_attempt(con, "open_meteo_archive", ok=False, error=str(e)) | |
| raise | |
| newest = con.execute( | |
| "select max(period_start) from observations where stream_id like 'weather.%' and is_forecast = 0" | |
| ).fetchone()[0] | |
| remaining = len(todo) - fetched | |
| if fetched or not todo: | |
| record_attempt(con, "open_meteo_archive", ok=True, latest_data_period=newest) | |
| else: | |
| record_attempt(con, "open_meteo_archive", ok=False, error=stopped) | |
| summary = {"level": level, "requests": fetched, "weighted_calls": round(spent, 1), "rows": rows, | |
| "year_chunks_remaining": remaining, "stopped": stopped, "newest_day": newest} | |
| if fetched: | |
| produce(con, run_id, DATA_STREAM, INTERFACE, {"kind": "weather_history", **summary}) | |
| return summary | |
| def fetch_forecast(con: sqlite3.Connection, level: str = "state", days: int = 16, | |
| fetch: Fetch = http_fetch, run_id: str = "manual") -> dict: | |
| issued = date.today().isoformat() | |
| rows = n = 0 | |
| try: | |
| for p in _points(con, level): | |
| payload = fetch(FORECAST_URL, { | |
| "latitude": p["lat"], "longitude": p["lon"], "forecast_days": days, | |
| "daily": ",".join(VARIABLES), "timezone": "Africa/Lagos", | |
| }) | |
| rows += _store(con, p["id"], payload, is_forecast=True, issued_at=issued) | |
| n += 1 | |
| except Exception as e: | |
| record_attempt(con, "open_meteo_forecast", ok=False, error=str(e)) | |
| raise | |
| record_attempt(con, "open_meteo_forecast", ok=True, latest_data_period=issued) | |
| summary = {"level": level, "points": n, "rows": rows, "issued_at": issued, | |
| "weighted_calls": round(n * call_weight(days), 1)} | |
| produce(con, run_id, DATA_STREAM, INTERFACE, {"kind": "weather_forecast", **summary}) | |
| return summary | |
| def estimate_backfill_calls(con: sqlite3.Connection, level: str = "state", start: date = DEFAULT_HISTORY_START) -> float: | |
| end = date.today() - timedelta(days=ARCHIVE_LAG_DAYS) | |
| return math.fsum(call_weight((b - a).days + 1) | |
| for p in _points(con, level) for a, b in _missing_years(con, p["id"], start, end)) | |