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>
137 lines
5.1 KiB
Python
137 lines
5.1 KiB
Python
"""Common tariff-shape schema.
|
|
|
|
Both AER-sourced plans (via aer_mapper.py) and the user's manually-entered
|
|
current plan are represented as a single `TariffPlan` shape, so the costing
|
|
engine only needs one evaluation path regardless of where a plan came from.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import date, datetime, time
|
|
from typing import Literal
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
DAYS: list[str] = ["MON", "TUE", "WED", "THU", "FRI", "SAT", "SUN"]
|
|
ALL_DAYS: list[str] = list(DAYS)
|
|
WEEKDAYS: list[str] = DAYS[:5]
|
|
WEEKEND: list[str] = DAYS[5:]
|
|
|
|
|
|
class TimeContext(BaseModel):
|
|
"""A (month, day-of-week, time-of-day) point used to match tariff windows.
|
|
|
|
Decoupled from a real `datetime` so the coverage validator can probe
|
|
synthetic points without constructing real calendar dates.
|
|
"""
|
|
|
|
month: int
|
|
day_code: str
|
|
time_of_day: time
|
|
|
|
@classmethod
|
|
def from_datetime(cls, dt: datetime) -> "TimeContext":
|
|
return cls(month=dt.month, day_code=DAYS[dt.weekday()], time_of_day=dt.time())
|
|
|
|
|
|
class TimeWindow(BaseModel):
|
|
days: list[str] = Field(default_factory=lambda: list(ALL_DAYS))
|
|
start_time: time
|
|
end_time: time # exclusive; if end_time == start_time the window covers all 24h
|
|
|
|
def contains(self, ctx: TimeContext) -> bool:
|
|
if ctx.day_code not in self.days:
|
|
return False
|
|
t = ctx.time_of_day
|
|
if self.start_time == self.end_time:
|
|
return True
|
|
if self.start_time < self.end_time:
|
|
return self.start_time <= t < self.end_time
|
|
# window wraps past midnight, e.g. 22:00-06:00
|
|
return t >= self.start_time or t < self.end_time
|
|
|
|
|
|
class UsageComponent(BaseModel):
|
|
label: str
|
|
rate_c_per_kwh: float
|
|
time_windows: list[TimeWindow] = Field(default_factory=list)
|
|
season_months: list[int] | None = None # 1-12; None = all year
|
|
|
|
def matches(self, ctx: TimeContext) -> bool:
|
|
if self.season_months is not None and ctx.month not in self.season_months:
|
|
return False
|
|
if not self.time_windows:
|
|
return True
|
|
return any(w.contains(ctx) for w in self.time_windows)
|
|
|
|
|
|
class FeedInComponent(UsageComponent):
|
|
pass
|
|
|
|
|
|
class TieredFeedIn(BaseModel):
|
|
"""A feed-in tariff paid at one rate up to a daily export threshold and
|
|
a different (usually lower) rate beyond it -- e.g. Alinta's "Solar
|
|
Balance" QLD plans: the first 10 kWh exported each day at 9c/kWh, the
|
|
remainder at 2c/kWh. This depends on each calendar day's *total* export
|
|
(not the time of day), so it's evaluated separately from the
|
|
time-of-use `feed_in_tariffs` list -- a plan should use one or the
|
|
other, not both.
|
|
"""
|
|
|
|
daily_threshold_kwh: float
|
|
rate_below_c_per_kwh: float
|
|
rate_above_c_per_kwh: float
|
|
|
|
|
|
class ConditionalDiscount(BaseModel):
|
|
description: str
|
|
percent: float | None = None
|
|
amount_dollars: float | None = None
|
|
applies_to: Literal["usage", "total"] = "total"
|
|
# True only for discounts that apply unconditionally (e.g. baked into the
|
|
# advertised rate already). Conditional discounts (pay-on-time, direct
|
|
# debit, first-year sign-up) default to False and are surfaced separately
|
|
# as a "potential additional saving" rather than folding into the
|
|
# headline comparison number.
|
|
auto_apply: bool = False
|
|
|
|
|
|
class TariffPlan(BaseModel):
|
|
plan_id: str
|
|
retailer: str
|
|
plan_name: str
|
|
supply_charge_c_per_day: float
|
|
usage_components: list[UsageComponent]
|
|
feed_in_tariffs: list[FeedInComponent] = Field(default_factory=list)
|
|
tiered_feed_in: TieredFeedIn | None = None # takes precedence over feed_in_tariffs when set
|
|
discounts: list[ConditionalDiscount] = Field(default_factory=list)
|
|
distributors: list[str] = Field(default_factory=list)
|
|
source: Literal["aer", "manual"] = "aer"
|
|
effective_from: date | None = None
|
|
# Set False (with modeling_notes explaining why) for plan structures the
|
|
# engine can't represent, e.g. tiered/block or demand tariffs, so they
|
|
# show as "not modeled" in the UI rather than silently mis-costed.
|
|
modelable: bool = True
|
|
modeling_notes: list[str] = Field(default_factory=list)
|
|
|
|
|
|
def find_usage_coverage_gaps(components: list[UsageComponent]) -> list[str]:
|
|
"""Sample every (month, weekday, half-hour) combination and report any
|
|
point matched by zero or more than one component, so structurally broken
|
|
TOU/seasonal definitions are caught before evaluating real usage data.
|
|
"""
|
|
issues: list[str] = []
|
|
for month in range(1, 13):
|
|
for day_code in DAYS:
|
|
for half_hour in range(48):
|
|
hh, mm = divmod(half_hour * 30, 60)
|
|
ctx = TimeContext(month=month, day_code=day_code, time_of_day=time(hh, mm))
|
|
matches = [c.label for c in components if c.matches(ctx)]
|
|
if len(matches) == 0:
|
|
issues.append(f"no usage component matches month={month} {day_code} {hh:02d}:{mm:02d}")
|
|
elif len(matches) > 1:
|
|
issues.append(
|
|
f"overlapping usage components at month={month} {day_code} {hh:02d}:{mm:02d}: {matches}"
|
|
)
|
|
return issues
|