Files
goodwe-addon/goodwe_controller/test_freeze.py
T
glenn schrooyenandClaude Opus 5 c040aac4b8 SAFETY-03: alarm when a computed setpoint exceeds the rating, do not just clip it
The clamp was never the missing half. specs.md 10.3 records an unclamped
integrator climbing ~130 W per 4 s past 10 kW and reporting 14 768 W pinned at
the 5 kW rail; clipped silently that is a flat, legal-looking 5000 W hold and
nothing says the loop asked for three times it.

write_setpoint() - the one path every write takes - now clamps to the VALIDATED
max_w and reports whether the raw value differed; check_output_clamp() raises
OUTPUT_CLAMP as a notification, an ERROR, an event row and a red status banner.
Not a trip: the clamped value is written and the loop keeps running.

The rating is max_w and nothing is min()ed at runtime - CFG-01 already refuses a
max_w outside the device range at startup. The device range is applied second as
defence in depth; it cannot bite after a successful validation, so if it ever
does it alarms too and names the bound. No new config key.

Decision grew i_clamped / clamped booleans, and that is the load-bearing part:
`reason` is one string that slew-limiting and the saturation freeze overwrite,
so an alarm keyed on it is silent on exactly the cycles a winding loop produces.
Mutating main.py back onto the string drops the 10.3 replay from 2 engagements
to 0.

Distinct from saturation by construction - no shared reason code, notification
id, status row or counter - and checked in the negative, after first asserting
the saturation freeze really did engage.

Two things found by running it rather than by reading it: a row per engagement
flushed the 40-row event log and evicted SAFETY-18's own METER FROZEN? row (now
coalesced per episode with an exact count), and a naive clamp turns a nan into a
command at the full rating because every comparison against nan is False.

test_clamp.py is new: assert-based, no framework. Six author mutations
reproduced red, including "clamp kept, alarm deleted". Two of the check's own
hypotheses about the freeze/clamp interaction were refused by its sweep and
rewritten to what it measured.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015VXjA3LWccN1rJsk3okyPC
2026-08-30 15:22:22 +02:00

555 lines
24 KiB
Python

