goodwe-addon/goodwe_controller/tools/tel11_pv_trace.py

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

TEL-11 adds 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.

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 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/tel11-head.json
    # …in a release/1.0 checkout, with this file copied in…
    python3 tools/tel11_pv_trace.py > /tmp/tel11-base.json
    python3 tools/tel11_pv_trace.py --compare /tmp/tel11-base.json \\
                                              /tmp/tel11-head.json

⚠️ 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, and any per-unit surface;
  * 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}}


# ⚠️ 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),
}
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):
    hass = Hass(grid_of, soc, pv_of)
    pub = Pub()
    opts = dict(ONE)
    if 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) -> dict:
    """`{"control": …, "telemetry": …}` for one setting of `pv_entity`."""
    _load()
    logging.disable(logging.CRITICAL)
    loop = asyncio.new_event_loop()
    control, tele = {}, {}
    for gname, gof in GRIDS.items():
        for pname, pof in PVS.items():
            key = f"{gname}/{pname}"
            c, hass, pub, ctrl = loop.run_until_complete(
                _run(gof, pof, pv_set))
            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"],
            }
    # 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))
    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)
    return {"control": control, "telemetry": tele}


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 _announcement(pv_set: bool) -> 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.
    """
    if pv_set and _supports_include():
        payloads = mqtt.discovery_payloads(
            include=mqtt.optional_keys({"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()}
    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()


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
    moved = [side for side, doc in (("A", a), ("B", b))
             if doc["set"] != doc["unset"]]
    if len(moved) != 1:
        print(f"REFUSED: `set` differs from `unset` on {moved or 'NEITHER'} "
              "side. Exactly one side must react to `pv_entity` - the HEAD. "
              "Neither means the instrument cannot see this card's material "
              "at all, and both means the baseline is not a 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} (it does not react to "
          f"`pv_entity`, which is what a tree without the option does).")
    print(f"  tree      {blabel} {_h(base['tree'])}")
    print(f"  tree      {hlabel} {_h(head['tree'])}")
    for what in ("control", "telemetry"):
        print(f"  {what:<9} {blabel} unset {_h(base['unset'][what])}")
        print(f"  {what:<9} {hlabel} unset {_h(head['unset'][what])}")
        print(f"  {what:<9} {hlabel} set   {_h(head['set'][what])}")

    rc = 0
    # 1. the acceptance criterion: unset is the baseline, in full.
    if head["unset"] == base["unset"]:
        print("OK  1/3  UNSET: control AND telemetry are byte-identical to the "
              "baseline tree. An install that does not set `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/3  UNSET {what} MOVED at {k!r}")
    # 2. the card's other half: setting it moves no DECISION.
    if head["set"]["control"] == head["unset"]["control"]:
        print("OK  2/3  SET: the control document is byte-identical with the "
              "channel switched on - PV reaches no decision.")
    else:
        rc = 1
        for k in sorted(head["set"]["control"]):
            if head["set"]["control"][k] != head["unset"]["control"][k]:
                print(f"FAIL 2/3  CONTROL MOVED at {k!r} when `pv_entity` was "
                      f"set")
    # 3. …and the instrument could have seen it. Without this the first two
    #    are an identity measured against a document that reaches nothing.
    if head["set"]["telemetry"] != head["unset"]["telemetry"]:
        n = sum(1 for k in head["set"]["telemetry"]
                if head["set"]["telemetry"][k]
                != head["unset"]["telemetry"].get(k))
        print(f"OK  3/3  CONTROL OF THE CONTROL: the telemetry document DOES "
              f"move when the option is set ({n} stimulus cells differ), so "
              f"1/3 and 2/3 are identities the instrument could have broken.")
    else:
        rc = 2
        print("REFUSED 3/3: the telemetry document did NOT move when "
              "`pv_entity` was set. This instrument cannot see the card'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),
                     "cycles": CYCLES},
        "unset": documents(False),
        "set": documents(True),
    }, sort_keys=True, indent=1))