goodwe-addon/goodwe_controller/test_telemetry.py

"""Runnable check for TEL-10: the loop's integrator, published.

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

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

⚠️ 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 Tuning, 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. The
# integrator moves `gain * error` = 3.6 kW on the first cycle and reaches its
# 5000 W bound on the second; the command may only move 400 W a cycle, so it is
# at 800 W when the integrator is already railed and needs twelve more cycles to
# get there. That gap is the fixture: for ten cycles the two numbers differ by
# kilowatts, and a row published from the wrong one is wrong by kilowatts.
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 LEAD TIME IS PART OF THE CLAIM, NOT A NICETY. "The published number
# reaches the bound" is satisfied by publishing the COMMAND too - it gets there
# as well, twelve cycles 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` this line is red and the value-only
# form was green.
check("⭐ THE PUBLISHED INTEGRATOR REACHES ITS BOUND, AND REACHES IT TEN CYCLES "
      f"BEFORE THE COMMAND DOES. A 6 kW export winds it to exactly "
      f"{BOUND_W:.0f} W - `integrator_max_w` is None, which control.py resolves "
      "to `max_w` - while the command is still travelling 400 W a cycle",
      first_at_bound is not None and i_row[first_at_bound] == -BOUND_W
      and first_cmd_at_bound - first_at_bound >= 10)

# ⚠️ 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 NOWHERE NEAR it 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(sp_row[first_at_bound]) <= BOUND_W / 2)
check("...and the gap is the slew limit and not an accident: the command is "
      "still travelling 400 W a cycle while the integrator stands still",
      abs(sp_row[first_at_bound + 1] - sp_row[first_at_bound])
      == slew_limit_w(Tuning(**LIVE)))

# 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 large while the command is catching up, so the row is not "
      f"a permanent zero: worst lead {min(lead_row):.0f} W",
      min(lead_row) <= -2000.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"] < -1000.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()
if fails:
    print(f"{len(fails)} FAILED:")
    for f in fails:
        print(f"  - {f}")
    sys.exit(1)
print("all checks passed")