goodwe-addon/goodwe_controller/test_watchdog.py

"""Runnable check for the sign-disagreement watchdog.  `python3 test_watchdog.py`

Every case here is either a fault this must catch or a piece of NORMAL behaviour
it must not mistake for one. The second list is the longer one on purpose: an
alarm that fires on ordinary operation is worse than no alarm, because the next
person switches it off.

The headline validation is not here - it is `sim/replay_sign.py` in the infra
repo, which drives this module over the two captured bus days. This file is the
table-driven half: boundaries, timers, the latch, and the arming gate.
"""

import sys
from datetime import datetime, timedelta, timezone

from app.watchdog import SignWatchdog, SignWatchdogConfig

fails = []
T0 = datetime(2026, 8, 20, 15, 0, 0, tzinfo=timezone.utc)


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


def drive(samples, cfg=None, following=True):
    """Feed (offset_s, commanded_w, grid_w) tuples. Returns (wd, trip_offset)."""
    wd = SignWatchdog(cfg)
    trip_at = None
    for item in samples:
        offset, cmd, grid = item[0], item[1], item[2]
        f = item[3] if len(item) > 3 else following
        if wd.evaluate(cmd, grid, f, T0 + timedelta(seconds=offset)) and trip_at is None:
            trip_at = offset
    return wd, trip_at


def opposed(duration_s, step=5, cmd=-1000, grid=1500, following=True, start=0):
    return [(start + t, cmd, grid, following)
            for t in range(0, int(duration_s) + 1, step)]


print("the fault it exists to catch")
wd, at = drive(opposed(300))
check("charging into an importing house trips", wd.tripped)
check("and trips at 180 s, not before", at == 180)
check("the reason names both signals", "-1000" in wd.reason and "+1500" in wd.reason)
print(f"       e.g. {wd.reason!r}")

wd, at = drive(opposed(300, cmd=2000, grid=-2000))
check("discharging into an exporting house trips too", wd.tripped and at == 180)

print()
print("the boundaries")
wd, _ = drive(opposed(600, cmd=-300, grid=1500))
check("300 W exactly does NOT qualify (strictly greater)", not wd.tripped)
wd, _ = drive(opposed(600, cmd=-301, grid=1500))
check("301 W does", wd.tripped)
wd, _ = drive(opposed(600, cmd=-1000, grid=300))
check("a 300 W meter does not qualify either", not wd.tripped)
wd, _ = drive(opposed(175))
check("175 s of disagreement does not trip", not wd.tripped)
wd, _ = drive(opposed(180))
check("180 s does (>= hold_s)", wd.tripped)

print()
print("continuity - the rule that produced zero false positives on real data")
broken = opposed(175) + [(180, -1000, -1500)] + opposed(175, start=185)
wd, _ = drive(broken)
check("179 s, one agreeing sample, then 175 s more: no trip", not wd.tripped)
check("and the window really was reset", wd.disagree_since is not None)
long_enough = broken + opposed(200, start=365)
wd, _ = drive(long_enough)
check("but a full 180 s after the reset does trip", wd.tripped)

print()
print("things that are NOT faults")
# The load-step transient. Measured on the healthy capture: longest 30 s.
wd, _ = drive(opposed(30) + [(35 + t, 800, 900) for t in range(0, 300, 5)])
check("a 30 s load-step crossover does not trip", not wd.tripped)

# SoC 10 %, Standby, declining a 1350 W discharge. The battery does nothing, so
# the meter stays where it was and the command agrees with it.
wd, _ = drive([(t, 1350, 1350) for t in range(0, 900, 5)])
check("a battery refusing a discharge command cannot trip it", not wd.tripped)

# A frozen meter drives command and reading into agreement, not disagreement.
wd, _ = drive([(t, 800, 800) for t in range(0, 900, 5)])
check("a frozen meter cannot trip it", not wd.tripped)

# Maintenance grid charge: 2500 W of charge against an importing house, for
# hours. Opposed, both huge - and entirely correct.
wd, _ = drive(opposed(7200, cmd=-2500, grid=3000, following=False))
check("two hours of maintenance grid charge does not trip", not wd.tripped)
check("and the timer never even started", wd.disagree_since is None)

wd, _ = drive([(t, -1000, None) for t in range(0, 900, 5)])
check("a missing meter reading does not trip", not wd.tripped)
wd, _ = drive(opposed(175) + [(180, -1000, None)] + opposed(600, start=185))
check("a missing reading mid-window RESETS it rather than accumulating",
      wd.tripped and wd.disagree_since is not None)
wd, at = drive(opposed(175) + [(180, -1000, None)] + opposed(600, start=185))
check("...so the trip lands 180 s after the gap, not after 180 s total",
      at == 365)

wd, _ = drive([(t, 0.0, 3000) for t in range(0, 900, 5)])
check("a 0 W command is not a wrong-direction command", not wd.tripped)

