goodwe-addon/goodwe_controller/test_clamp.py

"""Runnable check for SAFETY-03, the output clamp and its winding-loop alarm.

    python3 test_clamp.py

No framework, asserts only, same house style as test_freeze.py.

Five parts, and parts 3 and 4 are the ones that carry the acceptance criteria:

  1. THE CLAMP ARITHMETIC, on the shipped `clamp_to_rating()` - boundaries,
     symmetry, the device second stage, and the non-finite cases. Table driven.
  2. THE ALARM'S CADENCE - rising edge, re-assert, hysteresis - on the shipped
     `ClampWatchdog`. Every engagement counted, notifications not flooded.
  3. AC 1 + AC 3, WIRED. A computed setpoint that exceeds the rating, driven
     through the REAL `Controller.cycle()`: nothing over the rating reaches the
     device, and the alarm fires. Includes the historical runaway from
     specs.md 10.3 replayed against an unclamped transcription, so the check
     says what the clamp is actually worth in watts rather than asserting a
     constant.
  4. AC 2, AND IT IS THE LOAD-BEARING HALF. A loop that SATURATES WITHOUT
     CLAMPING - the inverter refusing to follow a perfectly legal command - must
     raise NO clamp alarm. If it does, the bug signal is inside the noise again
     and the whole ticket has bought nothing.
  5. NO PATH BYPASSES IT. write_setpoint() is the only route to the wire, so it
     is called directly with an out-of-range value the way the heartbeat and the
     operator's buttons call it.

------------------------------------------------------------------------------
WHY PART 4 ALSO ASSERTS ON `Decision`'s BOOLEANS AND NOT ON `reason`

`Decision.reason` is one string and the stages after the clamp overwrite it. The
check below drives a cycle that clamps the integrator AND is then slew limited,
and asserts that `reason` does NOT say so while `i_clamped` does. That is not a
style preference - an alarm keyed on the string is silent on exactly the cycles a
winding loop produces, and this is the check that would go red if someone
"simplified" main.py back onto it.

------------------------------------------------------------------------------
WHAT THIS CHECK DOES NOT COVER

- The firmware-side magnitude clamp (the card's "two clamps, not one"). This
  add-on's `write_setpoint()` is the add-on half; the ESP32 half is DRIVER-07's
  and is not in this repo's Python.
- A rig run. Nothing here needs one: the clamp is arithmetic on the write path
  and part 3 drives the shipped `Controller.cycle()` against a stub Home
  Assistant, which is the same surface test_freeze.py part 1-2 use. The wire-
  level invariant the card asks for (`|emitted| <= rating` on every frame) is
  asserted here over every value that reached `set_number()`, which is the last
  thing this process controls; proving it at the inverter needs the witness log
  and belongs with DRIVER-07.

⚠️ 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 math
import sys
from datetime import datetime, timedelta, timezone

from app.clamp import (
    CLEAR_CYCLES, REASON_CODE, REASSERT_S, ClampWatchdog, clamp_to_rating,
    is_commandable,
)
from app.control import Tuning, compute
from app.main import Controller

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

# The live install's numbers. NOTES.md and freeze.py both say max_w: 5000 there,
# which is also the device rating, so this is the bound the alarm is written at.
LIVE_MAX_W = 5000.0
CLAMP_ID = "goodwe_output_clamp"


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


# ==============================================================================
print("1. the clamp arithmetic, on the shipped expression")

# (requested, limit, expected applied, expected engaged, expected bound)
TABLE = [
    (5001.0, 5000.0, 5000.0, True, "max_w"),      # +limit+1
    (-5001.0, 5000.0, -5000.0, True, "max_w"),    # -limit-1, the export side
    (5000.0, 5000.0, 5000.0, False, ""),          # exactly +limit: NOT engaged
    (-5000.0, 5000.0, -5000.0, False, ""),        # exactly -limit: NOT engaged
    (0.0, 5000.0, 0.0, False, ""),                # the failsafe value
    (1234.0, 5000.0, 1234.0, False, ""),
    (14768.0, 5000.0, 5000.0, True, "max_w"),     # 10.3's reported runaway
]
for requested, limit, want_applied, want_engaged, want_bound in TABLE:
    r = clamp_to_rating(requested, limit)
    check(f"{requested:+.0f} W against a {limit:.0f} W rating -> "
          f"{r.applied_w:+.0f} W, engaged={r.engaged}, bound={r.bound!r}",
          r.applied_w == want_applied and r.engaged is want_engaged
          and r.bound == want_bound)

# It is a MAGNITUDE clamp (3.2), so import and export must be bound identically.
# Derived over a sweep rather than asserted on two hand-picked values.
asym = [m for m in range(0, 12001, 37)
        if clamp_to_rating(m, LIVE_MAX_W).applied_w
        != -clamp_to_rating(-m, LIVE_MAX_W).applied_w]
check(f"symmetric over {len(range(0, 12001, 37))} magnitudes "
      f"({len(asym)} asymmetric)", not asym)

# The engagement flag has to mean "the RAW value differed", which is the words
# AC 2 uses. Re-derived from the result's own fields, not from the table.
mismatch = [m for m in range(0, 12001, 37)
            for r in [clamp_to_rating(m, LIVE_MAX_W)]
            if r.engaged != (r.requested_w != r.applied_w)]
check("engaged is exactly 'raw differs from applied', over the same sweep",
      not mismatch)

# The device second stage. Only reachable when max_w is ABOVE the device range,
# i.e. when validate_startup_config() did not run or no longer holds.
r = clamp_to_rating(3000.0, 5000.0, dev_min=-2000.0, dev_max=2000.0)
check(f"a device narrower than max_w still bites, and says which bound "
      f"({r.applied_w:+.0f} W, {r.bound!r})",
      r.applied_w == 2000.0 and r.engaged and r.bound == "device")
r = clamp_to_rating(3000.0, 5000.0, dev_min=-5000.0, dev_max=5000.0)
check("...and is a no-op once the startup validation holds",
      r.applied_w == 3000.0 and not r.engaged)

# ⚠️ REGRESSION, REVIEW-FOUND. THE FIRST STAGE USED TO `return` AND THE DEVICE
# STAGE WAS THEN NEVER REACHED, so the harder the overshoot the less the guard
# did: at max_w 5000 against a +-4000 device, +9999 emitted +5000 - OUTSIDE THE
# DEVICE RANGE - while a mere +4500 was correctly held to +4000. That inverts
# the guard and it was a regression against the pre-PR `write_setpoint()`, whose
# comment read "never send a value the device will reject outright".
#
# The property is asserted over a sweep, not on the two witness values, because
# the defect was invisible to every single-value check that happened to sit
# inside one stage.
DEV_LO, DEV_HI = -4000.0, 4000.0
escapes = [(m, clamp_to_rating(m, 5000.0, DEV_LO, DEV_HI).applied_w)
           for m in range(-15000, 15001, 61)
           for a in [clamp_to_rating(m, 5000.0, DEV_LO, DEV_HI).applied_w]
           if not (DEV_LO <= a <= DEV_HI)]
big = clamp_to_rating(9999.0, 5000.0, DEV_LO, DEV_HI)
small = clamp_to_rating(4500.0, 5000.0, DEV_LO, DEV_HI)
print(f"    max_w 5000 vs a device of {DEV_LO:+.0f}..{DEV_HI:+.0f}: "
      f"+9999 -> {big.applied_w:+.0f} W ({big.bound}), "
      f"+4500 -> {small.applied_w:+.0f} W ({small.bound})")
check(f"BOTH stages always run - {big.requested_w:+.0f} W emits "
      f"{big.applied_w:+.0f} W, inside the device's {DEV_HI:+.0f} W, and names "
      f"both bounds ({big.bound!r})",
      big.applied_w == DEV_HI and big.bound == "max_w+device")
check(f"...and it holds over the whole range, so no overshoot escapes the "
      f"device ({len(escapes)} escapes in {len(range(-15000, 15001, 61))})",
      not escapes)
check("...a big overshoot is not clamped LESS than a small one",
      abs(big.applied_w) <= abs(small.applied_w) + 0.0)

print("   non-finite input is loss of input, not an out-of-range value")
for bad in (None, float("nan"), float("inf"), float("-inf"), "abc", object()):
    check(f"is_commandable({bad!r:.30}) is False", is_commandable(bad) is False)
for good in (0, -5000.0, 5000, 12.5):
    check(f"is_commandable({good!r}) is True", is_commandable(good) is True)

# ⚠️ WHY THAT GUARD EXISTS, AND THE ANSWER IS WORSE THAN "it passes a nan
# through". MEASURED HERE, not recalled: every comparison against a nan is
# False, so `min(limit, nan)` keeps its FIRST argument and a naive clamp turns a
# nan into a command at the FULL RATING - and then reports engaged=True, because
# limit != nan, so the winding-loop counter gets a transport bug filed in it.
# The number below is computed by the same expression the guard replaces.
nan = float("nan")
naive = max(-LIVE_MAX_W, min(LIVE_MAX_W, nan))
check(f"a naive clamp turns a nan into {naive:+,.0f} W and calls it "
      f"engaged={naive != nan} - hence the guard",
      naive == LIVE_MAX_W and (naive != nan) is True)


# ==============================================================================
print()
print("2. the alarm's cadence: every engagement counted, no flood")

wd = ClampWatchdog()
t = T0
fired = [wd.note(True, "winding", t + timedelta(seconds=i)) for i in range(10)]
check(f"the rising edge fires ({fired[0]})", fired[0] is True)
check(f"the next 9 one-second cycles fire {sum(fired[1:])} more times "
      f"(re-assert is {REASSERT_S:.0f} s)", sum(fired[1:]) == 0)
check(f"but every one of them is counted: engagements={wd.engagements}",
      wd.engagements == 10)
check(f"and it is one episode, not ten: episodes={wd.episodes}",
      wd.episodes == 1)

# --- THE RE-ASSERT CADENCE, PINNED IN BOTH DIRECTIONS ------------------------
# ⚠️ REVIEW-FOUND. This used to read `wd.note(True, ..., T0 + REASSERT_S + 9)`,
# which is green for ANY value of the constant - REASSERT_S = 3600 passed, and an
# hour of silence is precisely what the module promises not to do. A check whose
# stimulus is written in terms of the constant it is grading cannot grade it.
#
# The requirements are stated in wall-clock seconds and derived from what the
# alarm is FOR, so they do not move when the constant does:
#
#   floor - a loop pinned on the rail must speak at least MIN_PER_QUARTER times
#           inside the 900 s quarter that bills the capacity tariff. That window
#           is the same one freeze.py and DOCS.md reason about.
#   ceiling - it must not speak on more than MAX_FRACTION of cycles, or it is
#           the per-cycle flood the split exists to remove.
QUARTER_S = 900.0
MIN_PER_QUARTER = 5
MAX_FRACTION = 0.20

pinned = ClampWatchdog()
alarms = sum(1 for i in range(int(QUARTER_S))
             if pinned.note(True, "pinned", T0 + timedelta(seconds=i)))
print(f"    pinned on the rail for {QUARTER_S:.0f} s at 1 s cycles: "
      f"{pinned.engagements} engagements, {alarms} alarm(s) "
      f"({alarms / QUARTER_S:.1%} of cycles)")
check(f"a loop pinned for a whole billing quarter speaks at least "
      f"{MIN_PER_QUARTER} times (it spoke {alarms})",
      alarms >= MIN_PER_QUARTER)
check(f"...and on no more than {MAX_FRACTION:.0%} of cycles (it was "
      f"{alarms / QUARTER_S:.1%})", alarms <= MAX_FRACTION * QUARTER_S)
check("...while every single cycle is still counted",
      pinned.engagements == int(QUARTER_S))

# --- THE HYSTERESIS, PINNED AGAINST THE CHATTER DIRECTION --------------------
# ⚠️ REVIEW-FOUND. This used to issue `range(CLEAR_CYCLES - 1)` clean cycles,
# which at CLEAR_CYCLES = 1 issues ZERO of them and passes while printing
# "0 clean cycles do NOT clear it". Green for every value >= 1 - including 1,
# which IS the chatter the constant exists to prevent. The stimulus is a literal
# now: ONE clean cycle must not clear the alarm, whatever the constant says.
wd2 = ClampWatchdog()
wd2.note(True, "winding", T0)
wd2.note(False, "", T0 + timedelta(seconds=1))
check(f"ONE clean cycle does NOT clear the alarm (active={wd2.active}) - "
      f"a command dithering across the rail must not chatter",
      wd2.active is True)
survived = 1
while wd2.active and survived < 50:
    wd2.note(False, "", T0 + timedelta(seconds=1 + survived))
    survived += 1
print(f"    it took {survived} consecutive clean cycles to de-assert "
      f"(CLEAR_CYCLES = {CLEAR_CYCLES})")
check(f"...and it does de-assert, after {survived} of them, rather than "
      f"latching", wd2.active is False and survived == CLEAR_CYCLES)
check("...and the engagement count survives the clear, so the evidence does",
      wd2.engagements == 1)
check("a clean cycle on an idle watchdog says nothing",
      wd2.note(False, "", T0 + timedelta(seconds=300)) is False)


# ==============================================================================
print()
print("3. AC 1 + AC 3, wired: over the rating both clamps AND alarms")


class _Store:
    data = {}

    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:
    """Stub HA. `track` False models an inverter that is switched off entirely,
    which is the 10.3 runaway's own setup."""

    def __init__(self, grid, batt=0.0, track=True):
        self.grid = grid
        self.batt = batt
        self.track = track
        self.calls = []
        self.writes = []

    async def limits(self, entity):
        return (-LIVE_MAX_W, LIVE_MAX_W, 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))
        if self.track:
            self.batt = float(value)
        return True

    async def call(self, domain, service, data):
        self.calls.append((domain, service, data))
        return True


