"""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")