"""Runnable check for SAFETY-18, the frozen-but-fresh meter detector.
python3 test_freeze.py
Three parts, and the last two are the ones that matter:
1. the predicate's boundaries and the edge-trigger, table-driven;
2. SILENT ON REAL CONVERGED DATA. `scenarios/p1-age-2026-08-26.jsonl.gz` is
40 minutes of this house's live meter at the operating point, committed
here from the infra repo's `sim/scenarios/`. It is replayed through the
REAL `control.compute()` at the real cycle rule, and the detector must not
make a sound. This is the acceptance criterion TEL-05's objection is
about, and it is verified rather than argued;
3. FIRES ON A GENUINELY FROZEN METER, over real HTTP, through the REAL
`HomeWizardLocalSource` - the transport is what computes `unchanged_s`, so
a check that calls the predicate directly would not be testing the thing
that ships. Uses the rig's own `sim/hwsim.py --fault freeze` when the infra
repo is next to this one (or `$GOODWE_SIM`), and an equivalent local stub
otherwise. Both are exercised; the run says which one it used.
⚠️ Part 3 runs in WALL CLOCK - `unchanged_s` is `time.monotonic()` and the
transport really polls - so it drives the detector at `hold_s=10`, not the
shipped 300. The THRESHOLD is config; the MECHANISM is what is under test, and
the mechanism cannot be hurried. Part 2 uses the shipped 300 because that data
is replayed rather than lived.
⚠️ Part 3 also runs its NEGATIVE twin, and that is the load-bearing half: the
same railed command against a meter that is still MOVING, at the same 10 s
threshold. If that one alarms, the detector is a `unchanged_s` threshold with
extra steps and the whole reason for the conjunction is gone.
------------------------------------------------------------------------------
THE RIG RUN THIS STANDS IN FOR
Part 3 is the portable version of a run done on the real rig - the `goodwe-sim`
compose project on 192.168.2.147, under `sim/riglock.py`, with the driver staged
into `goodwe-sim-plant` so it polls that container's own hwsim while plantsim
drives it. Both phases passed there. To repeat it:
DOCKER_HOST=ssh://homelab@192.168.2.147 python sim/riglock.py status
⚠️ WITHOUT `DOCKER_HOST` SET, ON A WINDOWS HOST, `riglock status` REPORTS
`rig: UNKNOWN` - it shells out to the local `docker` CLI, which resolves to a
Docker Desktop npipe that is not running. That is INDISTINGUISHABLE from a
genuinely absent rig, and it is not one: the rig lives on the homelab box and was
up the whole time. `UNKNOWN` is the tool declining to observe, not an
observation. Do not read it as "no rig" (this cost this ticket a wrong caveat).
The rig run also produced a result part 3's stub cannot: under +8000 W of house
load the meter's POWER FIGURE never moved - 10130.0 W for 26 consecutive polls -
and only the kWh registers kept `unchanged_s` at zero. See the note in
freeze.py; it is why the fingerprint's register terms are the negative case
rather than a nicety.
"""
import asyncio
import gzip
import json
import os
import subprocess
import sys
import time
from datetime import datetime, timedelta, timezone
from pathlib import Path
import aiohttp
from app.control import Tuning, compute
from app.freeze import (
RAIL_FRAC, FreezeWatchdog, FreezeWatchdogConfig, looks_frozen, rail_w,
)
from app.main import Controller
from app.p1 import HomeWizardLocalSource, P1Ingest
fails = []
T0 = datetime(2026, 8, 26, 7, 0, 0, tzinfo=timezone.utc)
HERE = Path(__file__).resolve().parent
# The live install's numbers, which are the ones the exposure is written at.
LIVE_MAX_W = 5000.0
def check(name, cond):
print(f" {'ok ' if cond else 'FAIL'} {name}")
if not cond:
fails.append(name)
def tuning(max_w=LIVE_MAX_W):
"""The shipped defaults from config.yaml, at the live install's max_w."""
return Tuning(gain=0.6, max_w=max_w, slew_w=1000, deadband_w=15,
target_grid_w=-10, step_w=10, saturation_w=500,
saturation_cycles=3)
# --------------------------------------------------------------------------- #
# 1. the predicate and the edge
# --------------------------------------------------------------------------- #
print("predicate")
cfg = FreezeWatchdogConfig(hold_s=300)
check("both terms true -> frozen", looks_frozen(300, 5000, 5000, cfg))
check("unchanged long, command NOT railed -> silent (TEL-05's case)",
not looks_frozen(3600, -40, 5000, cfg))
check("command railed, meter still moving -> silent (a real 8 kW load)",
not looks_frozen(4, -5000, 5000, cfg))
check("neither -> silent", not looks_frozen(4, -40, 5000, cfg))
check("one second under the hold is still silent",
not looks_frozen(299.9, 5000, 5000, cfg))
check("just under the rail fraction is still silent",
not looks_frozen(3600, 5000 * RAIL_FRAC - 1, 5000, cfg))
check("at the rail fraction exactly, it fires",
looks_frozen(3600, 5000 * RAIL_FRAC, 5000, cfg))
check("the rail scales with max_w, not with a fixed watt figure",
looks_frozen(3600, 2000, 2000, cfg) and not looks_frozen(3600, 2000, 5000, cfg))
check("a charge rail counts too - the runaway can be either sign",
looks_frozen(3600, -5000, 5000, cfg))
check("no transport signal (unchanged_s None) -> silent, never assumed frozen",
not looks_frozen(None, 5000, 5000, cfg))
check("no command yet -> silent", not looks_frozen(3600, None, 5000, cfg))
check("rail_w tracks RAIL_FRAC", rail_w(5000) == 5000 * RAIL_FRAC)
print("the edge")
wd = FreezeWatchdog(cfg)
edges = [wd.evaluate(u, c, 5000, T0 + timedelta(seconds=i))
for i, (u, c) in enumerate(
[(10, 5000), (300, 5000), (305, 5000), (310, 5000),
(2, 5000), (300, 5000)])]
check("rises exactly once per episode", edges == [False, True, False, False,
False, True])
check("two episodes counted", wd.episodes == 2)
check("clearing the condition clears the alarm state",
not FreezeWatchdog(cfg).evaluate(2, 5000, 5000, T0))
check("the reason names both terms",
"unchanged over 300" in wd.reason and "command over 4750" in wd.reason)
# --------------------------------------------------------------------------- #
# 2. silent on 40 minutes of real converged behaviour
# --------------------------------------------------------------------------- #
print("converged replay - scenarios/p1-age-2026-08-26.jsonl.gz")
rows = [json.loads(line) for line in
gzip.open(HERE / "scenarios" / "p1-age-2026-08-26.jsonl.gz", "rt")]
check("the capture loaded", len(rows) > 1000)
# ⚠️ The capture records the POWER figure only. The shipped fingerprint also
# covers both kWh registers, so a real `unchanged_s` on this data can only be
# SHORTER than what is reconstructed here - this replay is the pessimistic
# version of the converged case, not a flattering one.
def replay(hold_s, max_w, cycle_floor_s=15.0):
"""Drive the real control law and the real detector over the capture."""
wd = FreezeWatchdog(FreezeWatchdogConfig(hold_s=hold_s))
t = tuning(max_w)
target, i_w, sat = 0.0, 0.0, 0
prev_pwr, changed_at = None, rows[0]["t"]
last_cycle = None
peak_cmd, peak_unchanged, fired_at = 0.0, 0.0, None
for r in rows:
now, pwr = r["t"], float(r["pwr"])
if prev_pwr is None or pwr != prev_pwr:
changed_at = now
arrival = (prev_pwr is None or pwr != prev_pwr)
prev_pwr = pwr
# main.run_control()'s rule: a cycle per changed reading, or the floor.
if arrival or last_cycle is None or now - last_cycle >= cycle_floor_s:
last_cycle = now
d = compute(prev_w=target, grid_w=pwr, actual_w=target, tuning=t,
sat_count=sat, i_w=i_w)
target, i_w, sat = d.target_w, d.i_w, d.sat_count
# ⚠️ The detector is evaluated on EVERY poll, not only on cycles, and
# that is deliberately harsher than the shipped wiring (which evaluates
# it inside cycle()). On an arrival-triggered cycle `unchanged_s` is 0
# by construction, so grading only there would be grading the detector
# on the one sample it cannot possibly fire on.
unchanged = now - changed_at
peak_cmd = max(peak_cmd, abs(target))
peak_unchanged = max(peak_unchanged, unchanged)
if wd.evaluate(unchanged, target, max_w,
T0 + timedelta(seconds=now - rows[0]["t"])) \
and fired_at is None:
fired_at = now - rows[0]["t"]
return wd, peak_cmd, peak_unchanged, fired_at
for mw in (2000.0, LIVE_MAX_W):
wd_r, peak_cmd, peak_unchanged, fired = replay(300.0, mw)
print(f" max_w {mw:.0f}: worst |command| {peak_cmd:.0f} W (rail "
f"{rail_w(mw):.0f} W), worst unchanged {peak_unchanged:.1f} s "
f"(hold 300 s), episodes {wd_r.episodes}")
check(f"max_w {mw:.0f}: never alarms on the converged capture", fired is None)
check(f"max_w {mw:.0f}: and the rail term had margin to spare",
peak_cmd < rail_w(mw))
# The margin on the OTHER term, as numbers rather than as trust. Both figures
# are quoted in config.yaml and freeze.py, so they are re-measured every run and
# a capture swapped underneath them cannot leave the prose saying the old thing.
_, _, worst_unchanged, _ = replay(300.0, LIVE_MAX_W)
gaps, last, prev = [], rows[0]["t"], None
for r in rows:
if prev is None or r["pwr"] != prev:
gaps.append(r["t"] - last)
last = r["t"]
prev = r["pwr"]
print(f" longest unchanged_s any sample would observe: {worst_unchanged:.1f} s")
print(f" longest interval serving no distinct reading: {max(gaps):.1f} s "
f"(spans an 11.4 s hole in the capture's own polling)")
check("worst observed unchanged_s is the 8 s the docs quote",
7.5 < worst_unchanged < 8.5)
check("longest distinct-reading interval is the 15.4 s the docs quote",
15.0 < max(gaps) < 16.0)
check("300 s is at least 15x the worst of the two", 300.0 >= 15 * max(gaps))
# --------------------------------------------------------------------------- #
# 2b. TEL-05's case, reconstructed, because the capture cannot contain it
# --------------------------------------------------------------------------- #
# ⚠️ BE HONEST ABOUT WHAT THE CAPTURE ABOVE PROVES AND WHAT IT DOES NOT. On this
# house's meter the power figure jitters at 1 W, so the worst `unchanged_s` any
# sample sees is 8 s - which means the capture alone does NOT refute a bare
# `unchanged_s > 300` threshold, and reporting it as though it did would be the
# confident explanation nobody checked. The case that refutes the bare threshold
# is the one p1.py describes and no capture of THIS meter contains: a converged
# meter whose power figure simply repeats, where the 1 Wh export register needs
# 360 s to advance at -10 W and the fingerprint therefore stands still.
#
# So it is constructed, from the operating point the capture measures (mean grid
# -14.2 W) rather than invented: one hour of a perfectly healthy converged meter
# that happens not to move.
print("TEL-05's low-power case - a converged meter that legitimately repeats")
tel05 = FreezeWatchdog(FreezeWatchdogConfig(hold_s=300))
t5 = tuning(LIVE_MAX_W)
tgt, iw, sc = 0.0, 0.0, 0
bare_would_fire = False
tel05_cmd = 0.0
for k in range(3600):
if k % 15 == 0: # cycle_floor_s, the only trigger left
d = compute(prev_w=tgt, grid_w=-14.2, actual_w=tgt, tuning=t5,
sat_count=sc, i_w=iw)
tgt, iw, sc = d.target_w, d.i_w, d.sat_count
tel05_cmd = max(tel05_cmd, abs(tgt))
# The REJECTED design, written out so the case can be shown to be real.
bare_would_fire = bare_would_fire or float(k) >= 300
tel05.evaluate(float(k), tgt, LIVE_MAX_W, T0 + timedelta(seconds=k))
print(f" an hour at -14.2 W: worst |command| {tel05_cmd:.0f} W, "
f"unchanged_s reaches 3599 s, episodes {tel05.episodes}")
check("the bare unchanged_s threshold TEL-05 rejected WOULD fire here",
bare_would_fire)
check("the conjunction does not - this is what the rail term buys",
tel05.episodes == 0)
check("...and the command stayed nowhere near the rail",
tel05_cmd < rail_w(LIVE_MAX_W))
# The structural version of the same point, re-derived rather than asserted in
# prose: the slowest register a HEALTHY meter can have while still being seen as
# unchanged for hold_s is 3600/hold_s watts, and at the shipped defaults that
# figure is inside the deadband, so the command there is parked at 0 W. If
# someone lowers meter_freeze_s or narrows the deadband, this is the check that
# should stop them.
DEADBAND_W = 15.0
slowest_healthy_w = 3600.0 / 300.0
check("a healthy meter slow enough to look unchanged is inside the deadband",
slowest_healthy_w < DEADBAND_W)
_park = compute(prev_w=0.0, grid_w=-slowest_healthy_w, actual_w=0.0,
tuning=tuning(LIVE_MAX_W), sat_count=0, i_w=0.0)
check("...and the shipped law really does park the command there",
_park.target_w == 0.0)
# ⚠️ THE PROOF THAT THE REPLAY IS CAPABLE OF FIRING AT ALL. Without this, "the
# converged capture is silent" is satisfied by a detector that is wired to
# nothing. Drop the hold to 1 s and the rail to nothing, and the SAME replay
# must alarm - so the silence above is the data's doing, not the harness's.
wd_hot, _, _, fired_hot = replay(1.0, 20.0)
check("...and the same replay DOES alarm once both terms are made reachable",
fired_hot is not None and wd_hot.episodes > 0)
# --------------------------------------------------------------------------- #
# 3. a genuinely frozen meter, over real HTTP, through the real transport
# --------------------------------------------------------------------------- #
def find_hwsim():
"""The rig's own fake meter, if the infra repo is reachable."""
cands = []
if os.environ.get("GOODWE_SIM"):
cands.append(Path(os.environ["GOODWE_SIM"]) / "hwsim.py")
cands.append(HERE.parents[1] / "goodwe" / "sim" / "hwsim.py")
for c in cands:
if c.exists():
return c
return None
STUB = r'''
import http.server, json, sys, threading, time
FROZEN = {"active_power_w": 700.0, "active_power_l1_w": 700.0,
"total_power_import_kwh": 1234.567, "total_power_export_kwh": 890.123}
moving = len(sys.argv) > 2 and sys.argv[2] == "moving"
t0 = time.monotonic()
class H(http.server.BaseHTTPRequestHandler):
def do_GET(self):
d = dict(FROZEN)
if moving:
# A REAL railed condition: a big steady import, so the power figure
# jitters and the 1 Wh registers advance the way physics makes them.
el = time.monotonic() - t0
d["active_power_w"] = 7000.0 + (int(el * 10) % 7)
d["active_power_l1_w"] = d["active_power_w"]
d["total_power_import_kwh"] = round(1234.567 + 7000 * el / 3.6e6, 3)
raw = json.dumps(d).encode()
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(raw)))
self.end_headers()
self.wfile.write(raw)
def log_message(self, *a): pass
http.server.ThreadingHTTPServer(("127.0.0.1", int(sys.argv[1])), H).serve_forever()
'''
HWSIM_DRIVER = r'''
import sys, threading, time
sys.path.insert(0, sys.argv[3])
import hwsim
hwsim.serve(int(sys.argv[1]))
mode = sys.argv[2]
hwsim.set_grid(7000.0)
if mode == "freeze":
hwsim.fault = "freeze" # exactly `hwsim.py --fault freeze`
while True:
# The plant keeps moving underneath either way. Under `freeze` hwsim's own
# set_grid() returns early, which is the whole point of that fault.
hwsim.set_grid(7000.0 + (int(time.monotonic() * 10) % 7))
time.sleep(0.2)
'''
async def live_run(port, mode, hold_s=10.0, seconds=16.0, max_w=LIVE_MAX_W):
"""Poll a real HTTP meter through the real transport and drive the detector."""
ingest = P1Ingest(phases=1, max_age_s=30)
wd = FreezeWatchdog(FreezeWatchdogConfig(hold_s=hold_s))
t = tuning(max_w)
target, i_w, sat = 0.0, 0.0, 0
fired_at, polls, worst_unchanged, worst_cmd = None, 0, 0.0, 0.0
async with aiohttp.ClientSession() as session:
src = HomeWizardLocalSource(session, ingest, "127.0.0.1", port=port,
poll_s=1.0)
start = time.monotonic()
while time.monotonic() - start < seconds:
if await src.poll_once():
polls += 1
grid = ingest.net_w
d = compute(prev_w=target, grid_w=grid, actual_w=target,
tuning=t, sat_count=sat, i_w=i_w)
target, i_w, sat = d.target_w, d.i_w, d.sat_count
worst_unchanged = max(worst_unchanged, src.unchanged_s or 0.0)
worst_cmd = max(worst_cmd, abs(target))
if wd.evaluate(src.unchanged_s, target, max_w,
datetime.now(timezone.utc)) and fired_at is None:
fired_at = time.monotonic() - start
await asyncio.sleep(1.0)
return fired_at, polls, worst_unchanged, worst_cmd
def with_server(argv, port, fn):
proc = subprocess.Popen([sys.executable, "-c"] + argv,
stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
try:
deadline = time.monotonic() + 10
while time.monotonic() < deadline:
try:
import socket
with socket.create_connection(("127.0.0.1", port), 0.3):
break
except OSError:
time.sleep(0.1)
else:
raise RuntimeError("the fake meter never came up")
return fn()
finally:
proc.kill()
proc.wait()
hw = find_hwsim()
print(f"frozen meter over HTTP - using {'the rig hwsim at ' + str(hw) if hw else 'the local stub'}")
if hw:
def argv_for(mode):
return [HWSIM_DRIVER, "8791" if mode == "freeze" else "8792", mode,
str(hw.parent)]
else:
def argv_for(mode):
return [STUB, "8791" if mode == "freeze" else "8792",
"frozen" if mode == "freeze" else "moving"]
fired, polls, unch, cmd = with_server(
argv_for("freeze"), 8791,
lambda: asyncio.run(live_run(8791, "freeze")))
print(f" freeze: {polls} polls, worst unchanged {unch:.1f} s, "
f"worst |command| {cmd:.0f} W, alarm at "
f"{'never' if fired is None else f'{fired:.1f} s'}")
check("the transport really polled a live server", polls >= 10)
check("a frozen meter raises the alarm", fired is not None)
check("...and only after the hold, not on the first railed poll",
fired is None or fired >= 10.0)
fired2, polls2, unch2, cmd2 = with_server(
argv_for("moving"), 8792,
lambda: asyncio.run(live_run(8792, "moving")))
print(f" moving: {polls2} polls, worst unchanged {unch2:.1f} s, "
f"worst |command| {cmd2:.0f} W, alarm at "
f"{'never' if fired2 is None else f'{fired2:.1f} s'}")
check("the moving meter also drove the command to the rail", cmd2 >= rail_w(LIVE_MAX_W))
check("a REAL railed condition against a live meter stays silent", fired2 is None)
check("...because the meter kept moving", unch2 < 10.0)
# --------------------------------------------------------------------------- #
# 4. the wiring, because a correct detector nobody calls is not a detector
# --------------------------------------------------------------------------- #
# ⚠️ EVERYTHING ABOVE PASSES WITH `app/freeze.py` IMPORTED BY NOTHING. That is
# the shape of defect this project keeps finding: the mechanism is right, the
# suite is green, and the shipped path never reaches it. So the real
# `Controller.cycle()` is driven here, against the real arbiter and the real
# control law, with a stand-in transport that reports `unchanged_s` the way
# HomeWizardLocalSource does - and the assertion is on the SERVICE CALL that
# leaves the process, not on the watchdog's own flag.
print("the wiring - Controller.cycle() actually raises the notification")
class _Store:
data: dict = {}
def set(self, *a):
pass
def get_time(self, *a):
return None
class _Pub:
enabled = False
def publish(self, *a, **k):
pass
def close(self):
pass
class _Hass:
def __init__(self, grid):
self.grid = grid
self.calls = []
self.writes = []
# The plant tracks the command instantly, as in test_loop.py. Without
# it the saturation freeze pins the command short of the rail and the
# runaway under test never happens - which would have made this a check
# that passes for the wrong reason.
self.batt = 0.0
async def limits(self, entity):
return (-5000.0, 5000.0, 1.0)
async def number(self, entity, invert=False):
return {"sensor.grid": self.grid, "sensor.soc": 50.0,
"sensor.batt": self.batt}.get(entity)
async def set_number(self, entity, value):
self.writes.append(float(value))
self.batt = float(value)
return True
async def call(self, domain, service, data):
self.calls.append((domain, service, data))
return True
class _Source:
"""Only the attribute the controller reads. Duck-typed like the real one."""
def __init__(self, unchanged_s):
self.unchanged_s = unchanged_s
def wired(unchanged_s, grid, cycles=60, **over):
async def go():
hass = _Hass(grid)
c = Controller(dict(
{"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
"soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
"meter_source": "off", "max_w": LIVE_MAX_W, "gain": 0.6,
"slew_w": 1000, "deadband_w": 15, "target_grid_w": -10,
"step_w": 10, "heartbeat_s": 10, "stale_input_s": 15},
**over), hass, _Store(), _Pub(), stop=asyncio.Event())
c.auto = True
c.dev_min, c.dev_max = -5000.0, 5000.0
c.p1_source = _Source(unchanged_s)
for _ in range(cycles):
await c.read_inputs()
await c.cycle()
return hass, c
return asyncio.run(go())
hass_f, ctl_f = wired(3600.0, 4000.0)
# ⚠️ FILTERED BY notification_id, NOT BY SERVICE, since SAFETY-03. This add-on
# now raises MORE THAN ONE kind of persistent notification, and the scenarios
# below drive a command to the rail on purpose - which is a winding loop, which
# legitimately raises the output-clamp alarm too. A bare `svc == "create"` test
# therefore counts a DIFFERENT guard's alarm as this one's, which is exactly the
# conflation SAFETY-03's second acceptance criterion exists to forbid. Keying on
# the id is what makes "this detector said nothing" mean it.
FROZEN_ID = "goodwe_meter_frozen"
def frozen_notes(hass):
return [d for dom, svc, d in hass.calls
if (dom, svc) == ("persistent_notification", "create")
and d.get("notification_id") == FROZEN_ID]
notes = frozen_notes(hass_f)
# The status row lives behind `p1_enabled`, which is what a homewizard_local
# install has. Flipped here rather than plumbing a whole P1 transport in - the
# row's own content is what is under test.
ctl_f.p1_enabled = True
rows = [c for c in ctl_f.checks() if "METER MAY BE FROZEN" in c["text"]]
print(f" frozen at +4000 W: command ran to {ctl_f.target:.0f} W, "
f"{len(notes)} notification(s), {len(rows)} red status row(s)")
check("the shipped cycle() raises exactly one HA notification", len(notes) == 1)
check("...with a fixed notification_id so repeats replace, not stack",
notes and notes[0].get("notification_id") == "goodwe_meter_frozen")
check("...and it says the controller is still running",
bool(notes) and "STILL RUNNING" in notes[0]["message"])
check("the status page carries a failed row saying so",
len(rows) == 1 and not rows[0]["ok"] and not rows[0]["warn"])
check("the event log names it", any("METER FROZEN" in e for e in ctl_f.events))
# ⚠️ NOTIFY, NOT TRIP. If a later cut turns this into a trip it must come here
# and change this expectation deliberately, not discover it in the field.
check("and it does NOT command 0 W - this cut is an alarm only",
abs(ctl_f.target) >= rail_w(LIVE_MAX_W))
hass_n, ctl_n = wired(3600.0, -14.2)
check("wired up, TEL-05's converged case still raises nothing",
not frozen_notes(hass_n))
hass_m, ctl_m = wired(2.0, 8000.0)
check("wired up, a real railed load against a moving meter raises nothing",
not frozen_notes(hass_m)
and abs(ctl_m.target) >= rail_w(LIVE_MAX_W))
check("an uncovered transport (no unchanged_s) raises nothing, and says so",
not frozen_notes(wired(None, 4000.0)[0]))
print()
if fails:
print(f"{len(fails)} FAILED: {', '.join(fails)}")
sys.exit(1)
print("all checks passed")