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>
86 lines
3.1 KiB
Python
86 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
from collections import defaultdict
|
|
from datetime import datetime, timedelta, timezone
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from fastapi import APIRouter, HTTPException, Query
|
|
|
|
from app.db.repositories import get_earliest_reading_ts, get_usage_series
|
|
from app.db.session import get_session
|
|
|
|
router = APIRouter(prefix="/api/usage", tags=["usage"])
|
|
|
|
BNE = ZoneInfo("Australia/Brisbane")
|
|
|
|
|
|
def _resolve_start(session, days: str, end: datetime) -> datetime:
|
|
if days == "all":
|
|
return get_earliest_reading_ts(session) or end
|
|
try:
|
|
n = int(days)
|
|
except ValueError:
|
|
raise HTTPException(400, "days must be an integer or 'all'")
|
|
if n < 1:
|
|
raise HTTPException(400, "days must be >= 1")
|
|
return end - timedelta(days=n)
|
|
|
|
|
|
def _bucket_key(local_date, granularity: str) -> str:
|
|
if granularity == "month":
|
|
return local_date.replace(day=1).isoformat()
|
|
if granularity == "week":
|
|
return (local_date - timedelta(days=local_date.weekday())).isoformat()
|
|
return local_date.isoformat()
|
|
|
|
|
|
@router.get("/daily")
|
|
def daily_usage(days: str = Query("30")) -> list[dict]:
|
|
end = datetime.now(timezone.utc)
|
|
with get_session() as session:
|
|
start = _resolve_start(session, days, end)
|
|
rows = get_usage_series(session, start, end)
|
|
|
|
# Plotting one bar per day is unreadable (and slow) once the range
|
|
# stretches into years, so roll up to weekly/monthly buckets for long
|
|
# ranges -- this is a display concern only, the underlying stored data
|
|
# stays at full 5-minute resolution.
|
|
span_days = (end - start).days
|
|
granularity = "month" if span_days > 550 else "week" if span_days > 120 else "day"
|
|
|
|
by_bucket: dict[str, dict[str, float]] = defaultdict(lambda: {"generation_kwh": 0.0, "load_kwh": 0.0, "import_kwh": 0.0, "export_kwh": 0.0})
|
|
for row in rows:
|
|
key = _bucket_key(row.ts.astimezone(BNE).date(), granularity)
|
|
bucket = by_bucket[key]
|
|
bucket["generation_kwh"] += row.generation_kwh
|
|
bucket["load_kwh"] += row.load_kwh
|
|
bucket["import_kwh"] += row.import_kwh
|
|
bucket["export_kwh"] += row.export_kwh
|
|
|
|
return [{"date": day, **{k: round(v, 3) for k, v in totals.items()}} for day, totals in sorted(by_bucket.items())]
|
|
|
|
|
|
@router.get("/summary")
|
|
def usage_summary(days: str = Query("30")) -> dict:
|
|
end = datetime.now(timezone.utc)
|
|
with get_session() as session:
|
|
start = _resolve_start(session, days, end)
|
|
rows = get_usage_series(session, start, end)
|
|
if not rows:
|
|
return {"days_with_data": 0}
|
|
|
|
dates_seen: set[str] = set()
|
|
totals = {"generation_kwh": 0.0, "load_kwh": 0.0, "import_kwh": 0.0, "export_kwh": 0.0}
|
|
for row in rows:
|
|
dates_seen.add(row.ts.astimezone(BNE).date().isoformat())
|
|
totals["generation_kwh"] += row.generation_kwh
|
|
totals["load_kwh"] += row.load_kwh
|
|
totals["import_kwh"] += row.import_kwh
|
|
totals["export_kwh"] += row.export_kwh
|
|
|
|
return {
|
|
"days_with_data": len(dates_seen),
|
|
"first_day": min(dates_seen),
|
|
"last_day": max(dates_seen),
|
|
**{k: round(v, 2) for k, v in totals.items()},
|
|
}
|