goodwe-addon/goodwe_controller/test_validate.py

"""Runnable check for the startup gate.  `python3 test_validate.py`

Two halves, and the second is the one that matters. The first asserts that the
three refusable configurations are refused. The second asserts that everything
on validate.py's "deliberately not refused" list still STARTS - because the way
a validator dies is not by missing a fault, it is by refusing something harmless
until somebody deletes the call.

⚠️ 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.maintenance import MAINT_OPTION_DEFAULTS, MaintConfig
from app.validate import (
    ACTUATOR_WRITE_KEYS, FIRMWARE_INGEST_VETO_S, ConfigRefused,
    actuator_writers, refuse, validate_startup_config,
)

fails = []


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


def base(**over):
    """A configuration that must start. Everything else is this, spoilt."""
    # ⚠️ The maintenance rates are named EXPLICITLY, at the values config.yaml
    # ships (MAINT-04). Under MAINT-04 that was load-bearing: with the key
    # absent, rule 6 fell back to a 2500 W literal that against this 2000 W
    # ceiling is a genuine refusal, so a base() without them was a config that
    # does not start and every check below would have been measuring the wrong
    # thing. CFG-03 closed that - the fallback is now config.yaml's own 2000 -
    # so they are here to say out loud what this fixture asks for rather than to
    # rescue it. The CFG-03 section at the foot of this file is what asserts
    # that absent and present are now THE SAME NUMBER; do not delete these two
    # keys, because a base() that agrees with the default by accident is a
    # fixture that cannot tell you when the default moves.
    o = {"setpoint_entity": "number.goodwe_setpoint", "max_w": 2000,
         "soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
         "meter_source": "off",
         "maintenance_charge_w": 2000, "maintenance_discharge_w": 2000}
    o.update(over)
    return o


def why(problems, needle):
    return any(needle in p for p in problems)


# --------------------------------------------------------------------------- #
print("the baseline must start, or nothing below means anything")
check("a good config with a 5000 W device is accepted",
      validate_startup_config(base(), -5000, 5000) == [])
check("...and with no device reachable yet, too",
      validate_startup_config(base(), None, None) == [])

# --------------------------------------------------------------------------- #
print("1. max_w against the device's own rating")
# ⚠️ THE ORIGINAL BUG. main.py used to log 'the device wins' and run anyway,
# with write_setpoint() clamping, so the configured ceiling was never the real
# one and the only evidence had scrolled off the log.
p = validate_startup_config(base(max_w=8000), -5000, 5000)
check("max_w above the device maximum is refused", len(p) == 1)
check("...and the message names the configured value", why(p, "8000"))
check("...and the device's value", why(p, "5000"))
check("...and says which one wins", why(p, "THE DEVICE WINS"))
check("max_w exactly at the device maximum is fine",
      validate_startup_config(base(max_w=5000), -5000, 5000) == [])
check("max_w under it is fine",
      validate_startup_config(base(max_w=2000), -5000, 5000) == [])
# The charge side is a separate rail and an asymmetric device is legal.
p = validate_startup_config(base(max_w=4000), -2500, 5000)
check("a max_w the device cannot take on the CHARGE side is refused too",
      len(p) == 1 and why(p, "charge side"))
check("...naming the device's floor", why(p, "-2500"))
# ⚠️ Not guessed at: the entity may simply not have booted yet. run_control()
# re-runs this the moment the real rating arrives.
check("an unreachable device skips the range check rather than assuming",
      validate_startup_config(base(max_w=8000), None, None) == [])
check("a non-numeric max_w is refused",
      why(validate_startup_config(base(max_w="lots"), -5000, 5000), "not a number"))

# --------------------------------------------------------------------------- #
# 1b. SAFETY-20's AC4 decision, PINNED HERE RATHER THAN ONLY WRITTEN DOWN
# --------------------------------------------------------------------------- #
# AC4 asked whether a device rating that disarms SAFETY-18's freeze rail
# additionally warrants a refusal of its own, the way `sign_disagree_w >= max_w`
# is refused. THE ANSWER IS NO, AND THE REASON IS RULE 2 ABOVE: the configuration
# is `max_w` above the device rating, which rule 2 already refuses on the
# identical predicate. A rail-specific rule (`0.95 * max_w > dev_max`) would fire
# on a strict SUBSET of what rule 2 refuses, so it could never be the rule that
# stopped a start.
#
# ⚠️ THAT DECISION HAS A PREMISE, AND THIS IS THE PREMISE, EXECUTED. If rule 2 is
# ever softened to a warning - which is exactly what main.py used to do before
# CFG-01 - the reasoning recorded in validate.py's "deliberately not refused"
# list silently stops being true, and SAFETY-18 is left with only its runtime
# NOT COVERED row. Then somebody has to come back and re-decide AC4.
from app.freeze import RAIL_FRAC  # noqa: E402  - imported here, beside its use

_DEV = 4700.0                       # the ticket's inverter
_CEIL = 5000.0                      # the ticket's configured ceiling
check("SAFETY-20's premise: a rating below max_w disarms the freeze rail, so "
      f"{RAIL_FRAC:g} * max_w is unreachable on the ticket's install",
      _DEV < RAIL_FRAC * _CEIL)
check("...and rule 2 REFUSES that whole configuration at startup, which is why "
      "SAFETY-20 added no rule of its own (AC4)",
      len(validate_startup_config(base(max_w=_CEIL), -_DEV, _DEV)) == 1)
check("...on the rating side specifically, not incidentally through some other "
      "rule", why(validate_startup_config(base(max_w=_CEIL), -_DEV, _DEV),
                  "THE DEVICE WINS"))
check("...and the case the refusal CANNOT reach - the rating unread because "
      "ESPHome was not up - still starts, which is precisely why SAFETY-20's "
      "answer had to be a runtime status row and not a startup gate",
      validate_startup_config(base(max_w=_CEIL), None, None) == [])

# --------------------------------------------------------------------------- #
print("2. an actuator to write to")
# A controller with no actuator: write_setpoint() returns on the blank entity,
# so no rating is read, no heartbeat is sent, layer 1 pins the inverter at 0 W,
# and the add-on looks perfectly healthy the whole time.
for blank in ("", "   ", None):
    p = validate_startup_config(base(setpoint_entity=blank), None, None)
    check(f"setpoint_entity {blank!r} is refused", len(p) == 1)
check("...and the message says why it is dangerous rather than untidy",
      why(validate_startup_config(base(setpoint_entity=""), None, None),
          "heartbeat"))
# --------------------------------------------------------------------------- #
# specs §1.5: TWO WRITERS ON ONE ACTUATOR, AND WHAT ACTUALLY GRADES IT
# --------------------------------------------------------------------------- #
# ⚠️ WHAT USED TO BE HERE, AND WHY IT WAS NOT ENOUGH. This was one line:
#
#     check("there is exactly one way to reach the actuator",
#           ACTUATOR_WRITE_KEYS == ("setpoint_entity",))
#
# and validate.py's comment said the §1.5 non-goal was ENFORCED by construction -
# exactly one such key, so a second writer is not expressible in the options
# schema at all. Both were true. Both graded THE SHAPE OF THE OPTIONS SCHEMA
# rather than the writes, and the difference is not academic: the config below
# has TWO WRITERS ON ONE ACTUATOR and leaves that tuple's length at 1. The check
# stays green while the property it protects is gone, which is this repository's
# named failure mode - a check that observes something other than what it claims
# to observe.
#
# So the first thing this section does is DEMONSTRATE the old check passing on
# the hazardous config, rather than assert in prose that it would have.
_PER_INVERTER_HAZARD = base(inverters=[
    {"id": "inverter-a", "setpoint_entity": "number.goodwe_setpoint"},
    # ⚠️ INVERTER B DOES NOT EXIST. Nothing is measured from it here; it is a
    # config fixture, and the only thing that matters about it is that its
    # `setpoint_entity` NAMES THE SAME ENTITY as the scalar key above - which is
    # the shape of the mistake, and the shape CFG-04 makes reachable.
    {"id": "inverter-b", "setpoint_entity": "number.goodwe_setpoint"}])

check("⚠️ THE DEFECT, DEMONSTRATED: on a per-inverter config with two writers "
      "aimed at ONE entity, the old proxy is still green - it counts option "
      "KEYS, and there is still exactly one key called setpoint_entity. This "
      "is the check that could not see the thing it was named after",
      len(ACTUATOR_WRITE_KEYS) == 1)

_census = actuator_writers(_PER_INVERTER_HAZARD)
print(f"    the same config, censused by writer: "
      f"{ {k: len(v) for k, v in _census.items()} }")
check("...and the replacement SEES it, because it counts writers per ACTUATOR "
      "rather than keys: three declarations, one entity",
      list(_census) == ["number.goodwe_setpoint"]
      and len(_census["number.goodwe_setpoint"]) == 3)
_hazard_problems = validate_startup_config(_PER_INVERTER_HAZARD, -5000, 5000)
check("AC 3: and the config is REFUSED, where today it starts",
      why(_hazard_problems, "one actuator"))
check("...and the refusal names the actuator and every writer aimed at it, so "
      "an operator can see WHICH declarations to remove rather than being told "
      "only that something is wrong",
      why(_hazard_problems, "number.goodwe_setpoint")
      and why(_hazard_problems, "inverters[0].setpoint_entity")
      and why(_hazard_problems, "inverters[1].setpoint_entity"))

# ⚠️ AND THE GATE MUST NOT REFUSE THE CONFIG CFG-04 IS ACTUALLY FOR. A validator
# that fires on the legitimate case gets switched off by the first person it
# inconveniences, and then none of this is enforced. Two inverters with two
# entities are two writers on TWO actuators, which is not the non-goal - it is
# the whole point of the per-inverter roster.
#
# ⚠️ THE ENTRIES BELOW GAINED `name` AND `max_w` WHEN CFG-04 LANDED, AND THE
# HAZARD FIXTURE ABOVE DELIBERATELY DID NOT. SAFETY-28 wrote both against a
# shape that did not exist yet and guessed `id`; CFG-04 chose `name`, and made
# `name`/`setpoint_entity`/`max_w` required of every entry (rule 9). The hazard
# fixture is left in the OLD guessed shape on purpose - the claim it makes is
# that `actuator_writers()` finds writers STRUCTURALLY, under a container and a
# key spelling this file was never told, and rewriting it into today's shape
# would quietly turn it into a check that the census knows about `inverters`.
# This one is the opposite claim - that a real, valid, two-inverter config is
# ACCEPTED - so it has to be a config the shipped validator actually accepts.
_PER_INVERTER_OK = {k: v for k, v in base().items() if k != "setpoint_entity"}
_PER_INVERTER_OK["inverters"] = [
    {"name": "inverter-a", "setpoint_entity": "number.goodwe_setpoint",
     "max_w": 2000, "bms_soc_floor": 10},
    # ⚠️ INVERTER B DOES NOT EXIST. 2000 W here is the SHIPPED default repeated,
    # not a rating measured off a second device - SPIKE-04 measures that when
    # there is one to measure. Nothing in this file depends on the number.
    # ⚠️ SAFETY-32 made `bms_soc_floor` a REQUIRED per-entry key, and the two
    # values here DIFFER (10 / 15) rather than being copied. 15 % is SIMULATED -
    # there is no second pack to read a cut-out off - and the reason it is not
    # simply 10 is MUTATION.md habit 1: a fixture whose two per-unit numbers are
    # equal cannot show a check reading the wrong unit's.
    {"name": "inverter-b", "setpoint_entity": "number.goodwe_setpoint_b",
     "max_w": 2000, "bms_soc_floor": 15}]
check("two inverters with their own entities are ACCEPTED - one writer each, "
      "no shared actuator, and that is CFG-04's whole purpose",
      validate_startup_config(_PER_INVERTER_OK, -5000, 5000) == [])
check("...and today's single-inverter config is untouched: one writer, one "
      "actuator, and the census finds it wherever it is declared",
      validate_startup_config(base(), -5000, 5000) == []
      and actuator_writers(base()) ==
      {"number.goodwe_setpoint": ["setpoint_entity"]})

# The census is found STRUCTURALLY, not by looking for a container called
# `inverters` - CFG-04 has not chosen that name yet, and a census that guessed
# wrong would go green while two writers existed, which is the same defect one
# step along.
check("a writer nested under a container this file has never heard of is "
      "still counted, because the walk is over the options TREE and not over a "
      "list of blessed key names",
      len(actuator_writers({"setpoint_entity": "number.x",
                            "whatever_cfg04_calls_it": {
                                "unit_b": {"setpoint_entity": "NUMBER.X"}}})
          ["number.x"]) == 2)
check("...and entity ids are folded to lower case, so a second writer cannot "
      "be smuggled past the census on Number.X versus number.x",
      why(validate_startup_config(base(inverters=[{"setpoint_entity":
                                                   "NUMBER.GOODWE_SETPOINT"}]),
                                  -5000, 5000), "one actuator"))
# ⚠️ THIS CHECK CHANGED SHAPE IN CFG-04, AND THE REASON IS A DECISION RATHER
# THAN AN ADJUSTMENT. It used to assert that a blank per-inverter entity left
# the whole config ACCEPTED - "blank is how an operator leaves a slot
# unconfigured". CFG-04 decided that is not true of a ROSTER ENTRY: an entry
# with no `setpoint_entity` is a unit that `allocator.plan()` still gives a
# share of the site target to and that nothing can command, on a status page
# reporting a site at full strength. Rule 9 refuses it by name. The claim about
# the CENSUS is untouched and is now asserted directly on the census, where it
# always belonged - a blank contributes no writer, so it cannot manufacture a
# two-writer refusal out of two unconfigured slots.
check("a BLANK per-inverter entity is not a writer - the same rule the scalar "
      "key gets, since write_setpoint() returns immediately on a blank entity",
      actuator_writers(base(inverters=[{"setpoint_entity": "  "},
                                       {"setpoint_entity": ""}]))
      == {"number.goodwe_setpoint": ["setpoint_entity"]})
check("...but the ENTRY is refused all the same, by rule 9 and not by the "
      "census: a roster slot nothing can write still takes a share of the site "
      "target every cycle and delivers none of it",
      why(validate_startup_config(base(inverters=[{"name": "inverter-b",
                                                   "max_w": 2000,
                                                   "setpoint_entity": "  "}]),
                                  -5000, 5000), "has no setpoint_entity"))

# --------------------------------------------------------------------------- #
print("3. a meter that can never read")
# ⚠️ These predicates belong to p1.validate_meter_options(), which build_source()
# also uses. Asserted here to prove the gate actually consults it - one entry
# point, not two.
check("meter_source off asks nothing of the P1 keys",
      validate_startup_config(base(meter_source="off"), -5000, 5000) == [])
check("homewizard_local with no p1_host is refused",
      why(validate_startup_config(base(meter_source="homewizard_local"),
                                  -5000, 5000), "p1_host"))
check("...and with a host it is accepted",
      validate_startup_config(base(meter_source="homewizard_local",
                                   p1_host="192.168.2.250"), -5000, 5000) == [])
check("a poll cadence at meter_max_age_s is refused at STARTUP, not just disabled",
      why(validate_startup_config(base(meter_source="homewizard_local",
                                       p1_host="m", meter_poll_s=30,
                                       meter_max_age_s=30), -5000, 5000),
          "meter_max_age_s"))
check("ha_signed with no p1_net_entity is refused",
      why(validate_startup_config(base(meter_source="ha_signed"), -5000, 5000),
          "p1_net_entity"))
check("a phase list disagreeing with meter_phases is refused",
      why(validate_startup_config(base(meter_source="ha_signed",
                                       p1_net_entity="sensor.n", meter_phases=3,
                                       p1_phase_net_entities=["a", "b"]),
                                  -5000, 5000), "phase-count"))
check("mqtt_p1 with no topic is refused",
      why(validate_startup_config(base(meter_source="mqtt_p1"), -5000, 5000),
          "meter_mqtt_topic"))
check("ha_dsmr with only half its entities is refused",
      why(validate_startup_config(base(meter_source="ha_dsmr",
                                       p1_import_entity="sensor.i"), -5000, 5000),
          "p1_export_entity"))
check("a misspelt meter_source is refused rather than silently ignored",
      why(validate_startup_config(base(meter_source="ha_signd"), -5000, 5000),
          "is not one of"))

# --------------------------------------------------------------------------- #
print("4. meter_max_age_s against the firmware's compile-time veto")
# ⚠️ The schema allows int(5,300); the firmware's threshold is baked in at 30 s
# and wins silently. Same shape as the max_w case: a configured bound overridden
# by a lower layer, with nothing saying so.
hw = dict(meter_source="homewizard_local", p1_host="m")
p = validate_startup_config(base(**hw, meter_max_age_s=120), -5000, 5000)
check("a meter_max_age_s above the firmware veto is refused", len(p) == 1)
check("...naming the configured value", why(p, "120"))
check("...naming the firmware's value", why(p, "30"))
check("...and saying which one wins", why(p, "THE FIRMWARE WINS"))
check("the tested pair 30/30 is accepted",
      validate_startup_config(base(**hw, meter_max_age_s=30), -5000, 5000) == [])
check("one second above it is refused",
      validate_startup_config(base(**hw, meter_max_age_s=31), -5000, 5000) != [])
# Below the veto the add-on gives up FIRST, which is fail-safe and visible on
# its own status page. Only above is the silent case.
check("below the veto is fine - the add-on being stricter is not a conflict",
      validate_startup_config(base(**hw, meter_max_age_s=10,
                                   meter_poll_s=4), -5000, 5000) == [])
# ⚠️ Scope: meter_max_age_s is read by nothing but P1Ingest, and the veto keys
# off a sensor only published when ingestion runs. With the meter off it governs
# nothing, so refusing would be pure trivia.
check("with meter_source off the number governs nothing and is not refused",
      validate_startup_config(base(meter_source="off", meter_max_age_s=300),
                              -5000, 5000) == [])
check("the constant is the firmware's, not a rounded guess",
      FIRMWARE_INGEST_VETO_S == 30.0)

# --------------------------------------------------------------------------- #
print("6. a maintenance charge the ceiling can actually accommodate (MAINT-04)")
# ⚠️ THE SHIPPED BUG. config.yaml shipped max_w 2000 against
# maintenance_charge_w 2500, so the cheap-window shaping asked for hi=-2500
# against the supervised lo=-2000 - lo > hi, arbiter clause 3, 0 W. Maintenance
# silently never charged in a cheap window on a default install.
p = validate_startup_config(base(max_w=2000, maintenance_charge_w=2500),
                            -5000, 5000)
check("maintenance_charge_w above max_w is refused", len(p) == 1)
check("...and the message names the charge rate", why(p, "2500"))
check("...and the ceiling", why(p, "2000"))
check("...and says which one wins", why(p, "THE CEILING WINS"))
# ⚠️ Equality is REACHABLE, not a near miss: lo=-2000, hi=-2000 is the single
# point -2000 W. Refusing it would be exactly the trivia that gets a validator
# deleted, and it is the boundary the rule is written `>` for.
check("maintenance_charge_w exactly at max_w starts",
      validate_startup_config(base(max_w=2000, maintenance_charge_w=2000),
                              -5000, 5000) == [])
check("below it starts",
      validate_startup_config(base(max_w=2000, maintenance_charge_w=1500),
                              -5000, 5000) == [])
# Not scoped to maintenance_enabled: the pair is contradictory either way, and
# scoping would leave the trap armed for the one action - ticking the box - that
# cannot be expected to change what a charge rate means.
check("...and it is refused with maintenance_enabled off, as shipped",
      why(validate_startup_config(
          base(max_w=2000, maintenance_charge_w=2500,
               maintenance_enabled=False), -5000, 5000), "THE CEILING WINS"))

# ⚠️ MAINT-05: AND AT TWO PACKS IT IS THE WEAKEST CAP, NOT THE FIRST ENTRY'S.
# The comment this replaces in validate.py said, in as many words, that it took
# the FIRST unit's cap because there was exactly one `Maintenance` and it could
# only ever recalibrate `inverters[0]` - and that "when `maint_unit` stops being
# `inverters[0]`, this line moves with it". It has. Every pack is now
# recalibrated on its own schedule, so the shaping claim can be filed on any of
# them and the cap that contradicts it first is the smallest.
#
# ⚠️ THE FIXTURE'S TWO CAPS ARE DELIBERATELY DIFFERENT AND THE RATE SITS
# BETWEEN THEM. Two equal caps could not tell "grades the first" from "grades
# the smallest", and a rate above both would be refused by either rule - so
# 3000 W against caps of 5000 W and 2000 W is the one arrangement where the old
# answer and the new one disagree. ⚠️ 2000 W for the second pack is a
# HYPOTHESIS: inverter B does not physically exist and nothing here is measured
# from one.
def _roster(a_max, b_max, **over):
    o = base(**over)
    for k in ("setpoint_entity", "soc_entity", "batt_entity", "max_w"):
        o.pop(k, None)
    o["inverters"] = [
        dict(name="inverter-a", setpoint_entity="number.sp_a",
             soc_entity="sensor.soc_a", batt_entity="sensor.batt_a",
             max_w=a_max),
        dict(name="inverter-b", setpoint_entity="number.sp_b",
             soc_entity="sensor.soc_b", batt_entity="sensor.batt_b",
             max_w=b_max)]
    return o


_two = _roster(5000, 2000, maintenance_charge_w=3000, maintenance_discharge_w=1000,
               sign_disagree_w=300, unit_sign_disagree_w=300)
_p_two = [p for p in validate_startup_config(_two, -5000, 5000)
          if "maintenance_charge_w" in p]
print(f"    5000 W and 2000 W packs, charge 3000 W -> {_p_two}")
check("⚠️ A CHARGE RATE THE SMALLEST PACK CANNOT TAKE IS REFUSED, even though "
      "the FIRST pack could. Before MAINT-05 the second pack was never "
      "recalibrated, so its cap could not contradict anything; it is now, and "
      "its cheap-window charge would ask for hi=-3000 against lo=-2000 - the "
      "`lo > hi` contradiction, 0 W, on the pack nobody was watching",
      len(_p_two) == 1 and why(_p_two, "max_w (inverter-b) is 2000 W"))
check("...and the message names the pack the number came off, not the first "
      "entry - an operator told to lower a number on the wrong device is being "
      "sent to the wrong device by the message that exists to stop that",
      not why(_p_two, "max_w (inverter-a)"))
check("...while the same rate against two packs that can both take it starts, "
      "so this is a comparison and not a blanket refusal of rosters",
      [p for p in validate_startup_config(
          _roster(5000, 4000, maintenance_charge_w=3000,
                  maintenance_discharge_w=1000, sign_disagree_w=300,
                  unit_sign_disagree_w=300), -5000, 5000)
       if "maintenance_charge_w" in p] == [])
check("...and at a ROSTER OF ONE `min()` over one cap is that cap: the same "
      "message, the same boundary, the same refusal the flat config always got",
      why(validate_startup_config(base(max_w=2000, maintenance_charge_w=2500),
                                  -5000, 5000), "THE CEILING WINS")
      and validate_startup_config(base(max_w=2000, maintenance_charge_w=2000),
                                  -5000, 5000) == [])

# ⚠️ THE SIBLING, CHECKED RATHER THAN ASSUMED. The shaping is NOT symmetric:
# drain files Claim.set at P_MAINTENANCE and owns the actuator, so the ceiling
# CLAMPS it (+2000 W) instead of contradicting it, and the arbiter's explain()
# names the clamp every cycle. Loud, and it still moves energy - so warned, not
# refused. If someone later "fixes the asymmetry" by making this a refusal,
# this check goes red and they have to justify it.
check("maintenance_discharge_w above max_w WARNS but starts",
      validate_startup_config(base(max_w=2000, maintenance_discharge_w=2500),
                              -5000, 5000) == [])

# The demonstration the rule is derived from, asserted rather than described.
from app.arbiter import (                   # noqa: E402
    Claim, P_LOOP, P_MAINTENANCE, P_MAINT_SHAPE, P_SAFETY, resolve,
)

def _charge(max_w, charge_w):
    return resolve([Claim.limit("supervised", P_SAFETY, -max_w, max_w, "max_w"),
                    Claim.limit("maintenance", P_MAINT_SHAPE, hi=-charge_w,
                                reason="cheap-window charge"),
                    Claim.set("loop", P_LOOP, 500.0, "grid")])

def _drain(max_w, disch_w):
    return resolve([Claim.limit("supervised", P_SAFETY, -max_w, max_w, "max_w"),
                    Claim.set("maintenance", P_MAINTENANCE, disch_w, "drain"),
                    Claim.set("loop", P_LOOP, 500.0, "grid")])

r = _charge(2000, 2500)
check("arbiter: the refused pair really does emit 0 W",
      r.target_w == 0.0 and r.contradiction)
r = _charge(2000, 2000)
check("arbiter: the permitted equality really does charge",
      r.target_w == -2000.0 and not r.contradiction)
r = _charge(3000, 2500)
check("arbiter: raising max_w above the rate charges at the rate",
      r.target_w == -2500.0 and not r.contradiction)
r = _drain(2000, 2500)
check("arbiter: the drain SET is clamped, not contradicted",
      r.target_w == 2000.0 and not r.contradiction)
check("...and the clamp names itself, which is why it is not refused",
      "supervised" in (r.bound_by or "") and "limited by" in r.explain())

# --------------------------------------------------------------------------- #
print("the SHIPPED defaults must pass the rule they are governed by (AC3)")
# ⚠️ Shipping a default the validator refuses would be worse than the bug it
# refuses. Read config.yaml itself rather than a transcription of it - a
# transcription is what lets the two drift apart silently. Regex, not PyYAML:
# the add-on image has no PyYAML (Dockerfile installs py3-aiohttp/py3-paho-mqtt
# only) and a check that skips on ImportError is not a check.
import os                                   # noqa: E402
import re                                   # noqa: E402

_CFG = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.yaml")
with open(_CFG, encoding="utf-8") as _fh:
    _txt = _fh.read()
_shipped = {k: float(v) for k, v in re.findall(
    r"^  (max_w|maintenance_charge_w|maintenance_discharge_w|sign_disagree_w): "
    r"(-?\d+(?:\.\d+)?)\s*$", _txt, re.M)}
check("config.yaml really was read (all four defaults found)",
      set(_shipped) == {"max_w", "maintenance_charge_w",
                        "maintenance_discharge_w", "sign_disagree_w"})
check(f"the shipped defaults are accepted {_shipped}",
      validate_startup_config(base(**_shipped), -5000, 5000) == [])
check("...and the shipped charge rate is at or under the shipped ceiling",
      _shipped["maintenance_charge_w"] <= _shipped["max_w"])
check("...and the shipped discharge rate too, so it warns about nothing",
      _shipped["maintenance_discharge_w"] <= _shipped["max_w"])

# --------------------------------------------------------------------------- #
print("the shipped meter_freeze_s, pinned (TEST-06 / AC3)")
# ⚠️ WHY A NUMBER GETS ITS OWN CHECK. `meter_freeze_s: 300` is the threshold
# SAFETY-18's entire argument is about, and before this it was pinned NOWHERE:
# 300 -> 3600 left test_freeze.py AND this file green, because the schema
# accepts int(60,3600) and no check ever read the shipped value. The schema
# bounds what is CONFIGURABLE; this pins what is SHIPPED, which is a different
# question and the one the docs answer.
#
# Read by regex like _shipped above, for the same reason: no PyYAML in the
# image, and a check that skips on ImportError is not a check.
_freeze_s = [float(v) for v in re.findall(
    r"^  meter_freeze_s: (\d+(?:\.\d+)?)\s*$", _txt, re.M)]
_freeze_schema = re.findall(r"^  meter_freeze_s: int\((\d+),(\d+)\)\s*$",
                            _txt, re.M)
check(f"config.yaml carries exactly one meter_freeze_s default {_freeze_s}",
      len(_freeze_s) == 1)
check("...and exactly one schema range for it", len(_freeze_schema) == 1)
_hold_s = _freeze_s[0] if _freeze_s else None
_lo, _hi = ((float(_freeze_schema[0][0]), float(_freeze_schema[0][1]))
            if _freeze_schema else (None, None))
check("the shipped meter_freeze_s is still the measured 300 s that freeze.py, "
      "config.yaml's own comment and test_freeze.py are all written at",
      _hold_s == 300.0)
check(f"...and it is inside its own schema range ({_lo}-{_hi} s)",
      None not in (_lo, _hi, _hold_s) and _lo <= _hold_s <= _hi)

# The two properties the number has to have, DERIVED rather than restated - so
# they go red on a value that is merely schema-legal. Both come straight from
# config.yaml's own comment on the key.
#
#   upper: the alert must land inside the 900 s quarter that bills the capacity
#          tariff, or the notification arrives after the money is spent;
#   lower: the slowest a HEALTHY meter can tick a 1 Wh register and still look
#          unchanged for hold_s is 3600/hold_s W, and that must stay inside the
#          deadband or the conjunction's two terms stop being mutually
#          exclusive (freeze.py, "WHERE hold_s COMES FROM").
_QUARTER_S = 900.0
_shipped_deadband = [float(v) for v in re.findall(
    r"^  deadband_w: (\d+(?:\.\d+)?)\s*$", _txt, re.M)]
check(f"config.yaml carries exactly one deadband_w default {_shipped_deadband}",
      len(_shipped_deadband) == 1)
_db = _shipped_deadband[0] if _shipped_deadband else None
_slowest_healthy_w = 3600.0 / _hold_s if _hold_s else None
print(f"    shipped meter_freeze_s {_hold_s} s: alert inside the "
      f"{_QUARTER_S:.0f} s billing quarter, and a healthy meter slow enough to "
      f"look unchanged for it is {_slowest_healthy_w} W against a "
      f"{_db} W deadband")
check(f"...the alarm still lands inside the {_QUARTER_S:.0f} s billing quarter",
      _hold_s is not None and _hold_s < _QUARTER_S)
check("...and a healthy meter that looks unchanged for it is still inside the "
      "deadband, so the two terms stay mutually exclusive",
      _slowest_healthy_w is not None and _db is not None
      and _slowest_healthy_w < _db)

# ⚠️ AND THE RANGE'S OWN BOUNDS. "300 sits inside the range" is worth nothing
# while the range itself is unpinned: widened to `int(1,86400)` an operator can
# configure `meter_freeze_s: 86400` and the guard is dead for a day at a time,
# with every check in this repo still green. That mutation survived the first
# cut of this PR and is what this block closes. Both numbers are argued for in
# config.yaml's own comment and in freeze.py, and each is re-derived here:
#
#   60 s   the floor freeze.py calls NOT A RECOMMENDATION - there a healthy
#          meter can look unchanged at 60 W, four times the shipped deadband, so
#          the conjunction's two terms stop being mutually exclusive and only
#          the margin is left;
#   3600 s the ceiling - already four times the billing quarter, so the top of
#          the range has given up detecting a freeze inside the quarter that
#          bills it. It is the last bound between a supported configuration and
#          a disabled guard, and moving it is a decision rather than a tidy-up.
print(f"    schema range {_lo}-{_hi} s: at the floor a healthy meter that looks "
      f"unchanged is "
      + (f"{3600.0 / _lo:.0f} W" if _lo else "?")
      + f" against a {_db} W deadband; the ceiling is "
      + (f"{_hi / _QUARTER_S:.1f}x" if _hi else "?")
      + " the billing quarter")
check(f"the schema range itself is still the argued 60-3600 s (found "
      f"{_lo}-{_hi})", (_lo, _hi) == (60.0, 3600.0))
check("...a floor that gives up the mutual exclusion and keeps only the "
      "margin, which is why config.yaml calls it not a recommendation",
      None not in (_lo, _db) and 3600.0 / _lo > _db)
check("...and a ceiling already past the billing quarter, so no value the "
      "schema still accepts can alarm inside the quarter that bills it",
      _hi is not None and _hi > _QUARTER_S)

# --------------------------------------------------------------------------- #
print("max_w must be a positive magnitude")
# Schema-blocked on a real install, but validate_startup_config() is also
# sim/inject.py's entry point and can be handed any dict at all.
for bad in (0, -1, -2000):
    p = validate_startup_config(base(max_w=bad), -5000, 5000)
    check(f"max_w {bad} is refused", len(p) == 1 and why(p, "must be positive"))
check("...and the message explains it would read as a trip, not a mistake",
      why(validate_startup_config(base(max_w=0), -5000, 5000), "contradiction"))
# ⚠️ SAFETY-02 changed what this case has to say, and the change is real rather
# than bookkeeping: with a 1 W ceiling the default 300 W sign-disagreement
# threshold is unreachable, so rule 5 refuses - correctly, because the watchdog
# would report itself armed while being incapable of firing. What this check is
# FOR is "a positive max_w is not itself a refusal", so it now names a sign
# threshold the ceiling can actually reach, and the interaction gets its own
# check below.
check("the smallest sane max_w still starts",
      validate_startup_config(base(max_w=1, sign_disagree_w=0.5,
                                   unit_sign_disagree_w=0.5,
                                   maintenance_charge_w=1), -5000, 5000) == [])
check("...but that same 1 W ceiling with the DEFAULT sign threshold is refused",
      why(validate_startup_config(base(max_w=1), -5000, 5000), "never fire"))
# ⚠️ SAFETY-30 ADDED A SECOND WAY TO DISARM A SIGN WATCHDOG BY CONFIGURATION,
# and the fixture above now has to name BOTH thresholds - which is the check
# doing its job rather than a bookkeeping edit. `unit_sign_disagree_w` bounds a
# unit's own SHARE, which the same `supervised` claim holds at +-max_w, so at or
# above the ceiling it is unsatisfiable by construction exactly as
# `sign_disagree_w` is. Refused even at the shipped roster of one, where this
# watchdog is not armed: a config that cannot work is refused when it is
# written, not on the day somebody adds a second inverter.
check("...and the per-inverter threshold has the same rule, named separately "
      "so an operator is told WHICH watchdog they disarmed",
      why(validate_startup_config(base(max_w=1, sign_disagree_w=0.5,
                                       maintenance_charge_w=1), -5000, 5000),
          "unit_sign_disagree_w"))
check("...and one watt under that ceiling is reachable, so accepted",
      validate_startup_config(base(max_w=2000, unit_sign_disagree_w=1999),
                              -5000, 5000) == [])

# --------------------------------------------------------------------------- #
print("what must NOT be refused - a validator that fires on trivia gets deleted")
# DOCS.md: the arbiter's `supervised` limit still binds the wire to +-max_w, so
# the only cost of the extra headroom is unwind latency after saturation.
check("integrator_max_w above max_w warns but starts",
      validate_startup_config(base(max_w=2000, integrator_max_w=9000),
                              -5000, 5000) == [])
check("integrator_max_w 0 (follow max_w) starts",
      validate_startup_config(base(integrator_max_w=0), -5000, 5000) == [])
# Fail-safe AND already a red NOT READY banner naming the entity: a refusal on
# top of that adds nothing an operator can act on.
check("a missing soc_entity is left to the readiness banner",
      validate_startup_config(base(soc_entity=""), -5000, 5000) == [])
check("a missing batt_entity likewise",
      validate_startup_config(base(batt_entity=""), -5000, 5000) == [])
# The transient case. ESPHome is routinely not up when the add-on boots, and
# refusing here would turn a device that arrives late into a restart loop.
check("a setpoint entity that is configured but not yet found starts",
      validate_startup_config(base(), None, None) == [])
check("meter_poll_s with thin but non-zero headroom starts",
      validate_startup_config(base(meter_source="homewizard_local", p1_host="m",
                                   meter_poll_s=20, meter_max_age_s=30),
                              -5000, 5000) == [])

# --------------------------------------------------------------------------- #
print("the refusal must be legible as a refusal, not as a crash")
p = validate_startup_config(base(max_w=8000, setpoint_entity=""), -5000, 5000)
check("several problems are all reported, not just the first", len(p) >= 1)
refuse(p)   # prints the banner; the eye is the assertion for its wording
err = ConfigRefused(p)
check("ConfigRefused carries every problem", err.problems == p)
check("...and str() names one of them", str(err) == p[0])

# --------------------------------------------------------------------------- #
print("the LATE refusal, when the device only shows up after the gate has run")
# ⚠️ This path had no test at all and no live demonstration - the review's
# catch. ESPHome is routinely not up when the add-on boots, so hass.limits()
# returns (None, None), the range checks are skipped, and run_control() is the
# first place the device's real rating is ever known. Everything below is about
# what happens when it turns out to conflict.
from app.main import Controller            # noqa: E402


class _LateHass:
    """Reports the range only run_control() gets to see."""

    def __init__(self, lo, hi):
        self.lo, self.hi = lo, hi
        self.writes = []

    async def limits(self, entity):
        return (self.lo, self.hi, 1.0)

    async def number(self, *a, **k):
        return 0.0

    async def set_number(self, entity, value):
        self.writes.append(value)
        return True


class _Store2:
    data = {}

    def set(self, *a):
        pass


class _Pub2:
    enabled = False

    def publish(self, *a, **k):
        pass

    def close(self):
        pass


def late(opts_over, lo, hi):
    """Drive run_control() against a device that only reports (lo, hi) now.

    ⚠️ Returns whether run_control() EXITED ON ITS OWN. That is the assertion
    that matters: on a refusal it must return without entering the while loop,
    and on a clean config it must not return at all. So the clean case is stopped
    from outside after it has had time to write, rather than waited on - a
    wait_for() here would just time out, which is the correct behaviour dressed
    up as a failure.
    """
    async def go():
        stop = asyncio.Event()
        hass = _LateHass(lo, hi)
        # A device narrower than the shipped 5000 W schema ceiling is exactly
        # the case that cannot be reached through the Supervisor on the live
        # install, so it is reached here instead.
        c = Controller(base(**opts_over), hass, _Store2(), _Pub2(), stop=stop)
        c.auto = True
        task = asyncio.create_task(c.run_control())
        for _ in range(30):                    # up to ~3 s
            if task.done():
                break
            await asyncio.sleep(0.1)
        exited = task.done()
        if not exited:
            c.stopping = True
        await asyncio.wait_for(task, timeout=5)
        return c, stop, hass, exited
    return asyncio.run(go())

c, stop, hass, exited = late({"max_w": 5000}, -3600.0, 3600.0)
check("a device narrower than max_w, discovered late, is refused",
      c.refused is not None and len(c.refused) == 1)
check("...naming both values", why(c.refused, "5000") and why(c.refused, "3600"))
check("...signalling amain to wind down", stop.is_set())
check("...and returning instead of looping forever", exited)
# The whole point: run_control() must not fall through into the while loop. A
# single write here would mean it commanded the inverter under a config it had
# just declared unusable.
check("...having never entered the control loop, so it wrote nothing",
      hass.writes == [])
check("...and it is recorded in the event log an operator reads",
      any("REFUSED" in e for e in c.events))

# The same path when there is no conflict: it must NOT refuse, and it must
# actually regulate - asserted by a write reaching the device, not by absence.
c2, stop2, hass2, exited2 = late({"max_w": 2000}, -5000.0, 5000.0)
check("a late device that agrees does not trip the re-check", c2.refused is None)
check("...and the control loop runs, writing to the device", hass2.writes != [])
check("...and does not signal a shutdown", not stop2.is_set() and not exited2)

# ⚠️ THE FOOTGUN THE REVIEW FOUND. `stop` used to default to None, so this
# construction was legal and the refusal path then returned into a live process
# with no control loop, no exit and no further output.
try:
    Controller(base(), _LateHass(-5000.0, 5000.0), _Store2(), _Pub2())
    check("a Controller with no stop Event is rejected at construction", False)
except TypeError:
    check("a Controller with no stop Event is rejected at construction", True)

# --------------------------------------------------------------------------- #
print("CFG-03: the key ABSENT is the SAME NUMBER everywhere, through real code")
# ⚠️ WHAT THIS REPLACES, AND WHY IT IS THE OPPOSITE ASSERTION. MAINT-04 left a
# TRIPWIRE here whose PASSING CONDITION WAS THE BUG: it asserted that
# config.yaml said 2000 while main.py's two `opts.get(..., 2500)` fallbacks and
# MaintConfig's dataclass default said 2500, and it went red if EITHER side
# moved - including when somebody closed the divergence. That check has been
# DELETED, as it and validate.py's ⚠️ block both instructed, and this is what
# stands in its place: the same four consumers, exercised the same way, now
# agreeing.
#
# The four are the four the defect ran through. The validator ACCEPTS a dict
# with the key missing; main.py BUILDS a MaintConfig from that same dict; the
# dataclass answers on its own with no dict at all; and the arbiter is what
# turns their disagreement into a number an operator can see. Any three of them
# agreeing was never the property that mattered.
_absent = {k: v for k, v in base().items()
           if not k.startswith("maintenance_")}
_ship_charge = _shipped["maintenance_charge_w"]
_ship_drain = _shipped["maintenance_discharge_w"]

check("the validator still ACCEPTS a dict with every maintenance key missing - "
      "the fixture is the reachable one, not a refusal in disguise",
      validate_startup_config(_absent, -5000, 5000) == [])

# 1. the mirror against the shipped file. Read, not transcribed: a transcription
#    is the thing that drifted.
check(f"the ONE in-code mirror equals what config.yaml ships "
      f"(charge {MAINT_OPTION_DEFAULTS['maintenance_charge_w']}, "
      f"shipped {_ship_charge:.0f})",
      float(MAINT_OPTION_DEFAULTS["maintenance_charge_w"]) == _ship_charge
      and float(MAINT_OPTION_DEFAULTS["maintenance_discharge_w"]) == _ship_drain)

# 2. MaintConfig on its own - the one that bites when main.py is bypassed
#    entirely, which is sim/inject.py and most of this suite.
check(f"MaintConfig() answers the shipped number, not a third one "
      f"({MaintConfig().charge_w:.0f} W)",
      MaintConfig().charge_w == _ship_charge
      and MaintConfig().discharge_w == _ship_drain)

# 3. main.py's own path, through a real Controller rather than through a regex
#    over its source. A source scan can only say the literal is gone; this says
#    the object an operator's install actually runs on holds the right number.
_ctl_absent = Controller(_absent, None, _Store2(), _Pub2(), stop=asyncio.Event())
print(f"    a Controller built from a dict with NO maintenance key: "
      f"charge_w={_ctl_absent.maint.cfg.charge_w:.0f} W, "
      f"discharge_w={_ctl_absent.maint.cfg.discharge_w:.0f} W")
check("main.py builds the shipped number from a dict that does not contain it",
      _ctl_absent.maint.cfg.charge_w == _ship_charge
      and _ctl_absent.maint.cfg.discharge_w == _ship_drain)
# ...and it is the SAME number when the operator supplies it explicitly, which
# is what "the default is not a special case" means.
_ctl_named = Controller(base(), None, _Store2(), _Pub2(), stop=asyncio.Event())
check("...and an explicit key builds the identical config, so absent and "
      "present are not two code paths",
      _ctl_named.maint.cfg == _ctl_absent.maint.cfg)

# 4. the arbiter - the consequence itself. Under MAINT-04 this pair was
#    (2000, 2500) and resolved to 0 W with contradiction=True; that was the
#    ACCEPTED-then-contradicted hole. The rates come out of the built config,
#    not out of a literal, so if the single source moves this moves with it.
_r_ok = _charge(float(_absent["max_w"]), _ctl_absent.maint.cfg.charge_w)
print(f"    arbiter with that config: {_r_ok.explain()}")
check("the arbiter CHARGES instead of contradicting itself - the hole MAINT-04 "
      "pinned open is closed at the only place it was ever reachable",
      _r_ok.target_w == -_ship_charge and not _r_ok.contradiction)
_r_drain = _drain(float(_absent["max_w"]), _ctl_absent.maint.cfg.discharge_w)
check("...and the drain SET is not clamped either, so the twin key the card "
      "did not name is closed too",
      _r_drain.target_w == _ship_drain and not _r_drain.contradiction)

# 5. nothing typed a number behind the mirror's back. Rule 6's fallback and
#    main.py's are the two sites the divergence lived at, so they are the two
#    sites a scan is worth running over.
_srcs = {name: open(os.path.join(os.path.dirname(os.path.abspath(__file__)),
                                 "app", name), encoding="utf-8").read()
         for name in ("main.py", "validate.py", "maintenance.py")}
_literals = {name: re.findall(
    r'"(maintenance_[a-z_]+)"\s*,\s*(-?\d+(?:\.\d+)?)\s*[,)]', src)
    for name, src in _srcs.items()}
check(f"no numeric fallback for any maintenance_* key survives in main.py, "
      f"validate.py or maintenance.py {_literals}",
      not any(_literals.values()))
check("...and MaintConfig's dataclass defaults carry no numeric literal either "
      "- they are indexed out of the mirror",
      not re.search(r"^\s*(charge_w|discharge_w):\s*float\s*=\s*-?\d",
                    _srcs["maintenance.py"], re.M))

# --------------------------------------------------------------------------- #
print("SPIKE-04a: inverter B's documented rating, and the provenance beside it")
# ⚠️ WHY A DOCUMENTATION CARD LEFT A CHECK BEHIND. Inverter B's rating was
# written down in THREE places - config.yaml's commented roster example,
# DOCS.md's, and tools/supervisor_shape_probe.py's "a blob an operator would
# POST" - as 3000 W, a figure nobody had measured, read off a datasheet, or
# sourced at all. It was wrong: SPIKE-04a (2026-09-04) identified inverter B
# from the owner's model number and serial as a GoodWe GW5048D-ES rated 4600 W
# on the battery path. Three transcriptions of one number drifting away from
# the truth together is this repo's recurring defect in its plainest form, and
# correcting them once WITHOUT PINNING THEM buys nothing: the next editor
# touches one of the three and the other two stay wrong, silently.
#
# ⚠️ AND THE PROVENANCE IS GRADED, NOT ONLY THE VALUE, WHICH IS THE HALF THAT
# MATTERS. 4600 W IS FROM A DATASHEET AND IS NOT A MEASUREMENT. The datasheet
# prints it under BATTERY INPUT DATA, so it is probably DC, while `max_w` is
# graded against 0x0518 - the inverter's own AC power - which would put the AC
# ceiling nearer 4324 W. An editor who tidies that caveat away leaves a paper
# number reading as an observation of hardware that does not exist yet, and
# THAT is the failure this card exists to prevent. So the caveat is what these
# checks reach for; a check that graded 4600 alone would watch the value and
# miss the lie.
#
# ⚠️ WHAT THIS DELIBERATELY DOES NOT GRADE: the test fixtures. Several files
# build a second inverter at 3000 W on purpose, because the property those
# checks consume is that B's rating is UNMISTAKABLY not A's, and the real pair
# (5000/4600) is a 400 W gap - narrower than `slew_w`. Those are illustrative,
# they say so at the line, and they are not statements of B's rating. The three
# sites below are the ones that ARE such statements.
_HERE = os.path.dirname(os.path.abspath(__file__))
_docs_txt = open(os.path.join(_HERE, "DOCS.md"), encoding="utf-8").read()
_probe_txt = open(os.path.join(_HERE, "tools", "supervisor_shape_probe.py"),
                  encoding="utf-8").read()

# One regex per site, each anchored on something only THAT site has, so a match
# cannot silently wander into a neighbouring blob. The probe's anchor is the
# operator blob's real entity name: its section D refusal entries use
# `number.b` and one of them carries a deliberate 9999, so an anchor loose
# enough to reach them would grade the wrong dict and still look green.
_b_rating = {
    "config.yaml": re.search(
        r"#\s+- name: inverter-b\b(?:[^\n]*\n)*?\s*#\s+max_w:\s*(\d+)", _txt),
    "DOCS.md": re.search(
        r"^  - name: inverter-b\b(?:[^\n]*\n)*?^    max_w:\s*(\d+)",
        _docs_txt, re.M),
    "supervisor_shape_probe.py": re.search(
        r'"name": "inverter-b",\s*"setpoint_entity": "number\.goodwe_b_setpoint_w"'
        r'(?:[^}]*?)"max_w":\s*(\d+)', _probe_txt),
}
check(f"all three documented rosters were FOUND "
      f"{ {k: bool(v) for k, v in _b_rating.items()} } - a regex that matches "
      "nothing is a green check that read nothing",
      all(_b_rating.values()))
# ⚠️ THE `len(...) == 3` IS NOT DECORATION. Without it a regex that stopped
# matching would shrink this dict, and "all the values agree" would go GREEN on
# the two sites that were still read - a check whose data no longer reaches the
# failure it names, which is the exact fault this whole section is about.
_b_vals = {k: int(m.group(1)) for k, m in _b_rating.items() if m}
check(f"...and all three state the SAME rating for inverter B {_b_vals}",
      len(_b_vals) == 3 and len(set(_b_vals.values())) == 1)
check(f"...and it is 4600 W - the GW5048D-ES datasheet figure, not the 3000 W "
      f"hypothesis that predated the unit being identified {_b_vals}",
      len(_b_vals) == 3 and set(_b_vals.values()) == {4600})

# The provenance triple. Each site must still say WHERE 4600 came from, that it
# is graded against an AC register, and what the DC-vs-AC ambiguity does to it -
# the last carried by the 4324 W figure, which is the one marker that cannot
# survive a careless "it's 4600 now, done" rewrite.
#
# ⚠️ SEARCHED IN A WINDOW AROUND THE ROSTER, NOT OVER THE WHOLE FILE, AND THE
# FIRST VERSION OF THIS CHECK DID THE LATTER. `0x0518` already appeared once in
# config.yaml and once in DOCS.md before this card, for unrelated reasons - so a
# whole-file `"0x0518" in src` was GREEN on those two files no matter what
# happened next to inverter B's rating. That is a check whose data cannot reach
# the failure it names, in the same file that exists to complain about them.
# The window is the text either side of the roster match, so deleting the caveat
# beside the number goes red even where the same word survives elsewhere.
_sites = {"config.yaml": _txt, "DOCS.md": _docs_txt,
          "supervisor_shape_probe.py": _probe_txt}
_MARKERS = ("datasheet", "0x0518", "4324")
_prov = {}
for _name, _src in _sites.items():
    _m = _b_rating[_name]
    _win = _src[max(0, _m.start() - 2500):_m.end() + 2500].lower() if _m else ""
    _prov[_name] = {k: k in _win for k in _MARKERS}
for _name, _got in sorted(_prov.items()):
    check(f"{_name} still carries B's provenance WITHIN 2500 chars of the "
          f"roster - the datasheet it came from, the 0x0518 it is graded "
          f"against, and the 4324 W the DC reading would make it {_got}",
          all(_got.values()))


# --------------------------------------------------------------------------- #
print("rules 5, 5b and 6 on a ROSTER SET - the smallest cap at EVERY position")
# ⚠️ SAFETY-32 ROUND 5, AND THE REASON THIS SECTION EXISTS AT ALL. PR #48 was
# rejected on FOUR MUTATIONS THAT SURVIVED A 19/19 GREEN SUITE, all of them in
# the three lines above:
#
#   validate.py:779  min(caps)   -> min(caps[:1])                  +4 B  GREEN
#   validate.py:779  min(caps)   -> min(caps[-1:])                 +5 B  GREEN
#   validate.py:778  caps = [cap for _, cap in caps_ix[:1]]        +4 B  GREEN
#   validate.py:893  min(caps_ix[-1:], key=lambda p: p[1])         +5 B  GREEN
#
# NONE of them is an equivalent mutant: each one deletes a safety refusal.
# `max_w: 5000` on inverter-a beside `2000` on inverter-b with
# `sign_disagree_w: 3000` is refused by the shipped validator with "the
# SAFETY-02 sign-disagreement watchdog can never fire" and ACCEPTED by every one
# of the four - the site starts, with a watchdog that reports `armed` on the
# status page and cannot trip on pack B. That is rule 5's own failure mode
# reached THROUGH rule 5.
#
# WHY THE OLD FIXTURE COULD NOT SEE IT, which is the whole lesson and is now
# MUTATION.md habit 10: the fixture was ONE roster, `[5000, 2000]`. On that
# roster `min` and `last` and `[-1:]` and `[:2]` all answer 2000. A fixture of N
# members cannot separate two reductions that agree on N members, and adding a
# member moves the hole instead of closing it.
#
# SO THE FIXTURE IS A SET, NOT A ROSTER: the smallest cap at EVERY POSITION of
# rosters of width 2, 3, 4 and 5. `[:1]` needs width 2 to be caught, `[:2]`
# needs 3, `[:3]` needs 4 - the sweep is what terminates the regress that a
# single fixture cannot. And the FAMILY TABLE below is COMPUTED AND PRINTED, so
# a member no fixture kills shows up as a NAMED SURVIVOR rather than as an
# absence from a list nobody re-derived.
#
# ⚠️ THE DIFFERING CAPS ARE A FIXTURE PROPERTY AND MUST STAY ONE. In production
# both packs are expected to floor at 10 % and SPIKE-04a puts inverter B at
# 4600 W against A's 5000 W - a 400 W gap, narrower than slew_w. A fixture built
# on the real pair would be degenerate on exactly the axis these checks measure.
# ⚠️ AND INVERTER B DOES NOT PHYSICALLY EXIST (SPIKE-04, ~2026-09-12). Nothing
# in this section is measured off a second unit; the rosters are simulated and
# are labelled so.
_BIG, _SMALL = 5000.0, 2000.0
_RATE = 3000.0                     # strictly between them, which is the point


def _roster_n(caps, **over):
    """A roster of len(caps) units, unit-0..unit-N, each with its own cap."""
    o = base(**over)
    for k in ("setpoint_entity", "soc_entity", "batt_entity", "max_w"):
        o.pop(k, None)
    o["inverters"] = [
        dict(name=f"unit-{i}", setpoint_entity=f"number.sp{i}",
             soc_entity=f"sensor.soc{i}", batt_entity=f"sensor.batt{i}",
             max_w=c, bms_soc_floor=10)
        for i, c in enumerate(caps)]
    o.update(over)
    return o


# The three rules that reduce over `unit_caps_indexed()`, each as
# (option name, the token its refusal is recognised by).
_CAP_RULES = (
    ("sign_disagree_w", "SAFETY-02 sign-disagreement"),
    ("unit_sign_disagree_w", "SAFETY-30 per-inverter sign-disagreement"),
    ("maintenance_charge_w", "maintenance charge in a cheap window"),
)
_QUIET = {"sign_disagree_w": 300.0, "unit_sign_disagree_w": 300.0,
          "maintenance_charge_w": 1000.0, "maintenance_discharge_w": 1000.0}

# THE SET: width x position of the one small cap.
_SET = []
for _n in (2, 3, 4, 5):
    for _pos in range(_n):
        _caps = [_BIG] * _n
        _caps[_pos] = _SMALL
        _SET.append((_n, _pos, tuple(_caps)))

# habit 1: the PRECONDITION as its own check. If a later edit makes the caps
# equal, this goes red rather than the claims below going quietly vacuous.
check("PRECONDITION: every fixture in the set has a UNIQUE smallest cap, that "
      "cap is at the stated position, and the rate under test sits STRICTLY "
      "between it and the others - two equal caps could not tell `min` from "
      "`first` from `last` and the whole set would be measuring nothing",
      len(_SET) == 14
      and all(c[p] == _SMALL and c.count(_SMALL) == 1 and _SMALL < _RATE < _BIG
              for _n2, p, c in _SET)
      and {p for _n2, p, _c in _SET} == {0, 1, 2, 3, 4})


def _verdict(opts, token):
    """(refused?, the unit the message NAMES, the cap it QUOTES)."""
    ps = [p for p in validate_startup_config(opts, -5000, 5000) if token in p]
    if not ps:
        return (False, None, None)
    m = re.search(r"max_w \(([^)]+)\) is (\d+) W", ps[0])
    return (True, m.group(1) if m else None, float(m.group(2)) if m else None)


# --- the shipped answer, on every member of the set -------------------------
_rows = []
for _opt, _token in _CAP_RULES:
    for _n, _pos, _caps in _SET:
        _o = _roster_n(list(_caps), **dict(_QUIET, **{_opt: _RATE}))
        _rows.append((_opt, _n, _pos, _caps, _verdict(_o, _token)))
print(f"    {len(_CAP_RULES)} rules x {len(_SET)} rosters = {len(_rows)} runs "
      f"of the REAL validator (rate {_RATE:.0f} W, caps {_SMALL:.0f}/"
      f"{_BIG:.0f} W, simulated rosters - inverter B does not exist):")
for _opt in sorted({r[0] for r in _rows}):
    _got = [(r[1], r[2], r[4]) for r in _rows if r[0] == _opt]
    print(f"      {_opt:<22} " + " ".join(
        f"n{n}p{p}:{'REFUSE ' + str(u) if ok else 'starts'}"
        for n, p, (ok, u, _c) in _got))

check("⚠️ EVERY ONE OF THE THREE CAP RULES REFUSES ON EVERY MEMBER OF THE SET, "
      "and NAMES THE UNIT THE NUMBER CAME OFF - at width 2, 3, 4 and 5, with "
      "the small cap at every position in turn. A reduction that answered "
      "`first`, `last`, `[:1]`, `[-1:]`, `[:2]` or `[:-1]` instead of `min` "
      "misses at least one member of this set, which is what four hand-run "
      "mutations against a 19/19 green suite proved the single `[5000, 2000]` "
      "fixture could not do",
      all(ok and unit == f"unit-{p}" and cap == _SMALL
          for _o2, _n2, p, _c2, (ok, unit, cap) in _rows))

check("...and it is a COMPARISON, not a blanket refusal of rosters: the same "
      "three rates against a roster whose caps can ALL take them start "
      "cleanly, at every width. Without this the line above is satisfied by a "
      "validator that refuses everything (habit 2 in the negative direction)",
      all(_verdict(_roster_n([_BIG] * n, **dict(_QUIET, **{opt: _RATE})),
                   tok) == (False, None, None)
          for opt, tok in _CAP_RULES for n in (1, 2, 3, 4, 5)))

# --- the POSITIONAL-SLICE FAMILY TABLE, computed and printed ----------------
# ⚠️ COMPUTED, NOT LISTED. Each candidate is applied to the SAME caps the
# validator saw, and its (unit, cap) answer is compared against what the shipped
# validator actually said on that roster. A candidate that agrees everywhere is
# a reduction this fixture set CANNOT distinguish from the shipped one, and it
# is printed as a SURVIVOR rather than left out of a table nobody re-derived.
def _pick(caps, fn):
    ix = list(enumerate(caps))
    try:
        i, c = fn(ix)
    except (ValueError, IndexError):
        return None
    return (f"unit-{i}", c)


_FAMILY = {
    "min (SHIPPED)":  lambda ix: min(ix, key=lambda p: p[1]),
    "max":            lambda ix: max(ix, key=lambda p: p[1]),
    "first":          lambda ix: ix[0],
    "last":           lambda ix: ix[-1],
    "min(caps[:1])":  lambda ix: min(ix[:1], key=lambda p: p[1]),
    "min(caps[-1:])": lambda ix: min(ix[-1:], key=lambda p: p[1]),
    "min(caps[:2])":  lambda ix: min(ix[:2], key=lambda p: p[1]),
    "min(caps[:3])":  lambda ix: min(ix[:3], key=lambda p: p[1]),
    "min(caps[1:])":  lambda ix: min(ix[1:], key=lambda p: p[1]),
    "min(caps[:-1])": lambda ix: min(ix[:-1], key=lambda p: p[1]),
    "min(caps[::2])": lambda ix: min(ix[::2], key=lambda p: p[1]),
    "mean":           lambda ix: (0, sum(c for _i, c in ix) / len(ix)),
}
_survivors = []
print("    the family table - each candidate against what the SHIPPED "
      "validator answered on the same roster:")
for _name, _fn in _FAMILY.items():
    _misses = []
    for _opt, _n, _pos, _caps, (_ok, _unit, _cap) in _rows:
        _pred = _pick(list(_caps), _fn)
        _pred_refuses = _pred is not None and _pred[1] > 0 and _RATE >= _pred[1]
        if _opt == "maintenance_charge_w":
            _pred_refuses = (_pred is not None and _pred[1] > 0
                             and _RATE > _pred[1])
        _said = ((True, _pred[0], _pred[1]) if _pred_refuses
                 else (False, None, None))
        if _said != (_ok, _unit, _cap):
            _misses.append(f"{_opt[:4]}/n{_n}p{_pos}")
    if not _misses and _name != "min (SHIPPED)":
        _survivors.append(_name)
    print(f"      {_name:<16} {'AGREES EVERYWHERE (survivor)' if not _misses else 'differs on ' + str(len(_misses)) + ' run(s), first: ' + _misses[0]}")
check("⚠️ AND THE FAMILY TABLE HAS NO SURVIVOR BUT THE SHIPPED REDUCTION. "
      "Eleven candidates - max, first, last, the mean, and every positional "
      "slice this card has actually been defeated by - are each computed "
      "against what the real validator answered on the same roster, and every "
      "one of them differs somewhere in the set. This is the check that would "
      "have taken PR #48 red four rounds ago: a member no fixture kills is "
      "PRINTED AS A NAMED SURVIVOR here instead of being absent from a list "
      f"nobody re-derived. survivors: {_survivors}",
      _survivors == [])

# --- the ROSTER OF ONE, where every one of them is the same thing -----------
_at_one = {n: _pick([_SMALL], f) for n, f in _FAMILY.items()}
print(f"    at a ROSTER OF ONE the twelve candidates answer: "
      f"{sorted({str(v) for v in _at_one.values()})}")
check("...and at a ROSTER OF ONE every candidate that still has a member to "
      "reduce answers THE SAME THING - which is exactly why no "
      "single-inverter fixture in this repository could ever have caught any "
      "of this, and why the shipped single-inverter refusals are what they "
      "always were. The only two that do NOT coincide are `caps[1:]` and "
      "`caps[:-1]`, which EMPTY the list and would raise: at N=1 they are "
      "crashes rather than silent wrong answers, so they are not survivors "
      "there either",
      {v for v in _at_one.values() if v is not None} == {("unit-0", _SMALL)}
      and {n for n, v in _at_one.items() if v is None}
      == {"min(caps[1:])", "min(caps[:-1])"}
      and _verdict(_roster_n([_SMALL], **dict(_QUIET, sign_disagree_w=_RATE)),
                   "SAFETY-02 sign-disagreement")[0])

# --- rule 5's SIBLING BRANCH, which was reading the FLAT max_w --------------
# ⚠️ FOUND BY READING THE THREE BRANCHES SIDE BY SIDE, NOT BY THE STRUCTURAL
# PIN - and said that way on purpose, because it is the shape of what the pin
# cannot see. `unit_sign_disagree_w` is SAFETY-30's, not this card's. It was
# graded against `_number(opts, "max_w", 2000.0)`, the FLAT option, three lines
# under a branch that had already stopped doing that. Both directions were
# measured before the fix and both are asserted here.
_leftover = _roster_n([_SMALL, _SMALL],
                      **dict(_QUIET, unit_sign_disagree_w=_RATE))
_leftover["max_w"] = _BIG          # what a single-inverter install leaves behind
_lv = _verdict(_leftover, "SAFETY-30 per-inverter sign-disagreement")
print(f"    leftover flat max_w {_BIG:.0f} beside a roster of two "
      f"{_SMALL:.0f} W units, unit_sign_disagree_w {_RATE:.0f} -> {_lv}")
check("⚠️ A LEFTOVER FLAT `max_w` NO LONGER DISARMS SAFETY-30's PER-UNIT "
      "WATCHDOG SILENTLY. Two 2000 W units beside a `max_w: 5000` left over "
      "from a single-inverter install and `unit_sign_disagree_w: 3000` used to "
      "produce ZERO problems - the site started, the status page said `armed`, "
      "and the watchdog could not trip on EITHER unit. It is refused now, "
      "against the weakest unit's own cap, and the message names that unit",
      _lv == (True, "unit-0", _SMALL))
check("...and the over-refusal in the OTHER direction is gone too: a roster of "
      "two 5000 W units with NO flat `max_w` was refused at 3000 W against a "
      "2000 W DEFAULT the operator never wrote, for a threshold both units can "
      "reach. It starts",
      _verdict(_roster_n([_BIG, _BIG],
                         **dict(_QUIET, unit_sign_disagree_w=_RATE)),
               "SAFETY-30 per-inverter sign-disagreement")
      == (False, None, None))
check("...and the flat single-inverter config it always refused is refused "
      "IDENTICALLY - `unit_caps()` at a roster of one is `[max_w]`, so this is "
      "the same comparison, the same boundary and the same sentence",
      why(validate_startup_config(base(max_w=2000, unit_sign_disagree_w=2000),
                                  -5000, 5000), "unit_sign_disagree_w")
      and validate_startup_config(base(max_w=2000, unit_sign_disagree_w=1999),
                                  -5000, 5000) == [])


# --------------------------------------------------------------------------- #
print()
if fails:
    print(f"{len(fails)} FAILED: " + "; ".join(fails))
    sys.exit(1)
print("all checks passed")