goodwe-addon/goodwe_controller/tools/tel11_pv_trace.py

"""TEL-11/TEL-12's behaviour-invariance instrument. `tools/tel11_pv_trace.py`

⚠️ THE FILE NAME STILL SAYS TEL-11 ON PURPOSE. It is quoted by
`DECISIONS/TEL-11-pv-production.md` §5 and by `test_pv.py`'s mutation table, and
a renamed instrument makes both of those references dangle for no gain. TEL-12
EXTENDED it rather than writing a second one, which is also the honest thing: the
two cards' claims are the same claim at two roster sizes, and one document set
graded by one `--compare` cannot drift apart the way two tools would.

TEL-11 added a PV production channel and claims two things a reviewer cannot
check by reading a diff:

  1. with `pv_entity` UNSET - which is every install today and most installs
     ever - the add-on behaves exactly as `release/1.0` does. Not "the control
     law is untouched": the whole observable surface, commands and telemetry
     alike.
  2. with `pv_entity` SET, the CONTROL decisions are still exactly those. The
     channel is instrumentation and must never become a control input, because
     a loop that reacted to production would destroy the very measurement the
     card exists to make.

TEL-12 makes the channel a ROSTER (`pv_entities`, several inverters, one site
total) and adds two more, in the same shape:

  3. with `pv_entity` set to ONE entity, the add-on behaves exactly as 0.11.0
     does - CONTROL AND TELEMETRY BOTH. That is a STRONGER claim than TEL-11
     could make, because TEL-11's baseline had no channel at all to be identical
     to, and it is the acceptance criterion for not breaking the live install:
     `pv.total()` passes a single entity's answer through verbatim for exactly
     this reason.
  4. with SEVERAL entities configured, and with one of several ABSENT, the
     control decisions are still exactly those. A PARTIAL total is a different
     number from a full one, so "a partial reaches no decision" is a separate
     claim rather than a corollary.

⚠️ AND TEL-12 HAD TO CHANGE `--compare`'s REFUSALS, BECAUSE TEL-11's WERE WRITTEN
AGAINST A BASELINE THAT NO LONGER EXISTS. TEL-11's rule was "exactly ONE side's
`set` document may differ from its own `unset`, and that side is the HEAD" - true
when `release/1.0` predated `pv_entity` entirely. `release/1.0` is now 0.11.0 and
BOTH sides react to `pv_entity`, so that refusal fires on every honest run. The
discriminator is the ROSTER instead: only the HEAD reacts to `pv_entities`, and
that is asserted rather than assumed.

This is what says so. It drives a fixed stimulus set through the SHIPPED
`Controller` and prints two documents per setting -

  `control`    every command written to an actuator, every `self.target`, every
               `inputs_ok`, the loop's own three fields and the arbiter's
               resolution, tick by tick. NO TEL-11/TEL-12 MATERIAL REACHES IT.
  `telemetry`  the published frames, the status rows and the announcement -
               i.e. everything the card is ALLOWED to move.

- and a `tree` fingerprint: sha256 and byte count of every `app/*.py` it
imported, so the comparison can say WHICH TREES it graded.

    python3 tools/tel11_pv_trace.py > /tmp/pv-head.json
    # …in a release/1.0 checkout, with this file copied in…
    python3 tools/tel11_pv_trace.py > /tmp/pv-base.json
    python3 tools/tel11_pv_trace.py --compare /tmp/pv-base.json \\
                                              /tmp/pv-head.json

⚠️ THREE DOCUMENT SETS PER TREE, AND THE THIRD IS WHY THE FIRST TWO STILL HASH
THE SAME AS THEY DID UNDER TEL-11. `unset` and `one` drive exactly the stimulus
cells TEL-11 drove, so their `control` hashes are comparable across the two
cards; the roster cells live in `multi`, a separate document, rather than being
appended to those. Adding cells to an existing document would have moved a hash
whose stability is itself the evidence.

⚠️ THE CONTROL IS PART OF THE TOOL, WHICH IS MUTATION.md HABIT 4 IN AN
INSTRUMENT - AND IT IS A DIFFERENT CONTROL FROM `safety56_behaviour_trace.py`'s,
ON PURPOSE. That card CHANGED sentences, so its `--compare` refuses to report
"behaviour unchanged" unless the sentences differ. THIS card's headline claim is
that with the option unset NOTHING differs at all, so "the documents are
identical" is the result rather than the warning sign - and the thing that has
to be proved instead is that the two runs came from DIFFERENT TREES and that the
instrument can see this card's material when it is switched on. So `--compare`
refuses unless:

  * the two `tree` fingerprints DIFFER (otherwise it is one tree twice, which
    is the trap ARCH-02's rehearsal harness walked into and SAFETY-56 names);
  * exactly ONE side's `set` document differs from its own `unset` document.
    That side is the HEAD; the other is the baseline, which has no such option
    and cannot react to it. Both, or neither, is ambiguous and is refused.

⚠️ WHAT IT CANNOT SEE, with shapes (habit 13):
  * a roster of more than one INVERTER (the battery kind - CFG-04's
    `inverters:`), and any per-unit surface. The PV roster IS driven, at three;
  * the real MQTT transport, Home Assistant, and the recorder. The publication
    is graded as the dict `Controller.publish()` hands the publisher plus the
    announcement `discovery_payloads()` returns - not as bytes on a broker;
  * wall-clock behaviour and the 1 s tick: every clock here is an argument;
  * the ingress HTML, the action log and the firmware;
  * anything a `pv_entity` whose sensor is HEALTHY does not exercise - the
    stimulus set drives a good read, a refused unit and a dead sensor, and
    nothing else;
  * ⚠️ and it is an IDENTITY over a fixed stimulus, so a divergence that needs
    an input this set never produces is invisible to it by construction. The
    stimulus is printed with the documents for exactly that reason.

⚠️ PURGE `__pycache__` BEFORE USE on a tree you have just edited.
`tools/mutation_run.sh` does, and MUTATION.md says why `python3 -B` alone does
not.
"""
import asyncio
import hashlib
import json
import logging
import sys
from pathlib import Path

