power-compare/scripts/backfill_isolarcloud.py

"""One-off historical backfill for a real iSolarCloud account (oauth2 mode).

Pulls day-by-day from --since (default: the plant's install date) up to
now, via the same `fetch_interval_data` the scheduler uses, and writes
straight into the `interval_reading` table.

Safe to interrupt and re-run: progress is checkpointed in `sync_state`
after each day, and re-inserting a day already covered is a no-op (rows
dedupe on (ts, source)).

The account's hourly API quota (2000 calls/hour on the Free plan) will be
hit partway through a multi-year backfill -- each day costs ~8 calls (one
per 3-hour chunk the historical endpoint allows). Rather than requiring a
manual re-run every time that happens, every API call in this script
(including the one-off plant-list lookup, not just the main loop) goes
through `call_with_retry`, which sleeps past the reset and retries on the
hour limit -- a full ~2.5 year backfill takes several such hourly cycles,
so expect this to run for multiple hours total, not minutes.

Usage:
  python scripts/backfill_isolarcloud.py                  # since plant install date
  python scripts/backfill_isolarcloud.py --since 2024-01-01
  python scripts/backfill_isolarcloud.py --days 365
"""

from __future__ import annotations

import argparse
import time
from datetime import datetime, timedelta, timezone
from typing import Callable, TypeVar

import httpx

from app.config import settings
from app.db.repositories import get_household_config, get_sync_state, set_sync_state, upsert_interval_readings
from app.db.session import get_session, init_db

PROGRESS_KEY = "isolarcloud_backfill_progress_date"
HOUR_LIMIT_WAIT_SECONDS = 65 * 60  # a bit over an hour, to be safe about the reset boundary
NETWORK_RETRY_WAIT_SECONDS = 30  # transient timeouts/connection drops -- worth a short retry, not a full hour

T = TypeVar("T")


def log(msg: str) -> None:
    print(msg, flush=True)


def call_with_retry(fn: Callable[[], T]) -> T:
    """Runs an iSolarCloud API call, sleeping past the hourly quota reset
    and retrying if it's hit, or backing off briefly on a transient network
    error; aborts (checkpoint intact) on the monthly quota."""
    from app.isolarcloud.oauth_client import ISolarCloudAuthError

    while True:
        try:
            return fn()
        except ISolarCloudAuthError as e:
            msg = str(e)
            if "E999" in msg:
                log(f"Hourly API limit hit -- sleeping {HOUR_LIMIT_WAIT_SECONDS // 60} min before retrying...")
                time.sleep(HOUR_LIMIT_WAIT_SECONDS)
                continue
            if "E998" in msg:
                raise SystemExit(
                    f"Monthly API limit reached: {msg}\n"
                    f"Progress is checkpointed -- re-run this script next month (or after a plan upgrade) to continue."
                )
            raise
        except httpx.TransportError as e:
            log(f"Network error ({e!r}) -- sleeping {NETWORK_RETRY_WAIT_SECONDS}s before retrying...")
            time.sleep(NETWORK_RETRY_WAIT_SECONDS)
            continue
        except httpx.HTTPStatusError as e:
            if e.response.status_code >= 500:
                log(
                    f"Server error ({e.response.status_code}) -- sleeping "
                    f"{NETWORK_RETRY_WAIT_SECONDS}s before retrying..."
                )
                time.sleep(NETWORK_RETRY_WAIT_SECONDS)
                continue
            raise


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--since", help="YYYY-MM-DD to start from (default: plant install date)")
    parser.add_argument("--days", type=int, help="alternative to --since: number of days back from today")
    args = parser.parse_args()

    init_db()
    if settings.use_mock_isolarcloud or settings.isolarcloud_auth_mode != "oauth2":
        raise SystemExit("This script requires USE_MOCK_ISOLARCLOUD=false and ISOLARCLOUD_AUTH_MODE=oauth2 in .env.")

    from app.isolarcloud.oauth_client import ISolarCloudOAuthClient

    client = ISolarCloudOAuthClient()

    with get_session() as session:
        ps_id = get_household_config(session).isolarcloud_ps_id or settings.isolarcloud_plant_id

    # Only auto-resume the *default* (install-date) target -- an explicit
    # --since/--days is a deliberate choice of range and shouldn't be
    # silently overridden by a checkpoint left by some earlier, different
    # backfill run (e.g. a small smoke-test range).
    resume_date: datetime | None = None
    if not args.since and not args.days:
        with get_session() as session:
            resume_str = get_sync_state(session, PROGRESS_KEY)
        if resume_str:
            resume_date = datetime.fromisoformat(resume_str)

    if args.since:
        start_date = datetime.strptime(args.since, "%Y-%m-%d").replace(tzinfo=timezone.utc)
    elif args.days:
        start_date = datetime.now(timezone.utc) - timedelta(days=args.days)
    elif resume_date:
        # No need to look up the install date at all -- we already know where to continue from.
        start_date = resume_date
        log(f"Resuming from previous progress: {start_date.date()}")
    else:
        plants = call_with_retry(client.get_plant_list)
        if not ps_id:
            ps_id = str(plants[0]["ps_id"])
        plant = next((p for p in plants if str(p["ps_id"]) == str(ps_id)), plants[0])
        start_date = datetime.strptime(plant["install_date"], "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc)

    if not ps_id:
        plants = call_with_retry(client.get_plant_list)
        ps_id = str(plants[0]["ps_id"])

    now = datetime.now(timezone.utc)
    total_inserted = 0
    total_points = 0
    cursor = start_date
    day = timedelta(days=1)

    log(f"Backfilling {ps_id} from {start_date.date()} to {now.date()} ({(now - start_date).days} days)...")
    while cursor < now:
        chunk_end = min(cursor + day, now)
        data_points = call_with_retry(lambda: client.fetch_interval_data(cursor, chunk_end))
        rows = [p.as_row("isolarcloud") for p in data_points]
        with get_session() as session:
            inserted = upsert_interval_readings(session, rows)
            set_sync_state(session, PROGRESS_KEY, chunk_end.isoformat())
        total_inserted += inserted
        total_points += len(data_points)
        log(f"  {cursor.date()}: {len(data_points)} points, {inserted} new rows (running total: {total_inserted})")
        cursor = chunk_end
        time.sleep(0.2)  # be a polite citizen of the API's rate limit

    log(f"Done. {total_points} points fetched, {total_inserted} new rows inserted.")


if __name__ == "__main__":
    main()