print()
print("the latch")
wd, _ = drive(opposed(200))
check("tripped after the fault", wd.tripped)
# The trip commands 0 W, which removes the command side of the comparison.
for t in range(200, 900, 5):
    wd.evaluate(0.0, 3000, True, T0 + timedelta(seconds=t))
check("stays tripped once the command is zeroed", wd.tripped)
for t in range(900, 1500, 5):
    wd.evaluate(1000, 1000, True, T0 + timedelta(seconds=t))
check("stays tripped even once everything looks healthy again", wd.tripped)
wd.reset()
check("reset() is the only way out", not wd.tripped and wd.disagree_since is None)
check("and the reason is cleared with it", wd.reason == "")

print()
print("the thresholds are configuration, not constants")
cfg = SignWatchdogConfig(min_w=800.0, hold_s=60.0)
wd, at = drive(opposed(300, cmd=-1000, grid=1500), cfg=cfg)
check("a 60 s hold trips at 60 s", wd.tripped and at == 60)
wd, _ = drive(opposed(600, cmd=-700, grid=1500), cfg=cfg)
check("an 800 W gate ignores a 700 W command", not wd.tripped)

print()
print("the configuration that would disarm it is refused")
from app.validate import validate_startup_config  # noqa: E402

base = {"setpoint_entity": "number.sp", "meter_source": "off",
        "meter_phases": 1, "meter_max_age_s": 30}
p = validate_startup_config({**base, "max_w": 2000, "sign_disagree_w": 300},
                            -5000, 5000)
check("300 W under a 2000 W ceiling is accepted", p == [])
p = validate_startup_config({**base, "max_w": 2000, "sign_disagree_w": 2000},
                            -5000, 5000)
check("a threshold AT max_w is refused", len(p) == 1 and "never fire" in p[0])
p = validate_startup_config({**base, "max_w": 500, "sign_disagree_w": 3000},
                            -5000, 5000)
check("a threshold above max_w is refused", len(p) == 1 and "never fire" in p[0])
p = validate_startup_config({**base, "max_w": 2000, "sign_disagree_w": 1999},
                            -5000, 5000)
check("one watt under the ceiling is reachable, so accepted", p == [])

print()
print("end to end: the trip is COMMAND ZERO, not alarm-only")
# ⚠️ THE FAULT INJECTED HERE IS A NEGATIVE GAIN, and that is not a contrivance -
# it is the minimal faithful model of what this watchdog is for. The check
# compares the loop's OUTPUT DIRECTION against the INPUT STATE it was regulating,
# so what it detects is a loop whose two ends have become decoupled in sign: a
# gain, a claim or a strategy with its polarity inverted, or an integrator so far
# wound up that unwinding takes minutes. (Illusmart's was the last one: it
# reached -14 768 W against a +-5000 W rail while writing once every 20-60 s, and
# that arithmetic alone is 5-15 minutes of wrong-direction command.)
#
# `gain` is float(0.05,1.0) in the schema, so this is not a reachable
# configuration - it is a stand-in for a code regression, which is the case a
# unit test can actually hold the line on.
import asyncio  # noqa: E402

from app.main import Controller  # noqa: E402


class _Hass:
    def __init__(self):
        self.writes = []

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

    async def number(self, entity, invert=False):
        # A house importing 2000 W, steadily, whatever the battery does.
        return {"sensor.grid": 2000.0, "sensor.soc": 50.0,
                "sensor.batt": 0.0}.get(entity)

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


class _S:
    data = {}

    def set(self, *a):
        pass

    def get_time(self, *a):
        return None


class _P:
    enabled = False

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

    def close(self):
        pass


async def _e2e():
    hass = _Hass()
    opts = {"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,
            "sign_disagree_w": 300, "sign_disagree_s": 1, "auto_start": True}
    c = Controller(opts, hass, _S(), _P(), stop=asyncio.Event())
    c.auto = True
    c.dev_min, c.dev_max = -5000.0, 5000.0
    seen_wrong_way = False
    for _ in range(12):
        await c.read_inputs()
        await c.cycle()
        if c.target < -300 and not c.sign_wd.tripped:
            seen_wrong_way = True
        await asyncio.sleep(0.25)
    return c, hass, seen_wrong_way


ctrl, hass, wrong_way = asyncio.run(_e2e())
check("the broken loop really did command charge into an import first", wrong_way)
check("the watchdog tripped", ctrl.sign_wd.tripped)
check("the commanded value is 0 W", ctrl.target == 0.0)
check("0 W actually reached the device", 0.0 in hass.writes)
check("and it got there through the failsafe claim, not by accident",
      "failsafe" in ctrl.reason)
print(f"       status line: {ctrl.reason!r}")
check("the status page says so in words",
      any("SIGN-DISAGREEMENT FAILSAFE" in c["text"] for c in ctrl.checks()))
check("the banner leads with it",
      "FAILSAFE" in ctrl.status_payload()["banner"])
check("MQTT reports the failsafe state", ctrl.sign_wd.tripped)
check("it stays at 0 W on later cycles too",
      hass.writes[-1] == 0.0 and hass.writes.count(0.0) >= 1)

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