admin/goodwe-addon
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity
Files
5a8d832eba916d9cfd752e0abe94f8f2420dd771
goodwe-addon/goodwe_controller/test_refusal_units.py
T
glenn schrooyenandClaude Opus 5.5 016ed5a5a3
checks / unit (pull_request) Successful in 10m1s
Details
infra-pin / fresh (pull_request) Successful in 1s
Details
SAFETY-73: a battery that refuses, per inverter, with two or more 
SAFETY-52's hold applied per unit at N>1 (owner, 2026-09-27, option R).
A unit that gives < 100 W for 12 cycles in one direction while asked
> 200 W, or trips its stale-reading gate on a ~0 W reading, gets a
P_SAFETY `refused` unit limit closing that direction, so the allocator
hands its share to the others within their own limits. Probe: a +-200 W
pin every 15 min, one unit at a time, only while the site asks 200 W that
way. All units refusing: the site law holds 0 W. Exits reseed the site
target from the wire. INFO plus one WARNING (REFUSES_ABOVE_CUT_OUT) above
the configured cut-out; SAFETY-42/43 no longer report a refusing unit.
N=1 unchanged.

test_refusal_units.py: cases a/b/c on flat and jitter registers against
the old behaviour, probes, recovery, exits; mutations U1-U13 all red.
TEST-19 no-fault campaign: 0 engagements in 5280 runs.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-28 00:13:26 +02:00

481 lines
20 KiB
Python
Raw Blame History

