goodwe-addon/goodwe_controller/test_pv.py

"""Runnable check for TEL-11 - the PV production channel.  `python3 test_pv.py`

OPT-10 measured that this site's battery reaches 90 % SoC on 2 days of 16 and
that 20.5 kWh of surplus left it while the pack had room - 12.6 kWh of that with
the battery not charging at all - and could not say WHY, because PV production
is recorded NOWHERE. Three causes need three different fixes: no irradiance
(nothing to fix), a charge-power limit, or the optimiser declining to charge
(the only software defect). This card adds the missing channel. It does not do
the attribution; that is OPT-10's, with a week of history behind it.

⚠️ WHAT THIS FILE REFUSES TO ASSERT, SAID FIRST. It does not grade that a row
EXISTS. `publish()` handing a dict with `pv` in it is satisfied by `pv: 0.0`
forever, and a PV channel that reports 0 W for a dead sensor MANUFACTURES the
conclusion the card exists to test - "there was no surplus to store" - in a week
of history nobody will re-derive. So every check below separates three states
that a weaker one would collapse into one:

    NO SOURCE      `pv_entity` unset. No read, no entity, no topic, no row.
    NO READING     configured, and the sensor did not answer. The topic carries
                   HA's own `None`, the entity's state is `unknown`, the
                   recorder records a GAP.
    ZERO WATTS     configured, and the sensor said 0. The topic carries `0.0`.

The last two are the pair the acceptance criterion is about and part 3 asserts
the PAYLOAD BYTES of both, because they differ by exactly one thing that no
check on the VALUE alone can see.

⚠️ AND THE OTHER HALF OF THE CARD IS A NEGATIVE, GRADED AS ONE. This is
instrumentation: it must not become a control input, or the loop would react to
the quantity being measured and the measurement would be worthless. Part 6 is
the in-suite half of that claim (the shipped command is bit-identical across
absent / 0 W / 8 kW of PV, `inputs_ok` never moves, and an AST scan says which
modules can even see the material). `tools/tel11_pv_trace.py` is the other half:
it hashes a control document on THIS tree and on `release/1.0`, with the option
set and unset.

  Part 1  `pv.watts()`: the unit table, and every refusal, by name.
  Part 2  the shipped read path, including what a FAILED read does to the
          value a good one left behind.
  Part 3  ⭐ publication: no source / no reading / zero watts, as payload bytes.
  Part 4  announced and published are the same key, from the same predicate.
  Part 5  the operator surfaces, and that an unset option adds none of them.
  Part 6  ⭐ NOT A CONTROL INPUT.
  Part 7  what this file cannot see, with sizes, and the mutation table.

⚠️ MUTATING A 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. Read MUTATION.md.
"""

import ast
import asyncio
import sys
from pathlib import Path

from app import mqtt, pv
from app.main import Controller

HERE = Path(__file__).resolve().parent
fails = []


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


def state(value, unit="W", **attrs):
    """A Home Assistant state object, in the shape `hass.state()` returns."""
    return {"state": value, "attributes": dict({"unit_of_measurement": unit},
                                               **attrs)}


# ==============================================================================
print("1. pv.watts(): the unit is read off the entity, and a refusal is named")
# ==============================================================================
# ⚠️ LITERALS ON THE RIGHT-HAND SIDE, NOT `raw * WATTS_PER[u]` (habit 2). An
# assertion written against the expression under test cannot tell a correct
# table from a table that is wrong in the same direction twice.
CASES = [
    (state("1234", "W"), 1234.0, "W"),
    (state("5.2", "kW"), 5200.0, "kW"),
    (state("0.0072", "MW"), 7200.0, "MW"),
    (state("4500000", "mW"), 4500.0, "mW"),
    (state("0.0000072", "GW"), 7200.0, "GW"),
    (state("0.0000000072", "TW"), 7200.0, "TW"),
    (state("-8", "W"), -8.0, "W"),
]
print("    state            unit      -> watts")
_bad = []
for st, want, unit in CASES:
    got, why = pv.watts(st)
    print(f"    {st['state']:>14}   {unit:<4}      -> {got}   ({why})")
    if got is None or abs(got - want) > 1e-6 or why != unit:
        _bad.append((st, got, want))
