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 ~35 SECONDS OF WALL CLOCK, on purpose. 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.

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

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):
        self.meter = meter
        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):
        if entity == "sensor.grid":
            return self.meter(time.monotonic() - self.t0)
        if entity == "sensor.batt":
            return self.batt
        if entity == "sensor.soc":
            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):
    """Run the real control loop for `seconds`; return (cycle times, writes)."""
    async def go():
        hass = FakeHass(meter)
        c = Controller(dict(BASE, cycle_floor_s=floor), 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
    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()
if fails:
    print(f"{len(fails)} FAILED: " + "; ".join(fails))
    sys.exit(1)
print("all checks passed")