353f154626
All checks were successful
Build and Push Docker Image / build-and-push (push) Successful in 20s
New card (hidden entirely for solar-only plants) shows: an SOC curve over Today/Last 3 days/This week, and an estimate of what the battery actually saved -- discharged energy valued at the import rate it avoided, minus charged energy valued at the feed-in credit given up, both matched against the current plan's own time-of-use windows. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
363 lines
14 KiB
Python
363 lines
14 KiB
Python
from datetime import datetime, time
|
|
from zoneinfo import ZoneInfo
|
|
|
|
import pytest
|
|
|
|
from app.costing.engine import BatteryReading, IntervalReading, evaluate, evaluate_battery_value
|
|
from app.costing.tariff_models import (
|
|
ALL_DAYS,
|
|
FeedInComponent,
|
|
TariffPlan,
|
|
TieredFeedIn,
|
|
TimeWindow,
|
|
UsageComponent,
|
|
find_usage_coverage_gaps,
|
|
)
|
|
|
|
BNE = ZoneInfo("Australia/Brisbane")
|
|
|
|
|
|
def bne(y, m, d, hh, mm=0) -> datetime:
|
|
return datetime(y, m, d, hh, mm, tzinfo=BNE)
|
|
|
|
|
|
def test_flat_plan_with_feed_in():
|
|
plan = TariffPlan(
|
|
plan_id="flat-1",
|
|
retailer="Test Co",
|
|
plan_name="Flat Saver",
|
|
supply_charge_c_per_day=100,
|
|
usage_components=[UsageComponent(label="flat", rate_c_per_kwh=30)],
|
|
feed_in_tariffs=[FeedInComponent(label="fit", rate_c_per_kwh=8)],
|
|
)
|
|
readings = [
|
|
IntervalReading(ts=bne(2026, 1, 1, 10), import_kwh=5, export_kwh=2),
|
|
IntervalReading(ts=bne(2026, 1, 1, 20), import_kwh=3, export_kwh=0),
|
|
IntervalReading(ts=bne(2026, 1, 2, 10), import_kwh=4, export_kwh=1),
|
|
]
|
|
|
|
result = evaluate(plan, readings)
|
|
|
|
assert result.days_with_data == 2
|
|
assert result.usage_breakdown[0].kwh == pytest.approx(12)
|
|
assert result.feed_in_breakdown[0].kwh == pytest.approx(3)
|
|
assert result.supply_charge_dollars == pytest.approx(2.00)
|
|
assert result.total_dollars == pytest.approx(5.36)
|
|
assert result.modeling_notes == []
|
|
|
|
|
|
def test_simple_time_of_use_plan():
|
|
peak = UsageComponent(
|
|
label="peak",
|
|
rate_c_per_kwh=40,
|
|
time_windows=[TimeWindow(days=ALL_DAYS, start_time=time(16, 0), end_time=time(20, 0))],
|
|
)
|
|
offpeak = UsageComponent(
|
|
label="offpeak",
|
|
rate_c_per_kwh=20,
|
|
time_windows=[TimeWindow(days=ALL_DAYS, start_time=time(20, 0), end_time=time(16, 0))],
|
|
)
|
|
plan = TariffPlan(
|
|
plan_id="tou-1",
|
|
retailer="Test Co",
|
|
plan_name="Time of Use",
|
|
supply_charge_c_per_day=80,
|
|
usage_components=[peak, offpeak],
|
|
feed_in_tariffs=[FeedInComponent(label="fit", rate_c_per_kwh=10)],
|
|
)
|
|
# No structural gaps: peak + offpeak windows partition the full 24h.
|
|
assert find_usage_coverage_gaps([peak, offpeak]) == []
|
|
|
|
readings = [
|
|
IntervalReading(ts=bne(2026, 3, 10, 6), import_kwh=2, export_kwh=0),
|
|
IntervalReading(ts=bne(2026, 3, 10, 12), import_kwh=3, export_kwh=5),
|
|
IntervalReading(ts=bne(2026, 3, 10, 18), import_kwh=4, export_kwh=0),
|
|
IntervalReading(ts=bne(2026, 3, 10, 22), import_kwh=1, export_kwh=0),
|
|
]
|
|
|
|
result = evaluate(plan, readings)
|
|
|
|
by_label = {c.label: c for c in result.usage_breakdown}
|
|
assert by_label["offpeak"].kwh == pytest.approx(6)
|
|
assert by_label["peak"].kwh == pytest.approx(4)
|
|
assert result.feed_in_breakdown[0].kwh == pytest.approx(5)
|
|
assert result.supply_charge_dollars == pytest.approx(0.80)
|
|
assert result.total_dollars == pytest.approx(3.10)
|
|
|
|
|
|
def test_seasonal_time_of_use_plan():
|
|
def peak(months, rate):
|
|
return UsageComponent(
|
|
label=f"peak-{months[0]}",
|
|
rate_c_per_kwh=rate,
|
|
season_months=months,
|
|
time_windows=[TimeWindow(days=ALL_DAYS, start_time=time(16, 0), end_time=time(20, 0))],
|
|
)
|
|
|
|
def offpeak(months, rate):
|
|
return UsageComponent(
|
|
label=f"offpeak-{months[0]}",
|
|
rate_c_per_kwh=rate,
|
|
season_months=months,
|
|
time_windows=[TimeWindow(days=ALL_DAYS, start_time=time(20, 0), end_time=time(16, 0))],
|
|
)
|
|
|
|
summer = [12, 1, 2]
|
|
winter = [3, 4, 5, 6, 7, 8, 9, 10, 11]
|
|
plan = TariffPlan(
|
|
plan_id="seasonal-tou-1",
|
|
retailer="Test Co",
|
|
plan_name="Seasonal TOU",
|
|
supply_charge_c_per_day=90,
|
|
usage_components=[
|
|
peak(summer, 50),
|
|
offpeak(summer, 22),
|
|
peak(winter, 35),
|
|
offpeak(winter, 18),
|
|
],
|
|
)
|
|
|
|
readings = [
|
|
IntervalReading(ts=bne(2026, 1, 15, 18), import_kwh=2, export_kwh=0), # summer peak
|
|
IntervalReading(ts=bne(2026, 1, 15, 8), import_kwh=3, export_kwh=0), # summer offpeak
|
|
IntervalReading(ts=bne(2026, 6, 15, 18), import_kwh=2, export_kwh=0), # winter peak
|
|
IntervalReading(ts=bne(2026, 6, 15, 8), import_kwh=3, export_kwh=0), # winter offpeak
|
|
]
|
|
|
|
result = evaluate(plan, readings)
|
|
|
|
assert result.days_with_data == 2
|
|
assert result.supply_charge_dollars == pytest.approx(1.80)
|
|
# 2*0.50 + 3*0.22 + 2*0.35 + 3*0.18 = 1.00 + 0.66 + 0.70 + 0.54 = 2.90
|
|
assert sum(c.cost_dollars for c in result.usage_breakdown) == pytest.approx(2.90)
|
|
assert result.total_dollars == pytest.approx(4.70)
|
|
assert result.modeling_notes == []
|
|
|
|
|
|
def test_time_of_use_feed_in_tariff():
|
|
plan = TariffPlan(
|
|
plan_id="tou-fit-1",
|
|
retailer="Test Co",
|
|
plan_name="Solar Soak",
|
|
supply_charge_c_per_day=70,
|
|
usage_components=[UsageComponent(label="flat", rate_c_per_kwh=25)],
|
|
feed_in_tariffs=[
|
|
FeedInComponent(
|
|
label="midday-soak",
|
|
rate_c_per_kwh=5,
|
|
time_windows=[TimeWindow(days=ALL_DAYS, start_time=time(10, 0), end_time=time(15, 0))],
|
|
),
|
|
FeedInComponent(
|
|
label="standard",
|
|
rate_c_per_kwh=12,
|
|
time_windows=[TimeWindow(days=ALL_DAYS, start_time=time(15, 0), end_time=time(10, 0))],
|
|
),
|
|
],
|
|
)
|
|
readings = [
|
|
IntervalReading(ts=bne(2026, 2, 1, 8), import_kwh=1, export_kwh=0.5),
|
|
IntervalReading(ts=bne(2026, 2, 1, 12), import_kwh=0, export_kwh=4),
|
|
IntervalReading(ts=bne(2026, 2, 1, 18), import_kwh=2, export_kwh=0),
|
|
]
|
|
|
|
result = evaluate(plan, readings)
|
|
|
|
by_label = {c.label: c for c in result.feed_in_breakdown}
|
|
assert by_label["midday-soak"].kwh == pytest.approx(4)
|
|
assert by_label["standard"].kwh == pytest.approx(0.5)
|
|
assert result.total_dollars == pytest.approx(1.19)
|
|
|
|
|
|
def test_gap_in_usage_windows_is_reported_and_uncharged():
|
|
peak_only = UsageComponent(
|
|
label="peak",
|
|
rate_c_per_kwh=40,
|
|
time_windows=[TimeWindow(days=ALL_DAYS, start_time=time(16, 0), end_time=time(20, 0))],
|
|
)
|
|
plan = TariffPlan(
|
|
plan_id="broken-1",
|
|
retailer="Test Co",
|
|
plan_name="Missing Offpeak",
|
|
supply_charge_c_per_day=0,
|
|
usage_components=[peak_only],
|
|
)
|
|
assert find_usage_coverage_gaps([peak_only]) != []
|
|
|
|
readings = [
|
|
IntervalReading(ts=bne(2026, 1, 1, 8), import_kwh=5, export_kwh=0), # falls in the gap
|
|
IntervalReading(ts=bne(2026, 1, 1, 18), import_kwh=2, export_kwh=0), # matches peak
|
|
]
|
|
|
|
result = evaluate(plan, readings)
|
|
|
|
assert result.usage_breakdown[0].kwh == pytest.approx(2)
|
|
assert result.total_dollars == pytest.approx(0.80)
|
|
assert any("did not match any tariff window" in note for note in result.modeling_notes)
|
|
|
|
|
|
def test_split_time_window_under_shared_label_is_not_double_counted():
|
|
# "Shoulder" needs two disjoint windows (7am-4pm and 9pm-10pm) under one
|
|
# label -- this must sum once, not double-count the cost.
|
|
shoulder_1 = UsageComponent(
|
|
label="Shoulder",
|
|
rate_c_per_kwh=36,
|
|
time_windows=[TimeWindow(days=ALL_DAYS, start_time=time(7, 0), end_time=time(16, 0))],
|
|
)
|
|
shoulder_2 = UsageComponent(
|
|
label="Shoulder",
|
|
rate_c_per_kwh=36,
|
|
time_windows=[TimeWindow(days=ALL_DAYS, start_time=time(21, 0), end_time=time(22, 0))],
|
|
)
|
|
peak = UsageComponent(
|
|
label="Peak",
|
|
rate_c_per_kwh=48,
|
|
time_windows=[TimeWindow(days=ALL_DAYS, start_time=time(16, 0), end_time=time(21, 0))],
|
|
)
|
|
offpeak = UsageComponent(
|
|
label="Off-peak",
|
|
rate_c_per_kwh=31,
|
|
time_windows=[TimeWindow(days=ALL_DAYS, start_time=time(22, 0), end_time=time(7, 0))],
|
|
)
|
|
plan = TariffPlan(
|
|
plan_id="split-shoulder-1",
|
|
retailer="Test Co",
|
|
plan_name="Split Shoulder",
|
|
supply_charge_c_per_day=0,
|
|
usage_components=[peak, shoulder_1, shoulder_2, offpeak],
|
|
)
|
|
assert find_usage_coverage_gaps([peak, shoulder_1, shoulder_2, offpeak]) == []
|
|
|
|
readings = [
|
|
IntervalReading(ts=bne(2026, 1, 1, 9), import_kwh=2, export_kwh=0), # shoulder window 1
|
|
IntervalReading(ts=bne(2026, 1, 1, 21, 30), import_kwh=3, export_kwh=0), # shoulder window 2
|
|
]
|
|
result = evaluate(plan, readings)
|
|
|
|
shoulder_rows = [c for c in result.usage_breakdown if c.label == "Shoulder"]
|
|
assert len(shoulder_rows) == 1
|
|
assert shoulder_rows[0].kwh == pytest.approx(5)
|
|
assert result.total_dollars == pytest.approx(5 * 0.36)
|
|
|
|
|
|
def test_tiered_daily_feed_in_tariff():
|
|
# Modeled on Alinta's QLD "Solar Balance" FiT: first 10 kWh/day at 9c,
|
|
# the remainder at 2c -- a *daily total export* threshold, not a
|
|
# time-of-use one.
|
|
plan = TariffPlan(
|
|
plan_id="tiered-fit-1",
|
|
retailer="Test Co",
|
|
plan_name="Solar Balance",
|
|
supply_charge_c_per_day=100,
|
|
usage_components=[UsageComponent(label="flat", rate_c_per_kwh=20)],
|
|
tiered_feed_in=TieredFeedIn(daily_threshold_kwh=10, rate_below_c_per_kwh=9, rate_above_c_per_kwh=2),
|
|
)
|
|
readings = [
|
|
# Day 1: 7 kWh exported total, all below the 10 kWh threshold.
|
|
IntervalReading(ts=bne(2026, 1, 1, 8), import_kwh=2, export_kwh=3),
|
|
IntervalReading(ts=bne(2026, 1, 1, 12), import_kwh=1, export_kwh=4),
|
|
# Day 2: 15 kWh exported total -- 10 below threshold, 5 above.
|
|
IntervalReading(ts=bne(2026, 1, 2, 8), import_kwh=1, export_kwh=5),
|
|
IntervalReading(ts=bne(2026, 1, 2, 12), import_kwh=1, export_kwh=10),
|
|
]
|
|
|
|
result = evaluate(plan, readings)
|
|
|
|
by_label = {c.label: c for c in result.feed_in_breakdown}
|
|
assert by_label["First 10 kWh/day"].kwh == pytest.approx(17) # 7 + min(15,10)
|
|
assert by_label["Remaining balance"].kwh == pytest.approx(5) # max(15-10, 0)
|
|
# credit = 17*0.09 + 5*0.02 = 1.53 + 0.10 = 1.63
|
|
# usage = (2+1+1+1)*0.20 = 1.00; supply = 2 days * $1.00 = $2.00
|
|
# total = 1.00 + 2.00 - 1.63 = 1.37
|
|
assert result.total_dollars == pytest.approx(1.37)
|
|
assert result.modeling_notes == []
|
|
|
|
|
|
def test_conditional_discount_kept_out_of_headline_total():
|
|
plan = TariffPlan(
|
|
plan_id="discount-1",
|
|
retailer="Test Co",
|
|
plan_name="Pay On Time",
|
|
supply_charge_c_per_day=0,
|
|
usage_components=[UsageComponent(label="flat", rate_c_per_kwh=100)],
|
|
discounts=[
|
|
{"description": "Pay on time discount", "percent": 10, "applies_to": "usage", "auto_apply": False}
|
|
],
|
|
)
|
|
readings = [IntervalReading(ts=bne(2026, 1, 1, 12), import_kwh=10, export_kwh=0)]
|
|
|
|
result = evaluate(plan, readings)
|
|
|
|
assert result.total_dollars == pytest.approx(10.00)
|
|
assert result.potential_conditional_savings_dollars == pytest.approx(1.00)
|
|
assert any("Pay on time discount" in note for note in result.modeling_notes)
|
|
|
|
|
|
def test_battery_value_flat_plan():
|
|
plan = TariffPlan(
|
|
plan_id="flat-1",
|
|
retailer="Test Co",
|
|
plan_name="Flat Saver",
|
|
supply_charge_c_per_day=100,
|
|
usage_components=[UsageComponent(label="flat", rate_c_per_kwh=30)],
|
|
feed_in_tariffs=[FeedInComponent(label="fit", rate_c_per_kwh=8)],
|
|
)
|
|
readings = [
|
|
# Discharged 2 kWh at 9am -> avoided 2 * 30c = $0.60 of import.
|
|
BatteryReading(ts=bne(2026, 1, 1, 9), charge_kwh=0, discharge_kwh=2),
|
|
# Charged 3 kWh at 1pm -> gave up 3 * 8c = $0.24 of export credit.
|
|
BatteryReading(ts=bne(2026, 1, 1, 13), charge_kwh=3, discharge_kwh=0),
|
|
]
|
|
|
|
result = evaluate_battery_value(plan, readings)
|
|
|
|
assert result.discharged_kwh == pytest.approx(2)
|
|
assert result.charged_kwh == pytest.approx(3)
|
|
assert result.avoided_import_dollars == pytest.approx(0.60)
|
|
assert result.foregone_export_dollars == pytest.approx(0.24)
|
|
assert result.net_value_dollars == pytest.approx(0.36)
|
|
assert result.days_with_data == 1
|
|
|
|
|
|
def test_battery_value_time_of_use_plan_matches_each_readings_own_window():
|
|
plan = TariffPlan(
|
|
plan_id="tou-1",
|
|
retailer="Test Co",
|
|
plan_name="TOU Plan",
|
|
supply_charge_c_per_day=0,
|
|
usage_components=[
|
|
UsageComponent(
|
|
label="Peak", rate_c_per_kwh=50,
|
|
time_windows=[TimeWindow(days=ALL_DAYS, start_time=time(16, 0), end_time=time(20, 0))],
|
|
),
|
|
UsageComponent(
|
|
label="Off-peak", rate_c_per_kwh=20,
|
|
time_windows=[TimeWindow(days=ALL_DAYS, start_time=time(20, 0), end_time=time(16, 0))],
|
|
),
|
|
],
|
|
)
|
|
readings = [
|
|
# Discharging at 6pm (peak) is worth more than at 10am (off-peak).
|
|
BatteryReading(ts=bne(2026, 1, 1, 18), charge_kwh=0, discharge_kwh=1),
|
|
BatteryReading(ts=bne(2026, 1, 1, 10), charge_kwh=0, discharge_kwh=1),
|
|
]
|
|
|
|
result = evaluate_battery_value(plan, readings)
|
|
|
|
assert result.avoided_import_dollars == pytest.approx(0.50 + 0.20)
|
|
|
|
|
|
def test_battery_value_tiered_feed_in_uses_above_threshold_rate():
|
|
plan = TariffPlan(
|
|
plan_id="tiered-1",
|
|
retailer="Test Co",
|
|
plan_name="Tiered FiT",
|
|
supply_charge_c_per_day=0,
|
|
usage_components=[UsageComponent(label="flat", rate_c_per_kwh=30)],
|
|
tiered_feed_in=TieredFeedIn(daily_threshold_kwh=10, rate_below_c_per_kwh=9, rate_above_c_per_kwh=2),
|
|
)
|
|
readings = [BatteryReading(ts=bne(2026, 1, 1, 13), charge_kwh=5, discharge_kwh=0)]
|
|
|
|
result = evaluate_battery_value(plan, readings)
|
|
|
|
# Approximated at the above-threshold (marginal) rate, not the capped one.
|
|
assert result.foregone_export_dollars == pytest.approx(5 * 0.02)
|