def wired(grid, cycles=40, batt=0.0, track=True, **over):
    async def go():
        hass = _Hass(grid, batt=batt, track=track)
        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": 0,
             "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 = -LIVE_MAX_W, LIVE_MAX_W
        # The red status row exists only WHILE the clamp is engaged, and the
        # hysteresis clears it a few cycles later - so it is captured live
        # rather than looked for after the run, when it is correctly gone.
        c.red_snapshot = []
        for _ in range(cycles):
            await c.read_inputs()
            await c.cycle()
            if c.clamp_wd.active and not c.red_snapshot:
                c.red_snapshot = c.checks()
        return hass, c
    return asyncio.run(go())


def clamp_notes(hass):
    return [d for dom, svc, d in hass.calls
            if (dom, svc) == ("persistent_notification", "create")
            and d.get("notification_id") == CLAMP_ID]


# THE 10.3 RUNAWAY, replayed. "With the inverter switched off entirely, an
# unclamped integrator climbed ~130 W per 4 s to over 10 kW and, pinned at the
# 5 kW rail, reported 14 768 W." `track=False` is the inverter switched off: it
# reports 0 W however hard it is commanded, so the error never closes.
hass_r, ctl_r = wired(6000.0, cycles=60, track=False)
emitted = [abs(w) for w in hass_r.writes]

