"""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