glenn schrooyenandClaude Opus 5
611d3dbcdf
1. notes_of() could not tell this guard's alarm from anybody else's. It selected by SERVICE, so against #12's app/ (which adds goodwe_output_clamp) eight assertions went red, AC1's core check among them. Blanket id-keying is not the repair and was measured, not guessed: selector vs #12's app/ id renamed by service (as committed) 8 red 1 red by id alone 3 red 6 red by id or title (shipped) 3 red 1 red all_notes_of() is the whole surface; notes_of() is this guard's, recognised by the id it should carry OR the title it writes - the `or` is what keeps the one assertion whose subject IS the id able to fail on its own. The three surface-wide claims stay surface-wide, each with a comment naming which surface it is about and why the next guard's author must decide rather than narrow them. They are the three that go red against #12, which is the correct signal. 2. The ordering comment named the wrong boundary. N3 - moving the call to immediately before write_setpoint() - is green and correctly so: self.target is assigned two lines earlier, write_setpoint() clamps only a local and can skip the write entirely. So the invariant is "after `self.target = resolution.target_w`", and the detector grades pre-clamp INTENT, not the wire. Corrected in test_freeze.py and in main.py's own note above the call (comment-only; I merge second and own the conflict). N8 - moving the call above the assignment - is RED 4. 3. The schema range's own bounds are now pinned. int(60,3600) -> int(1,86400) survived green: an operator could set meter_freeze_s: 86400 and the guard is dead with nothing red. Both bounds are re-derived rather than restated. 4. find_hwsim() crashed the whole suite on a checkout one level below the root (HERE.parents[1] -> IndexError, reproduced at C:/goodwe_controller). Guarded. And part 3 now names its transport in the summary: the rig hwsim and the stub are not equivalent - the rig path failed 1 run in 3 - so a green log has to say which one backed it. Also: the "revived transport re-rails" check is labelled a PRECONDITION like its sibling; freeze_rows() records that flipping p1_enabled exercises the row's content and never its gate; and the note about the deleted assertion is corrected - it was REDUNDANT, not undecidable. The reviewer falsified it in one line (move the hass.call above `if not fired: return`, RED 9). "I could not construct a falsifier, therefore it cannot fail" was exactly the inference this file exists to distrust. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015VXjA3LWccN1rJsk3okyPC
976 lines
47 KiB
Python
976 lines
47 KiB
Python
"""Runnable check for SAFETY-18, the frozen-but-fresh meter detector.
|
|
|
|
python3 test_freeze.py
|
|
|
|
Six parts. Parts 2 and 3 are the acceptance argument; parts 5 and 6 exist
|
|
because TEST-06 found three mutations that survived parts 1-4 green - the
|
|
evaluation POSITION inside cycle(), the NOT COVERED status string, and the
|
|
shipped `meter_freeze_s` itself:
|
|
|
|
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.
|
|
|
|
4. THE WIRING: the real `Controller.cycle()`, asserted on the service call
|
|
that leaves the process;
|
|
5. WHERE IN `cycle()` the detector is evaluated - by consequence, not by
|
|
source order. It runs after `write_setpoint()`, and that is what makes two
|
|
compositions safe (see the section for both);
|
|
6. WHAT THE STATUS PAGE SAYS about an uncovered transport, and the shipped
|
|
threshold read back out of `config.yaml`.
|
|
|
|
⚠️ 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 re
|
|
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
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# THE SHIPPED THRESHOLD, READ FROM config.yaml (TEST-06 / AC3)
|
|
# --------------------------------------------------------------------------- #
|
|
# ⚠️ `meter_freeze_s: 300` used to appear in this file only as the literal 300,
|
|
# retyped six times, so `300 -> 3600` in config.yaml left every check here
|
|
# green. Everything below is now driven off the SHIPPED value, and the value
|
|
# itself is pinned in test_validate.py the way MAINT-04 pins the other defaults.
|
|
#
|
|
# Regex, not PyYAML: the add-on image installs py3-aiohttp/py3-paho-mqtt only,
|
|
# and a check that skips on ImportError is not a check.
|
|
_CFG_TXT = (HERE / "config.yaml").read_text(encoding="utf-8")
|
|
|
|
|
|
def _cfg_default(key):
|
|
"""The single ` key: <number>` default line in config.yaml's options block."""
|
|
found = re.findall(rf"^ {key}: (-?\d+(?:\.\d+)?)\s*$", _CFG_TXT, re.M)
|
|
return float(found[0]) if len(found) == 1 else None
|
|
|
|
|
|
SHIPPED_HOLD_S = _cfg_default("meter_freeze_s")
|
|
SHIPPED_DEADBAND_W = _cfg_default("deadband_w")
|
|
SHIPPED_MAX_W = _cfg_default("max_w")
|
|
|
|
|
|
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)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# 1b. the SHIPPED threshold, and the two properties it has to have (AC3)
|
|
# --------------------------------------------------------------------------- #
|
|
# ⚠️ This is the number the whole ticket is an argument about, and it was pinned
|
|
# nowhere: `meter_freeze_s: 300 -> 3600` used to leave this file and
|
|
# test_validate.py both green. test_validate.py pins the literal in MAINT-04's
|
|
# manner; here it is pinned by its CONSEQUENCES, so a schema-legal value that
|
|
# gives up the argument goes red even though it is a legal value.
|
|
print("the shipped meter_freeze_s (AC3)")
|
|
check(f"config.yaml's numeric defaults were read (freeze {SHIPPED_HOLD_S}, "
|
|
f"deadband {SHIPPED_DEADBAND_W}, max_w {SHIPPED_MAX_W})",
|
|
None not in (SHIPPED_HOLD_S, SHIPPED_DEADBAND_W, SHIPPED_MAX_W))
|
|
check("the code default and the shipped default are the same number - "
|
|
"FreezeWatchdogConfig() is a default, config.yaml is the ship",
|
|
FreezeWatchdogConfig().hold_s == SHIPPED_HOLD_S)
|
|
check("this file's `tuning()` really is config.yaml's deadband, so the "
|
|
"mutual-exclusion arithmetic below is about the shipped law",
|
|
tuning().deadband_w == SHIPPED_DEADBAND_W)
|
|
|
|
# The upper bound is money: config.yaml's own comment says 300 s is "still
|
|
# inside the 900 s quarter that bills the capacity tariff". A hold that outlives
|
|
# the quarter notifies after the peak is already billed.
|
|
QUARTER_S = 900.0
|
|
check(f"the alarm still lands inside the {QUARTER_S:.0f} s billing quarter "
|
|
f"({SHIPPED_HOLD_S:.0f} s hold)", SHIPPED_HOLD_S < QUARTER_S)
|
|
# The lower bound is the conjunction itself - see the derivation below, which
|
|
# now runs off the shipped numbers rather than off two retyped literals.
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# 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(SHIPPED_HOLD_S, 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 {SHIPPED_HOLD_S:.0f} 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(SHIPPED_HOLD_S, 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(f"the shipped {SHIPPED_HOLD_S:.0f} s is at least 15x the worst of the two",
|
|
SHIPPED_HOLD_S >= 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=SHIPPED_HOLD_S))
|
|
t5 = tuning(LIVE_MAX_W)
|
|
tgt, iw, sc = 0.0, 0.0, 0
|
|
bare_would_fire = False
|
|
tel05_cmd = 0.0
|
|
# Long enough that the REJECTED bare threshold has to get its chance even if
|
|
# someone raises meter_freeze_s - the mutation must be caught by the pins above,
|
|
# not by this case going red for an incidental reason.
|
|
TEL05_S = max(3600, int(SHIPPED_HOLD_S) + 300)
|
|
for k in range(TEL05_S):
|
|
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) >= SHIPPED_HOLD_S
|
|
tel05.evaluate(float(k), tgt, LIVE_MAX_W, T0 + timedelta(seconds=k))
|
|
print(f" {TEL05_S / 60:.0f} min at -14.2 W: worst |command| {tel05_cmd:.0f} W, "
|
|
f"unchanged_s reaches {TEL05_S - 1} 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 = SHIPPED_DEADBAND_W # config.yaml's, not a retyped 15
|
|
slowest_healthy_w = 3600.0 / SHIPPED_HOLD_S
|
|
print(f" at the shipped {SHIPPED_HOLD_S:.0f} s hold, the slowest healthy "
|
|
f"meter that still looks unchanged is {slowest_healthy_w:.1f} W, against "
|
|
f"a {DEADBAND_W:.0f} W deadband")
|
|
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.
|
|
|
|
⚠️ `HERE.parents[1]` only exists when this package sits at least two levels
|
|
below the filesystem root - it raised IndexError and took the WHOLE suite
|
|
down on a checkout at `C:/goodwe_controller`, which is a failure of the
|
|
lookup, not of the thing being tested. Guarded, so a shallow checkout falls
|
|
through to the stub like any other missing sibling repo.
|
|
"""
|
|
cands = []
|
|
if os.environ.get("GOODWE_SIM"):
|
|
cands.append(Path(os.environ["GOODWE_SIM"]) / "hwsim.py")
|
|
if len(HERE.parents) > 1:
|
|
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()
|
|
# ⚠️ WHICH TRANSPORT BACKED PART 3 IS PART OF THE RESULT, and it is repeated in
|
|
# the summary at the end of the run. The two are not equivalent: pointed at the
|
|
# real `sim/hwsim.py` the rig path failed 1 run in 3 (a 16 s window against a
|
|
# 10 s hold, when hwsim.serve() is slow to start), while the stub is tuned never
|
|
# to fail. A green log that does not say which one ran is a green log that can
|
|
# have quietly substituted the weaker stand-in - the same shape as an assertion
|
|
# that cannot fail, from the other end.
|
|
PART3_TRANSPORT = f"the rig hwsim at {hw}" if hw else "the local stub"
|
|
print(f"frozen meter over HTTP - using {PART3_TRANSPORT}")
|
|
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
|
|
# ⚠️ The dead-poll-task case (AC1's second half): every input reads
|
|
# None, exactly as it does when the transport task has died while the
|
|
# controller keeps cycling.
|
|
self.dead = False
|
|
|
|
async def limits(self, entity):
|
|
return (-5000.0, 5000.0, 1.0)
|
|
|
|
async def number(self, entity, invert=False):
|
|
if self.dead:
|
|
return None
|
|
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
|
|
|
|
|
|
# ⚠️ meter_freeze_s is deliberately NOT in here. A controller built with the key
|
|
# absent must land on the shipped number by itself, and that is asserted below -
|
|
# supplying it here would test this dict instead of the shipped default.
|
|
STALE_INPUT_S = 15.0
|
|
|
|
|
|
def _opts(**over):
|
|
o = {"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": SHIPPED_DEADBAND_W,
|
|
"target_grid_w": -10, "step_w": 10, "heartbeat_s": 10,
|
|
"stale_input_s": STALE_INPUT_S}
|
|
o.update(over)
|
|
return o
|
|
|
|
|
|
# ⚠️ TWO SURFACES, AND EVERY ASSERTION BELOW SAYS WHICH ONE IT IS ABOUT.
|
|
# The first cut of this helper selected by SERVICE, which measured "any guard's
|
|
# notification" while every assertion built on it was named for THIS guard's.
|
|
# That was true only because `check_meter_frozen()` was the sole
|
|
# `persistent_notification.create` in `app/`; SAFETY-03 (#12) adds a second one
|
|
# (`goodwe_output_clamp`) raised by scenarios that rail the command on purpose,
|
|
# and eight assertions here went red against it - AC1's core check included.
|
|
#
|
|
# Blanket id-keying is NOT the repair. Three of those eight are claims about the
|
|
# WHOLE notification surface, and narrowing them to this guard's id would have
|
|
# silently shrunk a claim rather than fixed a conflation - which is the defect
|
|
# class this card exists to close. So both surfaces are available by name, and
|
|
# each assertion picks one deliberately.
|
|
FREEZE_NOTE_ID = "goodwe_meter_frozen"
|
|
|
|
|
|
def all_notes_of(hass):
|
|
"""EVERY HA persistent notification raised this run, whoever raised it."""
|
|
return [d for dom, svc, d in hass.calls
|
|
if (dom, svc) == ("persistent_notification", "create")]
|
|
|
|
|
|
def _is_freeze_note(d):
|
|
"""Is this notification the freeze detector's?
|
|
|
|
⚠️ THE ID **OR** THE TITLE, and the `or` is measured rather than tidy.
|
|
Keyed on the id alone, renaming the id empties every list in this file and
|
|
SIX assertions collapse together - so the one check whose subject IS the id
|
|
could no longer fail on its own, in the file about assertions that cannot
|
|
fail. Recognising the guard by what it says as well leaves the rename
|
|
exactly one red: the id's own check. Measured both ways:
|
|
|
|
selector vs #12's app/ id renamed
|
|
by service (the first cut) 8 red 1 red
|
|
by id alone 3 red 6 red
|
|
by id or title (this) 3 red 1 red
|
|
|
|
The union cannot re-admit another guard's alarm unless that guard also
|
|
calls itself frozen, which would be a confusable alarm on its own account.
|
|
"""
|
|
return (d.get("notification_id") == FREEZE_NOTE_ID
|
|
or "frozen" in str(d.get("title", "")).lower())
|
|
|
|
|
|
def notes_of(hass):
|
|
"""Only THIS guard's - see _is_freeze_note()."""
|
|
return [d for d in all_notes_of(hass) if _is_freeze_note(d)]
|
|
|
|
|
|
def notes_by_text_of(hass):
|
|
"""This guard's, recognised by what it SAYS and never by its id.
|
|
|
|
Exists for the two assertions whose subject IS the id: selected by the id,
|
|
both would be true by construction.
|
|
"""
|
|
return [d for d in all_notes_of(hass)
|
|
if "frozen" in str(d.get("title", "")).lower()]
|
|
|
|
|
|
def wired(unchanged_s, grid, cycles=60, **over):
|
|
"""Returns (hass, controller, trace).
|
|
|
|
`trace` is one row per cycle - (command after the cycle, the watchdog's
|
|
level, notifications raised so far) - which is what AC1 needs: the
|
|
QUESTION IS NOT WHETHER THE ALARM HAPPENS BUT ON WHICH CYCLE.
|
|
"""
|
|
async def go():
|
|
hass = _Hass(grid)
|
|
c = Controller(_opts(**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)
|
|
trace = []
|
|
for _ in range(cycles):
|
|
await c.read_inputs()
|
|
await c.cycle()
|
|
trace.append((c.target, c.freeze_wd.active, len(notes_of(hass))))
|
|
return hass, c, trace
|
|
return asyncio.run(go())
|
|
|
|
|
|
hass_f, ctl_f, trace_f = wired(3600.0, 4000.0)
|
|
notes = notes_of(hass_f) # THIS guard's
|
|
all_notes = all_notes_of(hass_f) # the whole surface
|
|
# 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)")
|
|
# SURFACE-WIDE, deliberately. "The controller alarmed once" is a different and
|
|
# stronger claim than "the freeze guard alarmed once", and it is the one this
|
|
# row has always made. When a second guard lands (SAFETY-03's
|
|
# `goodwe_output_clamp` is the first), this assertion is where somebody decides
|
|
# whether the scenario is still meant to raise exactly one alarm in total - a
|
|
# decision, made here, rather than a claim quietly narrowed to keep it green.
|
|
check("the whole notification surface carried exactly one alarm this run",
|
|
len(all_notes) == 1)
|
|
# THIS GUARD'S, keyed on the id.
|
|
check("...and the freeze detector is the one that raised it", len(notes) == 1)
|
|
# ⚠️ Selected by TEXT, not by id - see notes_by_text_of(). This is the only
|
|
# assertion whose subject is the id, so it is the one that must survive being
|
|
# the only thing a rename breaks.
|
|
check("...with a fixed notification_id so repeats replace, not stack",
|
|
[d.get("notification_id") for d in notes_by_text_of(hass_f)]
|
|
== [FREEZE_NOTE_ID])
|
|
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))
|
|
|
|
# ⚠️ THESE THREE ARE SURFACE-WIDE ON PURPOSE, and they are the three that go red
|
|
# against #12's `app/`. That is the correct signal and not a bug in them: each
|
|
# says "this scenario is quiet", which is a stronger claim than "this guard is
|
|
# quiet", and it is the claim a reader of a converged install wants. Whoever
|
|
# lands a second notifying guard has to come here and decide, per scenario,
|
|
# whether the new alarm belongs in it - `notes_of()` is one word away for the
|
|
# narrower reading, and choosing it must be a decision with a reason.
|
|
# (They also used to match `svc == "create"` in ANY domain, which was looser
|
|
# still.)
|
|
hass_n, ctl_n, _ = wired(3600.0, -14.2)
|
|
check("wired up, TEL-05's converged case leaves the whole surface silent",
|
|
not all_notes_of(hass_n))
|
|
hass_m, ctl_m, _ = wired(2.0, 8000.0)
|
|
check("wired up, a real railed load against a moving meter leaves the whole "
|
|
"surface silent",
|
|
not all_notes_of(hass_m) and abs(ctl_m.target) >= rail_w(LIVE_MAX_W))
|
|
check("an uncovered transport (no unchanged_s) leaves the whole surface silent",
|
|
not all_notes_of(wired(None, 4000.0)[0]))
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# 5. WHERE in cycle() the detector is evaluated (TEST-06 / AC1)
|
|
# --------------------------------------------------------------------------- #
|
|
# ⚠️ WHERE THE BOUNDARY ACTUALLY IS - AND IT IS NOT write_setpoint().
|
|
# The first version of this comment said the detector runs "after
|
|
# write_setpoint()" and grades "the command that has just gone on the wire".
|
|
# TEST-06's reviewer disproved both halves by mutation:
|
|
#
|
|
# N3 move check_meter_frozen() to immediately BEFORE write_setpoint()
|
|
# -> green, and CORRECTLY green
|
|
#
|
|
# because `self.target = resolution.target_w` is assigned two lines above the
|
|
# write, `write_setpoint()` clamps only a local `value`, and it can skip the
|
|
# write entirely (`if not (changed or force): return`). So what the detector
|
|
# grades is `self.target` - the arbiter's resolved intent, PRE device clamp and
|
|
# possibly never emitted - and the invariant these checks pin is:
|
|
#
|
|
# check_meter_frozen() runs AFTER `self.target = resolution.target_w`
|
|
#
|
|
# That distinction is not pedantry. A check more robust than its own
|
|
# explanation is how the next person moves the call ABOVE the assignment
|
|
# believing they had preserved the stated rule.
|
|
#
|
|
# Both properties below are asserted by an OBSERVABLE CONSEQUENCE rather than by
|
|
# source order, so semantic-preserving refactors stay green (measured by the
|
|
# reviewer: renaming the method, and calling it from a helper that itself runs
|
|
# last, both survive - the suite drives `await c.cycle()` and never names the
|
|
# method):
|
|
#
|
|
# (a) BOTH TERMS MUST DESCRIBE THE SAME MOMENT. The command it grades has to
|
|
# be the one THIS cycle resolved, judged against how long the meter that
|
|
# chose it has stood still. Evaluated above the assignment it grades the
|
|
# PREVIOUS cycle's command against THIS cycle's meter - a pairing that was
|
|
# never true of any moment - and the alarm arrives a whole cycle late.
|
|
# (b) THE DEAD-POLL-TASK COMPOSITION. When the poll task dies the inputs go
|
|
# stale, the arbiter resolves 0 W, and the detector - evaluated after that
|
|
# resolution lands in self.target - correctly sees an unrailed command and
|
|
# CLEARS. Evaluated above the assignment it would still be looking at the
|
|
# railed command from before the failsafe, and would keep the alarm
|
|
# standing over a controller that is already commanding 0 W.
|
|
print("WHERE the detector is evaluated in cycle() - the ordering AC1 protects")
|
|
|
|
_railed = [i for i, (t, _, _) in enumerate(trace_f)
|
|
if abs(t) >= rail_w(LIVE_MAX_W)]
|
|
_alarmed = [i for i, (_, _, n) in enumerate(trace_f) if n >= 1]
|
|
_first_railed = _railed[0] if _railed else None
|
|
_first_alarm = _alarmed[0] if _alarmed else None
|
|
print(f" ramp to the rail: first railed command on cycle {_first_railed}, "
|
|
f"first notification on cycle {_first_alarm}, over {len(trace_f)} cycles")
|
|
# Without this the equality below could be satisfied on cycle 0, where "the same
|
|
# cycle" and "one cycle later" are indistinguishable.
|
|
check("the command takes several cycles to reach the rail, so 'same cycle' and "
|
|
"'one cycle later' are different observations",
|
|
_first_railed is not None and _first_railed >= 1
|
|
and len(trace_f) > _first_railed + 1)
|
|
check("the alarm lands on the SAME cycle the command reaches the rail - "
|
|
"evaluated after `self.target = resolution.target_w` it grades the "
|
|
"command THIS cycle resolved; evaluated above that assignment it grades "
|
|
"the previous one and arrives a cycle late",
|
|
_first_alarm is not None and _first_alarm == _first_railed)
|
|
|
|
|
|
def dead_poll(after_cycles=20, dead_cycles=4, revived_cycles=20):
|
|
"""Alarm on a railed command, then kill the poll task and keep cycling.
|
|
|
|
The real thing: the transport task dies, every input reads None, and after
|
|
`stale_input_s` the arbiter's P_SAFETY stop resolves 0 W. `inputs_bad_since`
|
|
is backdated rather than slept through - the clock is wall clock, and the
|
|
mechanism rather than the wait is what is under test.
|
|
"""
|
|
async def go():
|
|
hass = _Hass(4000.0)
|
|
c = Controller(_opts(), hass, _Store(), _Pub(), stop=asyncio.Event())
|
|
c.auto = True
|
|
c.dev_min, c.dev_max = -5000.0, 5000.0
|
|
c.p1_source = _Source(3600.0)
|
|
for _ in range(after_cycles):
|
|
await c.read_inputs()
|
|
await c.cycle()
|
|
before = (c.target, c.freeze_wd.active, len(notes_of(hass)))
|
|
hass.dead = True
|
|
trace = []
|
|
for _ in range(dead_cycles):
|
|
await c.read_inputs()
|
|
if c.inputs_bad_since is not None:
|
|
c.inputs_bad_since = (datetime.now(timezone.utc)
|
|
- timedelta(seconds=STALE_INPUT_S + 1))
|
|
await c.cycle()
|
|
trace.append((c.target, c.freeze_wd.active, len(notes_of(hass))))
|
|
# ...and then the transport comes back, against a meter that is still
|
|
# frozen. The loop re-rails and this is a NEW episode - the alarm is
|
|
# allowed, indeed required, to speak again.
|
|
hass.dead = False
|
|
revived = []
|
|
for _ in range(revived_cycles):
|
|
await c.read_inputs()
|
|
await c.cycle()
|
|
revived.append((c.target, c.freeze_wd.active, len(notes_of(hass))))
|
|
# ⚠️ by TEXT: "the second episode REPLACED the first" is a claim about
|
|
# the id, so the list it is checked against must not be selected by the
|
|
# id - that would make it true by construction.
|
|
return before, trace, revived, notes_by_text_of(hass), c
|
|
return asyncio.run(go())
|
|
|
|
|
|
_before, _dead, _revived, _freeze_notes, _ctl_d = dead_poll()
|
|
_zeroed = [i for i, (t, _, _) in enumerate(_dead) if t == 0.0]
|
|
print(f" poll task killed: alarming at {_before[0]:+.0f} W "
|
|
f"(active={_before[1]}, {_before[2]} notification(s)) -> "
|
|
+ ", ".join(f"cycle {i}: {t:+.0f} W active={a} notes={n}"
|
|
for i, (t, a, n) in enumerate(_dead)))
|
|
check("the episode really was standing before the poll task died - otherwise "
|
|
"the release below is vacuous",
|
|
_before[1] and _before[2] == 1 and abs(_before[0]) >= rail_w(LIVE_MAX_W))
|
|
check("a dead poll task drives the command to 0 W",
|
|
bool(_zeroed) and _zeroed[0] == 0)
|
|
check("...and on THAT SAME cycle the alarm is already clear, because the "
|
|
"detector ran after 0 W was resolved into self.target, not before it",
|
|
bool(_zeroed) and not _dead[_zeroed[0]][1])
|
|
check("...and it stays clear for as long as the failsafe holds it there",
|
|
_dead and not any(a for _, a, _ in _dead))
|
|
# ⚠️ "and it raises nothing new while parked" is NOT asserted here, and the
|
|
# reason matters more than the deletion. It was dropped as REDUNDANT, not as
|
|
# undecidable: TEST-06's reviewer falsified it in one line - move the
|
|
# `hass.call` above `if not fired: return` so the detector notifies every cycle,
|
|
# and notifications appear in exactly these parked cycles (that mutation takes
|
|
# this file RED 9). My first note here claimed no mutation could break it, which
|
|
# was the very inference this file exists to distrust, made about this file.
|
|
# The deletion stands because the exact count below catches that same mutation
|
|
# nine times over - no coverage was lost - and the honest form of the claim is
|
|
# "I tried to break it and here is what happened", never "I could not, so it
|
|
# cannot".
|
|
|
|
# ⚠️ AND THE OTHER HALF, or "no second notification" is a claim no mutation can
|
|
# break: when the transport comes back against a meter that is STILL frozen the
|
|
# loop re-rails, and that is a new episode which MUST speak. What must not
|
|
# happen is a second notification stacking up beside the first.
|
|
_re_railed = [i for i, (t, _, _) in enumerate(_revived)
|
|
if abs(t) >= rail_w(LIVE_MAX_W)]
|
|
print(f" transport revived: command re-rails on cycle "
|
|
f"{_re_railed[0] if _re_railed else None}, "
|
|
f"{_ctl_d.freeze_wd.episodes} episode(s), {len(_freeze_notes)} "
|
|
f"notification(s) from this guard, "
|
|
f"ids {sorted({n.get('notification_id') for n in _freeze_notes})}")
|
|
# A PRECONDITION, not a discriminating check - the same role as "the command
|
|
# takes several cycles to reach the rail" above, and labelled the same way:
|
|
# without it the two checks below are satisfied by a controller that never came
|
|
# back. The crudest falsifier (never file the loop claim) takes most of this
|
|
# file red, which is the signature of a precondition rather than a check; the
|
|
# narrowest one measured is a stale-input stop that never releases.
|
|
check("PRECONDITION: a revived transport against a still-frozen meter re-rails "
|
|
"the command",
|
|
bool(_re_railed))
|
|
check("...which is a SECOND episode, and it alarms again on the cycle it rails",
|
|
_ctl_d.freeze_wd.episodes == 2
|
|
and bool(_re_railed) and _revived[_re_railed[0]][2] == _before[2] + 1)
|
|
# ⚠️ STABILITY, not the value - the value is pinned once, above. Renaming the
|
|
# id must fail that check and NOT this one; the two say different things and
|
|
# each has to be able to go red on its own.
|
|
check("...on the same notification_id, so the second REPLACES the first "
|
|
"instead of stacking a second one beside it",
|
|
len(_freeze_notes) == 2
|
|
and len({n.get("notification_id") for n in _freeze_notes}) == 1)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# 6. what the status page SAYS about coverage (TEST-06 / AC2)
|
|
# --------------------------------------------------------------------------- #
|
|
# ⚠️ The check above only ever asserted that an uncovered transport is SILENT,
|
|
# which a status page claiming `armed` on a guard that cannot run also satisfies.
|
|
# The AC was that the detector reports uncovered transports *and says so*, so
|
|
# the string is what gets asserted here, in both directions: the uncovered row
|
|
# must not claim `armed`, and the covered row must still claim it - or "does not
|
|
# say armed" would be satisfiable by deleting the armed branch outright.
|
|
print("what the status page SAYS about coverage")
|
|
|
|
|
|
def freeze_rows(unchanged_s, grid, **over):
|
|
"""The freeze-detector rows of the real checks(), on a p1 install.
|
|
|
|
⚠️ WHAT THIS DOES NOT COVER. The rows live behind `if self.p1_enabled:`, and
|
|
the flag is set here by hand rather than by building a real P1 transport -
|
|
inherited from part 4's fixture. So everything below proves "WHEN the row is
|
|
produced it says the right thing", and nothing below proves "the row is
|
|
produced on homewizard_local and not otherwise". That gate is uncovered;
|
|
AC2 is about the string, and widening the fixture belongs to whoever covers
|
|
the gate.
|
|
"""
|
|
_h, ctl, _t = wired(unchanged_s, grid, cycles=3, **over)
|
|
# p1_enabled is what a homewizard_local install has; the row's own content
|
|
# is what is under test, not the transport that feeds it.
|
|
ctl.p1_enabled = True
|
|
return ctl, [r for r in ctl.checks()
|
|
if r["text"].startswith("freeze detector")]
|
|
|
|
|
|
_ctl_u, _rows_u = freeze_rows(None, 4000.0, meter_source="ha_dsmr")
|
|
_ctl_c, _rows_c = freeze_rows(2.0, -14.2)
|
|
print(f" uncovered: {_rows_u[0]['text'] if _rows_u else '(no row at all)'}")
|
|
print(f" covered: {_rows_c[0]['text'] if _rows_c else '(no row at all)'}")
|
|
check("an uncovered transport gets exactly one freeze-detector row",
|
|
len(_rows_u) == 1)
|
|
check("...and it says NOT COVERED",
|
|
bool(_rows_u) and "NOT COVERED" in _rows_u[0]["text"])
|
|
check("...and it does NOT claim the guard is armed - that would be the status "
|
|
"page lying on behalf of a guard that cannot run",
|
|
bool(_rows_u) and "armed" not in _rows_u[0]["text"])
|
|
check("...and it names the meter_source it is not covering, so the operator "
|
|
"knows which install this is",
|
|
bool(_rows_u) and "ha_dsmr" in _rows_u[0]["text"])
|
|
check("...and it names homewizard_local as the transport that would be covered",
|
|
bool(_rows_u) and "homewizard_local" in _rows_u[0]["text"])
|
|
check("...flagged as a warning rather than a silent ok",
|
|
bool(_rows_u) and not _rows_u[0]["ok"] and _rows_u[0]["warn"])
|
|
|
|
check("a COVERED transport gets exactly one freeze-detector row",
|
|
len(_rows_c) == 1)
|
|
check("...and that one DOES say armed - without this, 'the uncovered row does "
|
|
"not say armed' is satisfied by deleting the armed branch",
|
|
bool(_rows_c) and "armed" in _rows_c[0]["text"]
|
|
and "NOT COVERED" not in _rows_c[0]["text"])
|
|
check("...and it quotes the hold and the rail it is armed at, derived from the "
|
|
"config the controller was actually built with",
|
|
bool(_rows_c)
|
|
and f"{_ctl_c.freeze_wd.cfg.hold_s:.0f} s unchanged" in _rows_c[0]["text"]
|
|
and f"{rail_w(LIVE_MAX_W):.0f} W" in _rows_c[0]["text"])
|
|
check("...as an ok row, since a covered armed guard is not a warning",
|
|
bool(_rows_c) and _rows_c[0]["ok"] and not _rows_c[0]["warn"])
|
|
|
|
# AC3's other half, through the built object rather than through a literal: the
|
|
# opts dict does NOT carry meter_freeze_s, so this is the number an install that
|
|
# leaves the key at its config.yaml default actually lands on.
|
|
print(f" a Controller built with meter_freeze_s ABSENT holds "
|
|
f"{_ctl_c.freeze_wd.cfg.hold_s:.0f} s (config.yaml ships "
|
|
f"{SHIPPED_HOLD_S:.0f} s)")
|
|
check("a Controller built with the key absent lands on the SHIPPED hold, so "
|
|
"config.yaml and main.py's fallback cannot drift apart unnoticed",
|
|
_ctl_c.freeze_wd.cfg.hold_s == SHIPPED_HOLD_S)
|
|
|
|
print()
|
|
print(f"part 3 (the live HTTP transport) ran against: {PART3_TRANSPORT}")
|
|
if fails:
|
|
print(f"{len(fails)} FAILED: {', '.join(fails)}")
|
|
sys.exit(1)
|
|
print("all checks passed")
|