HERE = Path(__file__).resolve().parent.parent      # the add-on directory
sys.path.insert(0, str(HERE))

Controller = mqtt = None


def _load():
    """The add-on's modules, imported LAZILY so `--compare` runs anywhere.

    ⚠️ NOT A STYLE CHOICE - `safety56_behaviour_trace.py`'s note is this one's
    too. `--compare` is the half a reviewer runs against two JSON files, in a
    directory with no `app/` in it, and a top-level import would make the
    comparison impossible exactly where it is most useful.
    """
    global Controller, mqtt
    from app.main import Controller                     # noqa: E402,F401
    from app import mqtt                                # noqa: E402,F401


A = 5000.0
TUNING = {"max_w": A, "gain": 0.6, "slew_w": 400.0, "target_grid_w": -10.0,
          "deadband_w": 15, "step_w": 10}
ONE = dict(TUNING, meter_source="off", meter_entity="sensor.grid",
           setpoint_entity="number.sp_a", soc_entity="sensor.soc_a",
           batt_entity="sensor.batt_a", heartbeat_s=10, stale_input_s=15,
           inverters=[], bms_soc_floor=10)
PV_ENTITY = "sensor.pv_power"


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

    def set(self, *a):
        pass

    def get_time(self, *a):
        return None

    def set_time(self, *a):
        pass