check(f"every unit in the table converts to watts against a literal, and the "
      f"reason string is the unit it was read in. Wrong: {_bad}", not _bad)
check("...and the table is HOME ASSISTANT's own power units, so a card that "
      "adds one has to say so here: mW, W, kW, MW, GW, TW",
      sorted(pv.WATTS_PER) == sorted(["mW", "W", "kW", "MW", "GW", "TW"]))
check("...and every factor is an exact power of ten - nothing in this table is "
      "a number anybody chose",
      all(abs(v * 1000 - round(v * 1000)) < 1e-9 and v > 0
          for v in pv.WATTS_PER.values()))

# ⚠️ THE CASE THE WHOLE CARD TURNS ON, AND IT IS ASSERTED IN BOTH DIRECTIONS.
check("⭐ ZERO WATTS IS A READING: a sensor that says 0 gives 0.0, not None - "
      "the pack can be full and the sun can be down and the honest report of "
      "the second one is a number",
      pv.watts(state("0", "W")) == (0.0, "W")
      and pv.watts(state("0.0", "kW")) == (0.0, "kW"))

REFUSALS = [
    ("no state object at all (entity missing, or the read failed)", None,
     "no state"),
    ("the entity is `unavailable`", state("unavailable"), "unavailable"),
    ("the entity is `unknown`", state("unknown"), "unknown"),
    ("an empty state", state(""), "empty"),
    ("a state that is not a number", state("sunny"), "not numeric"),
    ("no unit_of_measurement at all", state("1234", ""), "no unit"),
    ("a unit that is not power", state("1234", "%"), "'%'"),
    # ⚠️ IN HA's `UnitOfPower` ENUM AND DELIBERATELY NOT IN THE TABLE. Its
    # factor is a measured conversion rather than a decimal shift, and a wrong
    # constant in the channel that exists to attribute a 20.5 kWh shortfall is
    # worse than a refusal somebody can read.
    ("BTU/h, which IS one of HA's power units", state("1234", "BTU/h"),
     "'BTU/h'"),
]
_r = []
print("    refusal                                    -> (watts, why)")
for label, st, needle in REFUSALS:
    got, why = pv.watts(st)
    print(f"    {label:<42} -> ({got}, {why!r})")
    if got is not None or needle not in why:
        _r.append(label)
check(f"⭐ EVERY REFUSAL ANSWERS None AND SAYS WHICH REFUSAL IT WAS - never "
      f"0.0, and never a bare 'no'. A reason an operator cannot act on sends "
      f"them to look at the wrong box. Wrong: {_r}", not _r)
check("...and the refused states are the SAME list `hass.number()` refuses, "
      "imported rather than respelled - one path cannot start accepting what "
      "the other rejects",
      pv.BAD_STATES is __import__("app.hass", fromlist=["x"]).BAD_STATES)
# habit 1: the precondition, as its own check. Without it the row above could
# be true of a table that had quietly become empty.
check("PRECONDITION: the refusal list and the unit table are both non-empty, "
      "so the two checks above quantify over something",
      len(REFUSALS) >= 8 and len(pv.WATTS_PER) >= 6)


# ==============================================================================
print()
print("2. the shipped read path")
# ==============================================================================
LIVE = {"max_w": 5000.0, "gain": 0.6, "slew_w": 400.0, "target_grid_w": -10.0,
        "deadband_w": 15, "step_w": 10}