# What the SAME error would have done to an accumulator with no bound - the
# transcription is deliberately the unclamped velocity form 10.3 describes, so
# the figure below is computed here and not quoted from the spec.
runaway, err = 0.0, 6000.0
for _ in range(60):
    runaway += 0.6 * err
print(f"    unclamped, the same 60 cycles reach {runaway:,.0f} W; "
      f"the largest value that reached the device is {max(emitted):,.0f} W")
check(f"an unbounded accumulator really does run away here "
      f"({runaway:,.0f} W >> {LIVE_MAX_W:.0f} W)", runaway > 3 * LIVE_MAX_W)
check(f"AC 1: NOT ONE of the {len(emitted)} values that reached the device "
      f"exceeds the {LIVE_MAX_W:.0f} W rating",
      emitted and max(emitted) <= LIVE_MAX_W)
# ⚠️ AND THE CLAMP IS NOT THE ONLY THING HOLDING THAT LINE HERE, which is worth
# saying out loud rather than letting the number above imply otherwise. Against
# an inverter reporting 0 W the saturation freeze engages after
# `saturation_cycles` and pins the command well BELOW the rail - that is why the
# largest emitted value is what it is and not 5000 W. The clamp is what makes it
# bounded; the freeze is what makes it small. A check that read
# `max(emitted) == max_w` would have been asserting the freeze, not the clamp.
check(f"...and the saturation freeze is what pins it at {max(emitted):,.0f} W "
      f"rather than the rail", ctl_r.sat_count >= ctl_r.tuning.saturation_cycles)

