Blame for intel/ingest/weather.py

"""
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))