BASE_OPTS = {"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}


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."""
    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, an inverter that follows, and a scripted PV sensor.

    ⚠️ `states` IS A SCRIPT, ONE ENTRY PER CYCLE, so a check can drive a GOOD
    read and then a FAILED one and ask what the failure did to the value the
    good one left behind. A fixture that answers the same thing forever cannot
    reach that failure - MUTATION.md habit 9's shape, one surface over.
    """

    def __init__(self, grid=0.0, soc=50.0, states=(), follow=False):
        self.grid = grid
        # ⚠️ `follow` MAKES THE METER RESPOND TO THE BATTERY, and part 6 needs
        # it. With a FIXED meter the error never shrinks, the integrator winds
        # to its bound and the command rides `slew_w` to the rail on every
        # input - so every stimulus produces the same ramp and an identity over
        # them is free. `grid = base - batt` is the site: a battery discharging
        # 610 W into a 600 W import leaves the meter at the -10 W this add-on
        # rests at.
        self.follow = follow
        self.soc = soc
        self.batt = 0.0
        self.n = 0
        self.writes = []
        self.state_calls = []
        self.states = list(states)

    async def limits(self, entity):
        return (-LIVE["max_w"], LIVE["max_w"], 1.0)

    async def number(self, entity, invert=False):
        grid = self.grid - self.batt if self.follow else self.grid
        return {"sensor.grid": grid, "sensor.soc": self.soc,
                "sensor.batt": self.batt}.get(entity)

    async def state(self, entity):
        self.state_calls.append((self.n, entity))
        if not self.states:
            return None
        return self.states[min(self.n, len(self.states) - 1)]

    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):
        return True