notes_r = clamp_notes(hass_r)
print(f"    the loop clamped on {ctl_r.clamp_wd.engagements} of {60} cycles, "
      f"in {ctl_r.clamp_wd.episodes} episode(s), raising {len(notes_r)} "
      f"notification(s)")
check("AC 3: and it ALARMED - silently clipping this is the defect",
      ctl_r.clamp_wd.engagements > 0 and len(notes_r) >= 1)
check("...with its own notification_id, not shared with any other guard",
      all(d["notification_id"] == CLAMP_ID for d in notes_r))
check("...one episode, re-asserted rather than one per cycle",
      ctl_r.clamp_wd.episodes == 1 and len(notes_r) < ctl_r.clamp_wd.engagements)
check("...and it names itself a bug signal, not a limit in use",
      bool(notes_r) and "out of range" in notes_r[0]["message"])
rows_r = [e for e in ctl_r.events if REASON_CODE in e]
check(f"the action log records every engagement, coalesced into one row per "
      f"episode ({len(rows_r)} row(s), {ctl_r.clamp_wd.engagements} "
      f"engagement(s))",
      len(rows_r) == ctl_r.clamp_wd.episodes
      and f"{ctl_r.clamp_wd.engagements} engagement(s)" in rows_r[0])

# ⚠️ REGRESSION, AND IT WAS A REAL BUG BEFORE IT WAS A CHECK. `log_event()` keeps
# 40 rows. A row per engagement flushes that ring inside 40 cycles of a winding
# loop, and the first thing it evicted was SAFETY-18's own "METER FROZEN?" row -
# i.e. this guard silencing another guard's log. Driven here with a plant that
# TRACKS the command, so the saturation freeze never engages and the integrator
# clamps on essentially every cycle.
hass_l, ctl_l = wired(9000.0, cycles=80, track=True)
rows_l = [e for e in ctl_l.events if REASON_CODE in e]
other_l = [e for e in ctl_l.events if REASON_CODE not in e]
print(f"    80 winding cycles: {ctl_l.clamp_wd.engagements} engagement(s) -> "
      f"{len(rows_l)} log row(s), leaving room for {len(other_l)} other event(s)")
