"""Synthetic stand-in for the real iSolarCloud client. Lets Phases 1-4 (persistence, scheduler, web UI, costing) be built and demoed end-to-end before a real developer App Key/Secret exists (see SETUP.md). `RealISolarCloudClient` in client.py implements the exact same `fetch_interval_data` interface so the scheduler can swap between them via `settings.use_mock_isolarcloud` with no other code changes. """ from __future__ import annotations import math import random from datetime import datetime, timedelta from zoneinfo import ZoneInfo from app.isolarcloud.schemas import IntervalDataPoint BNE = ZoneInfo("Australia/Brisbane") RESOLUTION_MINUTES = 5 def _synthetic_point(ts: datetime) -> IntervalDataPoint: local = ts.astimezone(BNE) hour = local.hour + local.minute / 60 # Crude seasonal scaling: more sun and longer days in the QLD summer. seasonal_factor = 1.0 + 0.25 * math.cos(math.radians((local.month - 1) * 30)) daylight = max(0.0, math.sin(math.pi * (hour - 6) / 12)) if 6 <= hour <= 18 else 0.0 peak_kw = 5.0 * seasonal_factor # Seed on the timestamp (not wall-clock random) so re-generating the same # interval twice is stable-ish; harmless either way since the DB # deduplicates on (ts, source) and never overwrites an existing row. rng = random.Random(int(ts.timestamp()) // 60) generation_kw = peak_kw * daylight * rng.uniform(0.85, 1.15) base_load_kw = 0.4 morning_bump = 0.8 * math.exp(-((hour - 7) ** 2) / 2) evening_bump = 1.6 * math.exp(-((hour - 19) ** 2) / 2) load_kw = (base_load_kw + morning_bump + evening_bump) * rng.uniform(0.9, 1.1) step_hours = RESOLUTION_MINUTES / 60 generation_kwh = generation_kw * step_hours load_kwh = load_kw * step_hours import_kwh = max(0.0, load_kwh - generation_kwh) export_kwh = max(0.0, generation_kwh - load_kwh) return IntervalDataPoint( ts=ts, generation_kwh=round(generation_kwh, 4), load_kwh=round(load_kwh, 4), import_kwh=round(import_kwh, 4), export_kwh=round(export_kwh, 4), resolution_minutes=RESOLUTION_MINUTES, ) class MockISolarCloudClient: def fetch_interval_data(self, since: datetime, until: datetime) -> list[IntervalDataPoint]: points = [] ts = since step = timedelta(minutes=RESOLUTION_MINUTES) while ts < until: points.append(_synthetic_point(ts)) ts += step return points