goodwe-addon/goodwe_controller/test_freeze.py

"""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.
"""

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)
notes = [d for dom, svc, d in hass_f.calls
         if (dom, svc) == ("persistent_notification", "create")]
# 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 [1 for dom, svc, _ in hass_n.calls if svc == "create"])
hass_m, ctl_m = wired(2.0, 8000.0)
check("wired up, a real railed load against a moving meter raises nothing",
      not [1 for dom, svc, _ in hass_m.calls if svc == "create"]
      and abs(ctl_m.target) >= rail_w(LIVE_MAX_W))
check("an uncovered transport (no unchanged_s) raises nothing, and says so",
      not [1 for dom, svc, _ in wired(None, 4000.0)[0].calls if svc == "create"])

print()
if fails:
    print(f"{len(fails)} FAILED: {', '.join(fails)}")
    sys.exit(1)
print("all checks passed")