check(f"a long winding episode does NOT flood the 40-row event log "
      f"({len(rows_l)} rows for {ctl_l.clamp_wd.engagements} engagements)",
      ctl_l.clamp_wd.engagements > 40 and len(rows_l) <= ctl_l.clamp_wd.episodes)
check("...and the count in the row is the counter's, so nothing is lost",
      f"{ctl_l.clamp_wd.engagements} engagement(s)" in rows_l[-1])
# The RED row exists only while engaged, and the episode above is short, so it
# is captured mid-run rather than asserted after the hysteresis has cleared it.
check("the status page goes red WHILE it is engaged",
      any(not c["ok"] and not c["warn"] and "OUTPUT CLAMP ENGAGED" in c["text"]
          for c in ctl_r.red_snapshot))
check("...and once clear, a warn row still says how many times it engaged",
      any(c["warn"] and f"{ctl_r.clamp_wd.engagements} time(s)" in c["text"]
          for c in ctl_r.checks()))

# ⚠️ NOT A TRIP. 11.2's action is "clamp and alarm"; the loop keeps running and
# keeps writing the clamped value. If a later cut changes that it must come here.
check(f"it is NOT a trip - the command stands at {ctl_r.target:+.0f} W, not 0 W",
      ctl_r.target != 0.0 and ctl_r.target == hass_r.writes[-1])
check("...and the controller still reports itself running, not failsafe",
      ctl_r.auto and not ctl_r.sign_wd.tripped)

