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>
215 lines
9.6 KiB
Python
215 lines
9.6 KiB
Python
"""Maps a raw AER "Get Generic Plan Detail" JSON payload into the common
|
|
`TariffPlan` shape used by the costing engine.
|
|
|
|
Verified against real plan data while building this (AGL/EnergyAustralia
|
|
QLD plans): `electricityContract.tariffPeriod[]` holds one block per
|
|
season, each with `rateBlockUType` of "singleRate" or "timeOfUseRates" (an
|
|
"ALL" season shows as a start/end date spanning the full year, e.g.
|
|
"07-01".."06-30", which is treated as no seasonal restriction). Tiered/block
|
|
or demand-charge structures aren't represented by this shape and are marked
|
|
not-modelable rather than silently mis-costed.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import time as time_
|
|
|
|
from app.costing.tariff_models import (
|
|
ConditionalDiscount,
|
|
FeedInComponent,
|
|
TariffPlan,
|
|
TieredFeedIn,
|
|
TimeWindow,
|
|
UsageComponent,
|
|
)
|
|
|
|
|
|
def _parse_time(hhmm: str) -> time_:
|
|
hh, mm = (int(x) for x in hhmm.split(":"))
|
|
return time_(hh, mm)
|
|
|
|
|
|
def _end_time_exclusive(hhmm: str) -> time_:
|
|
"""AER end times are inclusive of their minute (e.g. "20:59" means up
|
|
to but not including 21:00) -- shift by one minute to get the exclusive
|
|
boundary `TimeWindow` expects."""
|
|
hh, mm = (int(x) for x in hhmm.split(":"))
|
|
total = (hh * 60 + mm + 1) % (24 * 60)
|
|
return time_(total // 60, total % 60)
|
|
|
|
|
|
def _season_months(start_mmdd: str, end_mmdd: str) -> list[int] | None:
|
|
"""A tariffPeriod's startDate/endDate ("MM-DD") approximate to whole
|
|
months here (the engine's TimeContext is month-granular, not
|
|
day-of-month-granular). A span covering all 12 months is treated as
|
|
"all year" (None) rather than an explicit list."""
|
|
start_month = int(start_mmdd.split("-")[0])
|
|
end_month = int(end_mmdd.split("-")[0])
|
|
months = []
|
|
m = start_month
|
|
for _ in range(12):
|
|
months.append(m)
|
|
if m == end_month:
|
|
break
|
|
m = m % 12 + 1
|
|
return None if len(months) >= 12 else months
|
|
|
|
|
|
def _time_windows(entries: list[dict]) -> list[TimeWindow]:
|
|
return [
|
|
TimeWindow(days=tou["days"], start_time=_parse_time(tou["startTime"]), end_time=_end_time_exclusive(tou["endTime"]))
|
|
for tou in entries
|
|
]
|
|
|
|
|
|
def _rate_cents(rates: list[dict]) -> tuple[float, bool]:
|
|
"""Returns (cents_per_kwh, is_tiered). Only the first tier is used for
|
|
tiered/inclining-block rate structures -- callers should mark the plan
|
|
not-modelable when is_tiered is True, since later tiers are ignored."""
|
|
return float(rates[0]["unitPrice"]) * 100, len(rates) > 1
|
|
|
|
|
|
def map_aer_plan(list_entry: dict, detail: dict) -> TariffPlan:
|
|
plan_id = list_entry["planId"]
|
|
retailer = list_entry.get("brandName") or list_entry.get("brand", "")
|
|
plan_name = detail.get("displayName", plan_id)
|
|
distributors = list_entry.get("geography", {}).get("distributors", [])
|
|
ec = detail.get("electricityContract", {})
|
|
|
|
notes: list[str] = []
|
|
modelable = True
|
|
usage_components: list[UsageComponent] = []
|
|
supply_charges_c: list[float] = []
|
|
|
|
for period in ec.get("tariffPeriod", []):
|
|
season_months = _season_months(period.get("startDate", "01-01"), period.get("endDate", "12-31"))
|
|
rate_type = period.get("rateBlockUType")
|
|
supply_charges_c.append(float(period.get("dailySupplyCharge", 0) or 0) * 100)
|
|
|
|
if rate_type == "singleRate":
|
|
block = period["singleRate"]
|
|
rate_c, tiered = _rate_cents(block["rates"])
|
|
usage_components.append(
|
|
UsageComponent(label=block.get("displayName", "Usage"), rate_c_per_kwh=rate_c, season_months=season_months)
|
|
)
|
|
if tiered:
|
|
modelable = False
|
|
notes.append(f"'{block.get('displayName', 'Usage')}' has tiered/step pricing; only the first tier is modeled.")
|
|
elif rate_type == "timeOfUseRates":
|
|
for tou_block in period["timeOfUseRates"]:
|
|
rate_c, tiered = _rate_cents(tou_block["rates"])
|
|
usage_components.append(
|
|
UsageComponent(
|
|
label=tou_block.get("displayName", tou_block.get("type", "TOU")),
|
|
rate_c_per_kwh=rate_c,
|
|
time_windows=_time_windows(tou_block.get("timeOfUse", [])),
|
|
season_months=season_months,
|
|
)
|
|
)
|
|
if tiered:
|
|
modelable = False
|
|
notes.append(f"'{tou_block.get('displayName', 'TOU')}' has tiered/step pricing; only the first tier is modeled.")
|
|
else:
|
|
modelable = False
|
|
notes.append(f"Tariff structure '{rate_type}' is not modeled (e.g. tiered/block or demand pricing).")
|
|
|
|
if ec.get("controlledLoad"):
|
|
notes.append(
|
|
"This plan also has a separately-metered controlled load tariff, not included "
|
|
"(usage data here is whole-house, not sub-metered)."
|
|
)
|
|
|
|
feed_in_tariffs: list[FeedInComponent] = []
|
|
tiered_feed_in: TieredFeedIn | None = None
|
|
retailer_fits = [fit for fit in ec.get("solarFeedInTariff", []) if fit.get("payerType") == "RETAILER"]
|
|
# Government/legacy premium schemes aren't available to new or existing
|
|
# standard sign-ups; only the retailer's own standard feed-in tariff(s)
|
|
# apply to a household comparing plans now.
|
|
single_tariff_fits = [fit for fit in retailer_fits if fit.get("tariffUType") == "singleTariff"]
|
|
|
|
def _has_volume(fit: dict) -> bool:
|
|
return "volume" in fit["singleTariff"]["rates"][0]
|
|
|
|
if len(single_tariff_fits) == 2 and _has_volume(single_tariff_fits[0]) and not _has_volume(single_tariff_fits[1]):
|
|
# A daily-volume-based feed-in tariff (e.g. "first 10kWh/day at 8c,
|
|
# thereafter at 4c") is represented in this data as two separate
|
|
# singleTariff entries rather than as tiers within one -- the first
|
|
# entry's "volume" is the daily kWh threshold, and the second
|
|
# (uncapped) entry is the remainder rate. Without this, both were
|
|
# treated as independent flat components and the first one in the
|
|
# list silently won every match, crediting 100% of exported energy
|
|
# at the (higher) capped-tier rate.
|
|
below_c, _ = _rate_cents(single_tariff_fits[0]["singleTariff"]["rates"])
|
|
above_c, _ = _rate_cents(single_tariff_fits[1]["singleTariff"]["rates"])
|
|
tiered_feed_in = TieredFeedIn(
|
|
daily_threshold_kwh=float(single_tariff_fits[0]["singleTariff"]["rates"][0]["volume"]),
|
|
rate_below_c_per_kwh=below_c,
|
|
rate_above_c_per_kwh=above_c,
|
|
)
|
|
else:
|
|
for fit in single_tariff_fits:
|
|
rate_c, tiered = _rate_cents(fit["singleTariff"]["rates"])
|
|
feed_in_tariffs.append(FeedInComponent(label=fit.get("displayName", "Feed-in tariff"), rate_c_per_kwh=rate_c))
|
|
if tiered:
|
|
modelable = False
|
|
notes.append(f"'{fit.get('displayName', 'Feed-in tariff')}' has tiered/step pricing; only the first tier is modeled.")
|
|
|
|
for fit in retailer_fits:
|
|
if fit.get("tariffUType") == "singleTariff":
|
|
continue # handled above
|
|
elif fit.get("tariffUType") == "timeVaryingTariffs":
|
|
for block in fit.get("timeVaryingTariffs", []):
|
|
rate_c, tiered = _rate_cents(block["rates"])
|
|
feed_in_tariffs.append(
|
|
FeedInComponent(
|
|
label=block.get("displayName", "Feed-in tariff"),
|
|
rate_c_per_kwh=rate_c,
|
|
# Unlike usage-side timeOfUseRates blocks (which key their
|
|
# windows as "timeOfUse"), the CDR solarFeedInTariff
|
|
# timeVaryingTariffs shape keys them as "timeVariations" --
|
|
# using the wrong key here silently made every time-varying
|
|
# feed-in tariff match all day (empty time_windows == "always").
|
|
time_windows=_time_windows(block.get("timeVariations", [])),
|
|
)
|
|
)
|
|
else:
|
|
notes.append(f"Feed-in tariff structure '{fit.get('tariffUType')}' not modeled; solar export credit may be understated.")
|
|
|
|
discounts: list[ConditionalDiscount] = []
|
|
for d in ec.get("discounts", []):
|
|
method = d.get("methodUType")
|
|
sub = d.get(method, {}) if method else {}
|
|
percent = float(sub["rate"]) * 100 if "rate" in sub else None
|
|
amount = float(sub["amount"]) if "amount" in sub else None
|
|
discounts.append(
|
|
ConditionalDiscount(
|
|
description=d.get("displayName", d.get("description", "Discount")),
|
|
percent=percent,
|
|
amount_dollars=amount,
|
|
applies_to="usage",
|
|
auto_apply=(d.get("type") == "GUARANTEED"),
|
|
)
|
|
)
|
|
|
|
for incentive in ec.get("incentives", []):
|
|
label = incentive.get("displayName") or incentive.get("description", "")
|
|
notes.append(f"One-off sign-up incentive not included in the cost estimate: {label}")
|
|
|
|
supply_charge_c_per_day = supply_charges_c[0] if supply_charges_c else 0.0
|
|
if len(set(supply_charges_c)) > 1:
|
|
notes.append("Supply charge varies by season in the source data; using the first season's rate.")
|
|
|
|
return TariffPlan(
|
|
plan_id=plan_id,
|
|
retailer=retailer,
|
|
plan_name=plan_name,
|
|
supply_charge_c_per_day=supply_charge_c_per_day,
|
|
usage_components=usage_components,
|
|
feed_in_tariffs=feed_in_tariffs,
|
|
tiered_feed_in=tiered_feed_in,
|
|
discounts=discounts,
|
|
distributors=distributors,
|
|
source="aer",
|
|
modelable=modelable and bool(usage_components),
|
|
modeling_notes=notes,
|
|
)
|