commit ccd4e7a48c1907183d595dd85962de1ea828c77d Author: rpotter6298 Date: Sun Aug 30 14:19:04 2026 +0200 Initial commit: astrape-oracle split out of the combined Astrape tree The forecasting runtime gibil pilots: weather ingest (Open-Meteo forecasts and the Ecowitt station), the oracle and weather stores, the snapshot daemon, the forecast evaluator, and the TCN training scheduler. classes/datasets.py holds the shapes the stores return for display. They sit beside the stores deliberately: a store's return type belongs to the store, and keeping them with the renderers would mean this repo importing the dashboard, which is backwards. Extracted from a single repository where the forecasting engine, the database plumbing, the hardware client, the ingest daemons and the web UI all imported each other freely. Verified that every module imports with gibil resolved from its own installed checkout. Co-Authored-By: Claude Opus 5 (1M context) diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..0650fbc --- /dev/null +++ b/.gitignore @@ -0,0 +1,5 @@ +__pycache__/ +*.py[cod] +*.egg-info/ +build/ +dist/ diff --git a/astrape_oracle/__init__.py b/astrape_oracle/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/astrape_oracle/classes/__init__.py b/astrape_oracle/classes/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/astrape_oracle/classes/datasets.py b/astrape_oracle/classes/datasets.py new file mode 100644 index 0000000..7b3c598 --- /dev/null +++ b/astrape_oracle/classes/datasets.py @@ -0,0 +1,28 @@ +"""Shapes the stores return for display. + +These live beside the stores rather than beside the renderers on purpose. A +store's return type belongs to the store: putting them in the dashboard would +mean astrape-oracle importing astrape-dashboard, which is backwards -- the +dashboard reads from the oracle, never the other way round. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +from gibil.classes.models import ( + WeatherForecastPoint, + WeatherResolvedTruth, + WeatherStationObservation, +) + + +@dataclass(frozen=True) +class WeatherDisplayDataset: + forecast_points: list[WeatherForecastPoint] + resolved_truth: list[WeatherResolvedTruth] + + +@dataclass(frozen=True) +class WeatherStationDisplayDataset: + observations: list[WeatherStationObservation] diff --git a/astrape_oracle/classes/oracle_store.py b/astrape_oracle/classes/oracle_store.py new file mode 100644 index 0000000..680c814 --- /dev/null +++ b/astrape_oracle/classes/oracle_store.py @@ -0,0 +1,1031 @@ +from __future__ import annotations + +from contextlib import contextmanager +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from os import environ +from typing import Iterator + +from gibil.classes.models import ( + ForecastKind, + NetPowerForecastPoint, + NetPowerForecastRun, + PowerForecastPoint, + PowerForecastRun, +) + + +class OracleStoreConfigurationError(RuntimeError): + pass + + +@dataclass(frozen=True) +class OracleStoreConfig: + database_url: str + + @classmethod + def from_env(cls) -> "OracleStoreConfig": + database_url = environ.get("ASTRAPE_DATABASE_URL") + if not database_url: + raise OracleStoreConfigurationError( + "ASTRAPE_DATABASE_URL is required for oracle storage" + ) + return cls(database_url=database_url) + + +class OracleStore: + """Persists generated oracle projection curves for later evaluation.""" + + def __init__(self, config: OracleStoreConfig) -> None: + self.config = config + + @classmethod + def from_env(cls) -> "OracleStore": + return cls(OracleStoreConfig.from_env()) + + def initialize(self) -> None: + with self._connection() as connection: + with connection.cursor() as cursor: + cursor.execute("CREATE EXTENSION IF NOT EXISTS timescaledb") + cursor.execute( + """ + CREATE TABLE IF NOT EXISTS oracle_power_forecast_points ( + issued_at TIMESTAMPTZ NOT NULL, + target_at TIMESTAMPTZ NOT NULL, + kind TEXT NOT NULL, + source TEXT NOT NULL, + model_version TEXT NOT NULL, + horizon_minutes INTEGER NOT NULL, + expected_power_w DOUBLE PRECISION NOT NULL, + p10_power_w DOUBLE PRECISION NOT NULL, + p50_power_w DOUBLE PRECISION NOT NULL, + p90_power_w DOUBLE PRECISION NOT NULL, + confidence DOUBLE PRECISION NOT NULL, + inserted_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (issued_at, target_at, kind, source, model_version) + ) + """ + ) + cursor.execute( + """ + SELECT create_hypertable( + 'oracle_power_forecast_points', + 'target_at', + if_not_exists => TRUE + ) + """ + ) + cursor.execute( + """ + CREATE TABLE IF NOT EXISTS oracle_net_forecast_points ( + issued_at TIMESTAMPTZ NOT NULL, + target_at TIMESTAMPTZ NOT NULL, + source TEXT NOT NULL, + horizon_minutes INTEGER NOT NULL, + expected_net_power_w DOUBLE PRECISION NOT NULL, + safe_net_power_w DOUBLE PRECISION NOT NULL, + p10_net_power_w DOUBLE PRECISION, + p50_net_power_w DOUBLE PRECISION, + p90_net_power_w DOUBLE PRECISION, + solar_p50_power_w DOUBLE PRECISION NOT NULL, + load_p50_power_w DOUBLE PRECISION NOT NULL, + solar_p10_power_w DOUBLE PRECISION NOT NULL, + solar_p90_power_w DOUBLE PRECISION, + load_p10_power_w DOUBLE PRECISION, + load_p90_power_w DOUBLE PRECISION NOT NULL, + inserted_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (issued_at, target_at, source) + ) + """ + ) + cursor.execute( + """ + ALTER TABLE oracle_net_forecast_points + ADD COLUMN IF NOT EXISTS p10_net_power_w DOUBLE PRECISION + """ + ) + cursor.execute( + """ + ALTER TABLE oracle_net_forecast_points + ADD COLUMN IF NOT EXISTS p50_net_power_w DOUBLE PRECISION + """ + ) + cursor.execute( + """ + ALTER TABLE oracle_net_forecast_points + ADD COLUMN IF NOT EXISTS p90_net_power_w DOUBLE PRECISION + """ + ) + cursor.execute( + """ + ALTER TABLE oracle_net_forecast_points + ADD COLUMN IF NOT EXISTS solar_p90_power_w DOUBLE PRECISION + """ + ) + cursor.execute( + """ + ALTER TABLE oracle_net_forecast_points + ADD COLUMN IF NOT EXISTS load_p10_power_w DOUBLE PRECISION + """ + ) + cursor.execute( + """ + SELECT create_hypertable( + 'oracle_net_forecast_points', + 'target_at', + if_not_exists => TRUE + ) + """ + ) + cursor.execute( + """ + CREATE TABLE IF NOT EXISTS oracle_forecast_evaluations ( + issued_at TIMESTAMPTZ NOT NULL, + target_at TIMESTAMPTZ NOT NULL, + kind TEXT NOT NULL, + source TEXT NOT NULL, + model_version TEXT NOT NULL, + horizon_minutes INTEGER NOT NULL, + expected_power_w DOUBLE PRECISION NOT NULL, + p10_power_w DOUBLE PRECISION, + p50_power_w DOUBLE PRECISION, + p90_power_w DOUBLE PRECISION, + realized_power_w DOUBLE PRECISION, + error_w DOUBLE PRECISION, + absolute_error_w DOUBLE PRECISION, + absolute_pct_error DOUBLE PRECISION, + covered_by_p10_p90 BOOLEAN, + sample_count INTEGER NOT NULL DEFAULT 0, + evaluated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + inserted_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY ( + issued_at, + target_at, + kind, + source, + model_version + ) + ) + """ + ) + cursor.execute( + """ + SELECT create_hypertable( + 'oracle_forecast_evaluations', + 'target_at', + if_not_exists => TRUE + ) + """ + ) + cursor.execute( + """ + CREATE INDEX IF NOT EXISTS oracle_forecast_evaluations_kind_horizon_idx + ON oracle_forecast_evaluations ( + kind, + horizon_minutes, + target_at DESC + ) + """ + ) + connection.commit() + + def save_runs( + self, + solar_run: PowerForecastRun, + load_run: PowerForecastRun, + net_run: NetPowerForecastRun, + ) -> int: + self.initialize() + power_rows = [ + ( + run.issued_at, + point.target_at, + run.kind.value, + run.source, + run.model_version, + point.horizon_minutes, + point.expected_power_w, + point.p10_power_w, + point.p50_power_w, + point.p90_power_w, + point.confidence, + ) + for run in (solar_run, load_run) + for point in run.points + ] + net_rows = [ + ( + net_run.issued_at, + point.target_at, + net_run.source, + point.horizon_minutes, + point.expected_net_power_w, + point.safe_net_power_w, + point.p10_net_power_w, + point.p50_net_power_w, + point.p90_net_power_w, + point.solar_p50_power_w, + point.load_p50_power_w, + point.solar_p10_power_w, + point.solar_p90_power_w, + point.load_p10_power_w, + point.load_p90_power_w, + ) + for point in net_run.points + ] + + with self._connection() as connection: + with connection.cursor() as cursor: + cursor.executemany( + """ + INSERT INTO oracle_power_forecast_points ( + issued_at, + target_at, + kind, + source, + model_version, + horizon_minutes, + expected_power_w, + p10_power_w, + p50_power_w, + p90_power_w, + confidence + ) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) + ON CONFLICT (issued_at, target_at, kind, source, model_version) + DO UPDATE SET + horizon_minutes = EXCLUDED.horizon_minutes, + expected_power_w = EXCLUDED.expected_power_w, + p10_power_w = EXCLUDED.p10_power_w, + p50_power_w = EXCLUDED.p50_power_w, + p90_power_w = EXCLUDED.p90_power_w, + confidence = EXCLUDED.confidence, + inserted_at = now() + """, + power_rows, + ) + cursor.executemany( + """ + INSERT INTO oracle_net_forecast_points ( + issued_at, + target_at, + source, + horizon_minutes, + expected_net_power_w, + safe_net_power_w, + p10_net_power_w, + p50_net_power_w, + p90_net_power_w, + solar_p50_power_w, + load_p50_power_w, + solar_p10_power_w, + solar_p90_power_w, + load_p10_power_w, + load_p90_power_w + ) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) + ON CONFLICT (issued_at, target_at, source) + DO UPDATE SET + horizon_minutes = EXCLUDED.horizon_minutes, + expected_net_power_w = EXCLUDED.expected_net_power_w, + safe_net_power_w = EXCLUDED.safe_net_power_w, + p10_net_power_w = EXCLUDED.p10_net_power_w, + p50_net_power_w = EXCLUDED.p50_net_power_w, + p90_net_power_w = EXCLUDED.p90_net_power_w, + solar_p50_power_w = EXCLUDED.solar_p50_power_w, + load_p50_power_w = EXCLUDED.load_p50_power_w, + solar_p10_power_w = EXCLUDED.solar_p10_power_w, + solar_p90_power_w = EXCLUDED.solar_p90_power_w, + load_p10_power_w = EXCLUDED.load_p10_power_w, + load_p90_power_w = EXCLUDED.load_p90_power_w, + inserted_at = now() + """, + net_rows, + ) + connection.commit() + + return len(power_rows) + len(net_rows) + + def load_latest_runs( + self, + ) -> tuple[PowerForecastRun, PowerForecastRun, NetPowerForecastRun] | None: + solar_run = self.load_latest_power_run("solar") + load_run = self.load_latest_power_run("load") + net_run = self.load_latest_net_run() + if solar_run is None or load_run is None or net_run is None: + return None + return solar_run, load_run, net_run + + def load_latest_power_run(self, kind: str) -> PowerForecastRun | None: + if kind not in {"solar", "load"}: + raise ValueError("kind must be 'solar' or 'load'") + + with self._connection() as connection: + with connection.cursor() as cursor: + cursor.execute( + """ + SELECT issued_at, kind, source, model_version + FROM oracle_power_forecast_points + WHERE kind = %s + GROUP BY issued_at, kind, source, model_version + ORDER BY max(inserted_at) DESC + LIMIT 1 + """, + (kind,), + ) + run_row = cursor.fetchone() + if run_row is None: + return None + + cursor.execute( + """ + SELECT + target_at, + horizon_minutes, + expected_power_w, + p10_power_w, + p50_power_w, + p90_power_w, + confidence + FROM oracle_power_forecast_points + WHERE issued_at = %s + AND kind = %s + AND source = %s + AND model_version = %s + ORDER BY target_at + """, + (run_row[0], run_row[1], run_row[2], run_row[3]), + ) + point_rows = cursor.fetchall() + + return PowerForecastRun( + issued_at=run_row[0], + kind=ForecastKind(run_row[1]), + source=run_row[2], + model_version=run_row[3], + points=[ + PowerForecastPoint( + target_at=row[0], + horizon_minutes=row[1], + expected_power_w=row[2], + p10_power_w=row[3], + p50_power_w=row[4], + p90_power_w=row[5], + confidence=row[6], + source=run_row[2], + model_version=run_row[3], + ) + for row in point_rows + ], + ) + + def load_latest_net_run(self) -> NetPowerForecastRun | None: + with self._connection() as connection: + with connection.cursor() as cursor: + cursor.execute( + """ + SELECT issued_at, source + FROM oracle_net_forecast_points + GROUP BY issued_at, source + ORDER BY max(inserted_at) DESC + LIMIT 1 + """ + ) + run_row = cursor.fetchone() + if run_row is None: + return None + + cursor.execute( + """ + SELECT + target_at, + horizon_minutes, + expected_net_power_w, + safe_net_power_w, + COALESCE(p10_net_power_w, safe_net_power_w), + COALESCE(p50_net_power_w, expected_net_power_w), + COALESCE(p90_net_power_w, expected_net_power_w), + solar_p50_power_w, + load_p50_power_w, + solar_p10_power_w, + COALESCE(solar_p90_power_w, solar_p50_power_w), + COALESCE(load_p10_power_w, load_p50_power_w), + load_p90_power_w + FROM oracle_net_forecast_points + WHERE issued_at = %s + AND source = %s + ORDER BY target_at + """, + (run_row[0], run_row[1]), + ) + point_rows = cursor.fetchall() + + return NetPowerForecastRun( + issued_at=run_row[0], + source=run_row[1], + points=[ + NetPowerForecastPoint( + target_at=row[0], + horizon_minutes=row[1], + expected_net_power_w=row[2], + safe_net_power_w=row[3], + p10_net_power_w=row[4], + p50_net_power_w=row[5], + p90_net_power_w=row[6], + solar_p50_power_w=row[7], + load_p50_power_w=row[8], + solar_p10_power_w=row[9], + solar_p90_power_w=row[10], + load_p10_power_w=row[11], + load_p90_power_w=row[12], + ) + for row in point_rows + ], + ) + + def load_recent_net_runs( + self, + lookback: timedelta = timedelta(hours=6), + limit: int = 6, + ) -> list[dict[str, object]]: + return self.load_lagged_net_runs( + lag_hours=[hour for hour in (1, 2, 6, 24, 48) if hour <= lookback.total_seconds() / 3600], + tolerance=timedelta(minutes=45), + limit=limit, + ) + + def load_lagged_net_runs( + self, + lag_hours: list[int] | None = None, + tolerance: timedelta = timedelta(minutes=45), + limit: int = 5, + ) -> list[dict[str, object]]: + if lag_hours is None: + lag_hours = [1, 2, 6, 24, 48] + + now = datetime.now(timezone.utc) + selected: list[tuple[int, datetime]] = [] + used_issued_at: set[datetime] = set() + + with self._connection() as connection: + with connection.cursor() as cursor: + for lag_hour in lag_hours: + target_issued_at = now - timedelta(hours=lag_hour) + cursor.execute( + """ + SELECT issued_at + FROM oracle_net_forecast_points + WHERE issued_at BETWEEN %s AND %s + GROUP BY issued_at + ORDER BY abs(extract(epoch FROM (issued_at - %s))) + LIMIT 1 + """, + ( + target_issued_at - tolerance, + target_issued_at + tolerance, + target_issued_at, + ), + ) + row = cursor.fetchone() + if row is None or row[0] in used_issued_at: + continue + selected.append((lag_hour, row[0])) + used_issued_at.add(row[0]) + if len(selected) >= limit: + break + + runs: list[dict[str, object]] = [] + for lag_hour, issued_at in selected: + cursor.execute( + """ + SELECT + target_at, + horizon_minutes, + expected_net_power_w, + safe_net_power_w, + COALESCE(p10_net_power_w, safe_net_power_w), + COALESCE(p50_net_power_w, expected_net_power_w), + p90_net_power_w, + solar_p50_power_w, + load_p50_power_w, + solar_p10_power_w, + solar_p90_power_w, + load_p10_power_w, + load_p90_power_w + FROM oracle_net_forecast_points + WHERE issued_at = %s + AND target_at >= %s + ORDER BY target_at + """, + (issued_at, issued_at), + ) + points = cursor.fetchall() + if not points: + continue + runs.append( + { + "lag_hours": lag_hour, + "issued_at": issued_at, + "points": [ + { + "target_at": row[0], + "horizon_minutes": row[1], + "expected_net_power_w": row[2], + "safe_net_power_w": row[3], + "p10_net_power_w": row[4], + "p50_net_power_w": row[5], + "p90_net_power_w": row[6], + "solar_p50_power_w": row[7], + "load_p50_power_w": row[8], + "solar_p10_power_w": row[9], + "solar_p90_power_w": row[10], + "load_p10_power_w": row[11], + "load_p90_power_w": row[12], + } + for row in points + ], + } + ) + + return runs + + def load_lagged_power_runs( + self, + kind: str, + lag_hours: list[int] | None = None, + tolerance: timedelta = timedelta(minutes=45), + limit: int = 5, + ) -> list[dict[str, object]]: + if kind not in {"solar", "load"}: + raise ValueError("kind must be 'solar' or 'load'") + if lag_hours is None: + lag_hours = [1, 2, 6, 24, 48] + + now = datetime.now(timezone.utc) + selected: list[tuple[int, datetime, str, str, str]] = [] + used_keys: set[tuple[datetime, str, str, str]] = set() + + with self._connection() as connection: + with connection.cursor() as cursor: + for lag_hour in lag_hours: + target_issued_at = now - timedelta(hours=lag_hour) + cursor.execute( + """ + SELECT issued_at, kind, source, model_version + FROM oracle_power_forecast_points + WHERE kind = %s + AND issued_at BETWEEN %s AND %s + GROUP BY issued_at, kind, source, model_version + ORDER BY abs(extract(epoch FROM (issued_at - %s))) + LIMIT 1 + """, + ( + kind, + target_issued_at - tolerance, + target_issued_at + tolerance, + target_issued_at, + ), + ) + row = cursor.fetchone() + if row is None: + continue + key = (row[0], row[1], row[2], row[3]) + if key in used_keys: + continue + selected.append((lag_hour, row[0], row[1], row[2], row[3])) + used_keys.add(key) + if len(selected) >= limit: + break + + runs: list[dict[str, object]] = [] + for lag_hour, issued_at, run_kind, source, model_version in selected: + cursor.execute( + """ + SELECT + target_at, + horizon_minutes, + expected_power_w, + p10_power_w, + p50_power_w, + p90_power_w, + confidence + FROM oracle_power_forecast_points + WHERE issued_at = %s + AND kind = %s + AND source = %s + AND model_version = %s + AND target_at >= %s + ORDER BY target_at + """, + (issued_at, run_kind, source, model_version, issued_at), + ) + points = cursor.fetchall() + if not points: + continue + runs.append( + { + "lag_hours": lag_hour, + "issued_at": issued_at, + "kind": run_kind, + "source": source, + "model_version": model_version, + "points": [ + { + "target_at": row[0], + "horizon_minutes": row[1], + "expected_power_w": row[2], + "p10_power_w": row[3], + "p50_power_w": row[4], + "p90_power_w": row[5], + "confidence": row[6], + } + for row in points + ], + } + ) + + return runs + + def evaluate_due_forecasts( + self, + actual_window: timedelta = timedelta(minutes=5), + lookback: timedelta = timedelta(days=7), + limit: int = 1000, + ) -> int: + self.initialize() + start_at = datetime.now(timezone.utc) - lookback + + with self._connection() as connection: + with connection.cursor() as cursor: + power_count = self._evaluate_due_power_forecasts( + cursor=cursor, + actual_window=actual_window, + start_at=start_at, + limit=limit, + ) + remaining_limit = max(limit - power_count, 0) + net_count = 0 + if remaining_limit > 0: + net_count = self._evaluate_due_net_forecasts( + cursor=cursor, + actual_window=actual_window, + start_at=start_at, + limit=remaining_limit, + ) + connection.commit() + + return power_count + net_count + + def load_evaluation_summary( + self, + lookback: timedelta = timedelta(days=7), + ) -> list[dict[str, object]]: + start_at = datetime.now(timezone.utc) - lookback + with self._connection() as connection: + with connection.cursor() as cursor: + cursor.execute( + """ + WITH bucketed AS ( + SELECT + *, + CASE + WHEN horizon_minutes < 120 THEN 1 + WHEN horizon_minutes < 240 THEN 2 + WHEN horizon_minutes < 480 THEN 3 + WHEN horizon_minutes < 960 THEN 4 + ELSE 5 + END AS horizon_bucket, + CASE + WHEN horizon_minutes < 120 THEN '0-2h' + WHEN horizon_minutes < 240 THEN '2-4h' + WHEN horizon_minutes < 480 THEN '4-8h' + WHEN horizon_minutes < 960 THEN '8-16h' + ELSE '16-24h' + END AS horizon_label + FROM oracle_forecast_evaluations + WHERE target_at >= %s + AND realized_power_w IS NOT NULL + ) + SELECT + kind, + source, + model_version, + horizon_bucket, + horizon_label, + min(horizon_minutes) AS min_horizon_minutes, + max(horizon_minutes) AS max_horizon_minutes, + count(*) AS evaluated_count, + avg(error_w) AS mean_error_w, + avg(absolute_error_w) AS mean_absolute_error_w, + percentile_cont(0.50) WITHIN GROUP ( + ORDER BY absolute_error_w + ) AS median_absolute_error_w, + avg(absolute_pct_error) AS mean_absolute_pct_error, + avg( + CASE + WHEN covered_by_p10_p90 IS NULL THEN NULL + WHEN covered_by_p10_p90 THEN 1.0 + ELSE 0.0 + END + ) AS interval_coverage + FROM bucketed + GROUP BY kind, source, model_version, horizon_bucket, horizon_label + ORDER BY kind, source, model_version, horizon_bucket + """, + (start_at,), + ) + rows = cursor.fetchall() + + return [ + { + "kind": row[0], + "source": row[1], + "model_version": row[2], + "horizon_bucket": row[3], + "horizon_label": row[4], + "min_horizon_minutes": row[5], + "max_horizon_minutes": row[6], + "evaluated_count": row[7], + "mean_error_w": row[8], + "mean_absolute_error_w": row[9], + "median_absolute_error_w": row[10], + "mean_absolute_pct_error": row[11], + "interval_coverage": row[12], + } + for row in rows + ] + + def _evaluate_due_power_forecasts( + self, + cursor: object, + actual_window: timedelta, + start_at: datetime, + limit: int, + ) -> int: + cursor.execute( + """ + WITH candidates AS ( + SELECT + forecast.issued_at, + forecast.target_at, + forecast.kind, + forecast.source, + forecast.model_version, + forecast.horizon_minutes, + forecast.expected_power_w, + forecast.p10_power_w, + forecast.p50_power_w, + forecast.p90_power_w + FROM oracle_power_forecast_points AS forecast + LEFT JOIN oracle_forecast_evaluations AS evaluation + ON evaluation.issued_at = forecast.issued_at + AND evaluation.target_at = forecast.target_at + AND evaluation.kind = forecast.kind + AND evaluation.source = forecast.source + AND evaluation.model_version = forecast.model_version + WHERE forecast.target_at >= %s + AND forecast.target_at <= now() - %s + AND ( + evaluation.issued_at IS NULL + OR evaluation.sample_count = 0 + ) + ORDER BY forecast.target_at, forecast.issued_at + LIMIT %s + ), + realized AS ( + SELECT + candidates.*, + actual.realized_power_w, + actual.sample_count + FROM candidates + LEFT JOIN LATERAL ( + SELECT + avg( + CASE candidates.kind + WHEN 'solar' THEN snapshot.solar_power_w + WHEN 'load' THEN snapshot.load_power_w + ELSE NULL + END + ) AS realized_power_w, + count(*) FILTER ( + WHERE CASE candidates.kind + WHEN 'solar' THEN snapshot.solar_power_w + WHEN 'load' THEN snapshot.load_power_w + ELSE NULL + END IS NOT NULL + ) AS sample_count + FROM sigen_plant_snapshots AS snapshot + WHERE snapshot.observed_at >= candidates.target_at + AND snapshot.observed_at < candidates.target_at + %s + ) AS actual ON TRUE + ) + INSERT INTO oracle_forecast_evaluations ( + issued_at, + target_at, + kind, + source, + model_version, + horizon_minutes, + expected_power_w, + p10_power_w, + p50_power_w, + p90_power_w, + realized_power_w, + error_w, + absolute_error_w, + absolute_pct_error, + covered_by_p10_p90, + sample_count, + evaluated_at + ) + SELECT + issued_at, + target_at, + kind, + source, + model_version, + horizon_minutes, + expected_power_w, + p10_power_w, + p50_power_w, + p90_power_w, + realized_power_w, + realized_power_w - p50_power_w, + abs(realized_power_w - p50_power_w), + CASE + WHEN abs(realized_power_w) < 1 THEN NULL + ELSE abs(realized_power_w - p50_power_w) / abs(realized_power_w) + END, + CASE + WHEN realized_power_w IS NULL THEN NULL + ELSE realized_power_w BETWEEN p10_power_w AND p90_power_w + END, + COALESCE(sample_count, 0), + now() + FROM realized + ON CONFLICT ( + issued_at, + target_at, + kind, + source, + model_version + ) + DO UPDATE SET + horizon_minutes = EXCLUDED.horizon_minutes, + expected_power_w = EXCLUDED.expected_power_w, + p10_power_w = EXCLUDED.p10_power_w, + p50_power_w = EXCLUDED.p50_power_w, + p90_power_w = EXCLUDED.p90_power_w, + realized_power_w = EXCLUDED.realized_power_w, + error_w = EXCLUDED.error_w, + absolute_error_w = EXCLUDED.absolute_error_w, + absolute_pct_error = EXCLUDED.absolute_pct_error, + covered_by_p10_p90 = EXCLUDED.covered_by_p10_p90, + sample_count = EXCLUDED.sample_count, + evaluated_at = EXCLUDED.evaluated_at, + updated_at = now() + """, + (start_at, actual_window, limit, actual_window), + ) + return cursor.rowcount + + def _evaluate_due_net_forecasts( + self, + cursor: object, + actual_window: timedelta, + start_at: datetime, + limit: int, + ) -> int: + cursor.execute( + """ + WITH candidates AS ( + SELECT + forecast.issued_at, + forecast.target_at, + 'net'::text AS kind, + forecast.source, + 'net_forecaster_v1'::text AS model_version, + forecast.horizon_minutes, + forecast.expected_net_power_w AS expected_power_w, + COALESCE(forecast.p10_net_power_w, forecast.safe_net_power_w) + AS p10_power_w, + COALESCE(forecast.p50_net_power_w, forecast.expected_net_power_w) + AS p50_power_w, + forecast.p90_net_power_w AS p90_power_w + FROM oracle_net_forecast_points AS forecast + LEFT JOIN oracle_forecast_evaluations AS evaluation + ON evaluation.issued_at = forecast.issued_at + AND evaluation.target_at = forecast.target_at + AND evaluation.kind = 'net' + AND evaluation.source = forecast.source + AND evaluation.model_version = 'net_forecaster_v1' + WHERE forecast.target_at >= %s + AND forecast.target_at <= now() - %s + AND ( + evaluation.issued_at IS NULL + OR evaluation.sample_count = 0 + ) + ORDER BY forecast.target_at, forecast.issued_at + LIMIT %s + ), + realized AS ( + SELECT + candidates.*, + actual.realized_power_w, + actual.sample_count + FROM candidates + LEFT JOIN LATERAL ( + SELECT + avg(snapshot.solar_power_w - snapshot.load_power_w) + AS realized_power_w, + count(*) FILTER ( + WHERE snapshot.solar_power_w IS NOT NULL + AND snapshot.load_power_w IS NOT NULL + ) AS sample_count + FROM sigen_plant_snapshots AS snapshot + WHERE snapshot.observed_at >= candidates.target_at + AND snapshot.observed_at < candidates.target_at + %s + ) AS actual ON TRUE + ) + INSERT INTO oracle_forecast_evaluations ( + issued_at, + target_at, + kind, + source, + model_version, + horizon_minutes, + expected_power_w, + p10_power_w, + p50_power_w, + p90_power_w, + realized_power_w, + error_w, + absolute_error_w, + absolute_pct_error, + covered_by_p10_p90, + sample_count, + evaluated_at + ) + SELECT + issued_at, + target_at, + kind, + source, + model_version, + horizon_minutes, + expected_power_w, + p10_power_w, + p50_power_w, + p90_power_w, + realized_power_w, + realized_power_w - p50_power_w, + abs(realized_power_w - p50_power_w), + CASE + WHEN abs(realized_power_w) < 1 THEN NULL + ELSE abs(realized_power_w - p50_power_w) / abs(realized_power_w) + END, + CASE + WHEN realized_power_w IS NULL OR p90_power_w IS NULL THEN NULL + ELSE realized_power_w BETWEEN p10_power_w AND p90_power_w + END, + COALESCE(sample_count, 0), + now() + FROM realized + ON CONFLICT ( + issued_at, + target_at, + kind, + source, + model_version + ) + DO UPDATE SET + horizon_minutes = EXCLUDED.horizon_minutes, + expected_power_w = EXCLUDED.expected_power_w, + p10_power_w = EXCLUDED.p10_power_w, + p50_power_w = EXCLUDED.p50_power_w, + p90_power_w = EXCLUDED.p90_power_w, + realized_power_w = EXCLUDED.realized_power_w, + error_w = EXCLUDED.error_w, + absolute_error_w = EXCLUDED.absolute_error_w, + absolute_pct_error = EXCLUDED.absolute_pct_error, + covered_by_p10_p90 = EXCLUDED.covered_by_p10_p90, + sample_count = EXCLUDED.sample_count, + evaluated_at = EXCLUDED.evaluated_at, + updated_at = now() + """, + (start_at, actual_window, limit, actual_window), + ) + return cursor.rowcount + + @contextmanager + def _connection(self) -> Iterator[object]: + try: + import psycopg + except ImportError as error: + raise OracleStoreConfigurationError( + "Install dependencies with `python3 -m pip install -r requirements.txt`" + ) from error + + with psycopg.connect(self.config.database_url) as connection: + yield connection diff --git a/astrape_oracle/classes/weather/__init__.py b/astrape_oracle/classes/weather/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/astrape_oracle/classes/weather/ecowitt.py b/astrape_oracle/classes/weather/ecowitt.py new file mode 100644 index 0000000..97ba9a1 --- /dev/null +++ b/astrape_oracle/classes/weather/ecowitt.py @@ -0,0 +1,151 @@ +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Any +from urllib.request import urlopen +import json +import re + +from gibil.classes.models import WeatherStationObservation + + +class EcowittGw3000Client: + """Fetches local live data from an Ecowitt GW3000/GW3000A gateway.""" + + def __init__(self, host: str, timeout_seconds: float = 5) -> None: + self.host = host.removeprefix("http://").removeprefix("https://").rstrip("/") + self.timeout_seconds = timeout_seconds + + def fetch_observation(self) -> WeatherStationObservation: + received_at = datetime.now(timezone.utc) + with urlopen(self._live_data_url(), timeout=self.timeout_seconds) as response: + payload = json.loads(response.read().decode("utf-8")) + + return EcowittGw3000Parser().parse( + payload=payload, + received_at=received_at, + ) + + def _live_data_url(self) -> str: + return f"http://{self.host}/get_livedata_info" + + +class EcowittGw3000Parser: + """Normalizes GW3000 live-data JSON into a station observation.""" + + common_id_names = { + "0x02": "temperature_c", + "0x07": "relative_humidity_pct", + "0x03": "dew_point_c", + "0x0B": "wind_speed_m_s", + "0x0C": "wind_gust_m_s", + "0x19": "wind_max_m_s", + "0x0A": "wind_direction_deg", + "0x15": "solar_radiation_w_m2", + "0x17": "uv_index", + } + rain_id_names = { + "0x0D": "rain_event_mm", + "0x0E": "rain_rate_mm_h", + "0x7C": "rain_hourly_mm", + "0x10": "rain_daily_mm", + "0x11": "rain_weekly_mm", + "0x12": "rain_monthly_mm", + "0x13": "rain_yearly_mm", + } + + def parse( + self, + payload: dict[str, Any], + received_at: datetime | None = None, + ) -> WeatherStationObservation: + if received_at is None: + received_at = datetime.now(timezone.utc) + + common = self._parse_id_list(payload.get("common_list", [])) + rain = self._parse_id_list(payload.get("rain", [])) + indoor = self._parse_indoor(payload.get("wh25", [])) + + return WeatherStationObservation( + observed_at=received_at, + received_at=received_at, + source="ecowitt_gw3000", + temperature_c=common.get("temperature_c"), + relative_humidity_pct=common.get("relative_humidity_pct"), + dew_point_c=common.get("dew_point_c"), + wind_speed_m_s=common.get("wind_speed_m_s"), + wind_gust_m_s=common.get("wind_gust_m_s"), + wind_max_m_s=common.get("wind_max_m_s"), + wind_direction_deg=common.get("wind_direction_deg"), + solar_radiation_w_m2=common.get("solar_radiation_w_m2"), + uv_index=common.get("uv_index"), + rain_rate_mm_h=rain.get("rain_rate_mm_h"), + rain_event_mm=rain.get("rain_event_mm"), + rain_hourly_mm=rain.get("rain_hourly_mm"), + rain_daily_mm=rain.get("rain_daily_mm"), + rain_weekly_mm=rain.get("rain_weekly_mm"), + rain_monthly_mm=rain.get("rain_monthly_mm"), + rain_yearly_mm=rain.get("rain_yearly_mm"), + indoor_temperature_c=indoor.get("indoor_temperature_c"), + indoor_humidity_pct=indoor.get("indoor_humidity_pct"), + absolute_pressure_hpa=indoor.get("absolute_pressure_hpa"), + relative_pressure_hpa=indoor.get("relative_pressure_hpa"), + metadata={ + "raw_payload": payload, + "debug": payload.get("debug", []), + }, + ) + + def _parse_id_list(self, rows: object) -> dict[str, float]: + parsed: dict[str, float] = {} + if not isinstance(rows, list): + return parsed + + for row in rows: + if not isinstance(row, dict): + continue + row_id = str(row.get("id", "")) + key = self.common_id_names.get(row_id) or self.rain_id_names.get(row_id) + if key is None: + continue + value = self._parse_number(row.get("val")) + if value is None: + continue + if key in {"wind_speed_m_s", "wind_gust_m_s", "wind_max_m_s"}: + value = self._knots_to_m_s(value) + parsed[key] = value + + return parsed + + def _parse_indoor(self, rows: object) -> dict[str, float]: + if not isinstance(rows, list) or not rows: + return {} + row = rows[0] + if not isinstance(row, dict): + return {} + + values = { + "indoor_temperature_c": self._parse_number(row.get("intemp")), + "indoor_humidity_pct": self._parse_number(row.get("inhumi")), + "absolute_pressure_hpa": self._parse_number(row.get("abs")), + "relative_pressure_hpa": self._parse_number(row.get("rel")), + } + return { + key: value + for key, value in values.items() + if value is not None + } + + def _parse_number(self, value: object) -> float | None: + if value is None: + return None + if isinstance(value, (int, float)): + return float(value) + + match = re.search(r"-?\d+(?:\.\d+)?", str(value)) + if match is None: + return None + return float(match.group(0)) + + def _knots_to_m_s(self, value: float) -> float: + return value * 0.514444 diff --git a/astrape_oracle/classes/weather/openmeteo.py b/astrape_oracle/classes/weather/openmeteo.py new file mode 100644 index 0000000..231116e --- /dev/null +++ b/astrape_oracle/classes/weather/openmeteo.py @@ -0,0 +1,254 @@ +from __future__ import annotations + +from datetime import date, datetime, timezone +from typing import Any +from urllib.parse import urlencode +from urllib.request import urlopen +import json + +from gibil.classes.models import ( + WeatherForecastPoint, + WeatherForecastRun, + WeatherResolvedTruth, +) + + +class OpenMeteoClient: + """Fetches external weather forecasts from Open-Meteo.""" + + base_url = "https://api.open-meteo.com/v1/forecast" + + def build_url( + self, + latitude: float, + longitude: float, + forecast_hours: int = 48, + timezone_name: str = "UTC", + ) -> str: + params = { + "latitude": latitude, + "longitude": longitude, + "hourly": ",".join( + [ + "temperature_2m", + "shortwave_radiation", + "cloud_cover", + ] + ), + "forecast_hours": forecast_hours, + "timezone": timezone_name, + } + return f"{self.base_url}?{urlencode(params)}" + + def fetch_forecast( + self, + latitude: float, + longitude: float, + forecast_hours: int = 48, + ) -> WeatherForecastRun: + url = self.build_url(latitude, longitude, forecast_hours) + with urlopen(url, timeout=10) as response: + payload = json.loads(response.read().decode("utf-8")) + + return OpenMeteoParser().parse_forecast( + payload=payload, + latitude=latitude, + longitude=longitude, + issued_at=datetime.now(timezone.utc), + ) + + +class OpenMeteoArchiveClient: + """Fetches historical weather data from Open-Meteo archive.""" + + base_url = "https://archive-api.open-meteo.com/v1/archive" + + def build_url( + self, + latitude: float, + longitude: float, + start_date: date, + end_date: date, + timezone_name: str = "UTC", + ) -> str: + params = { + "latitude": latitude, + "longitude": longitude, + "start_date": start_date.isoformat(), + "end_date": end_date.isoformat(), + "hourly": ",".join( + [ + "temperature_2m", + "shortwave_radiation", + "cloud_cover", + ] + ), + "timezone": timezone_name, + } + return f"{self.base_url}?{urlencode(params)}" + + def fetch_resolved_truth( + self, + latitude: float, + longitude: float, + start_date: date, + end_date: date, + ) -> list[WeatherResolvedTruth]: + url = self.build_url(latitude, longitude, start_date, end_date) + with urlopen(url, timeout=20) as response: + payload = json.loads(response.read().decode("utf-8")) + + return OpenMeteoArchiveParser().parse_resolved_truth(payload) + + +class OpenMeteoParser: + """Converts Open-Meteo JSON into clean external forecast records.""" + + def parse_forecast( + self, + payload: dict[str, Any], + latitude: float, + longitude: float, + issued_at: datetime, + ) -> WeatherForecastRun: + hourly = payload.get("hourly", {}) + times = hourly.get("time", []) + temperatures = hourly.get("temperature_2m", []) + radiation = hourly.get("shortwave_radiation", []) + cloud_cover = hourly.get("cloud_cover", []) + + points: list[WeatherForecastPoint] = [] + for index, raw_time in enumerate(times): + target_at = self._parse_time(raw_time) + horizon_hours = max( + 0, round((target_at - issued_at).total_seconds() / 3600) + ) + + points.append( + WeatherForecastPoint( + issued_at=issued_at, + target_at=target_at, + horizon_hours=horizon_hours, + temperature_c=self._at(temperatures, index), + shortwave_radiation_w_m2=self._at(radiation, index), + cloud_cover_pct=self._at(cloud_cover, index), + ) + ) + + return WeatherForecastRun( + issued_at=issued_at, + source="open_meteo", + latitude=latitude, + longitude=longitude, + points=points, + ) + + def _parse_time(self, raw_time: str) -> datetime: + parsed = datetime.fromisoformat(raw_time) + if parsed.tzinfo is None: + return parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) + + def _at(self, values: list[Any], index: int) -> float | None: + if index >= len(values): + return None + + value = values[index] + if value is None: + return None + + return float(value) + + +class OpenMeteoArchiveParser: + """Converts Open-Meteo archive JSON into resolved truth records.""" + + def parse_resolved_truth(self, payload: dict[str, Any]) -> list[WeatherResolvedTruth]: + hourly = payload.get("hourly", {}) + times = hourly.get("time", []) + temperatures = hourly.get("temperature_2m", []) + radiation = hourly.get("shortwave_radiation", []) + cloud_cover = hourly.get("cloud_cover", []) + + truth: list[WeatherResolvedTruth] = [] + for index, raw_time in enumerate(times): + truth.append( + WeatherResolvedTruth( + resolved_at=self._parse_time(raw_time), + temperature_c=self._at(temperatures, index), + shortwave_radiation_w_m2=self._at(radiation, index), + cloud_cover_pct=self._at(cloud_cover, index), + source="open_meteo_archive", + ) + ) + + return truth + + def _parse_time(self, raw_time: str) -> datetime: + parsed = datetime.fromisoformat(raw_time) + if parsed.tzinfo is None: + return parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) + + def _at(self, values: list[Any], index: int) -> float | None: + if index >= len(values): + return None + + value = values[index] + if value is None: + return None + + return float(value) + + +class WeatherBuilder: + """Builds a clean database-ready set of external weather forecast records.""" + + def build_forecast_run( + self, + source: str, + latitude: float, + longitude: float, + points: list[WeatherForecastPoint], + issued_at: datetime | None = None, + ) -> WeatherForecastRun: + if issued_at is None: + issued_at = datetime.now(timezone.utc) + + clean_points = [ + WeatherForecastPoint( + issued_at=issued_at, + target_at=point.target_at, + horizon_hours=max( + 0, round((point.target_at - issued_at).total_seconds() / 3600) + ), + temperature_c=point.temperature_c, + shortwave_radiation_w_m2=point.shortwave_radiation_w_m2, + cloud_cover_pct=point.cloud_cover_pct, + source=source, + ) + for point in sorted(points, key=lambda item: item.target_at) + ] + + return WeatherForecastRun( + issued_at=issued_at, + source=source, + latitude=latitude, + longitude=longitude, + points=clean_points, + ) + + def points_for_horizon( + self, + forecast_runs: list[WeatherForecastRun], + horizon_hours: int, + ) -> list[WeatherForecastPoint]: + points: list[WeatherForecastPoint] = [] + for run in forecast_runs: + points.extend( + point + for point in run.points + if point.horizon_hours == horizon_hours + ) + + return sorted(points, key=lambda point: point.target_at) diff --git a/astrape_oracle/classes/weather/store.py b/astrape_oracle/classes/weather/store.py new file mode 100644 index 0000000..ed73f19 --- /dev/null +++ b/astrape_oracle/classes/weather/store.py @@ -0,0 +1,640 @@ +from __future__ import annotations + +from contextlib import contextmanager +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from os import environ +from typing import Iterator + +from gibil.classes.models import ( + WeatherForecastPoint, + WeatherForecastRun, + WeatherResolvedTruth, + WeatherStationObservation, +) +from astrape_oracle.classes.datasets import ( + WeatherDisplayDataset, + WeatherStationDisplayDataset, +) + + +class WeatherStoreConfigurationError(RuntimeError): + pass + + +@dataclass(frozen=True) +class WeatherStoreConfig: + database_url: str + + @classmethod + def from_env(cls) -> "WeatherStoreConfig": + database_url = environ.get("ASTRAPE_DATABASE_URL") + if not database_url: + raise WeatherStoreConfigurationError( + "ASTRAPE_DATABASE_URL is required for weather storage" + ) + + return cls(database_url=database_url) + + +class WeatherStore: + """Persists external weather forecasts and resolved truth in TimescaleDB.""" + + def __init__(self, config: WeatherStoreConfig) -> None: + self.config = config + + @classmethod + def from_env(cls) -> "WeatherStore": + return cls(WeatherStoreConfig.from_env()) + + def initialize(self) -> None: + with self._connection() as connection: + with connection.cursor() as cursor: + cursor.execute("CREATE EXTENSION IF NOT EXISTS timescaledb") + cursor.execute( + """ + CREATE TABLE IF NOT EXISTS weather_forecast_points ( + issued_at TIMESTAMPTZ NOT NULL, + target_at TIMESTAMPTZ NOT NULL, + horizon_hours INTEGER NOT NULL, + source TEXT NOT NULL, + latitude DOUBLE PRECISION NOT NULL, + longitude DOUBLE PRECISION NOT NULL, + temperature_c DOUBLE PRECISION, + shortwave_radiation_w_m2 DOUBLE PRECISION, + cloud_cover_pct DOUBLE PRECISION, + inserted_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (issued_at, target_at, source) + ) + """ + ) + cursor.execute( + """ + SELECT create_hypertable( + 'weather_forecast_points', + 'target_at', + if_not_exists => TRUE + ) + """ + ) + cursor.execute( + """ + CREATE TABLE IF NOT EXISTS weather_resolved_truth ( + resolved_at TIMESTAMPTZ NOT NULL, + source TEXT NOT NULL, + temperature_c DOUBLE PRECISION, + shortwave_radiation_w_m2 DOUBLE PRECISION, + cloud_cover_pct DOUBLE PRECISION, + inserted_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (resolved_at, source) + ) + """ + ) + cursor.execute( + """ + ALTER TABLE weather_resolved_truth + ADD COLUMN IF NOT EXISTS cloud_cover_pct DOUBLE PRECISION + """ + ) + cursor.execute( + """ + SELECT create_hypertable( + 'weather_resolved_truth', + 'resolved_at', + if_not_exists => TRUE + ) + """ + ) + cursor.execute( + """ + CREATE TABLE IF NOT EXISTS weather_station_observations ( + observed_at TIMESTAMPTZ NOT NULL, + received_at TIMESTAMPTZ NOT NULL, + source TEXT NOT NULL, + temperature_c DOUBLE PRECISION, + relative_humidity_pct DOUBLE PRECISION, + dew_point_c DOUBLE PRECISION, + pressure_hpa DOUBLE PRECISION, + wind_speed_m_s DOUBLE PRECISION, + wind_gust_m_s DOUBLE PRECISION, + wind_max_m_s DOUBLE PRECISION, + wind_direction_deg DOUBLE PRECISION, + solar_radiation_w_m2 DOUBLE PRECISION, + uv_index DOUBLE PRECISION, + rain_rate_mm_h DOUBLE PRECISION, + rain_event_mm DOUBLE PRECISION, + rain_hourly_mm DOUBLE PRECISION, + rain_daily_mm DOUBLE PRECISION, + rain_weekly_mm DOUBLE PRECISION, + rain_monthly_mm DOUBLE PRECISION, + rain_yearly_mm DOUBLE PRECISION, + indoor_temperature_c DOUBLE PRECISION, + indoor_humidity_pct DOUBLE PRECISION, + absolute_pressure_hpa DOUBLE PRECISION, + relative_pressure_hpa DOUBLE PRECISION, + metadata JSONB NOT NULL DEFAULT '{}'::jsonb, + inserted_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (observed_at, source) + ) + """ + ) + cursor.execute( + """ + SELECT create_hypertable( + 'weather_station_observations', + 'observed_at', + if_not_exists => TRUE + ) + """ + ) + connection.commit() + + def save_forecast_run(self, forecast_run: WeatherForecastRun) -> int: + rows = [ + ( + point.issued_at, + point.target_at, + point.horizon_hours, + forecast_run.source, + forecast_run.latitude, + forecast_run.longitude, + point.temperature_c, + point.shortwave_radiation_w_m2, + point.cloud_cover_pct, + ) + for point in forecast_run.points + ] + if not rows: + return 0 + + with self._connection() as connection: + with connection.cursor() as cursor: + cursor.executemany( + """ + INSERT INTO weather_forecast_points ( + issued_at, + target_at, + horizon_hours, + source, + latitude, + longitude, + temperature_c, + shortwave_radiation_w_m2, + cloud_cover_pct + ) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) + ON CONFLICT (issued_at, target_at, source) + DO UPDATE SET + horizon_hours = EXCLUDED.horizon_hours, + latitude = EXCLUDED.latitude, + longitude = EXCLUDED.longitude, + temperature_c = EXCLUDED.temperature_c, + shortwave_radiation_w_m2 = EXCLUDED.shortwave_radiation_w_m2, + cloud_cover_pct = EXCLUDED.cloud_cover_pct, + inserted_at = now() + """, + rows, + ) + connection.commit() + + return len(rows) + + def save_resolved_truth(self, truth_points: list[WeatherResolvedTruth]) -> int: + rows = [ + ( + point.resolved_at, + point.source, + point.temperature_c, + point.shortwave_radiation_w_m2, + point.cloud_cover_pct, + ) + for point in truth_points + ] + if not rows: + return 0 + + with self._connection() as connection: + with connection.cursor() as cursor: + cursor.executemany( + """ + INSERT INTO weather_resolved_truth ( + resolved_at, + source, + temperature_c, + shortwave_radiation_w_m2, + cloud_cover_pct + ) + VALUES (%s, %s, %s, %s, %s) + ON CONFLICT (resolved_at, source) + DO UPDATE SET + temperature_c = EXCLUDED.temperature_c, + shortwave_radiation_w_m2 = EXCLUDED.shortwave_radiation_w_m2, + cloud_cover_pct = EXCLUDED.cloud_cover_pct, + inserted_at = now() + """, + rows, + ) + connection.commit() + + return len(rows) + + def save_zero_hour_forecast_as_truth( + self, forecast_run: WeatherForecastRun + ) -> int: + truth_points = [ + WeatherResolvedTruth( + resolved_at=point.issued_at, + source="open_meteo_zero_hour", + temperature_c=point.temperature_c, + shortwave_radiation_w_m2=point.shortwave_radiation_w_m2, + cloud_cover_pct=point.cloud_cover_pct, + ) + for point in forecast_run.points + if point.horizon_hours == 0 + ] + return self.save_resolved_truth(truth_points) + + def save_station_observation( + self, + observation: WeatherStationObservation, + ) -> int: + with self._connection() as connection: + with connection.cursor() as cursor: + try: + from psycopg.types.json import Jsonb + except ImportError as error: + raise WeatherStoreConfigurationError( + "Install dependencies with `python3 -m pip install -r requirements.txt`" + ) from error + + cursor.execute( + """ + INSERT INTO weather_station_observations ( + observed_at, + received_at, + source, + temperature_c, + relative_humidity_pct, + dew_point_c, + pressure_hpa, + wind_speed_m_s, + wind_gust_m_s, + wind_max_m_s, + wind_direction_deg, + solar_radiation_w_m2, + uv_index, + rain_rate_mm_h, + rain_event_mm, + rain_hourly_mm, + rain_daily_mm, + rain_weekly_mm, + rain_monthly_mm, + rain_yearly_mm, + indoor_temperature_c, + indoor_humidity_pct, + absolute_pressure_hpa, + relative_pressure_hpa, + metadata + ) + VALUES ( + %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, + %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, + %s, %s, %s, %s, %s + ) + ON CONFLICT (observed_at, source) + DO UPDATE SET + received_at = EXCLUDED.received_at, + temperature_c = EXCLUDED.temperature_c, + relative_humidity_pct = EXCLUDED.relative_humidity_pct, + dew_point_c = EXCLUDED.dew_point_c, + pressure_hpa = EXCLUDED.pressure_hpa, + wind_speed_m_s = EXCLUDED.wind_speed_m_s, + wind_gust_m_s = EXCLUDED.wind_gust_m_s, + wind_max_m_s = EXCLUDED.wind_max_m_s, + wind_direction_deg = EXCLUDED.wind_direction_deg, + solar_radiation_w_m2 = EXCLUDED.solar_radiation_w_m2, + uv_index = EXCLUDED.uv_index, + rain_rate_mm_h = EXCLUDED.rain_rate_mm_h, + rain_event_mm = EXCLUDED.rain_event_mm, + rain_hourly_mm = EXCLUDED.rain_hourly_mm, + rain_daily_mm = EXCLUDED.rain_daily_mm, + rain_weekly_mm = EXCLUDED.rain_weekly_mm, + rain_monthly_mm = EXCLUDED.rain_monthly_mm, + rain_yearly_mm = EXCLUDED.rain_yearly_mm, + indoor_temperature_c = EXCLUDED.indoor_temperature_c, + indoor_humidity_pct = EXCLUDED.indoor_humidity_pct, + absolute_pressure_hpa = EXCLUDED.absolute_pressure_hpa, + relative_pressure_hpa = EXCLUDED.relative_pressure_hpa, + metadata = EXCLUDED.metadata, + inserted_at = now() + """, + ( + observation.observed_at, + observation.received_at, + observation.source, + observation.temperature_c, + observation.relative_humidity_pct, + observation.dew_point_c, + observation.pressure_hpa, + observation.wind_speed_m_s, + observation.wind_gust_m_s, + observation.wind_max_m_s, + observation.wind_direction_deg, + observation.solar_radiation_w_m2, + observation.uv_index, + observation.rain_rate_mm_h, + observation.rain_event_mm, + observation.rain_hourly_mm, + observation.rain_daily_mm, + observation.rain_weekly_mm, + observation.rain_monthly_mm, + observation.rain_yearly_mm, + observation.indoor_temperature_c, + observation.indoor_humidity_pct, + observation.absolute_pressure_hpa, + observation.relative_pressure_hpa, + Jsonb(observation.metadata), + ), + ) + connection.commit() + + return 1 + + def load_latest_forecast_points( + self, + start_at: datetime, + end_at: datetime, + ) -> list[WeatherForecastPoint]: + with self._connection() as connection: + with connection.cursor() as cursor: + cursor.execute( + """ + SELECT + issued_at, + target_at, + horizon_hours, + source, + temperature_c, + shortwave_radiation_w_m2, + cloud_cover_pct + FROM ( + SELECT + issued_at, + target_at, + horizon_hours, + source, + temperature_c, + shortwave_radiation_w_m2, + cloud_cover_pct, + ROW_NUMBER() OVER ( + PARTITION BY target_at + ORDER BY issued_at DESC + ) as rn + FROM weather_forecast_points + WHERE target_at >= %s AND target_at <= %s + ) as ranked + WHERE rn = 1 + ORDER BY target_at + LIMIT 5000 + """, + (start_at, end_at), + ) + rows = cursor.fetchall() + + return [ + WeatherForecastPoint( + issued_at=row[0], + target_at=row[1], + horizon_hours=row[2], + source=row[3], + temperature_c=row[4], + shortwave_radiation_w_m2=row[5], + cloud_cover_pct=row[6], + ) + for row in rows + ] + + def load_forecast_history( + self, + start_at: datetime, + end_at: datetime, + ) -> dict[datetime, list[dict[str, object]]]: + """Every forecast issued for each target in the window, oldest first. + + Unlike `load_latest_forecast_points`, this keeps all vintages of a + forecast, so a training example can be shown the forecast that was + actually available at the time it was issued rather than the best one + known in hindsight. + """ + with self._connection() as connection: + with connection.cursor() as cursor: + cursor.execute( + """ + SELECT + issued_at, + target_at, + temperature_c, + shortwave_radiation_w_m2, + cloud_cover_pct + FROM weather_forecast_points + WHERE target_at >= %s + AND target_at <= %s + ORDER BY target_at, issued_at + """, + (start_at, end_at), + ) + rows = cursor.fetchall() + + by_target: dict[datetime, list[dict[str, object]]] = {} + for row in rows: + by_target.setdefault(row[1], []).append( + { + "issued_at": row[0], + "target_at": row[1], + "temperature_c": row[2], + "shortwave_radiation_w_m2": row[3], + "cloud_cover_pct": row[4], + } + ) + return by_target + + def load_display_dataset( + self, + start_at: datetime | None = None, + end_at: datetime | None = None, + ) -> WeatherDisplayDataset: + now = datetime.now(timezone.utc) + if start_at is None: + start_at = now - timedelta(hours=24) + if end_at is None: + end_at = now + timedelta(hours=48) + + with self._connection() as connection: + with connection.cursor() as cursor: + cursor.execute( + """ + SELECT + issued_at, + target_at, + horizon_hours, + source, + temperature_c, + shortwave_radiation_w_m2, + cloud_cover_pct + FROM ( + SELECT + issued_at, + target_at, + horizon_hours, + source, + temperature_c, + shortwave_radiation_w_m2, + cloud_cover_pct, + ROW_NUMBER() OVER (PARTITION BY target_at, horizon_hours ORDER BY issued_at DESC) as rn + FROM weather_forecast_points + WHERE target_at >= %s AND target_at <= %s + ) as ranked + WHERE rn = 1 + ORDER BY target_at, horizon_hours + LIMIT 5000 + """, + (start_at, end_at), + ) + forecast_rows = cursor.fetchall() + + cursor.execute( + """ + SELECT + resolved_at, + source, + temperature_c, + shortwave_radiation_w_m2, + cloud_cover_pct + FROM weather_resolved_truth + WHERE resolved_at >= %s AND resolved_at <= %s + ORDER BY resolved_at + LIMIT 5000 + """, + (start_at, now), + ) + truth_rows = cursor.fetchall() + + return WeatherDisplayDataset( + forecast_points=[ + WeatherForecastPoint( + issued_at=row[0], + target_at=row[1], + horizon_hours=row[2], + source=row[3], + temperature_c=row[4], + shortwave_radiation_w_m2=row[5], + cloud_cover_pct=row[6], + ) + for row in forecast_rows + ], + resolved_truth=[ + WeatherResolvedTruth( + resolved_at=row[0], + source=row[1], + temperature_c=row[2], + shortwave_radiation_w_m2=row[3], + cloud_cover_pct=row[4], + ) + for row in truth_rows + ], + ) + + def load_station_display_dataset( + self, + start_at: datetime | None = None, + end_at: datetime | None = None, + ) -> WeatherStationDisplayDataset: + now = datetime.now(timezone.utc) + if start_at is None: + start_at = now - timedelta(days=7) + if end_at is None: + end_at = now + + with self._connection() as connection: + with connection.cursor() as cursor: + cursor.execute( + """ + SELECT + observed_at, + received_at, + source, + temperature_c, + relative_humidity_pct, + dew_point_c, + pressure_hpa, + wind_speed_m_s, + wind_gust_m_s, + wind_max_m_s, + wind_direction_deg, + solar_radiation_w_m2, + uv_index, + rain_rate_mm_h, + rain_event_mm, + rain_hourly_mm, + rain_daily_mm, + rain_weekly_mm, + rain_monthly_mm, + rain_yearly_mm, + indoor_temperature_c, + indoor_humidity_pct, + absolute_pressure_hpa, + relative_pressure_hpa, + metadata + FROM weather_station_observations + WHERE observed_at >= %s AND observed_at <= %s + ORDER BY observed_at + LIMIT 100000 + """, + (start_at, end_at), + ) + rows = cursor.fetchall() + + return WeatherStationDisplayDataset( + observations=[ + WeatherStationObservation( + observed_at=row[0], + received_at=row[1], + source=row[2], + temperature_c=row[3], + relative_humidity_pct=row[4], + dew_point_c=row[5], + pressure_hpa=row[6], + wind_speed_m_s=row[7], + wind_gust_m_s=row[8], + wind_max_m_s=row[9], + wind_direction_deg=row[10], + solar_radiation_w_m2=row[11], + uv_index=row[12], + rain_rate_mm_h=row[13], + rain_event_mm=row[14], + rain_hourly_mm=row[15], + rain_daily_mm=row[16], + rain_weekly_mm=row[17], + rain_monthly_mm=row[18], + rain_yearly_mm=row[19], + indoor_temperature_c=row[20], + indoor_humidity_pct=row[21], + absolute_pressure_hpa=row[22], + relative_pressure_hpa=row[23], + metadata=row[24] or {}, + ) + for row in rows + ] + ) + + @contextmanager + def _connection(self) -> Iterator[object]: + try: + import psycopg + except ImportError as error: + raise WeatherStoreConfigurationError( + "Install dependencies with `python3 -m pip install -r requirements.txt`" + ) from error + + with psycopg.connect(self.config.database_url) as connection: + yield connection diff --git a/astrape_oracle/scripts/__init__.py b/astrape_oracle/scripts/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/astrape_oracle/scripts/daemons/__init__.py b/astrape_oracle/scripts/daemons/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/astrape_oracle/scripts/daemons/oracle_daemon.py b/astrape_oracle/scripts/daemons/oracle_daemon.py new file mode 100644 index 0000000..6639bf1 --- /dev/null +++ b/astrape_oracle/scripts/daemons/oracle_daemon.py @@ -0,0 +1,122 @@ +from __future__ import annotations + +import argparse +from dataclasses import dataclass +from os import environ +from sys import stderr +from time import sleep + +from gibil.classes.oracle_builder import EnergyOracleBuilder +from gibil.classes.forecast_config import EnergyForecastConfig +from astrape_ems.classes.env import EnvLoader +from astrape_oracle.classes.oracle_store import OracleStore +from astrape_ems.classes.store import SigenStore +from astrape_oracle.classes.weather.store import WeatherStore + + +@dataclass(frozen=True) +class OracleDaemonConfig: + poll_seconds: float + evaluate_forecasts: bool + evaluation_actual_window_minutes: float + evaluation_lookback_hours: float + evaluation_limit: int + + @classmethod + def from_env(cls) -> "OracleDaemonConfig": + return cls( + poll_seconds=float(environ.get("ASTRAPE_ORACLE_POLL_SECONDS", "300")), + evaluate_forecasts=environ.get( + "ASTRAPE_ORACLE_EVALUATE_FORECASTS", "1" + ).lower() + not in {"0", "false", "no"}, + evaluation_actual_window_minutes=float( + environ.get("ASTRAPE_ORACLE_EVALUATION_WINDOW_MINUTES", "5") + ), + evaluation_lookback_hours=float( + environ.get("ASTRAPE_ORACLE_EVALUATION_LOOKBACK_HOURS", "168") + ), + evaluation_limit=int(environ.get("ASTRAPE_ORACLE_EVALUATION_LIMIT", "1000")), + ) + + +class OracleDaemon: + """Periodically stores oracle projection curves for evaluation.""" + + def __init__( + self, + config: OracleDaemonConfig, + builder: EnergyOracleBuilder, + store: OracleStore, + ) -> None: + self.config = config + self.builder = builder + self.store = store + + @classmethod + def from_env(cls) -> "OracleDaemon": + return cls( + config=OracleDaemonConfig.from_env(), + builder=EnergyOracleBuilder( + weather_history=WeatherStore.from_env(), + plant_history=SigenStore.from_env(), + config=EnergyForecastConfig.from_env(), + ), + store=OracleStore.from_env(), + ) + + def run_once(self) -> int: + solar_run, load_run, net_run = self.builder.build() + saved_count = self.store.save_runs(solar_run, load_run, net_run) + if self.config.evaluate_forecasts: + from datetime import timedelta + + evaluated_count = self.store.evaluate_due_forecasts( + actual_window=timedelta( + minutes=self.config.evaluation_actual_window_minutes + ), + lookback=timedelta(hours=self.config.evaluation_lookback_hours), + limit=self.config.evaluation_limit, + ) + return saved_count + evaluated_count + return saved_count + + def run_forever(self) -> None: + self.store.initialize() + while True: + try: + saved_count = self.run_once() + print(f"stored_oracle_records={saved_count}", flush=True) + except Exception as error: + print(f"oracle_poll_error={error}", file=stderr, flush=True) + sleep(self.config.poll_seconds) + + +def main() -> None: + try: + EnvLoader().load() + args = parse_args() + daemon = OracleDaemon.from_env() + if args.once: + print(f"stored_oracle_records={daemon.run_once()}", flush=True) + return + daemon.run_forever() + except Exception as error: + print(f"oracle_daemon_startup_error={error}", file=stderr) + raise SystemExit(1) from error + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Store Astrape oracle projection curves." + ) + parser.add_argument( + "--once", + action="store_true", + help="Save one set of oracle curves and exit.", + ) + return parser.parse_args() + + +if __name__ == "__main__": + main() diff --git a/astrape_oracle/scripts/daemons/weather_forecast_daemon.py b/astrape_oracle/scripts/daemons/weather_forecast_daemon.py new file mode 100644 index 0000000..d25c737 --- /dev/null +++ b/astrape_oracle/scripts/daemons/weather_forecast_daemon.py @@ -0,0 +1,137 @@ +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from os import environ +from sys import stderr +from time import sleep + +from astrape_ems.classes.env import EnvLoader +from astrape_oracle.classes.weather.openmeteo import ( + OpenMeteoArchiveClient, + OpenMeteoClient, + WeatherBuilder, +) +from astrape_oracle.classes.weather.store import WeatherStore + + +@dataclass(frozen=True) +class DbDaemonConfig: + latitude: float + longitude: float + forecast_hours: int + truth_lookback_days: int + truth_end_delay_days: int + poll_seconds: int + + @classmethod + def from_env(cls) -> "DbDaemonConfig": + return cls( + latitude=float(_required_env("ASTRAPE_LATITUDE")), + longitude=float(_required_env("ASTRAPE_LONGITUDE")), + forecast_hours=int(environ.get("ASTRAPE_WEATHER_FORECAST_HOURS", "48")), + truth_lookback_days=int( + environ.get("ASTRAPE_WEATHER_TRUTH_LOOKBACK_DAYS", "14") + ), + truth_end_delay_days=int( + environ.get("ASTRAPE_WEATHER_TRUTH_END_DELAY_DAYS", "5") + ), + poll_seconds=int(environ.get("ASTRAPE_WEATHER_POLL_SECONDS", "3600")), + ) + + +class DbDaemon: + """Runs builder components that populate Astrape's database.""" + + def __init__( + self, + config: DbDaemonConfig, + weather_client: OpenMeteoClient, + archive_client: OpenMeteoArchiveClient, + weather_builder: WeatherBuilder, + weather_store: WeatherStore, + ) -> None: + self.config = config + self.weather_client = weather_client + self.archive_client = archive_client + self.weather_builder = weather_builder + self.weather_store = weather_store + + @classmethod + def from_env(cls) -> "DbDaemon": + return cls( + config=DbDaemonConfig.from_env(), + weather_client=OpenMeteoClient(), + archive_client=OpenMeteoArchiveClient(), + weather_builder=WeatherBuilder(), + weather_store=WeatherStore.from_env(), + ) + + def initialize(self) -> None: + self.weather_store.initialize() + + def run_once(self) -> tuple[int, int]: + raw_run = self.weather_client.fetch_forecast( + latitude=self.config.latitude, + longitude=self.config.longitude, + forecast_hours=self.config.forecast_hours, + ) + forecast_run = self.weather_builder.build_forecast_run( + source=raw_run.source, + latitude=raw_run.latitude, + longitude=raw_run.longitude, + points=raw_run.points, + issued_at=raw_run.issued_at, + ) + forecast_count = self.weather_store.save_forecast_run(forecast_run) + zero_hour_truth_count = self.weather_store.save_zero_hour_forecast_as_truth( + forecast_run + ) + + today = datetime.now(timezone.utc).date() + truth_end = today - timedelta(days=self.config.truth_end_delay_days) + truth_start = truth_end - timedelta(days=self.config.truth_lookback_days) + truth_points = self.archive_client.fetch_resolved_truth( + latitude=self.config.latitude, + longitude=self.config.longitude, + start_date=truth_start, + end_date=truth_end, + ) + archive_truth_count = self.weather_store.save_resolved_truth(truth_points) + + return forecast_count, zero_hour_truth_count + archive_truth_count + + def run_forever(self) -> None: + self.initialize() + while True: + forecast_count, truth_count = self.run_once() + print( + f"stored_weather_forecast_points={forecast_count} " + f"stored_weather_resolved_truth={truth_count}", + flush=True, + ) + sleep(self.config.poll_seconds) + + +def main() -> None: + try: + EnvLoader().load() + daemon = DbDaemon.from_env() + daemon.run_forever() + except Exception as error: + print(f"db_daemon_startup_error={error}", file=stderr) + raise SystemExit(1) from error + + +def _required_env(name: str) -> str: + value = environ.get(name) + if not value: + raise RuntimeError( + f"{name} is required. Set ASTRAPE_DATABASE_URL, " + "ASTRAPE_LATITUDE, and ASTRAPE_LONGITUDE before starting db_daemon." + ) + return value + + +if __name__ == "__main__": + main() diff --git a/astrape_oracle/scripts/daemons/weather_station_daemon.py b/astrape_oracle/scripts/daemons/weather_station_daemon.py new file mode 100644 index 0000000..086bc66 --- /dev/null +++ b/astrape_oracle/scripts/daemons/weather_station_daemon.py @@ -0,0 +1,113 @@ +from __future__ import annotations + +import argparse +from dataclasses import dataclass +from os import environ +from sys import stderr +from time import sleep + +from astrape_ems.classes.env import EnvLoader +from astrape_oracle.classes.weather.ecowitt import EcowittGw3000Client +from astrape_oracle.classes.weather.store import WeatherStore + + +@dataclass(frozen=True) +class WeatherStationDaemonConfig: + host: str + poll_seconds: float + timeout_seconds: float + + @classmethod + def from_env(cls) -> "WeatherStationDaemonConfig": + host = environ.get("ECOWITT_GW3000_HOST") + if not host: + raise RuntimeError("ECOWITT_GW3000_HOST is required") + + return cls( + host=host, + poll_seconds=float(environ.get("ECOWITT_GW3000_POLL_SECONDS", "60")), + timeout_seconds=float(environ.get("ECOWITT_GW3000_TIMEOUT_SECONDS", "5")), + ) + + +class WeatherStationDaemon: + """Polls local weather station observations into TimescaleDB.""" + + def __init__( + self, + config: WeatherStationDaemonConfig, + client: EcowittGw3000Client, + weather_store: WeatherStore, + ) -> None: + self.config = config + self.client = client + self.weather_store = weather_store + + @classmethod + def from_env(cls) -> "WeatherStationDaemon": + config = WeatherStationDaemonConfig.from_env() + return cls( + config=config, + client=EcowittGw3000Client( + host=config.host, + timeout_seconds=config.timeout_seconds, + ), + weather_store=WeatherStore.from_env(), + ) + + def initialize(self) -> None: + self.weather_store.initialize() + + def run_once(self) -> int: + observation = self.client.fetch_observation() + return self.weather_store.save_station_observation(observation) + + def run_forever(self) -> None: + self.initialize() + while True: + try: + saved_count = self.run_once() + print( + f"stored_weather_station_observations={saved_count}", + flush=True, + ) + except Exception as error: + print(f"weather_station_poll_error={error}", file=stderr, flush=True) + + sleep(self.config.poll_seconds) + + +def main() -> None: + try: + EnvLoader().load() + daemon = WeatherStationDaemon.from_env() + args = parse_args() + if args.once: + daemon.initialize() + saved_count = daemon.run_once() + print( + f"stored_weather_station_observations={saved_count}", + flush=True, + ) + return + + daemon.run_forever() + except Exception as error: + print(f"weather_station_daemon_startup_error={error}", file=stderr) + raise SystemExit(1) from error + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Poll local Ecowitt weather station observations into Astrape." + ) + parser.add_argument( + "--once", + action="store_true", + help="Initialize storage, save one observation, and exit.", + ) + return parser.parse_args() + + +if __name__ == "__main__": + main() diff --git a/astrape_oracle/scripts/debug/__init__.py b/astrape_oracle/scripts/debug/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/astrape_oracle/scripts/debug/debug_duplicates.py b/astrape_oracle/scripts/debug/debug_duplicates.py new file mode 100644 index 0000000..d846dd1 --- /dev/null +++ b/astrape_oracle/scripts/debug/debug_duplicates.py @@ -0,0 +1,31 @@ +#!/usr/bin/env python3 +"""Debug script to find duplicate target times in forecast data.""" + +from astrape_ems.classes.env import EnvLoader +from astrape_oracle.classes.weather.store import WeatherStore +from collections import defaultdict + +EnvLoader().load() + +store = WeatherStore.from_env() +dataset = store.load_display_dataset() + +# Group by (target_at, horizon_hours) to find duplicates +by_key = defaultdict(list) +for point in dataset.forecast_points: + key = (point.target_at, point.horizon_hours) + by_key[key].append(point) + +# Find duplicates +duplicates = {k: v for k, v in by_key.items() if len(v) > 1} + +print(f"\nTotal forecast points: {len(dataset.forecast_points)}") +print(f"Unique (target_at, horizon) pairs: {len(by_key)}") +print(f"Duplicate (target_at, horizon) pairs: {len(duplicates)}") + +if duplicates: + print("\nFirst 3 duplicates:") + for (target_at, horizon), points in list(duplicates.items())[:3]: + print(f"\n target_at={target_at}, horizon={horizon}h ({len(points)} points):") + for i, p in enumerate(points): + print(f" [{i}] issued_at={p.issued_at}, temp={p.temperature_c}, source={p.source}") diff --git a/astrape_oracle/scripts/debug/debug_energy_forecast.py b/astrape_oracle/scripts/debug/debug_energy_forecast.py new file mode 100644 index 0000000..b81027c --- /dev/null +++ b/astrape_oracle/scripts/debug/debug_energy_forecast.py @@ -0,0 +1,74 @@ +#!/usr/bin/env python3 +"""Debug baseline energy forecast curves.""" + +from __future__ import annotations + +import argparse +from datetime import timezone + +from gibil.classes.oracle_builder import EnergyForecastBuilder +from gibil.classes.forecast_config import EnergyForecastConfig +from astrape_ems.classes.env import EnvLoader +from astrape_ems.classes.store import SigenStore +from astrape_oracle.classes.weather.store import WeatherStore +from gibil.classes.models import PowerForecastPoint + + +def main() -> None: + EnvLoader().load() + args = parse_args() + solar_run, load_run, net_run = EnergyForecastBuilder( + weather_history=WeatherStore.from_env(), + plant_history=SigenStore.from_env(), + config=EnergyForecastConfig.from_env(), + ).build() + + print( + f"issued_at={net_run.issued_at.astimezone(timezone.utc).isoformat(timespec='seconds')}" + ) + print( + f"solar_model={solar_run.model_version} " + f"load_model={load_run.model_version} points={len(net_run.points)}" + ) + print( + "target_at solar_p10 solar_p50 solar_p90 " + "load_p10 load_p50 load_p90 net_p10 net_p50 net_p90" + ) + solar_by_target = _by_target(solar_run.points) + load_by_target = _by_target(load_run.points) + for point in net_run.points[: args.limit]: + solar_point = solar_by_target[point.target_at] + load_point = load_by_target[point.target_at] + print( + f"{point.target_at.astimezone(timezone.utc).isoformat(timespec='minutes'):25} " + f"{solar_point.p10_power_w:9.0f} " + f"{solar_point.p50_power_w:9.0f} " + f"{solar_point.p90_power_w:9.0f} " + f"{load_point.p10_power_w:8.0f} " + f"{load_point.p50_power_w:8.0f} " + f"{load_point.p90_power_w:8.0f} " + f"{point.p10_net_power_w:7.0f} " + f"{point.p50_net_power_w:7.0f} " + f"{point.p90_net_power_w:7.0f}" + ) + + +def _by_target(points: list[PowerForecastPoint]) -> dict[object, PowerForecastPoint]: + return {point.target_at: point for point in points} + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Print baseline solar/load/net forecast curves." + ) + parser.add_argument( + "--limit", + type=int, + default=24, + help="Number of forecast points to show. Defaults to 24.", + ) + return parser.parse_args() + + +if __name__ == "__main__": + main() diff --git a/astrape_oracle/scripts/debug/debug_oracle_evaluations.py b/astrape_oracle/scripts/debug/debug_oracle_evaluations.py new file mode 100644 index 0000000..9fb49cf --- /dev/null +++ b/astrape_oracle/scripts/debug/debug_oracle_evaluations.py @@ -0,0 +1,5 @@ +from astrape_oracle.scripts.evaluator import main + + +if __name__ == "__main__": + main() diff --git a/astrape_oracle/scripts/evaluator.py b/astrape_oracle/scripts/evaluator.py new file mode 100644 index 0000000..4a377e0 --- /dev/null +++ b/astrape_oracle/scripts/evaluator.py @@ -0,0 +1,102 @@ +from __future__ import annotations + +import argparse +from datetime import timedelta + +from astrape_ems.classes.env import EnvLoader +from astrape_oracle.classes.oracle_store import OracleStore + + +def main() -> None: + EnvLoader().load() + args = parse_args() + store = OracleStore.from_env() + + if args.evaluate: + evaluated_count = store.evaluate_due_forecasts( + actual_window=timedelta(minutes=args.actual_window_minutes), + lookback=timedelta(hours=args.lookback_hours), + limit=args.limit, + ) + print(f"evaluated_oracle_forecasts={evaluated_count}") + + if args.summary: + rows = store.load_evaluation_summary( + lookback=timedelta(hours=args.lookback_hours) + ) + print(f"oracle_evaluation_summary_rows={len(rows)}") + for row in rows: + print( + " ".join( + [ + f"kind={row['kind']}", + f"model={row['model_version']}", + f"horizon={row.get('horizon_label') or _format_horizon(row)}", + f"n={row['evaluated_count']}", + f"bias={_format_w(row['mean_error_w'])}", + f"mae={_format_w(row['mean_absolute_error_w'])}", + f"median_ae={_format_w(row['median_absolute_error_w'])}", + f"mape={_format_pct(row['mean_absolute_pct_error'])}", + f"coverage={_format_pct(row['interval_coverage'])}", + ] + ) + ) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Evaluate stored Astrape oracle predictions against Sigen actuals." + ) + parser.add_argument( + "--evaluate", + action="store_true", + help="Evaluate stored predictions whose target time has passed.", + ) + parser.add_argument( + "--summary", + action="store_true", + help="Print evaluation quality by kind/model/horizon bucket.", + ) + parser.add_argument( + "--actual-window-minutes", + type=float, + default=5, + help="Minutes after each target timestamp to average as realized actuals.", + ) + parser.add_argument( + "--lookback-hours", + type=float, + default=168, + help="Only evaluate/summarize predictions with target times this recent.", + ) + parser.add_argument( + "--limit", + type=int, + default=1000, + help="Maximum unevaluated predictions to process.", + ) + args = parser.parse_args() + if not args.evaluate and not args.summary: + args.evaluate = True + args.summary = True + return args + + +def _format_w(value: object) -> str: + if value is None: + return "n/a" + return f"{float(value):.0f}W" + + +def _format_horizon(row: dict[str, object]) -> str: + return f"{row['min_horizon_minutes']}-{row['max_horizon_minutes']}m" + + +def _format_pct(value: object) -> str: + if value is None: + return "n/a" + return f"{float(value) * 100:.1f}%" + + +if __name__ == "__main__": + main() diff --git a/astrape_oracle/scripts/tcn/__init__.py b/astrape_oracle/scripts/tcn/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/astrape_oracle/scripts/tcn/schedule.py b/astrape_oracle/scripts/tcn/schedule.py new file mode 100644 index 0000000..5ad65ab --- /dev/null +++ b/astrape_oracle/scripts/tcn/schedule.py @@ -0,0 +1,301 @@ +from __future__ import annotations + +import argparse +import json +import os +from dataclasses import asdict, dataclass, field +from datetime import datetime, timedelta, timezone +from pathlib import Path +from time import monotonic + +from astrape_ems.classes.env import EnvLoader +from astrape_ems.classes.store import SigenStore +from astrape_oracle.classes.weather.store import WeatherStore +from gibil.classes.usage_sequence_dataset import ( + UsageSequenceDatasetBuilder, + UsageSequenceDatasetConfig, +) +from gibil.classes.usage_training import train_model + + +PROJECT_ROOT = Path(__file__).resolve().parents[3] +DEFAULT_MODEL_PATH = PROJECT_ROOT / "models" / "usage_sequence_tcn_v1.pt" +DEFAULT_STATE_PATH = PROJECT_ROOT / "models" / "usage_sequence_training_state.json" +DEFAULT_LOG_PATH = PROJECT_ROOT / "logs" / "usage_sequence_training.jsonl" + + +@dataclass(frozen=True) +class TrainingRegimeConfig: + artifact_path: Path = DEFAULT_MODEL_PATH + state_path: Path = DEFAULT_STATE_PATH + log_path: Path = DEFAULT_LOG_PATH + full_epochs: int = 20 + finetune_epochs: int = 1 + batch_size: int = 32 + full_learning_rate: float = 0.001 + finetune_learning_rate: float = 0.0002 + seed: int = 7 + threshold_seconds: float = 10 * 60 + terminal_threshold_seconds: float = 60 * 60 + nightly_interval_days: float = 1.0 + max_full_retrain_interval_days: float = 7.0 + target_finetune_seconds: float = 5 * 60 + finetune_min_samples: int = 32 + + @classmethod + def from_env(cls) -> "TrainingRegimeConfig": + EnvLoader().load() + return cls( + artifact_path=Path( + os.environ.get( + "ASTRAPE_USAGE_SEQUENCE_ARTIFACT_PATH", + str(DEFAULT_MODEL_PATH), + ) + ), + state_path=Path( + os.environ.get( + "ASTRAPE_USAGE_SEQUENCE_TRAINING_STATE_PATH", + str(DEFAULT_STATE_PATH), + ) + ), + log_path=Path( + os.environ.get( + "ASTRAPE_USAGE_SEQUENCE_TRAINING_LOG_PATH", + str(DEFAULT_LOG_PATH), + ) + ), + full_epochs=int(os.environ.get("ASTRAPE_USAGE_SEQUENCE_FULL_EPOCHS", "20")), + finetune_epochs=int( + os.environ.get("ASTRAPE_USAGE_SEQUENCE_FINETUNE_EPOCHS", "1") + ), + batch_size=int(os.environ.get("ASTRAPE_USAGE_SEQUENCE_BATCH_SIZE", "32")), + full_learning_rate=float( + os.environ.get("ASTRAPE_USAGE_SEQUENCE_FULL_LR", "0.001") + ), + finetune_learning_rate=float( + os.environ.get("ASTRAPE_USAGE_SEQUENCE_FINETUNE_LR", "0.0002") + ), + seed=int(os.environ.get("ASTRAPE_USAGE_SEQUENCE_SEED", "7")), + threshold_seconds=float( + os.environ.get("ASTRAPE_USAGE_SEQUENCE_FULL_T_SECONDS", str(10 * 60)) + ), + terminal_threshold_seconds=float( + os.environ.get("ASTRAPE_USAGE_SEQUENCE_FULL_T2_SECONDS", str(60 * 60)) + ), + max_full_retrain_interval_days=float( + os.environ.get("ASTRAPE_USAGE_SEQUENCE_MAX_FULL_INTERVAL_DAYS", "7") + ), + target_finetune_seconds=float( + os.environ.get("ASTRAPE_USAGE_SEQUENCE_TARGET_FINETUNE_SECONDS", str(5 * 60)) + ), + finetune_min_samples=int( + os.environ.get("ASTRAPE_USAGE_SEQUENCE_FINETUNE_MIN_SAMPLES", "32") + ), + ) + + +@dataclass +class TrainingRegimeState: + last_full_training_seconds: float | None = None + last_full_sample_count: int | None = None + last_full_completed_at: str | None = None + last_finetune_completed_at: str | None = None + full_retrain_locked: bool = False + run_count: int = 0 + history: list[dict[str, object]] = field(default_factory=list) + + @classmethod + def load(cls, path: Path) -> "TrainingRegimeState": + if not path.exists(): + return cls() + return cls(**json.loads(path.read_text())) + + def save(self, path: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(asdict(self), indent=2, sort_keys=True) + "\n") + + +class UsageSequenceTrainingScheduler: + def __init__(self, config: TrainingRegimeConfig) -> None: + self.config = config + + def run(self, force_mode: str | None = None, dry_run: bool = False) -> dict[str, object]: + state = TrainingRegimeState.load(self.config.state_path) + decision = self._decision(state, force_mode=force_mode) + if dry_run: + event = { + "created_at": self._now_iso(), + "dry_run": True, + "decision": decision, + "state": asdict(state), + } + self._log_event(event) + print(json.dumps(event, sort_keys=True)) + return event + + dataset_config = UsageSequenceDatasetConfig.from_env() + builder = UsageSequenceDatasetBuilder( + config=dataset_config, + plant_history=SigenStore.from_env(), + weather_history=WeatherStore.from_env(), + ) + examples = builder.build(limit=decision["sample_limit"]) + if not examples: + raise SystemExit("No usage sequence examples available for scheduled training.") + + artifact_path = self.config.artifact_path + base_artifact_path = artifact_path if decision["mode"] == "finetune" else None + if decision["mode"] == "finetune" and not artifact_path.exists(): + decision = self._full_decision(reason="missing_artifact") + base_artifact_path = None + + started = monotonic() + result = train_model( + examples=examples, + builder=builder, + epochs=decision["epochs"], + batch_size=self.config.batch_size, + learning_rate=decision["learning_rate"], + artifact_path=artifact_path, + seed=self.config.seed, + base_artifact_path=base_artifact_path, + mode=decision["mode"], + ) + training_seconds = monotonic() - started + + event = { + "created_at": self._now_iso(), + "mode": result.mode, + "reason": decision["reason"], + "training_seconds": training_seconds, + "sample_count": result.sample_count, + "training_examples": result.training_examples, + "validation_examples": result.validation_examples, + "final_train_loss": result.final_train_loss, + "final_validation_loss": result.final_validation_loss, + "artifact_path": str(result.artifact_path), + "next_full_interval_days": self._full_interval_days(training_seconds), + } + self._update_state(state, event) + state.save(self.config.state_path) + self._log_event(event) + print(json.dumps(event, sort_keys=True)) + return event + + def _decision( + self, + state: TrainingRegimeState, + force_mode: str | None, + ) -> dict[str, object]: + if force_mode == "full": + return self._full_decision(reason="forced_full") + if force_mode == "finetune": + return self._finetune_decision(state, reason="forced_finetune") + if state.full_retrain_locked: + return self._finetune_decision(state, reason="full_retrain_locked") + if state.last_full_training_seconds is None: + return self._full_decision(reason="no_full_training_history") + if self._due_for_full_retrain(state): + return self._full_decision(reason="full_retrain_due") + return self._finetune_decision(state, reason="nightly_finetune") + + def _due_for_full_retrain(self, state: TrainingRegimeState) -> bool: + if state.last_full_completed_at is None: + return True + completed_at = datetime.fromisoformat(state.last_full_completed_at) + due_after = completed_at + timedelta( + days=self._full_interval_days(state.last_full_training_seconds) + ) + return datetime.now(timezone.utc) >= due_after + + def _full_decision(self, reason: str) -> dict[str, object]: + return { + "mode": "full", + "reason": reason, + "sample_limit": None, + "epochs": self.config.full_epochs, + "learning_rate": self.config.full_learning_rate, + } + + def _finetune_decision( + self, + state: TrainingRegimeState, + reason: str, + ) -> dict[str, object]: + return { + "mode": "finetune", + "reason": reason, + "sample_limit": self._finetune_sample_limit(state), + "epochs": self.config.finetune_epochs, + "learning_rate": self.config.finetune_learning_rate, + } + + def _finetune_sample_limit(self, state: TrainingRegimeState) -> int | None: + if not state.last_full_training_seconds or not state.last_full_sample_count: + return None + samples_per_second = state.last_full_sample_count / state.last_full_training_seconds + sample_limit = round(samples_per_second * self.config.target_finetune_seconds) + return max(self.config.finetune_min_samples, sample_limit) + + def _full_interval_days(self, training_seconds: float | None) -> float: + if training_seconds is None or training_seconds <= self.config.threshold_seconds: + return self.config.nightly_interval_days + if training_seconds >= self.config.terminal_threshold_seconds: + return self.config.max_full_retrain_interval_days + span = self.config.terminal_threshold_seconds - self.config.threshold_seconds + progress = (training_seconds - self.config.threshold_seconds) / span + return self.config.nightly_interval_days + progress * ( + self.config.max_full_retrain_interval_days - self.config.nightly_interval_days + ) + + def _update_state( + self, + state: TrainingRegimeState, + event: dict[str, object], + ) -> None: + state.run_count += 1 + if event["mode"] == "full": + state.last_full_training_seconds = float(event["training_seconds"]) + state.last_full_sample_count = int(event["sample_count"]) + state.last_full_completed_at = str(event["created_at"]) + if state.last_full_training_seconds >= self.config.terminal_threshold_seconds: + state.full_retrain_locked = True + else: + state.last_finetune_completed_at = str(event["created_at"]) + state.history.append(event) + state.history = state.history[-30:] + + def _log_event(self, event: dict[str, object]) -> None: + self.config.log_path.parent.mkdir(parents=True, exist_ok=True) + with self.config.log_path.open("a") as handle: + handle.write(json.dumps(event, sort_keys=True) + "\n") + + def _now_iso(self) -> str: + return datetime.now(timezone.utc).isoformat() + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Adaptive full-retrain/fine-tune scheduler for the usage TCN." + ) + parser.add_argument("--dry-run", action="store_true") + parser.add_argument( + "--force-mode", + choices=("full", "finetune"), + default=None, + help="Override the adaptive decision for this run.", + ) + return parser.parse_args() + + +def main() -> None: + EnvLoader().load() + args = parse_args() + UsageSequenceTrainingScheduler(TrainingRegimeConfig.from_env()).run( + force_mode=args.force_mode, + dry_run=args.dry_run, + ) + + +if __name__ == "__main__": + main() diff --git a/astrape_oracle/scripts/tcn/train.py b/astrape_oracle/scripts/tcn/train.py new file mode 100644 index 0000000..9cc8a45 --- /dev/null +++ b/astrape_oracle/scripts/tcn/train.py @@ -0,0 +1,158 @@ +from __future__ import annotations + +import argparse +from os import environ +from pathlib import Path + +from astrape_ems.classes.env import EnvLoader +from astrape_ems.classes.store import SigenStore +from astrape_oracle.classes.weather.store import WeatherStore +from gibil.classes.usage_sequence_dataset import ( + UsageSequenceDatasetBuilder, + UsageSequenceDatasetConfig, +) +from gibil.classes.usage_training import train_model + + +def main() -> None: + EnvLoader().load() + environ.setdefault("TORCH_CPP_LOG_LEVEL", "ERROR") + args = parse_args() + config = UsageSequenceDatasetConfig.from_env() + builder = UsageSequenceDatasetBuilder( + config=config, + plant_history=SigenStore.from_env(), + weather_history=WeatherStore.from_env(), + ) + examples = builder.build(limit=args.limit) + + print(f"usage_sequence_examples={len(examples)}") + print( + "minimum_history_hours=" + f"{builder.max_past_hours + config.future_hours}" + ) + print(f"past_features={len(builder.past_feature_names)}") + for scale in config.past_scales: + print( + f"past_scale={scale.name} " + f"hours={scale.hours} " + f"step_seconds={scale.step_seconds} " + f"steps={builder.past_steps(scale)}" + ) + print(f"future_steps={builder.future_steps}") + print(f"future_features={len(builder.future_feature_names)}") + + if examples: + first = examples[0] + last = examples[-1] + print(f"first_issued_at={first.issued_at.isoformat()}") + print(f"last_issued_at={last.issued_at.isoformat()}") + for name, rows in first.past_by_scale.items(): + print(f"first_past_{name}_shape={len(rows)}x{len(rows[0])}") + token_count = sum( + len(tokens) + for tokens in first.past_tokens_by_scale[name] + ) + print(f"first_past_{name}_tokens={token_count}") + print( + "first_future_feature_shape=" + f"{len(first.future_features)}x{len(first.future_features[0])}" + ) + print( + "first_future_tokens=" + f"{sum(len(tokens) for tokens in first.future_tokens)}" + ) + print(f"first_targets={len(first.targets)}") + print( + "first_target_preview=" + + ",".join(f"{value:.0f}" for value in first.targets[:8]) + ) + + if args.dry_run: + return + + if not examples: + raise SystemExit("No usage sequence examples available for training yet.") + + result = train_model( + examples=examples, + builder=builder, + epochs=args.epochs, + batch_size=args.batch_size, + learning_rate=args.learning_rate, + artifact_path=args.artifact_path, + seed=args.seed, + base_artifact_path=args.base_artifact_path, + mode=args.mode, + ) + print(f"training_mode={result.mode}") + print(f"training_examples={result.training_examples}") + print(f"validation_examples={result.validation_examples}") + if result.final_train_loss is not None: + print(f"final_train_pinball_loss={result.final_train_loss:.4f}") + if result.final_validation_loss is not None: + print(f"final_validation_pinball_loss={result.final_validation_loss:.4f}") + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Build training windows for the sequence usage oracle." + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="Only build examples and print dataset shape.", + ) + parser.add_argument( + "--limit", + type=int, + default=None, + help="Optional maximum number of examples to build.", + ) + parser.add_argument( + "--epochs", + type=int, + default=20, + help="Training epochs.", + ) + parser.add_argument( + "--batch-size", + type=int, + default=32, + help="Training batch size.", + ) + parser.add_argument( + "--learning-rate", + type=float, + default=0.001, + help="Adam learning rate.", + ) + parser.add_argument( + "--artifact-path", + type=Path, + default=Path("models/usage_sequence_tcn_v1.pt"), + help="Where to save the trained TCN artifact.", + ) + parser.add_argument( + "--base-artifact-path", + type=Path, + default=None, + help="Existing artifact to continue training from.", + ) + parser.add_argument( + "--mode", + choices=("full", "finetune"), + default="full", + help="Training mode recorded in the saved artifact.", + ) + parser.add_argument( + "--seed", + type=int, + default=7, + help="Deterministic shuffle seed.", + ) + return parser.parse_args() + + +if __name__ == "__main__": + main() diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..ebd30f3 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,13 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "astrape-oracle" +version = "0.1.0" +description = "Forecasting runtime: weather ingest, oracle snapshots, TCN training" +requires-python = ">=3.12" +dependencies = ["gibil", "astrape-ems", "psycopg[binary]>=3.2,<4"] + +[tool.setuptools.packages.find] +include = ["astrape_oracle*"]