# The export side, because the clamp is a magnitude clamp and a sign bug in it
# would be invisible on the import case alone.
hass_x, ctl_x = wired(-6000.0, cycles=60, track=False)
check(f"the export side clamps and alarms too "
      f"({min(hass_x.writes):+,.0f} W, {len(clamp_notes(hass_x))} note(s), "
      f"{ctl_x.clamp_wd.engagements} engagement(s))",
      min(hass_x.writes) < 0 and abs(min(hass_x.writes)) <= LIVE_MAX_W
      and clamp_notes(hass_x) and ctl_x.clamp_wd.engagements > 0)


# ==============================================================================
print()
print("4. AC 2: saturation WITHOUT clamping must NOT raise a clamp alarm")

# A legal command the inverter will not follow. The meter sits just outside the
# deadband so the loop moves a little and then rests; `batt` is pinned far from
# whatever is commanded, which is the saturation condition (|cmd - actual| >
# saturation_w) and nothing else. The integrator never approaches the rating.
hass_s, ctl_s = wired(60.0, cycles=40, batt=2500.0, track=False)
peak = max(abs(w) for w in hass_s.writes)
print(f"    saturated for {ctl_s.sat_count} consecutive cycles; the command "
      f"peaked at {peak:.0f} W, {peak / LIVE_MAX_W:.1%} of the rating")
# ⚠️ THE POSITIVE HALF FIRST. A negative check that saw nothing happen is not an
# observation - if the saturation freeze never engaged, "no clamp alarm" is
# vacuous and this check is worthless.
check(f"the saturation freeze REALLY DID engage (sat_count="
      f"{ctl_s.sat_count} >= {ctl_s.tuning.saturation_cycles})",
      ctl_s.sat_count >= ctl_s.tuning.saturation_cycles)
check("...and the event log says so, in the saturation's own words",
      any("saturation freeze" in e for e in ctl_s.events))
check(f"...while the command stayed well inside the rating "
      f"({peak:.0f} W < {LIVE_MAX_W:.0f} W)", peak < LIVE_MAX_W)
# Now the negative, which is only worth anything because of the three above.
check("AC 2: NO clamp notification was raised", not clamp_notes(hass_s))
check("AC 2: the clamp counted ZERO engagements", ctl_s.clamp_wd.engagements == 0)
check("AC 2: no OUTPUT_CLAMP row in the action log",
      not [e for e in ctl_s.events if REASON_CODE in e])
check("AC 2: and no clamp row on the status page",
      not [c for c in ctl_s.checks() if "output clamp" in c["text"].lower()])

print("   ...and the two alarms do not share a notification_id")
ids_clamp = {d["notification_id"] for d in clamp_notes(hass_r)}
ids_other = {d.get("notification_id") for dom, svc, d in hass_r.calls
             if svc == "create"} - ids_clamp
check(f"clamp ids {sorted(ids_clamp)} are disjoint from every other alarm's "
      f"{sorted(x for x in ids_other if x)}", not (ids_clamp & ids_other))

print("   ...and the alarm cannot be keyed on Decision.reason")
# A cycle that clamps the integrator AND is then slew limited. `reason` reports
# the LAST stage, so a string-keyed alarm is silent on it; the boolean is not.
t = Tuning(max_w=2000.0, slew_w=100.0, gain=0.6)
d = compute(prev_w=1000.0, grid_w=6000.0, actual_w=1000.0, tuning=t, i_w=1900.0)
print(f"    reason={d.reason!r}  i_clamped={d.i_clamped}  clamped={d.clamped}")
check("the integrator DID clamp on that cycle", d.i_clamped is True)
check("...but `reason` says 'slew-limited' - the clamp was overwritten "
      "downstream, so a string-keyed alarm is SILENT here",
      d.reason == "slew-limited" and "clamp" not in d.reason)
# And the same for the OUTPUT clamp, on a config that lets the integrator run
# above the rail - then both flags fire and `reason` still reports neither.
t2 = Tuning(max_w=2000.0, integrator_max_w=3000.0, slew_w=100.0, gain=0.6)
d2 = compute(prev_w=1000.0, grid_w=6000.0, actual_w=1000.0, tuning=t2, i_w=2900.0)
print(f"    reason={d2.reason!r}  i_clamped={d2.i_clamped}  clamped={d2.clamped}")
check("both clamps engaged, and `reason` reports neither",
      d2.i_clamped and d2.clamped and "clamp" not in d2.reason)
