84c4922bbc
FastAPI app that pulls interval usage from Sungrow iSolarCloud (OAuth2), caches AER Consumer Data Right electricity plan data, and compares actual metered usage against every available plan to estimate real annual cost (TOU, tiered feed-in, and flat tariffs). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
120 lines
3.9 KiB
Python
120 lines
3.9 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.dialects.sqlite import insert as sqlite_insert
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.costing.engine import IntervalReading
|
|
from app.costing.tariff_models import TariffPlan
|
|
from app.db.models import (
|
|
CurrentPlanConfigRow,
|
|
HouseholdConfigRow,
|
|
IntervalReadingRow,
|
|
PlanCacheRow,
|
|
SyncStateRow,
|
|
)
|
|
|
|
|
|
_INSERT_BATCH_SIZE = 500 # keeps well under SQLite's default SQL variable limit
|
|
|
|
|
|
def upsert_interval_readings(session: Session, rows: list[dict]) -> int:
|
|
"""Insert interval readings, ignoring rows that already exist for the
|
|
same (ts, source) so repeated polling / restarts never duplicate data."""
|
|
inserted = 0
|
|
for i in range(0, len(rows), _INSERT_BATCH_SIZE):
|
|
batch = rows[i : i + _INSERT_BATCH_SIZE]
|
|
stmt = sqlite_insert(IntervalReadingRow).values(batch)
|
|
stmt = stmt.on_conflict_do_nothing(index_elements=["ts", "source"])
|
|
result = session.execute(stmt)
|
|
inserted += result.rowcount or 0
|
|
return inserted
|
|
|
|
|
|
def get_interval_readings(session: Session, start: datetime, end: datetime) -> list[IntervalReading]:
|
|
stmt = (
|
|
select(IntervalReadingRow)
|
|
.where(IntervalReadingRow.ts >= start, IntervalReadingRow.ts < end)
|
|
.order_by(IntervalReadingRow.ts)
|
|
)
|
|
return [
|
|
IntervalReading(ts=row.ts, import_kwh=row.import_kwh, export_kwh=row.export_kwh)
|
|
for row in session.scalars(stmt)
|
|
]
|
|
|
|
|
|
def get_earliest_reading_ts(session: Session) -> datetime | None:
|
|
return session.scalar(select(func.min(IntervalReadingRow.ts)))
|
|
|
|
|
|
def get_usage_series(session: Session, start: datetime, end: datetime) -> list[IntervalReadingRow]:
|
|
stmt = (
|
|
select(IntervalReadingRow)
|
|
.where(IntervalReadingRow.ts >= start, IntervalReadingRow.ts < end)
|
|
.order_by(IntervalReadingRow.ts)
|
|
)
|
|
return list(session.scalars(stmt))
|
|
|
|
|
|
def upsert_plan(session: Session, plan: TariffPlan, raw_detail: dict, fuel_type: str = "ELECTRICITY") -> None:
|
|
row = session.get(PlanCacheRow, plan.plan_id)
|
|
values = dict(
|
|
retailer=plan.retailer,
|
|
plan_name=plan.plan_name,
|
|
fuel_type=fuel_type,
|
|
distributors_json=json.dumps(plan.distributors),
|
|
last_refreshed=datetime.now(timezone.utc),
|
|
raw_detail_json=json.dumps(raw_detail),
|
|
parsed_tariff_json=plan.model_dump_json(),
|
|
)
|
|
if row is None:
|
|
session.add(PlanCacheRow(plan_id=plan.plan_id, **values))
|
|
else:
|
|
for k, v in values.items():
|
|
setattr(row, k, v)
|
|
|
|
|
|
def get_cached_plans(session: Session) -> list[TariffPlan]:
|
|
return [TariffPlan.model_validate_json(row.parsed_tariff_json) for row in session.scalars(select(PlanCacheRow))]
|
|
|
|
|
|
def get_current_plan(session: Session) -> TariffPlan | None:
|
|
row = session.get(CurrentPlanConfigRow, 1)
|
|
if row is None:
|
|
return None
|
|
return TariffPlan.model_validate_json(row.tariff_json)
|
|
|
|
|
|
def set_current_plan(session: Session, retailer_name: str, plan_name: str, plan: TariffPlan) -> None:
|
|
row = session.get(CurrentPlanConfigRow, 1)
|
|
if row is None:
|
|
row = CurrentPlanConfigRow(id=1)
|
|
session.add(row)
|
|
row.retailer_name = retailer_name
|
|
row.plan_name = plan_name
|
|
row.tariff_json = plan.model_dump_json()
|
|
|
|
|
|
def get_household_config(session: Session) -> HouseholdConfigRow:
|
|
row = session.get(HouseholdConfigRow, 1)
|
|
if row is None:
|
|
row = HouseholdConfigRow(id=1)
|
|
session.add(row)
|
|
session.flush()
|
|
return row
|
|
|
|
|
|
def get_sync_state(session: Session, key: str) -> str | None:
|
|
row = session.get(SyncStateRow, key)
|
|
return row.value if row else None
|
|
|
|
|
|
def set_sync_state(session: Session, key: str, value: str) -> None:
|
|
row = session.get(SyncStateRow, key)
|
|
if row is None:
|
|
session.add(SyncStateRow(key=key, value=value))
|
|
else:
|
|
row.value = value
|