"""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 ] # AER's Consumer Data Right "Get Generic Plan Detail" unitPrice fields are # GST-EXCLUSIVE -- confirmed by comparing a real, currently-effective # EnergyAustralia contract's own quoted GST-inclusive rates against this # plan's cached AER data: every usage rate and the supply charge matched # the cached value multiplied by exactly 1.1, precisely (four independent # figures, not a coincidence), and a real AGL bill separately confirmed # the same convention (its line items are excl-GST, with GST added back as # one combined amount at the end). Feed-in/solar buy-back rates are the # exception -- confirmed GST-free in that same EnergyAustralia contract's # own terms ("The rate at which we purchase your electricity does not # include GST") -- so this multiplier applies only to usage rates and the # supply charge, never to feed_in_tariffs/tiered_feed_in. GST_MULTIPLIER = 1.1 def _rate_cents(rates: list[dict]) -> tuple[float, bool]: """Returns (cents_per_kwh, is_tiered), GST-exclusive as given by AER -- callers apply GST_MULTIPLIER themselves where it applies (usage rates, not feed-in). 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 * GST_MULTIPLIER) 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 * GST_MULTIPLIER, 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 * GST_MULTIPLIER, 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}") # Cost modeling can't tell you whether you actually qualify for a plan -- # AER's data sometimes states this explicitly (e.g. "must purchase an # eligible solar PV system from an Origin approved solar installer", # "new residential customers only"), and a plan ranking well on price # alone is useless if you don't meet its eligibility criteria. Surface # whatever AER provides so at least the stated restrictions are visible # in the comparison -- this can't catch requirements AER leaves # unstated (e.g. concession/seniors card plans with no eligibility # block at all), so a plan showing no note here isn't a guarantee of # eligibility, just that AER didn't document a restriction. for elig in ec.get("eligibility", []): description = elig.get("description") or elig.get("information") if description: notes.append(f"Eligibility requirement: {description}") 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, )