# ⚠️ HOW THE FREEZE AND THE CLAMP INTERACT, MEASURED RATHER THAN ASSUMED, AND
# THE MEASUREMENT REFUSED TWO GUESSES BEFORE THIS ONE. Written first as "a
# frozen cycle never clamps" (false), then as "only the i_w == 0 reseed
# exception clamps" (also false - at i_w == 0 with a matching actual the cycle
# is not saturated at all, so it is never frozen). What the sweep actually
# shows: a frozen cycle clamps exactly when the integrator REVERSES. control.py
# permits that on purpose - "it may fall, cross zero, or reverse outright" -
# because the alternative deadlocks the loop, and a reversal is a full swing
# that can clear the bound on the far side in one step.
#
# That matters for this ticket rather than being trivia: it is the one shape
# where the saturation freeze and a clamp engagement coincide, so it is exactly
# where an alarm that shared a channel with saturation would lose the clamp.
grid_sweep = (-9000.0, -3000.0, 3000.0, 9000.0)
grid = [(pp, g) for pp in (-1900.0, -500.0, 0.0, 500.0, 1900.0)
        for g in grid_sweep]
runs = {(pp, g): compute(pp, g, 0.0, t, sat_count=5, i_w=pp) for pp, g in grid}
frozen = [k for k, d in runs.items() if d.frozen]
frozen_clamps = [k for k in frozen if runs[k].i_clamped]
# The property, stated independently of the control law's own branches: the
# correction pushes AGAINST the standing integrator, and the resulting swing
# clears the bound.
reversals = [(pp, g) for pp, g in frozen
             if pp * g < 0 and abs(pp + t.gain * g) > t.max_w]
print(f"    {len(frozen)} of {len(grid)} sweep cells are frozen; "
      f"those that also clamp: {frozen_clamps}")
check(f"a frozen cycle CAN clamp, and only by reversing "
      f"({len(frozen_clamps)} of {len(frozen)} frozen cells)",
      frozen_clamps and frozen_clamps == reversals)
check("...and a frozen cycle pushing the way it already leans never clamps - "
      "that is the freeze doing its job",
      not [k for k in frozen if runs[k].i_clamped and k[0] * k[1] > 0])
check("a healthy cycle sets neither flag",
      not any([compute(0.0, 100.0, 0.0, t).i_clamped,
               compute(0.0, 100.0, 0.0, t).clamped]))


# ==============================================================================
print()
print("5. no path reaches the wire around the clamp")


async def _direct():
    hass = _Hass(0.0)
    c = Controller({"setpoint_entity": "number.sp", "max_w": 2000,
                    "meter_source": "off"},
                   hass, _Store(), _Pub(), stop=asyncio.Event())
    c.dev_min, c.dev_max = -LIVE_MAX_W, LIVE_MAX_W
    # The shape the heartbeat, shutdown() and handle_action() all use.
    await c.write_setpoint(9999.0, force=True)
    await c.write_setpoint(-9999.0, force=True)
    await c.write_setpoint(float("nan"), force=True)
    await c.write_setpoint(1500.0, force=True)
    return hass, c


hass_d, ctl_d = asyncio.run(_direct())
print(f"    asked for 9999 / -9999 / nan / 1500 W; the device saw "
      f"{hass_d.writes}")
check("a direct out-of-range write is clamped, not passed through",
      hass_d.writes[:2] == [2000.0, -2000.0])
check("a nan is refused and becomes 0 W, per 11.1 rule 1",
      hass_d.writes[2] == 0.0)
check("...and is NOT counted as a clamp engagement - it is a transport bug",
      sum(1 for e in ctl_d.events if REASON_CODE in e) == 2)
check("an in-range value is untouched", hass_d.writes[3] == 1500.0)
check("every emitted frame satisfies |emitted| <= the rating",
      all(abs(w) <= 2000.0 for w in hass_d.writes))

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