goodwe-addon/goodwe_controller/test_loop.py

"""Runnable check for the control loop's cadence.  `python3 test_loop.py`

SAFETY-12: the loop must evaluate on a clock as well as on a change. Before the
fix, `run_control()` ran `cycle()` only when the meter value differed from the
previous poll, so a meter that repeats a value ran NO control cycles while the
heartbeat kept re-emitting the last command - measured on the rig during TEL-05
as 70 s at a literally constant 350.0 W with 14 arrivals and zero cycles.

⚠️ THIS TEST TAKES ~95 SECONDS OF WALL CLOCK (measured 93 s), on purpose, and
that figure is the only thing standing between it and a future developer who
decides a slow test is a broken one. The thing under test is a wall-clock floor
inside a loop that sleeps 1 s; mocking the clock would leave the assertion
testing the mock. It drives the real `run_control()` and counts what actually
happened. Eight runs of the real loop is what the eight scenarios cost.

⚠️ If you shorten the windows, re-measure and update this number and the one in
HANDOVER.md. An understated runtime reads as a hang.

`cycle_floor_s` is read from options with no lower bound in code, so a huge
value reproduces the pre-fix behaviour exactly. Every claim below is therefore
an A/B on one binary, not a comparison against a remembered old one.

⚠️ MUTATING THE MODULE THIS FILE GRADES? PURGE `__pycache__` FIRST - or run
`tools/mutation_run.sh`, which does. `python3 -B` alone does NOT save you: it
stops bytecode being WRITTEN, not a stale one being READ, and a size-preserving
edit reverted inside one second is invisible to CPython's staleness test. That
is TEST-15, and it has already turned one live mutation into a false survivor.
Read MUTATION.md before you report a survivor.
"""

import asyncio
import sys
import time

from app.main import Controller

fails = []
NEVER = 1e9          # cycle_floor_s so large the floor can never elapse: pre-fix


def check(name, cond):
    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
    if not cond:
        fails.append(name)


class FakeHass:
    """A scripted meter and a plant that tracks the command instantly."""

    def __init__(self, meter, soc_dies_at=None):
        self.meter = meter
        self.soc_dies_at = soc_dies_at
        self.batt = 0.0
        self.writes = []
        self.t0 = time.monotonic()

    async def limits(self, entity):
        return (-5000.0, 5000.0, 1.0)

    async def number(self, entity, invert=False):
        now = time.monotonic() - self.t0
        if entity == "sensor.grid":
            return self.meter(now)
        if entity == "sensor.batt":
            return self.batt
        if entity == "sensor.soc":
            if self.soc_dies_at is not None and now >= self.soc_dies_at:
                return None
            return 50.0
        return None

    async def set_number(self, entity, value):
        self.writes.append(float(value))
        self.batt = float(value)
        return True


class _Store:
    data = {}

    def set(self, *a):
        pass


class _Pub:
    enabled = False

    def publish(self, *a, **k):
        pass

    def close(self):
        pass


BASE = {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
        "soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
        "meter_source": "off", "max_w": 2000, "gain": 0.6, "slew_w": 1000,
        "deadband_w": 15, "target_grid_w": -10, "step_w": 10,
        "heartbeat_s": 10, "stale_input_s": 15}


def drive(meter, floor, seconds, soc_dies_at=None, **over):
    """Run the real control loop for `seconds`; return (cycle times, writes)."""
    async def go():
        hass = FakeHass(meter, soc_dies_at)
        c = Controller(dict(BASE, cycle_floor_s=floor, **over), hass, _Store(),
                       _Pub(), stop=asyncio.Event())
        c.auto = True
        stamps = []
        real = c.cycle

        async def counted():
            stamps.append(time.monotonic() - hass.t0)
            await real()

        c.cycle = counted
        task = asyncio.create_task(c.run_control())
        await asyncio.sleep(seconds)
        c.stopping = True
        await asyncio.wait_for(task, timeout=10)
        return stamps, hass.writes, c.target, c.reason
    return asyncio.run(go())


def worst_gap(stamps, seconds):
    """Longest stretch with no cycle, including the tail after the last one."""
    edges = [0.0] + list(stamps) + [seconds]
    return max(b - a for a, b in zip(edges, edges[1:]))


# --------------------------------------------------------------------------- #
print("a meter that repeats the same value")
FLAT = lambda t: 350.0                                          # noqa: E731
FLOOR, SECONDS = 3.0, 12.0

pre_stamps, pre_writes, _, _ = drive(FLAT, NEVER, SECONDS)
# The defect itself, demonstrated rather than described: one cycle at startup
# and then nothing for the rest of the window, while writes keep happening.
check("pre-fix: a constant meter runs exactly one cycle, at startup",
      len(pre_stamps) == 1)