class Pub:
    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, a following inverter, and a scripted PV sensor."""

    def __init__(self, grid_of, soc, pv_of):
        self.grid_of, self.soc, self.pv_of = grid_of, soc, pv_of
        self.batt = 0.0
        self.n = 0
        self.writes = []

    async def limits(self, entity):
        return (-A, A, 1.0)

    async def number(self, entity, invert=False):
        v = {"sensor.grid": self.grid_of(self.n), "sensor.soc": self.soc,
             "sensor.soc_a": self.soc, "sensor.batt_a": self.batt,
             "sensor.batt": self.batt}.get(entity)
        return None if v is None else (-v if invert else v)

    async def state(self, entity):
        return self.pv_of(self.n)

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

    async def call(self, domain, service, data):
        return True


def _pv_state(watts, unit="W"):
    return {"state": str(watts),
            "attributes": {"unit_of_measurement": unit}}


def _by_entity(mapping):
    """A `hass.state()` that answers each entity its OWN scripted state."""
    async def state(entity):
        return mapping.get(entity)
    return state


# ⚠️ THE STIMULUS SET, PRINTED WITH THE DOCUMENTS. An identity is only as wide
# as what was driven through it, so the worlds and the PV schedules are part of
# the output rather than a fact about the source.
#
# Each PV schedule is exercised against each grid world. A schedule that only
# ever answered healthily could not separate "the loop ignores production" from
# "the loop ignores a production channel that is always saying the same thing".
GRIDS = {
    "EXPORT-6K": lambda t: -6000.0,
    "IMPORT-3K": lambda t: 3000.0,
    "REST": lambda t: -10.0,
    "SWINGING": lambda t: (-4000.0 if (t // 5) % 2 else 4000.0),
    "METER-GAPS": lambda t: (None if t % 7 == 3 else -1500.0),
}
# --- TEL-12: THE ROSTER, AND THE VALUES ARE THE DESIGN ------------------------
# Three entities, distinct non-zero contributions in two different units, so the
# site total is separable from every partial of itself AND from max/min/first/
# last/mean over the three. ⚠️ NOT ONE OF THEM IS ZERO: a member contributing 0 W
# could go absent without moving the sum of the rest, and a stimulus like that
# cannot tell "published the absence" from "published the partial sum" - which is
# MUTATION.md habit 23 with a coinciding value in the fixture rather than in the
# plant.
PV_ENTITIES = ["sensor.pv_east", "sensor.pv_west", "sensor.pv_garage"]
_R = {"sensor.pv_east": (1200.0, "W"), "sensor.pv_west": (2.1, "kW"),
      "sensor.pv_garage": (450.0, "W")}
ROSTER_TOTAL_W = 3750.0          # 1200 + 2100 + 450, in watts


def _roster(dead=(), zero=()):
    """`{entity: state}` for the three - `dead` answers nothing, `zero` says 0."""
    return {e: (None if e in dead
                else _pv_state(0, "W") if e in zero else _pv_state(v, u))
            for e, (v, u) in _R.items()}


# ⚠️ EACH CELL CARRIES BOTH HALVES OF THE CLAIM, AS LITERALS: the number that
# must NEVER appear on the PV row, and the number that MUST. `--compare` asserts
# against these rather than re-deriving anything from the code under test
# (habit 2), and a cell with no dead member has no partial to leak - which is why
# `never` is `None` there rather than the true total. That distinction cost a run:
# labelling WEST-ZERO's true total as its "partial" made 5/5 refuse the shipped
# tree.
#
#            dead members            zero members            never / always (W)
ROSTERS = {
    "ALL-THREE": ((), (), None, ROSTER_TOTAL_W),
    "EAST-DEAD": (("sensor.pv_east",), (), 2550.0, None),
    "WEST-DEAD": (("sensor.pv_west",), (), 1650.0, None),
    "GARAGE-DEAD": (("sensor.pv_garage",), (), 3300.0, None),
    # ⚠️ 0.0 IS THE NUMBER `sum([])` PRODUCES, and it is the one a roster with
    # nothing reporting must never publish: it is the "there was no sun" the
    # channel exists to test.
    "ALL-DEAD": (tuple(_R), (), 0.0, None),
    # ⚠️ AND THE MIRROR OF THIS CARD'S DEFECT: a roster that blanked the site
    # whenever one member honestly read 0 W would report an absence on every
    # clear evening. All three ANSWER here, so 1650 W MUST be published.
    "WEST-ZERO": ((), ("sensor.pv_west",), None, 1650.0),
}

PVS = {
    "none": lambda t: None,
    "zero": lambda t: _pv_state(0),
    "noon": lambda t: _pv_state(8000),
    "ramp": lambda t: _pv_state(t * 250),
    "kilowatts": lambda t: _pv_state(4.25, "kW"),
    "bad-unit": lambda t: _pv_state(1234, "%"),
    "flapping": lambda t: (None if t % 3 else _pv_state(3300)),
    "negative": lambda t: _pv_state(-4200),
}
CYCLES = 24


async def _run(grid_of, pv_of, pv_set, soc=50.0, stop_at=None, roster=None):
    hass = Hass(grid_of, soc, pv_of)
    if roster is not None:
        # ⚠️ PER-ENTITY ANSWERS, which a roster needs and `pv_of` cannot give:
        # a scripted function of the tick alone answers every entity the same
        # thing, and three equal numbers cannot separate a sum from any other
        # reduction over three, nor drive ONE member absent.
        hass.state = _by_entity(roster)
    pub = Pub()
    opts = dict(ONE)
    if roster is not None:
        opts["pv_entities"] = list(PV_ENTITIES)
    elif pv_set:
        opts["pv_entity"] = PV_ENTITY
    c = Controller(opts, hass, MemStore(), pub, stop=asyncio.Event())
    c.auto = True
    c.dev_min, c.dev_max = -A, A
    c.inverters[0].dev_min, c.inverters[0].dev_max = -A, A
    ctrl = []
    for t in range(CYCLES):
        if stop_at is not None and t == stop_at:
            c.auto = False
        await c.read_inputs()
        await c.cycle()
        c.publish()
        # ⚠️ THE CONTROL DOCUMENT, AND NOTHING FROM THIS CARD IS IN IT. Every
        # field here existed on release/1.0 and is a DECISION: what was
        # commanded, what the site is aiming at, whether the inputs were good
        # enough to act on, what the loop did and which claim won.
        ctrl.append([hass.writes[-1] if hass.writes else None,
                     c.target, c.inputs_ok, round(c.i_w, 6),
                     round(c.loop_i_w, 6), round(c.loop_i_lead, 6),
                     c.loop_reason, c.reason, c.following,
                     c.maint.phase, len(c.events)])
        hass.n += 1
    return c, hass, pub, ctrl


def documents(pv_set: bool, roster: bool = False) -> dict:
    """`{"control": …, "telemetry": …}` for one setting of the PV channel.

    ⚠️ `roster=True` DRIVES ITS OWN STIMULUS CELLS AND LANDS IN ITS OWN DOCUMENT
    (TEL-12). It does NOT append cells to the `unset`/`one` documents, because
    those two documents' `control` hashes are the evidence TEL-11 recorded and a
    new cell would move them for a reason that is not behaviour. Same reason
    `_supports_include()` lives in `tree` rather than in a document.
    """
    _load()
    logging.disable(logging.CRITICAL)
    loop = asyncio.new_event_loop()
    control, tele, partials = {}, {}, {}
    _cells = ([(f"{g}/{r}", gof, None, _roster(dead, zero), (never, always))
               for g, gof in GRIDS.items()
               for r, (dead, zero, never, always) in ROSTERS.items()]
              if roster else
              [(f"{g}/{pn}", gof, pof, None, None)
               for g, gof in GRIDS.items() for pn, pof in PVS.items()])
    for key, gof, pof, rmap, partial in _cells:
        c, hass, pub, ctrl = loop.run_until_complete(
            _run(gof, pof, pv_set, roster=rmap))
        control[key] = {
            "ticks": ctrl,
            "writes": hass.writes,
            "sha": hashlib.sha256(
                json.dumps(ctrl, sort_keys=True).encode()).hexdigest(),
        }
        rows = c.checks()
        tele[key] = {
            "frames": pub.frames,
            "rows": [[r["ok"], r["warn"], r["text"]] for r in rows],
            "n_rows": len(rows),
            "page": c.status_payload()["rows"],
            "level": c.status_payload()["level"],
            "banner": c.status_payload()["banner"],
        }
        if partial is not None:
            # ⚠️ TEL-12. THE NUMBER THAT MUST NEVER APPEAR ON THE PV ROW,
            # carried in the output so `--compare` asserts against a literal
            # rather than re-deriving it from the code under test (habit 2).
            # `wire` is what `str()` of it looks like on the topic, because that
            # is the comparison that matters.
            #
            # ⚠️ IN ITS OWN TOP-LEVEL MAP AND *NOT* INSIDE THE TELEMETRY CELL.
            # It is a fact about the STIMULUS, produced by this tool and
            # identical on both trees, so putting it in the telemetry document
            # made every roster cell differ from its unset reference for a
            # reason that is not behaviour - and the HEAD/baseline discriminator
            # then fired on both sides. Measured, on the first run.
            never, always = partial
            partials[key] = {
                "never": None if never is None else str(round(never, 1)),
                "always": None if always is None else str(round(always, 1))}
    # A stopped run, because the "stopped or blind" branch writes its own rows.
    c, hass, pub, ctrl = loop.run_until_complete(
        _run(GRIDS["EXPORT-6K"], PVS["noon"], pv_set, stop_at=8,
             roster=dict(_roster()) if roster else None))
    control["STOPPED"] = {"ticks": ctrl, "writes": hass.writes,
                          "sha": hashlib.sha256(
                              json.dumps(ctrl, sort_keys=True).encode()
                          ).hexdigest()}
    tele["STOPPED"] = {"frames": pub.frames,
                       "rows": [[r["ok"], r["warn"], r["text"]]
                                for r in c.checks()]}
    loop.close()
    # The announcement is telemetry too, and it is the surface an install's
    # entity registry is keyed on.
    tele["ANNOUNCEMENT"] = _announcement(pv_set, roster)
    out = {"control": control, "telemetry": tele}
    if partials:
        out["partials"] = partials
    return out


def _supports_include() -> bool:
    """Does THIS tree's `discovery_payloads()` take TEL-11's `include`?

    ⚠️ A PROPERTY OF THE TREE, AND IT IS RECORDED IN `tree` RATHER THAN IN A
    DOCUMENT. `release/1.0` has neither the argument nor `optional_keys()`, so
    the tool has to run on a tree that predates both or it cannot produce a
    baseline at all. Putting the answer in the documents instead would make
    `unset` differ between the two trees for a reason that is not behaviour -
    which is the "check that goes red for the wrong reason" MUTATION.md keeps
    finding - so it goes beside the file hashes, where a reader comparing two
    trees is already looking.
    """
    import inspect
    return ("include"
            in inspect.signature(mqtt.discovery_payloads).parameters
            and hasattr(mqtt, "optional_keys"))


def _supports_roster() -> bool:
    """Does THIS tree know `pv_entities` at all? TEL-12.

    ⚠️ IN `tree` FOR `_supports_include()`'s OWN REASON. A tree that predates
    TEL-12 answers False; a tree that predates TEL-11 has no `app/pv.py` either,
    so the import is guarded rather than assumed.
    """
    try:
        from app import pv as _p
    except ImportError:
        return False
    return callable(getattr(_p, "entities", None))


def _announcement(pv_set: bool, roster: bool = False) -> list:
    """The retained discovery set, in announcement order.

    ⚠️ AT `pv_set` FALSE THIS IS THE BARE CALL ON BOTH TREES, which is the
    point: an install that never set the option must receive the SAME
    announcement from both, and that identity is graded rather than argued.

    ⚠️ AND AT `roster` TRUE THE OPTS CARRY `pv_entities` AND NOTHING ELSE. On a
    tree that predates TEL-12 `optional_keys()` reads only `pv_entity`, so it
    answers `()` and no PV row is announced - which is exactly how a baseline is
    supposed to fail to react to a key it has never heard of, and is what
    `--compare` 4/5 asserts rather than assumes.
    """
    if (pv_set or roster) and _supports_include():
        payloads = mqtt.discovery_payloads(
            include=mqtt.optional_keys({"pv_entities": list(PV_ENTITIES)}
                                       if roster
                                       else {"pv_entity": PV_ENTITY}))
    else:
        payloads = mqtt.discovery_payloads()
    return [[t, json.dumps(cfg, sort_keys=True)] for t, cfg in payloads]


def tree() -> dict:
    """sha256 and byte count of every `app/*.py` this run imported from."""
    _load()
    out = {"supports_include": _supports_include(),
           # TEL-12. Does THIS tree know the roster at all? A property of the
           # tree, recorded beside the file hashes for `_supports_include()`'s
           # own reason: putting it in a document would make `multi` differ
           # between trees for something that is not behaviour.
           "supports_roster": _supports_roster()}
    for f in sorted((HERE / "app").glob("*.py")):
        raw = f.read_bytes()
        out[f"app/{f.name}"] = [hashlib.sha256(raw).hexdigest(), len(raw)]
    return out


def _h(obj) -> str:
    return hashlib.sha256(json.dumps(obj, sort_keys=True).encode()).hexdigest()


# ⚠️ WHY `multi` IS GRADED PER GRID WORLD AND `one` IS NOT. `one` drives the same
# stimulus cells `unset` does - `GRID/pv-schedule` - so the two documents have the
# same keys and compare directly. `multi`'s cells are `GRID/roster-state`, a
# different naming over a different stimulus, so a whole-document comparison
# would report a difference that is only the key names. The comparable thing is
# the GRID: for one grid world, `unset` produces the SAME control ticks for every
# PV schedule (nothing is read), so that grid has ONE control document and each
# roster cell in the same grid must equal it. That per-grid collapse is asserted
# as a precondition before it is used - without it "equal to the unset document
# for this grid" names nothing.
def _grid(key: str) -> str:
    return key.split("/")[0]


def _collapse(doc_part: dict):
    """`{grid: value}` if every cell in a grid agrees, else `None`.

    ⚠️ RETURNS `None` RATHER THAN A BEST GUESS. A caller that got a dict has a
    well-defined per-grid reference; a caller that got `None` has to say so and
    refuse, which is what `compare()` does.
    """
    out = {}
    for key, value in doc_part.items():
        if "/" not in key:
            continue
        if _grid(key) in out and out[_grid(key)] != value:
            return None
        out[_grid(key)] = value
    return out


def _reduces_to_unset(doc: dict, what: str) -> bool:
    """Is this side's `multi` document its own `unset` one, grid for grid?

    True of a tree that has never heard of `pv_entities`: it reads nothing,
    announces nothing and shows no row, so the roster cells are the unset cells.
    False of a tree that implements TEL-12. This is the HEAD/baseline
    discriminator, and it is MEASURED rather than taken from
    `tree.supports_roster`, which is the tree describing itself.
    """
    ref = _collapse(doc["unset"][what])
    if ref is None:
        return False
    return all(value == ref.get(_grid(key))
               for key, value in doc["multi"][what].items() if "/" in key)


def compare(a_path, b_path) -> int:
    a = json.loads(Path(a_path).read_text(encoding="utf-8"))
    b = json.loads(Path(b_path).read_text(encoding="utf-8"))

    # --- the control, first: two DIFFERENT trees, and one of them reacts -----
    if a["tree"] == b["tree"]:
        print("REFUSED: the two runs have the SAME `app/` fingerprint, so they "
              "are one tree twice. An identity is evidence only when the two "
              "sides could have differed.")
        return 2
    # ⚠️ TEL-12 HAD TO CHANGE THIS DISCRIMINATOR AND HERE IS WHY. TEL-11's rule
    # was "exactly ONE side's `set` document differs from its own `unset`" -
    # true while `release/1.0` predated `pv_entity` entirely. `release/1.0` is
    # 0.11.0 now, BOTH sides react to `pv_entity`, and that refusal would fire
    # on every honest run. The ROSTER is the key only the HEAD has heard of.
    for side, doc in (("A", a), ("B", b)):
        for which in ("unset", "one", "multi"):
            if which not in doc:
                print(f"REFUSED: side {side} has no {which!r} document. Both "
                      f"sides must be produced by THIS version of the tool - "
                      f"copy it into the baseline checkout rather than using "
                      f"the one that is committed there.")
                return 2
    for side, doc in (("A", a), ("B", b)):
        for what in ("control", "telemetry"):
            if _collapse(doc["unset"][what]) is None:
                print(f"REFUSED: on side {side} the `unset` {what} document "
                      f"does NOT collapse per grid world - two PV schedules in "
                      f"one grid gave different answers with the channel off. "
                      f"Nothing is read in that state, so this is the "
                      f"instrument being wrong rather than the tree.")
                return 2
    moved = [side for side, doc in (("A", a), ("B", b))
             if not _reduces_to_unset(doc, "telemetry")]
    if len(moved) != 1:
        print(f"REFUSED: `multi` fails to reduce to `unset` on "
              f"{moved or 'NEITHER'} side. Exactly one side may react to "
              f"`pv_entities` - the HEAD. Neither means the instrument cannot "
              f"see TEL-12's material at all; both means the baseline is not a "
              f"baseline.")
        return 2
    head, base = (b, a) if moved == ["B"] else (a, b)
    hlabel = "B" if moved == ["B"] else "A"
    blabel = "A" if hlabel == "B" else "B"
    print(f"  HEAD is {hlabel}; baseline is {blabel} - its roster cells ARE its "
          f"unset cells, which is what a tree that has never heard of "
          f"`pv_entities` does with them.")
    print(f"  supports_roster   {blabel} "
          f"{base['tree'].get('supports_roster')} / {hlabel} "
          f"{head['tree'].get('supports_roster')}   (self-reported; the verdict "
          f"above is MEASURED)")
    print(f"  tree      {blabel} {_h(base['tree'])}")
    print(f"  tree      {hlabel} {_h(head['tree'])}")
    for what in ("control", "telemetry"):
        for side, doc, which in ((blabel, base, "unset"),
                                 (hlabel, head, "unset"),
                                 (blabel, base, "one"), (hlabel, head, "one"),
                                 (blabel, base, "multi"),
                                 (hlabel, head, "multi")):
            print(f"  {what:<9} {side} {which:<5} {_h(doc[which][what])}")

    rc = 0
    # 1. TEL-11's acceptance criterion: unset is the baseline, in full.
    if head["unset"] == base["unset"]:
        print("OK  1/5  UNSET: control AND telemetry are byte-identical to the "
              "baseline tree. An install that configures no PV entity sees no "
              "change of any kind.")
    else:
        rc = 1
        for what in ("control", "telemetry"):
            for k in sorted(set(head["unset"][what]) | set(base["unset"][what])):
                if head["unset"][what].get(k) != base["unset"][what].get(k):
                    print(f"FAIL 1/5  UNSET {what} MOVED at {k!r}")
    # 2. ⚠️ TEL-12's OWN ACCEPTANCE CRITERION, AND IT IS STRONGER THAN ANYTHING
    #    TEL-11 COULD ASSERT. The baseline is 0.11.0, which HAS the channel, so
    #    "one entity behaves exactly as 0.11.0 does" is a real cross-tree
    #    identity over BOTH documents rather than an argument about f-strings.
    #    The install that reported this defect has `pv_entity` set; this is the
    #    line that says the upgrade does not move it.
    if head["one"] == base["one"]:
        print("OK  2/5  ONE ENTITY: control AND telemetry are byte-identical to "
              "0.11.0's single-entity path - the same commands, the same "
              "payloads, the same status rows, the same announcement.")
    else:
        rc = 1
        for what in ("control", "telemetry"):
            for k in sorted(set(head["one"][what]) | set(base["one"][what])):
                if head["one"][what].get(k) != base["one"][what].get(k):
                    print(f"FAIL 2/5  ONE-ENTITY {what} MOVED at {k!r}")
                    print(f"          base {base['one'][what].get(k)!r}")
                    print(f"          head {head['one'][what].get(k)!r}")
    # 3. the card's other half at ONE entity - TEL-11's 2/3, kept verbatim in
    #    substance because it is the claim that must not decay.
    if head["one"]["control"] == head["unset"]["control"]:
        print("OK  3/5  CONTROL, ONE ENTITY: the control document is "
              "byte-identical with the channel switched on - PV reaches no "
              "decision.")
    else:
        rc = 1
        for k in sorted(head["one"]["control"]):
            if head["one"]["control"][k] != head["unset"]["control"].get(k):
                print(f"FAIL 3/5  CONTROL MOVED at {k!r} with one entity set")
    # 4. ⚠️ AND AT A ROSTER OF THREE, AND WITH A HOLE IN IT. A PARTIAL total is
    #    a DIFFERENT number from a full one, so "a partial reaches no decision"
    #    is a separate claim and not a corollary of 3/5.
    _cref = _collapse(head["unset"]["control"])
    _bad = [k for k, v in head["multi"]["control"].items()
            if "/" in k and v != _cref.get(_grid(k))]
    # ⚠️ AND THE SAME DOCUMENT ACROSS THE TWO TREES, which is the strongest form
    # of the claim available: the baseline runs the roster cells with a key it
    # has never heard of, so its control document for them is 0.11.0's. If
    # HEAD's matches it byte for byte, the control path under a roster IS
    # 0.11.0's control path.
    _xtree = head["multi"]["control"] == base["multi"]["control"]
    if not _xtree:
        rc = 1
        print("FAIL 4/5  the roster control document is NOT the baseline "
              "tree's. A roster changed a decision.")
    if not _bad and _xtree and (head["multi"]["control"].get("STOPPED")
                                == head["unset"]["control"].get("STOPPED")):
        print(f"OK  4/5  CONTROL, THREE ENTITIES: every roster cell's control "
              f"document equals its grid world's `unset` one - "
              f"{len(head['multi']['control']) - 1} cells over "
              f"{len(_cref)} grid worlds, covering three healthy entities, one "
              f"dead at EVERY position, all three dead and one honestly "
              f"reporting 0 W. Neither the total nor any PARTIAL total reaches "
              f"a decision, and the document is byte-identical to the BASELINE "
              f"tree's for the same cells.")
    else:
        rc = 1
        for k in _bad or ["STOPPED"]:
            print(f"FAIL 4/5  CONTROL MOVED at {k!r} under a roster")
    # 5. …and the instrument could have seen it, twice over: the telemetry
    #    document DOES move, and the partial sum a defect would have published
    #    is on no PV row anywhere in it.
    n = sum(1 for k, v in head["multi"]["telemetry"].items()
            if "/" in k
            and v != _collapse(head["unset"]["telemetry"]).get(_grid(k)))
    leaks, missing = [], []
    for key, spec in head["multi"].get("partials", {}).items():
        rows = [str(f["pv"]) for f in head["multi"]["telemetry"][key]["frames"]
                if "pv" in f]
        if spec["never"] is not None and spec["never"] in rows:
            leaks.append((key, spec["never"]))
        # ⚠️ AND THE OTHER DIRECTION, because "never publishes 1650" is also
        # satisfied by a channel that publishes nothing ever. A cell whose
        # members all ANSWER must publish its total, including the one where a
        # member honestly reports 0 W.
        if spec["always"] is not None and set(rows) != {spec["always"]}:
            missing.append((key, spec["always"], sorted(set(rows))[:4]))
    if not head["multi"].get("partials"):
        print("REFUSED 5/5: the HEAD run carries no `partials` map, so the "
              "partial-sum assertion quantified over nothing.")
        return 2
    if n and not leaks and not missing:
        print(f"OK  5/5  CONTROL OF THE CONTROL: the telemetry document DOES "
              f"move under a roster ({n} stimulus cells differ from their grid "
              f"world's unset telemetry), so 1/5-4/5 are identities this "
              f"instrument could have broken. AND in no cell with a dead member "
              f"does the PV row carry the partial sum of the survivors, while "
              f"every cell whose members all ANSWER publishes its total - "
              f"including the one where a member honestly reports 0 W.")
    else:
        rc = 2
        if leaks:
            print(f"REFUSED 5/5: THE PARTIAL SUM IS ON THE WIRE at "
                  f"{leaks[:6]}{' …' if len(leaks) > 6 else ''} - a smaller "
                  f"number indistinguishable from a real one, which is the "
                  f"defect this card exists to prevent.")
        if missing:
            print(f"REFUSED 5/5: a cell whose members ALL ANSWERED did not "
                  f"publish its total: {missing[:6]}. An absence published for "
                  f"a healthy roster defeats the channel from the other end - "
                  f"'no PV reported' on every clear evening.")
        if not leaks and not missing:
            print("REFUSED 5/5: the telemetry document did NOT move under a "
                  "roster. This instrument cannot see TEL-12's material, so "
                  "nothing above it means anything.")
    return rc


if __name__ == "__main__":
    if len(sys.argv) == 4 and sys.argv[1] == "--compare":
        sys.exit(compare(sys.argv[2], sys.argv[3]))
    print(json.dumps({
        "tree": tree(),
        "stimulus": {"grids": sorted(GRIDS), "pv": sorted(PVS),
                     "rosters": sorted(ROSTERS), "entities": PV_ENTITIES,
                     "roster_total_w": ROSTER_TOTAL_W, "cycles": CYCLES},
        "unset": documents(False),
        # ⚠️ `one` WAS CALLED `set` UNDER TEL-11 AND IS THE SAME STIMULUS SET,
        # renamed because there are now two ways to switch the channel on and
        # "set" no longer says which. Its `control` hashes are comparable to the
        # ones TEL-11 recorded; nothing about the cells moved.
        "one": documents(True),
        "multi": documents(False, roster=True),
    }, sort_keys=True, indent=1))