"""Runnable check for SAFETY-73. `python3 test_refusal_units.py`
SAFETY-52 per inverter. With two or more inverters, a battery that refuses
(gives < 100 W for 12 cycles while asked > 500 W, or trips the stale-reading
gate on a ~0 W reading) is closed in that direction by its own P_SAFETY limit,
so the allocator hands its share to the others. It is probed at 200 W every
15 minutes, one inverter at a time. When every inverter refuses the same way
the site holds 0 W. When a refusal ends the site target restarts from the watts
on the wire. DECISIONS/SAFETY-73 has the record.
Drives the REAL `Controller` (read_inputs + cycle, the real freshness gate,
allocator, OPT-09 pack claim and SAFETY-42 detector) with two inverters in
simulated time. At the cut-out a battery reads -4 W for any request >= -4 W
and accepts charge. Two register shapes: FLAT at -4 W (the 09-06 capture) and
JITTER +-3 W (the gate cannot arm). `OLD` runs the same fixture with the
per-unit refusal switched off, which is 652097a's behaviour at N>1.
Mutations (MUTATION.md, `python3 -B`, copies purged of `__pycache__`), graded
over this file / test_delivery / test_refusal. Every one is red here:
# mutation reds
--- --------------------------------------------------------- -------
U1 main.py: the per-unit engage clause never fires 7/2/0
U2 main.py: the N>1 stale-reading seed returns at once 14/0/0
U3 main.py: no `refused` claim is filed 13/2/0
U4 main.py: a probe never pins (always the closed limit) 7/0/0
U5 main.py: a probe pin stands whatever the site target 1/0/0
U6 main.py: no reseed of the site target when a refusal ends 1/0/0
U7 main.py: no site hold when every inverter refuses 2/0/0
U8 main.py: the other-direction exit removed (charge counter) 1/0/0
U9 main.py: the delivery exit removed 3/0/0
U10 main.py: probes are no longer one at a time 1/0/0
U11 main.py: the above-cut-out WARNING condition inverted 1/0/0
U12 main.py: SAFETY-42's account no longer sees a small ask 1/0/0
U13 main.py: a unit engages above `saturation_w`, not PROBE_W 1/0/0
On 652097a itself this file stops at the first sample (`Inverter` has no
`refused`); its OLD arms reproduce that tree's N>1 behaviour instead, and
every PRECONDITION line is graded against them.
"""
import asyncio
import logging
import sys
from datetime import datetime, timedelta, timezone
from app import control
from app.control import PROBE_EVERY_S, PROBE_S, PROBE_W
from app.main import Controller
from app.arbiter import P_LOOP, P_PEAK, P_SAFETY, Claim, resolve
import app.main as m
fails = []
def check(name, cond):
print(f" {'ok ' if cond else 'FAIL'} {name}")
if not cond:
fails.append(name)
TICK = 5.0
CUT_W = -4.0
H = 3600.0
BASE = datetime(2026, 9, 6, 19, 30, tzinfo=timezone.utc)
UNITS = ("a", "b")
class _Store:
def __init__(self):
from app.store import DEFAULTS
self.data = dict(DEFAULTS)
def set(self, k, v):
self.data[k] = v
def get_time(self, *a):
return None
class _Pub:
enabled = False
def publish(self, *a, **k):
pass
def close(self):
pass
class _Hass:
"""A house, one meter, and two batteries that follow unless cut out."""
def __init__(self, load, cut, soc, jitter):
self.load, self.cut, self.soc, self.jitter = load, cut, soc, jitter
self.cmd = {u: 0.0 for u in UNITS}
self.t = 0.0
def batt(self, u):
c = self.cmd[u]
if self.cut[u](self.t) and c >= CUT_W:
if self.jitter:
return CUT_W + ((int(self.t / TICK) % 3) - 1) * 3.0
return CUT_W
return float(round(c))
def grid(self):
return self.load(self.t) - sum(self.batt(u) for u in UNITS)
async def limits(self, e):
return (-2000.0, 2000.0, 1.0)
async def call(self, *a):
return True
async def number(self, e, invert=False):
if e == "sensor.grid":
return self.grid()
u = e[-1]
if e.startswith("sensor.soc"):
v = self.soc[u]
return v(self.t) if callable(v) else v
if e.startswith("sensor.batt"):
return self.batt(u)
return None
async def set_number(self, e, v):
self.cmd[e[-1]] = float(v)
return True
def _off(c):
"""652097a's N>1 behaviour: nothing refuses per unit."""
c._unit_refusals = lambda: None
c._unit_refusal_start = lambda *a, **k: None
def run(load, cut, dur, *, soc=None, jitter=False, old=False, events=None,
extra=None):
"""Real cycles every TICK of simulated time. Rows are dicts per cycle."""
async def go():
hass = _Hass(load, cut, soc or {"a": 50.0, "b": 50.0}, jitter)
inv = [{"name": f"inverter-{u}", "setpoint_entity": f"number.sp_{u}",
"soc_entity": f"sensor.soc_{u}",
"batt_entity": f"sensor.batt_{u}",
"max_w": 2000, "bms_soc_floor": 10} for u in UNITS]
o = {"meter_entity": "sensor.grid", "meter_source": "off",
"inverters": inv, "gain": 0.6, "slew_w": 400,
"target_grid_w": -10, "deadband_w": 15, "step_w": 10,
"heartbeat_s": 10, "stale_input_s": 15, "auto_start": True}
o.update(extra or {})
c = Controller(o, hass, _Store(), _Pub(), stop=asyncio.Event())
for i in c.inverters:
i.dev_min, i.dev_max = -2000.0, 2000.0
c.auto = True
if old:
_off(c)
real_dt = m.datetime
clock = [0.0]
class _DT(real_dt):
@classmethod
def now(cls, tz=None):
return BASE + timedelta(seconds=clock[0])
m.datetime = _DT
rows = []
trips = {u: 0 for u in UNITS}
was = {u: False for u in UNITS}
pending = dict(events or {})
try:
t = 0.0
while t < dur:
hass.t = clock[0] = t
for et in sorted(pending):
if et <= t:
pending.pop(et)(c)
await c.read_inputs(now_mono=t)
for u, i in zip(UNITS, c.inverters):
s = i.batt_fresh.stale
trips[u] += s and not was[u]
was[u] = s
await c.cycle()
rows.append(dict(
t=t, grid=hass.grid(), site=c.target, reason=c.loop_reason,
a=hass.cmd["a"], b=hass.cmd["b"],
ra=c.inverters[0].refused, rb=c.inverters[1].refused,
da=c.inverters[0].delivery_wd.tripped,
db=c.inverters[1].delivery_wd.tripped))
t += TICK
finally:
m.datetime = real_dt
return rows, trips, c
return asyncio.run(go())
def after(rows, t0, t1=None):
return [r for r in rows if r["t"] >= t0 and (t1 is None or r["t"] < t1)]
def mean(rows, k):
return sum(r[k] for r in rows) / len(rows) if rows else float("nan")
def never(t):
return False
def cut_at(t0, t1=None):
return lambda t: t >= t0 and (t1 is None or t < t1)
def probes(rows, k):
"""Start times of runs where refused unit `k` is asked exactly PROBE_W."""
out, prev = [], False
for r in rows:
on = abs(r[k]) == PROBE_W and r["r" + k] != 0
if on and not prev:
out.append(r["t"])
prev = on
return out
SHAPES = (("flat", False), ("jitter", True))
# ==============================================================================
print("1. case a: house 3000 W, both give power, A cuts out at 120 s")
# ==============================================================================
for shape, jit in SHAPES:
new, ntr, cn = run(lambda t: 3000.0, {"a": cut_at(120), "b": never}, H,
jitter=jit)
old, otr, _ = run(lambda t: 3000.0, {"a": cut_at(120), "b": never}, H,
jitter=jit, old=True)
tail_n, tail_o = after(new, 300), after(old, 300)
held = [r for r in tail_n if abs(r["a"]) != PROBE_W]
print(f" {shape}: old B {mean(tail_o, 'b'):.0f} W grid "
f"{mean(tail_o, 'grid'):+.0f} W trips A {otr['a']}; new B "
f"{mean(tail_n, 'b'):.0f} W grid {mean(tail_n, 'grid'):+.0f} W "
f"trips A {ntr['a']}")
check(f"{shape}: PRECONDITION: on the old law B never takes over (mean "
f"under 1950 W) while A gives nothing",
mean(tail_o, "b") < 1950 and mean(tail_o, "grid") > 1050)
check(f"{shape}: A is refused and asked 0 W outside its probes, B carries "
f"its rating, and the grid imports only what B cannot give",
cn.inverters[0].refused == 1 and all(r["a"] == 0.0 for r in held)
and min(r["b"] for r in tail_n) == 2000.0
and abs(mean(tail_n, "grid") - 1004) < 20)
check(f"{shape}: A's stale-reading gate trips at most once, the site is "
f"never stopped after the first 5 minutes, and SAFETY-42 does not "
f"report A as not delivering",
ntr["a"] <= 1
and all(r["reason"] != "not-running" for r in tail_n)
and not any(r["da"] for r in new))
# ==============================================================================
print("2. case b: house 1200 W, which B alone can carry, A cuts out")
# ==============================================================================
for shape, jit in SHAPES:
new, ntr, cn = run(lambda t: 1200.0, {"a": cut_at(120), "b": never}, H,
jitter=jit)
old, _, _ = run(lambda t: 1200.0, {"a": cut_at(120), "b": never}, H,
jitter=jit, old=True)
tail_n, tail_o = after(new, 300), after(old, 300)
imp = sum(max(0.0, r["grid"]) for r in tail_n) * TICK / H
print(f" {shape}: old grid {mean(tail_o, 'grid'):+.0f} W, new grid "
f"{mean(tail_n, 'grid'):+.0f} W, new import {imp:.0f} Wh")
check(f"{shape}: PRECONDITION: the old law leaves a standing import",
mean(tail_o, "grid") > 250)
check(f"{shape}: B takes over the whole house and the grid sits near its "
f"target", abs(mean(tail_n, "grid") + 10) < 25 and imp < 10)
pa = probes(new, "a")
in_probe = [r for r in new
if any(p <= r["t"] < p + PROBE_S for p in pa)]
check(f"{shape}: A is probed at 200 W every 15 minutes while B carries "
f"the rest, and the two never pull opposite ways",
len(pa) >= 3
and all(abs(q - p - PROBE_EVERY_S - PROBE_S) <= 2 * TICK
for p, q in zip(pa, pa[1:]))
and all(r["b"] > 0 for r in in_probe)
and all(r["a"] * r["b"] >= 0 for r in new))
# ==============================================================================
print("3. case c: both cut out; the site holds 0 W, probes one at a time")
# ==============================================================================
for shape, jit in SHAPES:
new, ntr, cn = run(lambda t: 3000.0, {"a": cut_at(120), "b": cut_at(150)},
1.5 * H, jitter=jit)
old, _, _ = run(lambda t: 3000.0, {"a": cut_at(120), "b": cut_at(150)},
1.5 * H, jitter=jit, old=True)
tail_n, tail_o = after(new, 600), after(old, 600)
pa, pb = probes(new, "a"), probes(new, "b")
print(f" {shape}: old mean asks A/B {mean(tail_o, 'a'):.0f}/"
f"{mean(tail_o, 'b'):.0f} W; new {mean(tail_n, 'a'):.0f}/"
f"{mean(tail_n, 'b'):.0f} W, probes A {len(pa)} B {len(pb)}")
check(f"{shape}: PRECONDITION: the old law keeps asking two empty "
f"batteries", mean(tail_o, "a") > 150 and mean(tail_o, "b") > 150)
check(f"{shape}: the site holds 0 W: the loop says `battery refuses` or "
f"`battery probe` and each battery is asked 0 W or its probe",
all(r["reason"] in (control.REFUSED, control.PROBING)
for r in tail_n)
and all(r["a"] in (0.0, PROBE_W) and r["b"] in (0.0, PROBE_W)
for r in tail_n))
check(f"{shape}: both batteries are probed, never at the same time",
len(pa) >= 3 and len(pb) >= 3
and not any(r["a"] and r["b"] for r in new if r["t"] > 300))
# ==============================================================================
print("4. recovery: the reseed stops the parked site target reaching the grid")
# ==============================================================================
# Both cut, the house drops to 940 W at 1 h, A recovers at 1.5 h with no sign
# change. Phase A, with the envelope but no reseed and no site hold: -1059 W.
ld = (lambda t: 3000.0 if t < H else 940.0)
for shape, jit in SHAPES:
rows, _, c = run(ld, {"a": cut_at(120, 1.5 * H), "b": cut_at(150)},
2 * H, jitter=jit)
win = after(rows, 1.5 * H, 1.5 * H + 900)
worst = min(r["grid"] for r in win)
print(f" {shape}: worst grid after A recovers {worst:+.0f} W; A "
f"refused {c.inverters[0].refused}")
check(f"{shape}: A's recovery is found (at its next probe at the latest) "
f"and never exports more than 250 W",
c.inverters[0].refused == 0 and worst > -250
and any("BATTERY REFUSES ENDED inverter-a" in e for e in c.events))
# Case a: B at its rating, the site target parked at 3600 W while the house
# still imports; the house falls to 2500 W and A recovers. Without the reseed A
# is handed the parked 1600 W at once, 1100 W more than the house needs.
ld2 = (lambda t: 3000.0 if t < 0.5 * H else 2500.0)
rows, _, c = run(ld2, {"a": cut_at(120, 0.75 * H), "b": never}, 1.5 * H)
win = after(rows, 0.75 * H, 0.75 * H + 900)
print(f" a: worst grid when A recovers {min(r['grid'] for r in win):+.0f} W")
check("case a: a recovery hands A no parked site target (worst export under "
"250 W)", min(r["grid"] for r in win) > -250
and any("BATTERY REFUSES ENDED inverter-a" in e for e in c.events))
# ==============================================================================
print("5. exits, words and the circulation guard")
# ==============================================================================
# Solar: the house exports from 30 min, so the loop wants to charge.
rows, _, c = run(lambda t: 3000.0 if t < 0.5 * H else -2500.0,
{"a": cut_at(120), "b": never}, H)
check("the grid asking the other direction lets A charge at once (its limit "
"closes only the refused direction) and a minute of charge ends the "
"refusal",
c.inverters[0].refused == 0 and min(r["a"] for r in after(rows, 0.6 * H))
< -500 and any("the other way" in e for e in c.events))
def _stop(c):
c.auto = False
rows, _, c = run(lambda t: 3000.0, {"a": cut_at(120), "b": never}, 0.5 * H,
events={1200.0: _stop})
check("stopping control ends the refusal",
c.inverters[0].refused == 0
and any("control stopped" in e for e in c.events))
# A small house: the site asks under 200 W, so a probe may not pin A (B would
# have to charge while A gives power). The probe waits instead.
rows, _, c = run(lambda t: 3000.0 if t < 600 else 120.0,
{"a": cut_at(120), "b": never}, 1.2 * H)
check("with the site asking under 200 W no probe is sent and no inverter is "
"ever asked the opposite way to the other",
c.inverters[0].refused == 1
and probes(after(rows, 700), "a") == []
and all(r["a"] * r["b"] >= 0 for r in rows)
and not any("ARBITER" in e for e in c.events))
# The hold binds every strategy's ask: a peak-guard SET at P_PEAK on the site
# is still placed around a refusing unit's closed limit.
_res = resolve([Claim.set("peak", P_PEAK, 3000.0, "quarter peak"),
Claim.set("loop", P_LOOP, 500.0, "tracking"),
Claim.limit("supervised", P_SAFETY, -2000.0, 2000.0,
actuator="inverter-a"),
Claim.limit("supervised", P_SAFETY, -2000.0, 2000.0,
actuator="inverter-b"),
Claim.limit("refused", P_SAFETY, hi=0.0,
reason=control.REFUSED, actuator="inverter-a")],
units=("inverter-a", "inverter-b"),
live=("inverter-a", "inverter-b"))
check("a P_SAFETY refusal limit binds a higher-priority strategy too: the "
"peak guard's 3000 W goes to B, A gets 0 W",
_res.shares.get("inverter-a") == 0 and _res.shares.get("inverter-b")
== 2000)
class _Grab(logging.Handler):
def __init__(self):
super().__init__()
self.lines = []
def emit(self, record):
self.lines.append((record.levelno, record.getMessage()))
_g = _Grab()
_lg = logging.getLogger("goodwe")
_lv = _lg.level
_lg.setLevel(logging.INFO)
_lg.addHandler(_g)
run(lambda t: 3000.0, {"a": cut_at(120), "b": never}, 600)
run(lambda t: 3000.0, {"a": cut_at(120), "b": never}, 600,
soc={"a": 10.0, "b": 50.0})
_lg.removeHandler(_g)
_lg.setLevel(_lv)
warns = [x for lv, x in _g.lines if lv == logging.WARNING
and "REFUSES_ABOVE_CUT_OUT" in x]
infos = [x for lv, x in _g.lines if lv == logging.INFO
and "BATTERY_REFUSES" in x]
print(f" {warns} / {infos}")
check("a refusal above the configured cut-out level logs INFO and one WARNING "
"naming the inverter, and a battery at its cut-out logs neither "
"(OPT-09's limit closes it first)",
len(warns) == 1 and warns[0].startswith("inverter-a ")
and len(infos) == 1 and infos[0].startswith("inverter-a:")
and len(warns[0].split()) <= 25)
# A night load of 400 W: each unit is asked about 400 W once the loop has
# made up A's missing half, under `saturation_w`. Engaging at PROBE_W is what
# refuses A here; at 500 W it would keep a phantom share that SAFETY-43 names.
rows, _, c = run(lambda t: 400.0, {"a": cut_at(0), "b": never}, 1200)
check("a 400 W night load: A is refused at a share under 500 W, B carries the "
"house and nothing reports A as not delivering",
c.inverters[0].refused == 1 and not any(r["da"] for r in rows)
and abs(mean(after(rows, 300), "grid") + 10) < 25
and mean(after(rows, 300), "b") > 380)
# A night load of 150 W: A's share never passes PROBE_W, so it is never
# refused, and SAFETY-43 must not name it as not delivering either: a battery
# asked under 200 W that gives under 100 W is a refusal too small to call.
rows, _, c = run(lambda t: 150.0, {"a": cut_at(0), "b": never}, 1200)
check("a small share on an empty battery is neither refused nor reported as "
"not delivering, and the grid stays near its target",
c.inverters[0].refused == 0 and not any(r["da"] for r in rows)
and abs(mean(after(rows, 300), "grid") + 10) < 25)
rows, _, c = run(lambda t: 3000.0, {"a": cut_at(120), "b": never}, 600)
notes = c.liveness_notes()
check("the status page names the refusing inverter",
any(n.startswith("inverter-a: the battery refuses to give power")
for n in notes))
# Shadow mode writes nothing, so nothing is asked and nothing can refuse.
rows, _, c = run(lambda t: 3000.0, {"a": cut_at(120), "b": never}, 1200,
extra={"shadow_mode": True})
check("shadow mode: no inverter is refused and both stay at 0 W",
c.inverters[0].refused == 0 and c.inverters[1].refused == 0
and all(r["a"] == 0.0 and r["b"] == 0.0 for r in rows))
# ==============================================================================
print("6. one inverter is untouched")
# ==============================================================================
def _boom(*a, **k):
raise AssertionError("SAFETY-73 code ran at N=1")
_saved = {k: getattr(Controller, k) for k in (
"_unit_refusals", "_unit_refusal_start", "_unit_refusal_end",
"_unit_probe", "_site_refused")}
for k in _saved:
setattr(Controller, k, _boom)
try:
import test_refusal as _t1
_one_ok = not _t1.fails
except (AssertionError, SystemExit) as e:
_one_ok = isinstance(e, SystemExit) and not e.code
finally:
for k, v in _saved.items():
setattr(Controller, k, v)
check("SAFETY-52's whole single-inverter suite passes with every SAFETY-73 "
"method made to raise, so none of it runs at N=1", _one_ok)
print()
if fails:
print(f"{len(fails)} FAILED: " + ", ".join(fails))
sys.exit(1)
print("all checks passed")
Reference in New Issue View Git Blame Copy Permalink