check("...while the heartbeat keeps writing anyway, so nothing looks wrong",
      len(pre_writes) >= 2)

post_stamps, post_writes, _, _ = drive(FLAT, FLOOR, SECONDS)
check("post-fix: the same constant meter runs cycles throughout",
      len(post_stamps) >= int(SECONDS / FLOOR) - 1)
# The actual acceptance criterion: a BOUND on the gap, not a count.
gap = worst_gap(post_stamps, SECONDS)
check(f"...with no gap longer than the floor ({gap:.1f}s <= {FLOOR + 1.5:.1f}s)",
      gap <= FLOOR + 1.5)
check("...and the loop actually corrected, instead of holding a stale command",
      len(post_writes) > len(pre_writes))

# --------------------------------------------------------------------------- #
print("a meter that changes normally - nothing may move")
# ⚠️ THE REGRESSION GUARD, and the more important half. The floor is a deadline
# on the gap between cycles, reset by every cycle whichever trigger fired - not
# a second scheduler. On a healthy meter it must never elapse, so the cycle
# count, the write count and the commanded values must be IDENTICAL either side
# of the fix. Anything else is a new write cadence, which SAFETY-12 forbids:
# the deadband, step_w and the 10 s force-write decide what reaches the wire,
# and this change is only allowed to alter when the loop thinks.
STEPS = [0.0, 300.0, 850.0, -400.0, 120.0, -900.0, 60.0]
LIVE = lambda t: STEPS[int(t // 2) % len(STEPS)]                # noqa: E731

a_stamps, a_writes, _, _ = drive(LIVE, NEVER, 12.0)
b_stamps, b_writes, _, _ = drive(LIVE, FLOOR, 12.0)
check("a changing meter runs the same number of cycles as before the fix",
      len(a_stamps) == len(b_stamps))
check("...commanding exactly the same sequence of values", a_writes == b_writes)

# --------------------------------------------------------------------------- #
print("a meter parked at the target - the converged operating point")
# The case TEL-05 warned about: at the converged target the power figure
# legitimately repeats for minutes. The loop must still run, and must still
# write nothing but the heartbeat, because the deadband - not the arrival
# trigger - is what holds the command still.
AT_TARGET = lambda t: -10.0                                     # noqa: E731
c_stamps, c_writes, _, _ = drive(AT_TARGET, NEVER, 12.0)
d_stamps, d_writes, _, _ = drive(AT_TARGET, FLOOR, 12.0)
check("the loop evaluates at the converged point too", len(d_stamps) > len(c_stamps))
check("...and the deadband still holds the command, so writes are unchanged",
      d_writes == c_writes)

# --------------------------------------------------------------------------- #
print("an input dies while the meter repeats - the safety claim must still fire")
# ⚠️ THE STRONGEST PROPERTY IN SAFETY-12, and the one that makes it a safety fix
# rather than a latency fix. The `inputs missing -> 0 W` claim lives INSIDE
# cycle() (main.py, the stale_input_s branch and the P_SAFETY claim below it).
# Pre-fix, on a meter whose value happened to repeat, cycle() was never called
# again - so that claim could not fire AT ALL. The controller held its last
# command through an input outage of unbounded length with the status line still
# reading "loop -> N W".
#
# Layer 1 does not cover this either on a `meter_source: off` install: no P1 age
# sensor is published (see Controller.publish), so the firmware has nothing to
# veto on, and the heartbeat keeps its watchdog fed the whole time.
#
# stale_input_s is shortened to 3 s here so the outage resolves inside a test
# rather than inside a quarter of a minute. The mechanism is the default one.
c2_stamps, c2_writes, pre_target, pre_reason = drive(
    FLAT, NEVER, 10.0, soc_dies_at=4.0, stale_input_s=3)
check("pre-fix: a dead input during a constant meter never reaches the safety claim",
      pre_target != 0.0 and "safety" not in pre_reason)
check(f"...so it holds its last command instead ({pre_target:g} W, {pre_reason!r})",
      pre_target == 220.0)

d2_stamps, d2_writes, post_target, post_reason = drive(
    FLAT, FLOOR, 10.0, soc_dies_at=4.0, stale_input_s=3)
check("post-fix: the same outage commands 0 W", post_target == 0.0)
check(f"...through the safety claim, not by accident ({post_reason!r})",
      "safety" in post_reason)
check("...and the 0 W actually reached the device", 0.0 in d2_writes)

# --------------------------------------------------------------------------- #
print()
if fails:
    print(f"{len(fails)} FAILED: " + "; ".join(fails))
    sys.exit(1)
print("all checks passed")