def run(cycles=3, grid=0.0, states=(), auto=True, follow=False, **over):
    """`cycles` real read/cycle/publish turns. Returns (controller, hass, pub)."""
    async def go():
        hass = _Hass(grid=grid, states=states, follow=follow)
        pub = _Pub()
        c = Controller(dict(BASE_OPTS, **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):
            await c.read_inputs()
            await c.cycle()
            c.publish()
            hass.n += 1
        return c, hass, pub
    return asyncio.run(go())


PV_ON = {"pv_entity": "sensor.pv_power"}

c_off, h_off, p_off = run(cycles=3)
check("⚠️ UNSET IS NOT READ AT ALL: with no `pv_entity` the add-on makes NO "
      "state call, for any entity, on any cycle - not a read that fails "
      "quietly, no read. This is what 'no warning spam' is made of",
      h_off.state_calls == [] and c_off.pv_w is None
      and c_off.pv_why == "not configured")

c_on, h_on, _ = run(cycles=3, states=[state("3210", "W")], **PV_ON)
print(f"    state calls: {h_on.state_calls}")
check("configured: the entity is read once per cycle, by name, and the value "
      "reaches `self.pv_w` in watts",
      h_on.state_calls == [(0, "sensor.pv_power"), (1, "sensor.pv_power"),
                           (2, "sensor.pv_power")]
      and c_on.pv_w == 3210.0 and c_on.pv_why == "W")

c_kw, _, _ = run(cycles=1, states=[state("3.21", "kW")], **PV_ON)
check("...and a kW sensor is read in kW: 3.21 kW is 3210 W, not 3.21 W. There "
      "is no magnitude heuristic anywhere on this path - 30 W of dusk and "
      "30 kW of noon are both ordinary readings",
      c_kw.pv_w == 3210.0 and c_kw.pv_why == "kW")

# ⚠️ THE ONE THAT SEPARATES "ABSENT" FROM "STALE", AND IT NEEDS TWO CYCLES.
c_drop, _, p_drop = run(cycles=2, states=[state("4100", "W"), None], **PV_ON)
print(f"    published pv by cycle: {[f['pv'] for f in p_drop.frames]}")
check("⭐ A FAILED READ CLEARS THE PREVIOUS GOOD ONE. The cycle after a 4100 W "
      "reading disappears publishes an ABSENCE, not 4100 W again - a value "
      "that survives its own sensor is indistinguishable from a real one at "
      "the moment somebody is working out whether the sun was shining",
      [f["pv"] for f in p_drop.frames] == [4100.0, None]
      and c_drop.pv_w is None and "no state" in c_drop.pv_why)


# ==============================================================================
print()
print("3. ⭐ publication: no source, no reading, zero watts - as payload bytes")
# ==============================================================================
class FakeClient:
    def __init__(self):
        self.msgs = []

    def publish(self, topic, payload=None, retain=False):
        self.msgs.append((topic, payload, retain))


def wire(values, include=("pv",)):
    """What the SHIPPED `MqttPublisher.publish()` hands paho for `values`."""
    pub = mqtt.MqttPublisher.__new__(mqtt.MqttPublisher)
    pub.omit, pub.units, pub.include, pub.enabled = set(), [], list(include), True
    pub.client = FakeClient()
    pub.publish(values)
    return pub.client.msgs


_zero = run(cycles=1, states=[state("0", "W")], **PV_ON)[2].frames[-1]
_absent = run(cycles=1, states=[None], **PV_ON)[2].frames[-1]
_nosrc = p_off.frames[-1]
print(f"    no source : 'pv' in frame = {'pv' in _nosrc}")
def pv_msg(values):
    """The `goodwe_ctl/pv` message the shipped publisher emits, or None."""
    got = [m for m in wire(values) if m[0] == "goodwe_ctl/pv"]
    return got[0] if got else None


print(f"    no reading: {_absent['pv']!r}  -> wire {pv_msg(_absent)}")
print(f"    zero watts: {_zero['pv']!r}  -> wire {pv_msg(_zero)}")
check("NO SOURCE: an install that never set `pv_entity` publishes no `pv` key "
      "at all - there is no entity to feed, so feeding one would be a topic "
      "nothing announced",
      "pv" not in _nosrc)
check("NO READING: the key is PRESENT and its value is None, which is a "
      "different frame from the one above. 'the option is off' and 'the sensor "
      "did not answer' are two facts and one field cannot carry both",
      "pv" in _absent and _absent["pv"] is None)
check("ZERO WATTS: 0.0, published as a number", _zero["pv"] == 0.0)
check("⭐ AND ON THE WIRE THE LAST TWO ARE DIFFERENT BYTES ON THE SAME TOPIC: "
      "`None` - which is Home Assistant's own payload for 'this sensor has no "
      "value', so the entity goes to `unknown` and the recorder records a GAP "
      "- against `0.0`. THIS IS THE ACCEPTANCE CRITERION: a week later, no PV "
      "reported is not PV reported 0 W",
      pv_msg(_absent) == ("goodwe_ctl/pv", "None", False)
      and pv_msg(_zero) == ("goodwe_ctl/pv", "0.0", False)
      and pv_msg(_absent) != pv_msg(_zero)
      # ...and the pv row is the ONLY thing that differs between the two
      # frames on the wire, so the claim above is about that row and not
      # about some other number the two fixtures happened to move.
      and [m for m in wire(_absent) if m[0] != "goodwe_ctl/pv"]
      == [m for m in wire(_zero) if m[0] != "goodwe_ctl/pv"])
check("...and the payload is HA's literal, from the module's own constant, not "
      "a string spelled here", mqtt.PAYLOAD_NONE == "None")
# ⚠️ AND THE BLAST RADIUS IS ONE KEY. The None-dropping in `publish()` is what
# `circulation` depends on - its note in main.py says a missing per-unit reading
# must leave the last figure standing rather than publish a zero that reads as
# health - so widening it would be a behaviour change to another card's row.
check("⭐ AND NO OTHER ROW CHANGED BEHAVIOUR: `circulation: None` is still "
      "DROPPED, not published as `None` - TEL-11 widened the rule for exactly "
      "one key and `ABSENT_AS_UNKNOWN` is the list",
      wire({"circulation": None, "grid": None, "loop_reason": None}) == []
      and mqtt.ABSENT_AS_UNKNOWN == frozenset({"pv"}))


# ==============================================================================
print()
print("4. announced and published are the same key, from the same predicate")
# ==============================================================================
# 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-11's row, and
# it is taken off the SHIPPED announcement and the SHIPPED `publish()`.
def announced(opts):
    return {t.rsplit("/", 2)[-2].replace(f"{mqtt.BASE}_", "")
            for t, _c in mqtt.discovery_payloads(
                include=mqtt.optional_keys(opts))}


check("with a source: `pv` is announced AND published",
      "pv" in announced(PV_ON) and "pv" in _absent)
check("without one: announced by nothing and published by nothing",
      "pv" not in announced({}) and "pv" not in _nosrc)
check("...and the predicate is ONE expression, `mqtt.optional_keys()`, read by "
      "the announcement and by `Controller.publish()` alike - two spellings is "
      "how an entity gets announced that nothing feeds",
      mqtt.optional_keys({}) == ()
      and mqtt.optional_keys({"pv_entity": ""}) == ()
      and mqtt.optional_keys({"pv_entity": "   "}) == ()
      and mqtt.optional_keys({"pv_entity": "sensor.x"}) == ("pv",))
check("...and a whitespace-only option is the SAME as an empty one everywhere "
      "- the controller does not read it either",
      run(cycles=1, pv_entity="   ")[1].state_calls == [])

# ⚠️ AND THE RESTART, WHICH IS THE OTHER HALF OF "RECORDABLE". Home Assistant
# keeps the entity_id it composed at FIRST discovery and keys its registry on
# `unique_id`, so a row whose id moved between two starts is a NEW entity beside
# a dead one and the history stays behind on the old id. Two independent
# announcements are asserted byte-identical, and the composed id is modelled
# INDEPENDENTLY of app/mqtt.py (the model is `tools/rehearse_roster.py`'s, and
# it is anchored there on the one composed id that was verified against the live
# system).
import re as _re                                                   # noqa: E402


def compose(device_name, entity_name):
    s_ = _re.sub(r"[^a-z0-9]+", "_", f"{device_name} {entity_name}".lower())
    return "sensor." + s_.strip("_")


_a1 = mqtt.discovery_payloads(include=mqtt.optional_keys(PV_ON))
_a2 = mqtt.discovery_payloads(include=mqtt.optional_keys(PV_ON))
_pv_cfg = [c for _t, c in _a1 if c["unique_id"] == "goodwe_ctl_pv"]
print(f"    composed entity_id: "
      f"{compose(mqtt.DEVICE['name'], _pv_cfg[0]['name'])}")
check("a restart re-announces the SAME bytes, so Home Assistant sees the entity "
      "it already has rather than a new one beside a dead one - the history is "
      "continuous across a restart of the add-on",
      [(t, __import__("json").dumps(c, sort_keys=True)) for t, c in _a1]
      == [(t, __import__("json").dumps(c, sort_keys=True)) for t, c in _a2]
      and len(_pv_cfg) == 1)
check("...and the id Home Assistant composes for it is "
      "`sensor.goodwe_pv_power`, which is what DOCS.md tells an operator to put "
      "on a dashboard - device name plus entity name, NOT the object_id",
      compose(mqtt.DEVICE["name"], _pv_cfg[0]["name"])
      == "sensor.goodwe_pv_power"
      and _pv_cfg[0]["state_topic"] == "goodwe_ctl/pv")


# ==============================================================================
print()
print("5. the operator surfaces")
# ==============================================================================
def rows_of(c):
    return [r["text"] for r in c.checks()]


def page_rows(c):
    return c.status_payload()["rows"]


_off_rows, _on_rows = rows_of(c_off), rows_of(c_on)
print(f"    unset: {len(_off_rows)} status rows; set: {len(_on_rows)}")
check("⚠️ AN UNSET OPTION ADDS NO STATUS ROW AND NO PAGE ROW - not a caution, "
      "not a 'no PV configured' nag. `peak_forecast_entity` DOES nag when it "
      "is unset, because an unset forecast leaves a money guard without an "
      "input; this is instrumentation and a permanent caution on every install "
      "that never asked for it is the warning spam the card forbids",
      not [t for t in _off_rows if "PV" in t]
      and not [r for r in page_rows(c_off) if r[0] == "PV"])
check("...and the whole row list is otherwise the same length, so nothing else "
      "moved either", len(_on_rows) == len(_off_rows) + 1)
_pv_row = [t for t in _on_rows if t.startswith("PV production")]
print(f"    {_pv_row}")
# habit 14: the ordered pair - value, entity and unit each in its OWN slot in
# the literal sentence, so a transposition is visible rather than merely
# present.
check("a good read says the watts, the entity it came from and the unit it was "
      "reported in, each in its own slot",
      _pv_row == ["PV production: 3210 W from sensor.pv_power "
                  "(reported in W)"]
      and ("PV", "3 210 W") in page_rows(c_on))

c_bad, _, _ = run(cycles=1, states=[state("1234", "%")], **PV_ON)
_bad_row = [r for r in c_bad.checks() if r["text"].startswith("PV production")]
print(f"    {[r['text'] for r in _bad_row]}")
check("a refused read says WHICH refusal, and names the entity - 'no reading' "
      "on its own sends an operator to look at a working sensor",
      len(_bad_row) == 1
      and _bad_row[0]["text"] == ("PV production: no reading from "
                                  "sensor.pv_power - unit '%' is not one of "
                                  "mW, W, kW, MW, GW, TW - it cannot be read "
                                  "in watts"))
check("⭐ ...and it is a CAUTION, never a `bad` row: `status_payload()` builds "
      "NOT READY out of rows that are neither ok nor warn, and a telemetry "
      "channel that could put a battery site into NOT READY would be "
      "observability with a foot in the control path",
      _bad_row[0] == {"ok": False, "warn": True, "text": _bad_row[0]["text"]}
      and not c_bad.status_payload()["banner"].startswith("NOT READY")
      and c_bad.status_payload()["level"] != "bad"
      and c_bad.status_payload()["banner"]
      == c_off.status_payload()["banner"])
check("...and the page row renders an absence as an em dash, not as 0 W",
      ("PV", "—") in page_rows(c_bad))


# ==============================================================================
print()
print("6. ⭐ NOT A CONTROL INPUT")
# ==============================================================================
# ⚠️ THE CLAIM IS NOT "pv is not in control.py". It is that the COMMANDS do not
# move, and that is graded by driving the same stimulus with the PV channel
# absent, at zero, and at 8 kW - a spread that would be impossible to miss if
# any term of the control law had learned to read it.
#
# ⚠️ AND THE FIRST VERSION OF THIS FIXTURE COULD NOT HAVE SEEN IT, WHICH IS
# MUTATION.md HABIT 23 AND THE CARD'S OWN BATTERY FOUND IT. The stimulus was a
# 6 kW export over 12 cycles; the card's central mutation - `self.grid -=
# self.pv_w` in `read_inputs()`, +106 B, on disk, purged, genuinely running -
# came back a SURVIVOR. Not because the property held: because at 6 kW of
# export the loop is against its rail and the command is moving at `slew_w`
# every cycle, so -6000 W and -14000 W of apparent error produce the SAME
# ramp, tick for tick. The mutant ran and could not change the answer on the
# data it was given.
#
# The stimulus is now a SMALL import the loop can actually settle on, far
# inside the rail, and the separation is asserted as its own precondition
# below: a check that cannot go red on the one mutation the card is about is
# the card's own habit-4.
_stim = dict(cycles=40, grid=600.0, follow=True)
_c_none, _h_none, _ = run(**_stim)
_c_absent, _h_absent, _ = run(**_stim, states=[None], **PV_ON)
_c_zero, _h_zero, _ = run(**_stim, states=[state("0", "W")], **PV_ON)
_c_big, _h_big, _ = run(**_stim, states=[state("8000", "W")], **PV_ON)
_w_none, _w_big = _h_none.writes, _h_big.writes
print(f"    unset   : {_w_none}  -> settled at {_c_none.target:g} W")
print(f"    8000 W  : {_w_big}  -> settled at {_c_big.target:g} W")
check("⭐ THE COMMANDS ARE BIT-IDENTICAL ACROSS unset / no reading / 0 W / "
      "8 kW of production, over 40 cycles of a real `cycle()` against a 600 W "
      "import the loop SETTLES on - every write, in order, and the same "
      "resting target. The PV channel changes no number that reaches an "
      "inverter",
      _w_none == _h_absent.writes == _h_zero.writes == _w_big
      and _c_none.target == _c_absent.target == _c_zero.target
      == _c_big.target)
check("PRECONDITION: that stimulus really does command something, and RESTS "
      "well inside the rail rather than riding the slew limit to it - at the "
      "rail every input produces the same ramp and the identity above would "
      "be free (habit 1)",
      len(_w_none) >= 4 and abs(_c_none.target) > 100
      and abs(_c_none.target) < 0.5 * LIVE["max_w"])
# ⚠️ THE HABIT-23 PRECONDITION, AND IT IS THE ONE THAT MAKES THE IDENTITY MEAN
# SOMETHING. "What would the mutation have had to change, and does the fixture
# reach that value?" Here: if the 8 kW HAD been subtracted from the meter
# reading, would this stimulus produce a different command? It is one line to
# ask, and on the first version of this fixture the answer was NO - see the
# note above `_stim`.
_c_shift, _h_shift, _ = run(cycles=40, grid=600.0 - 8000.0, follow=True)
print(f"    meter shifted by the same 8 kW -> settled at "
      f"{_c_shift.target:g} W")
check("⭐ PRECONDITION (habit 23): the same stimulus with the METER shifted by "
      "that 8 kW rests somewhere visibly different, so 'the commands did not "
      "move' is a fact about the code and not about a rail that would have "
      "flattened any input",
      _h_shift.writes != _w_big
      and abs(_c_shift.target - _c_big.target) > 4000)
check("...and `inputs_ok` does not move either: a PV sensor that never answers "
      "is not a missing INPUT, because nothing computes with it. An install "
      "whose solar integration breaks keeps regulating",
      _c_none.inputs_ok is True and _c_absent.inputs_ok is True)

# --- the structural half, and its scope is COMPUTED (habit 12) ---------------
APP = sorted((HERE / "app").glob("*.py"))
NEEDLES = ("pv_entity", "pv_w", "pv_why", "optional_keys", "ABSENT_AS_UNKNOWN")


def touches(src: str) -> bool:
    """Does this module MENTION TEL-11's material, by any spelling?

    ⚠️ A TEXT SCAN ON PURPOSE, AND IT IS THE CONSERVATIVE DIRECTION HERE. An
    AST scan would have to enumerate the forms a reference can take (attribute,
    subscript, `getattr`, a string key) and habit 12 says a hand-written list of
    forms is the same defect as a hand-written scope. Text over-reports - a
    comment counts - and over-reporting is safe for a check whose claim is
    "these modules do NOT mention it".
    """
    return any(n in src for n in NEEDLES) or "from . import pv" in src \
        or "from .pv import" in src


_touch = sorted(f.name for f in APP if touches(f.read_text(encoding="utf-8")))
print(f"    {len(APP)} modules under app/; TEL-11 material is in: {_touch}")
check(f"PRECONDITION: the scan found the whole package, not a subset - "
      f"{len(APP)} modules", len(APP) >= 25 and "control.py" in
      [f.name for f in APP])
check("⭐ EXACTLY THREE MODULES CAN SEE THE MATERIAL: `pv.py` (the reader), "
      "`main.py` (the read and the two operator surfaces) and `mqtt.py` (the "
      "publication). `control.py`, `arbiter.py`, `allocator.py`, `peakguard.py`, "
      "`arbitrage.py`, `maintenance.py`, `clamp.py`, `freeze.py`, "
      "`watchdog.py`, `delivery.py` and `circulation.py` do not mention it in "
      "any spelling, comments included",
      _touch == ["main.py", "mqtt.py", "pv.py"])
# habit 4 in the instrument: a scanner that cannot go red is not a scanner.
check("...and the scan CAN go red - a module body with the material in it is "
      "caught, so the green above is the tree speaking and not the scanner "
      "being blind",
      touches("x = opts['pv_entity']\n") and touches("from . import pv\n")
      and not touches("# nothing to see, a pvc pipe\n"))


# ==============================================================================
print()
print("7. what this file CANNOT see, with sizes - and the mutation table")
# ==============================================================================
# habit 13. An enumerated list with an honest gap is worth more than a claim of
# completeness a reviewer falsifies in ten minutes.
for line in [
    "Home Assistant itself. That `None` on the state topic renders as state",
    "  `unknown`, and that the recorder keeps it, is READ OUT OF HA's source",
    "  (mqtt/const.py PAYLOAD_NONE, mqtt/sensor.py's `_message_received`,",
    "  retrieved 2026-09-24) and quoted in app/mqtt.py. No HA runs in this",
    "  suite, so what is graded here is the BYTES this add-on publishes.",
    "Whether the operator's PV sensor is telling the truth. A sensor wired",
    "  backwards reads -4 kW at noon; this file asserts that the sign is",
    "  published unchanged, never that it is right. That is what the history",
    "  the card creates is FOR - see DECISIONS/TEL-11-pv-production.md §2.",
    "The control-path claim OUTSIDE app/. `tools/`, the ESPHome firmware and",
    "  the ingress page are not scanned by part 6's census.",
    "A roster of more than one. Every fixture here is the shipped roster of",
    "  one; the row is a SITE row and there is nothing per-unit to separate.",
    "Wall-clock behaviour: no fixture here sleeps, and the read is driven by",
    "  the caller's loop rather than by the 1 s tick.",
]:
    print("    " + line)

print()
print("    mutation                                          bytes  reds")
print("    ------------------------------------------------  -----  --------")
for row in [
    ("M1  the pv row moved into SENSORS", "-79 B", "3 test_mqtt + 1 here"),
    ("M2  publish() drops a None pv row again", "-98 B", "1 here"),
    ("M3  ABSENT_AS_UNKNOWN widened past the one key", "+23 B", "1 here"),
    ("M4  a dead sensor reported as 0 W", "-1 B", "4 here"),
    ("M5  a failed read leaves the last value standing", "+46 B", "2 here"),
    ("M6  the unit read and then ignored", "-18 B", "2 here"),
    ("M7  `self.grid -= self.pv_w` in read_inputs()", "+106 B",
     "2 here + the trace"),
    ("M8  the row announced and fed unconditionally", "-52 B", "3 here"),
    ("M9  the pv caution made a `bad` row (NOT READY)", "+1 B", "1 here"),
    ("M10 the read gate removed - unset is read anyway", "-10 B",
     "2 here + 2 crashes"),
]:
    print(f"    {row[0]:<48}  {row[1]:>5}  {row[2]}")
print("    ⚠️ M7 WAS A SURVIVOR ON THE FIRST BATTERY and part 6's note says why:")
print("    the fixture drove the loop to its RAIL, where every input produces")
print("    the same ramp. MUTATION.md habit 23, with a saturated actuator in")
print("    place of a coinciding value - and it is in that file now.")
print("    ⚠️ M10's two extra reds are CRASHES, not named checks: every")
print("    fixture in test_inverter.py and test_telemetry.py predates the PV")
print("    read and has no `state()` on its Hass double, so an unconditional")
print("    read is an AttributeError. A red, but an unnamed one - habit 20's")
print("    shape - and it is recorded rather than tidied away.")
print("    Every row above was RUN, in the CI image, on a purged tree, one at a")
print("    time - see DECISIONS/TEL-11-pv-production.md §5 for the transcript.")
print("    ⚠️ A RED COUNT IS A PROPERTY OF THE SUITE AS MUCH AS OF THE MUTATION:")
print("    these name THIS tree. Re-measure rather than inherit.")

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