goodwe-addon/goodwe_controller/test_telemetry.py

"""Runnable check for TEL-10 and SAFETY-48.  `python3 test_telemetry.py`

TEL-10 publishes the loop's integrator. SAFETY-48 bounds it. They are graded in
one file because they are one observable: the number SAFETY-48 bounds is the
number TEL-10 added, and a check on either that could not see the other would
be grading its own fixture.

`self.i_w` is carried between cycles of `Controller.cycle()` and, before this
card, left the process nowhere at all - no entity, no MQTT attribute, no log
line. The last investigation into what it does had to RECONSTRUCT it from the
grid trace; the reconstruction reproduced about a third of the engagements it
was checked against and was discarded.

⚠️ WHAT THIS FILE REFUSES TO DO, SAID FIRST. It does not assert that a row
EXISTS. `publish()` handing a dict with `loop_i` in it is satisfied by
`loop_i: 0.0` forever, by `loop_i: setpoint`, and by a row that reports the
command under the integrator's name - which is exactly the confusion the card
is about, because on a default install the two are within a quantisation step
of each other for most of every day. Every check below therefore drives REAL
CYCLES through the real `Controller.cycle()` until the two quantities are
MEASURABLY DIFFERENT, and grades the published number at a tick where reading
the wrong one gives the wrong answer. MUTATION.md habit 4.

VERIFIED BY BREAKING IT. Every check below was driven by the mutation that
defeats the claim it names, in the CI image, on a tree purged of `__pycache__`
before each run, with the file's sha256 taken before, after, and again after
the revert (MUTATION.md's reporting rules, and habit 18's unique backup path).

  #   mutation                                        bytes  reds  file
  --  ----------------------------------------------  -----  ----  -----------
  M1  `self.loop_i_w = loop_prev` - publish the         -3 B    5   this file
      COMMAND under the integrator's name
  M2  `decision.i_w - self.target` for the lead         +2 B    0   SURVIVOR
  M2b `decision.i_w - decision.target_w` for the lead   +8 B    1   this file
  M3  `"loop_reason": self.reason` - the ARBITER's      -5 B    5   this file
      sentence in the loop's row
  M4  delete `self.loop_reason = "not-running"`        -45 B    1   this file
  M5  delete the not-running branch's two zeroes       -67 B    1   this file
  M6  `"loop_i": round(self.target, 1)`                 -2 B    4   this file
  M7  the `loop_i` row deleted from `SENSORS`          -96 B 10+1   test_mqtt
                                                                      + this file
  M8  `loop_i` announced PER INVERTER                  +89 B    8   test_mqtt
  M9  `loop_i` announced BEFORE release/1.0's nine      +0 B    7   test_mqtt

⚠️ A RED COUNT IS A PROPERTY OF THE SUITE AS MUCH AS OF THE MUTATION, SO EVERY
ONE ABOVE NAMES A TREE - this one, at this commit. A later card that adds a
check, re-baselines one or deletes one moves the number with the mutation
untouched. Re-measure rather than inherit: `tools/mutation_run.sh` on a `cp -a`
of the tree, one file at a time, and say which tree. M7's count was written as
8 and is 10 on a re-run, on both trees - the direction was right and the count
was decoration, which is the half of MUTATION.md's reporting rule 3 that is
easiest to let rot.

⚠️ AND SAFETY-48 HAS ALREADY MOVED ONE OF THEM. M1 defeats FIVE checks on
TEL-10's own tree and FOUR on this one: this card replaced the "...the gap is
the slew limit and not an accident" check, because the gap it asserted is now
bounded and the assertion is written against the bound instead. The mutation
did not change; the suite did. Both counts are right, for different trees, and
neither is right without one.

⚠️ M9 IS ZERO BYTES AND CHANGES NO PAYLOAD. Every id, name, unit and topic is
identical; only the ORDER of the retained messages moves. No fingerprint and no
byte delta can see it, which is why part 1 asserts the announcement as an
ordered LIST and not as a set.

⚠️ M2 IS A REAL SURVIVOR AND IS REPORTED AS ONE. At the line it edits,
`loop_prev` and `self.target` are the same value: `cycle()` does not reassign
`self.target` until `self.target = resolution.target_w`, further down. The
mutation therefore changes nothing that runs, on a purged tree, which is not
the direction stale bytecode fabricates. `loop_prev` is kept anyway - it is the
name the `compute()` call is made with, so the lead cannot be measured against
a different cycle's command by anybody moving a line - and M2b is the mutation
that DOES change the value, aimed at exactly that.

⚠️ AND THE FIRST DRAFT OF THIS FILE HAD TWO CHECKS OF ITS OWN THAT COULD NOT
REACH THE FAILURE THEY NAMED. Both are MUTATION.md habit 4:

  * "the published integrator reaches its bound" was GREEN under M1, because
    the command reaches the bound too - twelve cycles later. The check now
    asserts the LEAD TIME, not the value.
  * "a stopped controller publishes `not-running`" was GREEN under M4 and M5,
    because the fixture STARTED stopped and `__init__` writes the same values.
    It stops the controller part-way through a run that has already wound the
    integrator to -5000 W, so a row left standing is a WRONG row.

SAFETY-48's OWN BATTERY, same discipline, against `app/control.py`:

  #   mutation                                        bytes  reds
  --  ----------------------------------------------  -----  --------------------
  L1  the whole SAFETY-48 block deleted              -195 B  5 here + 4 control
  L2  `SLEW_LEAD_STEPS = 0` - the velocity form        +0 B  4 here + 3 control
  L3  `SLEW_LEAD_STEPS = 12` - a cap above the rail    +1 B  4 here + 3 control
  L4  the cap applied unconditionally, outside the   -92 B  1 here   ⚠️ SURVIVOR
      slew branch                                             of all 30 files
                                                              until part 4's
                                                              `integrator_max_w`
                                                              case existed
  L5  `min`/`max` swapped - the cap on the wrong      +0 B  5 here + 5 control
      side                                                  (or 5 + 4; see below)
  L6  the lead measured against `prev_w` rather than  +0 B  6 here + 3 control
      the command it emits

⚠️ L4, L5 AND L6 ARE ALL ZERO- OR NEGATIVE-BYTE EDITS THAT CHANGE NO PAYLOAD
AND NO OPTION, AND L4 WAS BIT-IDENTICAL ON THE SHIPPED CONFIGURATION - 0 of
256 246 ticks over the 16-day recording. It is separable only where
`integrator_max_w` is above `max_w`, which is the stimulus part 4 drives; the
COUNT of separable states is a property of whatever grid was swept and three
sweeps disagree about it, so DECISIONS/SAFETY-48 §4 carries all three rather
than one.

⚠️ AND L5 IS TWO MUTATIONS WEARING ONE NAME, WHICH IS WHY ITS COUNT NEEDS A
SPELLING BESIDE IT. "min/max swapped" can mean the FUNCTION names exchanged
(`max(i_w, slewed + cap)` / `min(i_w, slewed - cap)`) - 5 here + 5 in
test_control.py, one of the five being SAFETY-13's own "frozen integrator never
deadlocks" sweep over 13 041 states - or the whole BRANCH bodies exchanged
(`max(i_w, slewed - cap)` / `min(i_w, slewed + cap)`), which is 5 + 4 and does
NOT reach the deadlock sweep. Both are red, both are the cap on the wrong side,
and a table that writes one number for the pair is reporting a mutation nobody
can re-run. The first spelling is the one the row above quotes.

⚠️ 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

from app import mqtt
from app.control import SLEW_LEAD_STEPS, Tuning, compute, slew_limit_w
from app.main import Controller

fails = []

# The live install's shape, and every number in it is measured rather than
# chosen: `max_w` and `slew_w` are read straight off the recorder (the setpoint
# sensor's extremes are exactly +-5000 and 11 226 recorded steps are exactly
# 400 W), `gain` and `target_grid_w` are what the options carry.
LIVE = {"max_w": 5000.0, "gain": 0.6, "slew_w": 400.0, "target_grid_w": -10.0,
        "deadband_w": 15, "step_w": 10}
# The integrator's bound on a default install: `integrator_max_w` is None, which
# control.py resolves to `max_w`. Written as a literal on purpose (habit 2) -
# `== c.tuning.max_w` would compare the thing under test against itself.
BOUND_W = 5000.0
# `slew_w / gain`: the error above which the integrator moves further in one
# cycle than the command is allowed to. 400 / 0.6.
OUTRUN_W = 666.67


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


class _Store:
    def __init__(self):
        self.data = {}

    def set(self, *a):
        pass

    def get_time(self, *a):
        return None


class _Pub:
    """Captures what `Controller.publish()` hands the publisher, every cycle."""

    enabled = False

    def __init__(self):
        self.frames = []

    def publish(self, values):
        self.frames.append(dict(values))

    def close(self):
        pass


class _Hass:
    """A scripted meter and an inverter that follows what it is told.

    ⚠️ THE PLANT FOLLOWS EXACTLY AND WITH NO LAG, WHICH IS DELIBERATE AND IS
    THE HARDER CASE FOR EVERY CHECK HERE. A lagging inverter trips the
    SATURATION freeze (`|prev_w - actual_w| > saturation_w`), and the
    saturation freeze already holds the integrator - so a laggy fixture would
    let SAFETY-13's mechanism pass for SAFETY-48's. Nothing below may be
    explained by saturation, and part 4 asserts `frozen` is False throughout.
    """

    def __init__(self, grid, soc=50.0):
        self.grid_seq = grid if callable(grid) else (lambda _n: grid)
        self.soc = soc
        self.batt = 0.0
        self.n = 0
        self.writes = []
        self.calls = []

    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_seq(self.n), "sensor.soc": self.soc,
                "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


def run(grid, cycles=30, auto=True, stop_after=None, **over):
    """`cycles` real `Controller.cycle()`s, and every published frame.

    `stop_after` presses stop part-way through - which is not the same run as
    starting stopped, and is the only one that can tell a row that was WRITTEN
    on the not-running branch from a row that was never written at all.
    """
    async def go():
        hass = _Hass(grid)
        pub = _Pub()
        c = Controller(dict(
            {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
             "soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
             "meter_source": "off", "heartbeat_s": 10, "stale_input_s": 15},
            **LIVE, **over),
            hass, _Store(), pub, stop=asyncio.Event())
        c.auto = auto
        c.dev_min, c.dev_max = -LIVE["max_w"], LIVE["max_w"]
        for k in range(cycles):
            if stop_after is not None and k == stop_after:
                c.auto = False
            await c.read_inputs()
            await c.cycle()
            c.publish()
            hass.n += 1
        return c, hass, pub.frames
    return asyncio.run(go())


# ==============================================================================
print("1. TEL-10: the integrator is published, and it is NOT the command")
# ==============================================================================
# A standing 6 kW export against a 5 kW rail and a 400 W slew step.
#
# ⚠️ THIS FIXTURE WAS BUILT AGAINST THE UNBOUNDED INTEGRATOR AND SAFETY-48 TOOK
# ITS SEPARATION AWAY, WHICH IS WORTH SAYING RATHER THAN QUIETLY RE-TUNING. On
# the law TEL-10 shipped against, the integrator reached its 5000 W bound on
# cycle 1 and the command needed twelve more cycles to arrive, so the two rows
# differed by KILOWATTS and any check at all could see it. SAFETY-48 bounds the
# lead at one slew step, so the gap is now 800 W and the integrator arrives ONE
# cycle early instead of twelve. Both surviving separations are graded: the
# ARRIVAL ORDER, and the 800 W plateau that holds for the whole ramp.
ctl, hass, frames = run(-6000.0, cycles=30)
i_row = [f["loop_i"] for f in frames]
sp_row = [f["setpoint"] for f in frames]
lead_row = [f["loop_lead"] for f in frames]
why_row = [f["loop_reason"] for f in frames]
print("    cycle  loop_i   setpoint  loop_lead  loop_reason")
for k in range(8):
    print(f"      {k:<5d}{i_row[k]:8.0f} {sp_row[k]:9.0f} {lead_row[k]:10.0f}"
          f"  {why_row[k]}")

first_at_bound = next((k for k, v in enumerate(i_row) if abs(v) >= BOUND_W), None)
first_cmd_at_bound = next((k for k, v in enumerate(sp_row) if abs(v) >= BOUND_W),
                          len(sp_row))
print(f"    the integrator reaches {-BOUND_W:.0f} W at cycle "
      f"{first_at_bound}; the command reaches it at cycle "
      f"{first_cmd_at_bound}")

# ⚠️ THE ARRIVAL ORDER IS THE CLAIM, NOT THE VALUE. "The published number
# reaches the bound" is satisfied by publishing the COMMAND too - it gets there
# as well, a cycle later - so a check written on the value alone grades nothing
# and the mutation that publishes `prev_w` walks straight through it. Measured:
# with `self.loop_i_w = loop_prev` the value-only form was GREEN and this one is
# red.
check("⭐ THE PUBLISHED INTEGRATOR REACHES ITS BOUND, AND REACHES IT BEFORE THE "
      f"COMMAND DOES. A 6 kW export winds it to exactly {BOUND_W:.0f} W - "
      "`integrator_max_w` is None, which control.py resolves to `max_w` - while "
      "the command is still a slew step short",
      first_at_bound is not None and i_row[first_at_bound] == -BOUND_W
      and first_cmd_at_bound > first_at_bound)

# ⚠️ THE PRECONDITION, AS ITS OWN CHECK (habit 1). If the command happened to be
# at the bound too, the line above would be satisfied by publishing EITHER
# number and would be grading nothing.
check("...and the command is NOT at the bound at that tick, so the two are "
      f"separable: setpoint {sp_row[first_at_bound]:.0f} W against an "
      f"integrator of {i_row[first_at_bound]:.0f} W",
      abs(BOUND_W - abs(sp_row[first_at_bound])) >= 200.0)

# ⚠️ AND THE PLATEAU, WHICH IS THE SEPARATION THAT LASTS. Through the whole
# ramp the integrator stands exactly two slew steps ahead of the command the
# previous cycle left standing - one step is `SLEW_LEAD_STEPS` against the
# command this cycle EMITS, and the emitted command is itself one step ahead of
# `prev_w`. It is graded as a RUN of identical values rather than as a maximum,
# because a maximum is reached by one tick and a plateau is the shape.
_plateau = [k for k in range(1, 13) if lead_row[k] == -800.0]
print(f"    the lead plateau: {lead_row[1:13]}")
check("⭐ AND THE TWO ROWS ARE SEPARATED FOR THE WHOLE RAMP, not at one tick: "
      f"the lead sits at exactly -800 W (two slew steps) on "
      f"{len(_plateau)} of cycles 1..12 while both numbers travel 400 W a "
      "cycle", len(_plateau) >= 8)
check("...and the command really is travelling a full slew step a cycle "
      "through that plateau, so the plateau is not a stalled loop",
      all(round(abs(sp_row[k] - sp_row[k - 1]), 6)
          == slew_limit_w(Tuning(**LIVE)) for k in _plateau))

# The lead is `i_w - prev_w`, and `prev_w` is the command the LAST cycle left
# standing - which is the setpoint this file captured one frame earlier. Graded
# as that identity rather than as a magnitude, because a magnitude cannot tell
# a lead from the integrator itself.
ident = [k for k in range(1, 12)
         if round(lead_row[k], 1) != round(i_row[k] - sp_row[k - 1], 1)]
check("⭐ AND `loop_lead` IS THE INTEGRATOR MINUS THE COMMAND IT WAS COMPUTED "
      "AGAINST - the PREVIOUS cycle's setpoint, not this one's. Checked as an "
      f"identity over cycles 1..11 ({len(ident)} disagreements)", not ident)
check("...and it is never a permanent zero while the loop is working: worst "
      f"lead {min(lead_row):.0f} W",
      min(lead_row) <= -800.0)


# ==============================================================================
print()
print("2. TEL-10: the reason row is compute()'s OWN word, not the arbiter's")
# ==============================================================================
# ⚠️ THE TWO SENTENCES EXIST SIDE BY SIDE AND ONE OF THEM ALREADY HAD A ROW.
# `self.reason` is `Resolution.explain()` - "loop -> -800 W" - and it prints the
# winning claim's NAME, never the claim's reason. So a publisher wired to
# `self.reason` would produce a row that is never wrong-looking and never once
# says `slew-limited`, which is the value SAFETY-48 is about.
print(f"    the arbiter's sentence: {ctl.reason!r}")
print(f"    the loop's own word   : {ctl.loop_reason!r}")
check("⭐ THE LOOP'S REASON REACHES THE WIRE, and on a slew-limited cycle it is "
      "the string `slew-limited` - a value the arbiter's sentence does not "
      "contain at all",
      "slew-limited" in why_row and "slew-limited" not in ctl.reason)
check("...and the two rows are DIFFERENT STRINGS on the same cycle, so one "
      "cannot have been published from the other",
      frames[-1]["loop_reason"] != ctl.reason)
check("...and every published reason is one `compute()` can actually produce",
      set(why_row) <= {"tracking", "deadband", "i-clamped", "clamped",
                       "slew-limited", "saturated-freeze", "freeze-released",
                       "not-running"})

# A converged site: the error is inside the deadband and the law says so.
calm = run(-10.0, cycles=6)[2]
check("...and a site resting inside the deadband publishes `deadband`, not the "
      "last exciting word the loop said", calm[-1]["loop_reason"] == "deadband")

# ⚠️ THE NOT-RUNNING PATH IS A DIFFERENT BRANCH OF `cycle()` AND IS GRADED AS
# ONE (habit 9). That branch zeroes `self.i_w`, so a row that kept repeating the
# last integrator there would say the loop was winding while nothing is being
# controlled at all.
#
# ⚠️ AND `auto: False` IS NOT THAT BRANCH, WHICH IS WHY BOTH ARE HERE. The
# shipped `shadow_mode: auto` means a STOPPED controller still runs the whole
# decision path and records it - SAFETY-10 - so the loop is live, the
# integrator is real, and a check that assumed otherwise would have been
# asserting `not-running` about a loop that was running. It took `auto: False`
# AND `shadow_mode: off` to reach the arm that zeroes it.
shadowed = run(-6000.0, cycles=6, auto=False)[2]
print(f"    stopped, shadow_mode auto: status {shadowed[-1]['status']!r}, "
      f"loop_i {shadowed[-1]['loop_i']:+.0f} W, "
      f"reason {shadowed[-1]['loop_reason']!r}")
check("a STOPPED controller in shadow mode still publishes a live integrator - "
      "the decision path runs, so the row must report it rather than a zero "
      "that would read as 'nothing is winding'",
      shadowed[-1]["status"] == "shadow"
      and shadowed[-1]["loop_i"] < -100.0
      and shadowed[-1]["loop_reason"] != "not-running")

# ⚠️ AND IT IS STOPPED PART-WAY, NOT STARTED STOPPED, WHICH IS THE WHOLE
# DIFFERENCE. A controller that never ran carries `__init__`'s values, so a run
# that begins stopped is satisfied by a branch that writes NOTHING - measured:
# deleting `self.loop_reason = "not-running"` from `cycle()` outright was a
# -45 B SURVIVOR of the first draft of this check. The loop has to wind first,
# so that a row left standing is a row that says the wrong thing.
stopped = run(-6000.0, cycles=12, stop_after=6, shadow_mode="off")[2]
print(f"    while running  : loop_i {stopped[5]['loop_i']:+.0f} W, "
      f"reason {stopped[5]['loop_reason']!r}, status {stopped[5]['status']!r}")
print(f"    after stopping : loop_i {stopped[-1]['loop_i']:+.0f} W, "
      f"reason {stopped[-1]['loop_reason']!r}, status {stopped[-1]['status']!r}")
check("...and the precondition: the loop really was winding before stop was "
      "pressed, so a stale row afterwards would be a WRONG row and not merely "
      "an unset one",
      stopped[5]["loop_i"] <= -1000.0
      and stopped[5]["loop_reason"] != "not-running")
check("⭐ AND ONCE THE DECISION PATH IS GENUINELY OFF THE ROWS SAY SO: "
      "`not-running` and a zero integrator - the same zero `cycle()` writes to "
      "`self.i_w` on that branch, not the number it was carrying a cycle ago",
      stopped[-1]["status"] == "stopped"
      and stopped[-1]["loop_reason"] == "not-running"
      and stopped[-1]["loop_i"] == 0.0 and stopped[-1]["loop_lead"] == 0.0)


# ==============================================================================
print()
print("3. TEL-10: announced and published are the same three topics")
# ==============================================================================
# mqtt.py's `omit` note: an entity that exists and is never fed is a worse
# signal than one that does not exist. This is that claim for TEL-10's rows,
# taken off the SHIPPED announcement and the SHIPPED `publish()`.
announced = {t.rsplit("/", 2)[-2].replace(f"{mqtt.BASE}_", "")
             for t, _ in mqtt.discovery_payloads()}
published = set(frames[-1])
TEL10 = {"loop_i", "loop_lead", "loop_reason"}
print(f"    announced keys: {sorted(announced & TEL10)}")
print(f"    published keys: {sorted(published & TEL10)}")
check("all three rows are announced AND published - no entity nothing feeds, "
      "no topic nothing declared",
      TEL10 <= announced and TEL10 <= published)
check("...and they are on the SITE device, not per unit - there is one "
      "integrator",
      all(c["device"] == mqtt.DEVICE
          for t, c in mqtt.discovery_payloads(["inverter-a", "inverter-b"])
          if c["unique_id"].replace(f"{mqtt.BASE}_", "") in TEL10))
# ==============================================================================
print()
print("4. SAFETY-48: the slew step ceiling stops the integrator outrunning it")
# ==============================================================================
# ⚠️ THE ARITHMETIC THE CARD IS ABOUT, RE-DERIVED RATHER THAN QUOTED. The
# integrator moves `gain * error` a cycle and the command moves at most
# `slew_w`, so the integrator outruns the command whenever
# `|error| > slew_w / gain`. At the live shape that is 667 W, and the recorded
# median error at the onset of an episode is 1114 W.
T = Tuning(**LIVE)
check(f"the outrun threshold is `slew_w / gain` = {OUTRUN_W:.0f} W, and it is "
      "the shipped pair that produces it, not a constant written here",
      abs(T.slew_w / T.gain - OUTRUN_W) < 0.5)

# ⚠️ TWO LEADS, AND CONFUSING THEM IS HOW THIS CHECK WOULD BE WRITTEN WRONG.
# The bound is on `i_w - target`, the command this cycle EMITS - that is what
# makes the unwind one cycle, because the command can cover one slew step in
# one cycle. `loop_lead`, and `Decision.i_w - prev_w`, are measured against the
# command the PREVIOUS cycle left standing, and the emitted command is itself
# one slew step further on - so the published figure is bounded by TWO steps,
# not one. Both are asserted, against their own bound.
_SLEW = slew_limit_w(T)
d = compute(prev_w=0.0, grid_w=-6000.0, actual_w=0.0, tuning=T, i_w=-100.0)
print(f"    one cycle at a 5990 W error: command {d.target_w:+.0f} W, "
      f"integrator {d.i_w:+.0f} W, ahead of the EMITTED command by "
      f"{d.i_w - d.target_w:+.0f} W and of `prev_w` by {d.i_w:+.0f} W, "
      f"reason {d.reason!r}")
check(f"⭐ THE INTEGRATOR MAY LEAD THE COMMAND IT EMITS BY AT MOST "
      f"{SLEW_LEAD_STEPS} x slew_w = {SLEW_LEAD_STEPS * _SLEW:.0f} W - so the "
      "worst-case unwind is ONE cycle, not `integrator_max_w / slew_w` (12.5 "
      "at this shape)",
      abs(d.i_w - d.target_w) <= SLEW_LEAD_STEPS * _SLEW + 1e-9)

# ...and over a real run, which is where release/1.0 walks to the rail.
leads = [abs(f["loop_lead"]) for f in frames]
print(f"    over {len(frames)} driven cycles: worst published lead "
      f"{max(leads):.0f} W, worst integrator {min(i_row):.0f} W")
check(f"⭐ AND IT HOLDS ON EVERY DRIVEN CYCLE: the published lead - measured "
      f"against the PREVIOUS command - never exceeds "
      f"{(SLEW_LEAD_STEPS + 1) * _SLEW:.0f} W, one step for the bound and one "
      "for the step the command took. On release/1.0 this same run reaches "
      "4600 W",
      max(leads) <= (SLEW_LEAD_STEPS + 1) * _SLEW + 1e-9)
check("...and release/1.0's figure is not merely larger, it is over the bound: "
      "the run that produced 4600 W here would be red on the line above",
      4600.0 > (SLEW_LEAD_STEPS + 1) * _SLEW)
check("...and nothing here is the SATURATION freeze doing SAFETY-48's work: "
      "the inverter tracks exactly, so `frozen` is never true",
      not any(f["loop_reason"] in ("saturated-freeze", "freeze-released")
              for f in frames))

# ⚠️ AND THE CAP IS NOT ZERO, WHICH IS ITS OWN CHECK BECAUSE NOTHING ELSE
# NOTICES. `i_w = slewed` - a cap of zero - satisfies every bound above and is
# what control.py's docstring calls "the exact velocity form it replaced":
# the accumulator stops being a separate quantity at all. The claim is
# therefore that the integrator is STRICTLY ahead of the command it emits, by
# exactly one step, on a slew-limited cycle.
check(f"⭐ AND IT IS NOT A BACK-CALCULATION: on a slew-limited cycle the "
      f"integrator stands EXACTLY {SLEW_LEAD_STEPS} slew step(s) ahead of the "
      f"command it emits ({d.i_w - d.target_w:+.0f} W), not on top of it - "
      "a cap of zero would collapse this function to the velocity form it "
      "replaced",
      d.reason == "slew-limited"
      and round(abs(d.i_w - d.target_w), 6) == SLEW_LEAD_STEPS * _SLEW)

# ⚠️ AND IT BINDS IN BOTH DIRECTIONS OF TRAVEL, WHICH IS NOT OBVIOUS FROM THE
# SPELLING. `if target > slewed ... elif target < slewed` reads like two cases
# and is one: a command FALLING from +2400 W toward an integrator at +100 W has
# an integrator 2300 W ahead of it in the direction of travel, and it is capped
# exactly as a rising one is. A check written only on the rising case would be
# green for a gate that had lost half of itself.
_dn = compute(prev_w=2400.0, grid_w=-200.0, actual_w=2400.0, tuning=T,
              sat_count=0, i_w=100.0)
print(f"    descending: prev +2400 W, integrator +100 W -> command "
      f"{_dn.target_w:+.0f} W, integrator {_dn.i_w:+.0f} W, "
      f"reason {_dn.reason!r}")
check("⭐ THE CAP BINDS ON A FALLING COMMAND TOO: descending from +2400 W the "
      f"integrator is pulled to exactly one slew step beyond the command it "
      f"emits ({_dn.i_w - _dn.target_w:+.0f} W), not left 2300 W below it",
      _dn.reason == "slew-limited"
      and round(abs(_dn.i_w - _dn.target_w), 6) == SLEW_LEAD_STEPS * _SLEW)

# ⚠️ THE CAP IS INSIDE THE SLEW BRANCH, AND THIS IS THE ONLY STIMULUS THAT CAN
# TELL THAT FROM AN UNCONDITIONAL ONE. Applying it to every cycle - `i_w =
# max(target - cap, min(target + cap, i_w))` after `target = slewed` - is a
# -92 B edit that is BIT-IDENTICAL on the shipped configuration, and it was a
# SURVIVOR of all 30 files until this check existed. It is measured: 0 of
# 256 246 ticks differ over the 16-day recording, and 2050 of 257 400 swept
# states do - every one of them at `integrator_max_w` ABOVE `max_w`, the
# configuration control.py says is the only way its `clamped` reason is
# reachable. There the integrator is DELIBERATELY allowed above the rail, and a
# cap that fired on every cycle would drag it back to the command and make the
# option inert.
_HI = Tuning(gain=0.6, max_w=2000.0, slew_w=400.0, target_grid_w=-10.0,
             integrator_max_w=5000.0)
_hi = compute(prev_w=-2400.0, grid_w=-8000.0, actual_w=-6000.0, tuning=_HI,
              sat_count=0, i_w=-2400.0)
print(f"    integrator_max_w 5000 above a max_w of 2000: command "
      f"{_hi.target_w:+.0f} W, integrator {_hi.i_w:+.0f} W, "
      f"reason {_hi.reason!r}")
check("⭐ THE CAP DOES NOT FIRE WHERE THE SLEW DID NOT BIND: with "
      "`integrator_max_w` deliberately above `max_w` the integrator keeps its "
      "own bound (-5000 W) instead of being dragged to the command - an "
      "unconditional cap would put it at -2400 W and make the option inert",
      _hi.i_w == -5000.0 and _hi.reason == "clamped")

# ⚠️ THE TWO RECORDED DEADLOCKS, AS TEST CASES. control.py names both at
# length; a gate written on the integrator is exactly the edit that brings them
# back, so each is driven rather than argued.
#
# (a) THE `i_w == 0.0` CASE. `main.py` resets the integrator to exactly 0.0 on
#     every stop and every reseed, and "may not wind further in the direction it
#     is already pushing" has no referent there. A gate that files zero under
#     one sign blocks the first push toward the other one forever.
z = compute(prev_w=0.0, grid_w=-6000.0, actual_w=0.0, tuning=T, sat_count=0,
            i_w=0.0)
check("at `i_w == 0.0` the first push toward charging is NOT blocked - the "
      f"command leaves zero ({z.target_w:+.0f} W) and the integrator moves "
      f"({z.i_w:+.0f} W)", z.target_w < 0.0 and z.i_w < 0.0)

# (b) THE SHRINK-ONLY ENCODING. "Only corrections that shrink |i_w|" is
#     unsatisfiable for both signs whenever the correction is larger than twice
#     the integrator, and the loop then sits at its last value forever. Driven
#     as the measured case: 0 W held into a 2 kW import.
imp = run(2000.0, cycles=12)[2]
print(f"    2 kW import from rest: setpoints "
      f"{[f['setpoint'] for f in imp[:6]]}")
check("a 2 kW import from rest recovers rather than sitting at its last value "
      "- the shrink-only encoding held 0 W indefinitely here",
      imp[-1]["setpoint"] >= 1000.0)

# ⚠️ AND THE GATE MUST NOT BE A DELAY. A loop that is slew-limited for a long
# ramp has to arrive, or the fix has bought export back with standing error.
far = run(-6000.0, cycles=40)[2]
print(f"    a 6 kW export, 40 cycles: setpoint reaches "
      f"{far[-1]['setpoint']:.0f} W of the {-BOUND_W:.0f} W rail")
check("⭐ THE COMMAND STILL ARRIVES: a railed export drives the command to the "
      "rail, one slew step a cycle, with the integrator bounded the whole way "
      "- the gate bounds the LEAD, it does not stop the loop",
      far[-1]["setpoint"] == -BOUND_W)


print()
if fails:
    print(f"{len(fails)} FAILED:")
    for f in fails:
        print(f"  - {f}")
    sys.exit(1)
print("all checks passed")