glenn schrooyenandClaude Opus 5
5184a2cfc2
The deadband is a one-way ratchet: any resting point inside it holds indefinitely. Import and export are separate registers on the meter, so a loop resting at +14 W bills 0.34 kWh/day while behaving perfectly. target_grid_w (default -10 W) moves that residue onto the export register. Worst billed rest point drops from 15 W to under 5 W. Behaviour is unchanged at target_grid_w: 0. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
125 lines
5.3 KiB
Python
125 lines
5.3 KiB
Python
"""Runnable check for the control law. `python3 test_control.py`
|
|
|
|
No framework, no fixtures - it needs to run on a tech's laptop and in CI with
|
|
nothing installed. Every assert here corresponds to a rule that exists because
|
|
its absence caused an observed failure on real hardware.
|
|
|
|
If you change control.py, run this. If it fails, the inverter would have done
|
|
something you did not intend.
|
|
"""
|
|
|
|
import sys
|
|
|
|
from app.control import Tuning, compute, maintenance_charge_floor, peak_at_risk
|
|
|
|
T = Tuning()
|
|
fails = []
|
|
|
|
|
|
def check(name, cond):
|
|
if cond:
|
|
print(f" ok {name}")
|
|
else:
|
|
print(f" FAIL {name}")
|
|
fails.append(name)
|
|
|
|
|
|
print("control law")
|
|
|
|
# Deadband: inside meter noise, hold exactly - do not drift.
|
|
d = compute(prev_w=900, grid_w=10, actual_w=900, tuning=T)
|
|
check("deadband holds the command", d.target_w == 900 and d.reason == "deadband")
|
|
|
|
d = compute(prev_w=900, grid_w=20, actual_w=900, tuning=T)
|
|
check("outside deadband it acts", d.target_w != 900)
|
|
|
|
# Proportional: 0 + 0.6*500 = 300
|
|
d = compute(prev_w=0, grid_w=500, actual_w=0, tuning=T)
|
|
check("proportional step (gain 0.6)", d.target_w == 300)
|
|
|
|
# Sign: exporting (negative grid) must CHARGE (negative target).
|
|
d = compute(prev_w=0, grid_w=-500, actual_w=0, tuning=T)
|
|
check("export drives charging", d.target_w == -300)
|
|
|
|
# Clamp
|
|
d = compute(prev_w=1900, grid_w=1000, actual_w=1900, tuning=Tuning(max_w=2000, slew_w=5000))
|
|
check("clamped to max_w", d.target_w == 2000)
|
|
|
|
# Slew: from 0 with a huge error, no more than slew_w in one cycle.
|
|
d = compute(prev_w=0, grid_w=5000, actual_w=0, tuning=Tuning(max_w=5000, slew_w=1000))
|
|
check("slew limits one cycle", d.target_w == 1000)
|
|
|
|
# Saturation needs DURATION: one diverging cycle must NOT freeze.
|
|
t = Tuning(saturation_w=500, saturation_cycles=3)
|
|
d1 = compute(prev_w=2000, grid_w=500, actual_w=1000, tuning=t, sat_count=0)
|
|
check("one saturated cycle does not freeze", not d1.frozen and d1.sat_count == 1)
|
|
d2 = compute(prev_w=2000, grid_w=500, actual_w=1000, tuning=t, sat_count=d1.sat_count)
|
|
d3 = compute(prev_w=2000, grid_w=500, actual_w=1000, tuning=t, sat_count=d2.sat_count)
|
|
check("three consecutive saturated cycles freeze", d3.frozen)
|
|
check("freeze forbids raising magnitude", d3.target_w <= 2000)
|
|
|
|
# ...and one good cycle clears the counter immediately.
|
|
d4 = compute(prev_w=2000, grid_w=500, actual_w=1990, tuning=t, sat_count=3)
|
|
check("counter resets when tracking resumes", d4.sat_count == 0 and not d4.frozen)
|
|
|
|
# Freeze must still allow the magnitude to FALL (that is the escape route).
|
|
d = compute(prev_w=2000, grid_w=-800, actual_w=1000, tuning=t, sat_count=3)
|
|
check("freeze still allows magnitude to fall", d.target_w < 2000)
|
|
|
|
# Maintenance shaping (charge-only, cheap-window floor) is no longer this
|
|
# function's business - it is expressed as limit claims. See test_arbiter.py.
|
|
|
|
# Quantisation
|
|
d = compute(prev_w=0, grid_w=7, actual_w=0, tuning=Tuning(deadband_w=1, step_w=10))
|
|
check("quantised to step_w", d.target_w % 10 == 0)
|
|
|
|
print("capacity tariff")
|
|
check("no forecast means no cap", maintenance_charge_floor(2500, None, 3500) == 2500)
|
|
check("headroom caps the charge", maintenance_charge_floor(2500, 2000, 3500) == 1500)
|
|
check("no headroom means no charge", maintenance_charge_floor(2500, 4000, 3500) == 0)
|
|
check("peak risk detected", peak_at_risk(4000, 3500) is True)
|
|
check("peak risk off without forecast", peak_at_risk(None, 3500) is False)
|
|
|
|
print("behaviour: 2 kW load step converges")
|
|
# Closed-loop sim. The plant is modelled as first-order-ish: it moves most of
|
|
# the way to the command each cycle (measured: 94 % by 3.3 s against a ~5 s
|
|
# cycle). House load steps by 2000 W at t=0.
|
|
prev, actual, sat, load = 0.0, 0.0, 0, 2000.0
|
|
cycles = 0
|
|
for i in range(12):
|
|
grid = load - actual # what the meter sees
|
|
d = compute(prev, grid, actual, T, sat)
|
|
prev, sat = d.target_w, d.sat_count
|
|
actual = actual + 0.94 * (prev - actual) # plant follows
|
|
cycles += 1
|
|
if abs(load - actual) < T.deadband_w:
|
|
break
|
|
check(f"converges within deadband in {cycles} cycles (<=6)", cycles <= 6)
|
|
check("no overshoot past the load", actual <= load + T.deadband_w)
|
|
|
|
print("grid bias: the deadband must not rest on the import register")
|
|
# The billed asymmetry: import and export are separate registers, so a resting
|
|
# point inside the deadband on the import side is paid for every second it
|
|
# holds. 14 W held all day is 0.34 kWh.
|
|
T0 = Tuning(target_grid_w=0.0)
|
|
TB = Tuning(target_grid_w=-10.0)
|
|
check("unbiased, +14 W import rests forever",
|
|
compute(500.0, 14.0, 500.0, T0).reason == "deadband")
|
|
d = compute(500.0, 14.0, 500.0, TB)
|
|
check("biased, the same +14 W is corrected", d.reason != "deadband" and d.target_w > 500.0)
|
|
check("biased, a small export rests", compute(500.0, -10.0, 500.0, TB).reason == "deadband")
|
|
check("biased, the band still ends before -25 W",
|
|
compute(500.0, -30.0, 500.0, TB).reason != "deadband")
|
|
# Worst-case billed leak: the band is [bias - deadband, bias + deadband], so it
|
|
# drops from 15 W to 5 W. Set target_grid_w to -deadband_w to remove it entirely,
|
|
# at the cost of giving that much away as export.
|
|
check("worst billed rest point falls from 15 W to under 5 W",
|
|
compute(500.0, 4.9, 500.0, TB).reason == "deadband"
|
|
and compute(500.0, 5.0, 500.0, TB).reason != "deadband")
|
|
|
|
print()
|
|
if fails:
|
|
print(f"{len(fails)} FAILED: {', '.join(fails)}")
|
|
sys.exit(1)
|
|
print("all checks passed")
|