goodwe-addon/goodwe_controller/tools/safety56_behaviour_trace.py

"""SAFETY-56's behaviour-invariance instrument. `tools/safety56_behaviour_trace.py`

SAFETY-56 rewrote three operator-facing sentences and nothing else. This is what
says so: it drives a fixed stimulus set through the SHIPPED modules and prints
two documents -

  `behaviour`  every verdict, every `reading()`, every `releases`, every
               `inputs_ok` and every readiness row's (ok, warn). NO SENTENCE
               REACHES IT, so a tree whose only change is wording must produce
               this document byte for byte.
  `sentences`  the four operator-facing surfaces, from the same stimulus.

Run it in two trees and compare:

    python3 tools/safety56_behaviour_trace.py > /tmp/after.json
    python3 tools/safety56_behaviour_trace.py --compare /tmp/before.json /tmp/after.json

⚠️ THE CONTROL IS PART OF THE TOOL AND NOT A SEPARATE STEP, WHICH IS MUTATION.md
HABIT 4 IN AN INSTRUMENT. `--compare` REFUSES to report "behaviour unchanged"
unless the `sentences` halves DIFFER. Without that, comparing a tree with itself
- which this card's first run did, by archiving a branch whose edits were not
committed yet - gives a perfect result that means nothing. An identity is
evidence only when the two sides could have differed, and ARCH-02's rehearsal
harness is the worked example of what happens when nobody checks.

⚠️ IT GRADES BEHAVIOUR, NEVER WORDING. Two trees that agree here can still print
different sentences - that is the point. `test_freshness.py` part 11 grades the
sentences themselves.

⚠️ WHAT IT CANNOT SEE, with shapes (habit 13): anything outside `freshness.py`'s
verdict, `_read_unit()`'s gating and `checks()`'s rows; a roster of more than
one; the MQTT and ingress surfaces; and wall-clock behaviour, since every clock
here is the caller's monotonic argument. It is a pin on the three things this
card is forbidden to move, not a full behavioural equivalence.

⚠️ PURGE 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 importlib.util
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))

BOUNDS = Freshness = Controller = DEFAULTS = None


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

    ⚠️ NOT A STYLE CHOICE. `--compare` is the half a reviewer runs on a laptop,
    against two JSON files, in a directory that has no `app/` in it - and a
    top-level import would make the comparison impossible exactly where it is
    most useful.
    """
    global BOUNDS, Freshness, Controller, DEFAULTS
    from app.freshness import BOUNDS, Freshness     # noqa: E402,F401
    from app.main import Controller                 # noqa: E402,F401
    from app.store import DEFAULTS                  # noqa: E402,F401

A = 5000.0
ONE = dict(meter_source="off", meter_entity="sensor.grid",
           setpoint_entity="number.sp_a", soc_entity="sensor.soc_a",
           batt_entity="sensor.batt_a", max_w=A, gain=0.6, slew_w=0,
           deadband_w=15, target_grid_w=0, step_w=10, heartbeat_s=10,
           stale_input_s=15, inverters=[], bms_soc_floor=10)


class MemStore:
    def __init__(self, d=None):
        self.data = dict(DEFAULTS, **(d or {}))

    def set(self, k, v):
        self.data[k] = v

    def get_time(self, *a):
        return None

    def set_time(self, *a):
        pass


class Pub:
    enabled = False

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

    def close(self):
        pass


class Hass:
    def __init__(self, numbers=None, limits=None):
        self.numbers = dict(numbers or {})
        self.limits_by = dict(limits or {})
        self.served = []
        self.writes = []

    async def number(self, entity, invert=False):
        v = self.numbers.get(entity)
        self.served.append((entity, v))
        return None if v is None else (-v if invert else v)

    async def limits(self, entity):
        return self.limits_by.get(entity, (None, None, None))

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

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


def one_site(soc=50.0):
    hass = Hass({"sensor.grid": 0.0, "sensor.soc_a": soc,
                 "sensor.batt_a": -1200.0},
                {"number.sp_a": (-A, A, 1.0)})
    c = Controller(dict(ONE), hass, MemStore(), Pub(), stop=asyncio.Event())
    c.inverters[0].dev_min, c.inverters[0].dev_max = -A, A
    return c, hass


async def _world(name, soc, reg_of, cmd_of, n):
    c, hass = one_site(soc)
    seen = []
    for t in range(n):
        hass.numbers["sensor.batt_a"] = reg_of(t)
        c.inverters[0].last_write = cmd_of(t)
        await c.read_inputs(now_mono=float(t))
        seen.append([c.inverters[0].batt, c.inputs_ok,
                     c.inverters[0].batt_fresh.stale])
    return {"name": name, "reads": seen,
            "rows": [[r["ok"], r["warn"]] for r in c.checks()],
            "n_rows": len(c.checks()), "events": len(c.events),
            "inputs_bad": c.inputs_bad_since is not None,
            "releases": [i.batt_fresh.releases for i in c.inverters]}


# The command and register schedules. Each of the conjunction's three terms is
# exercised alone and with the others, and the last two schedules take a verdict
# all the way through SAFETY-57's release and back into a re-trip.
CMDS = [lambda t: 0.0, lambda t: 1300.0, lambda t: -1200.0,
        lambda t: 1300.0 if t >= 1 else -1200.0,
        lambda t: 249.0 if t >= 1 else 0.0,
        lambda t: 251.0 if t >= 1 else 0.0,
        lambda t: 1300.0 if t < 40 else 0.0,
        lambda t: 5.0 if t >= 1 else 1300.0,
        lambda t: (t % 7) * 200.0 - 600.0,
        lambda t: (-1200.0 if t < 1 else 1300.0) if t < 45 else (
            0.0 if t < 90 else 1300.0),
        lambda t: (-1200.0 if t < 1 else 1300.0) if t < 45 else 100.0]
REGS = [lambda t: -3.0, lambda t: -1200.0, lambda t: -1200.0 + (t % 3),
        lambda t: None if t % 5 == 0 else -1200.0,
        lambda t: 0.0, lambda t: float((t * 37) % 11) - 5.0]

WORLDS = [
    ("DEAD", 50.0, lambda t: -3.0, lambda t: 800.0 if t < 1 else 1600.0, 40),
    ("REFUSED", 11.0, lambda t: -3.0, lambda t: 800.0 if t < 1 else 1600.0, 40),
    ("ABSENT", 50.0, lambda t: None, lambda t: 0.0, 3),
    ("HEALTHY", 50.0, lambda t: -1200.0 + (t % 3),
     lambda t: 1300.0 if t % 2 else -1200.0, 60),
    ("RELEASE", 50.0, lambda t: -1200.0,
     lambda t: (-1200.0 if t < 1 else 1300.0) if t < 40 else 0.0, 60),
    ("RETRIP", 50.0, lambda t: -1200.0,
     lambda t: (-1200.0 if t < 1 else 1300.0) if t < 40 else
     (0.0 if t < 80 else 1300.0), 130),
]


def behaviour():
    """Everything this card is forbidden to move. No sentence may enter it."""
    _load()
    logging.disable(logging.CRITICAL)
    out = {}

    # 1. the committed live capture, read by read.
    spec = importlib.util.spec_from_file_location(
        "_s56decode", HERE / "tools" / "decode_bus_capture.py")
    dec = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(dec)
    cap_file = (HERE / "scenarios" / "safety57-2026-09-21"
                / "bus-a-2026-09-21-1030-1110.log.gz")
    events, _census = dec.decode(cap_file)
    sp, bt = dec.series(events, "setpoint_w"), dec.series(events, "batt_w")
    f = Freshness(BOUNDS["batt"])
    t = float(int(max(sp[0][0], bt[0][0])) + 1)
    t1 = int(min(sp[-1][0], bt[-1][0]))
    cap = []
    while t <= t1:
        f.submit(dec.held_at(bt, t), dec.held_at(sp, t), t)
        cap.append([t, f.stale, f.reading(), f.releases, f.held,
                    f.rested_since, f.graded_from, f.standing_at_grade,
                    f.released_at])
        t += 1.0
    out["capture_reads"] = len(cap)
    out["capture_trace"] = hashlib.sha256(
        json.dumps(cap, sort_keys=True).encode()).hexdigest()
    out["capture_edges"] = [r[:4] for i, r in enumerate(cap)
                            if i == 0 or r[1] != cap[i - 1][1]]

    # 2. the swept grid over the policy object.
    sweep = []
    for ci, cf in enumerate(CMDS):
        for ri, rf in enumerate(REGS):
            g = Freshness(BOUNDS["batt"])
            row = []
            for n in range(120):
                g.submit(rf(n), cf(n), float(n))
                row.append([g.stale, g.reading(), g.releases])
            sweep.append([ci, ri, row])
    out["sweep_cells"] = len(sweep)
    out["sweep_trace"] = hashlib.sha256(
        json.dumps(sweep, sort_keys=True).encode()).hexdigest()
    out["sweep_stale_reads"] = sum(1 for _c, _r, row in sweep
                                   for s, _v, _n in row if s)
    out["sweep_missing_reads"] = sum(1 for _c, _r, row in sweep
                                     for _s, v, _n in row if v is None)
    out["sweep_releases"] = sum(row[-1][2] for _c, _r, row in sweep)

    # 3. the shipped read path and every readiness row's verdict.
    loop = asyncio.new_event_loop()
    out["worlds"] = [loop.run_until_complete(_world(*w)) for w in WORLDS]
    loop.close()
    return out


def sentences():
    """The four operator-facing surfaces, from the REFUSED world."""
    _load()
    recs = []

    class _H(logging.Handler):
        def emit(self, record):
            recs.append(record.getMessage())

    logging.disable(logging.NOTSET)
    log = logging.getLogger("goodwe")
    log.addHandler(_H())
    log.setLevel(logging.DEBUG)

    async def go():
        c, hass = one_site(11.0)
        for t in range(40):
            hass.numbers["sensor.batt_a"] = -3.0
            c.inverters[0].last_write = 800.0 if t < 1 else 1600.0
            await c.read_inputs(now_mono=float(t))
        return c

    loop = asyncio.new_event_loop()
    c = loop.run_until_complete(go())
    loop.close()
    return {
        "reason": c.inverters[0].batt_fresh.reason,
        "log": [m for m in recs if "SAFETY-37" in m],
        "stale_note": c.inverters[0].stale_note(),
        "not_ready_row": [r["text"] for r in c.checks()
                          if r["text"].startswith("battery power")],
    }


def compare(a_path, b_path):
    a = json.loads(Path(a_path).read_text(encoding="utf-8"))
    b = json.loads(Path(b_path).read_text(encoding="utf-8"))
    beh_same = a["behaviour"] == b["behaviour"]
    sent_same = a["sentences"] == b["sentences"]
    ha = hashlib.sha256(json.dumps(a["behaviour"], sort_keys=True).encode())
    hb = hashlib.sha256(json.dumps(b["behaviour"], sort_keys=True).encode())
    print(f"behaviour A {ha.hexdigest()}")
    print(f"behaviour B {hb.hexdigest()}")
    if sent_same:
        # ⚠️ THE REFUSAL, AND IT IS THE POINT OF THE TOOL. Two runs whose
        # sentences agree are either the same tree twice or a card that changed
        # nothing an operator reads. Either way "behaviour unchanged" is not a
        # result, and reporting it would be an identity measured against itself.
        print("REFUSED: the `sentences` halves are IDENTICAL, so these two runs "
              "cannot be evidence that behaviour survived a WORDING change - "
              "they may be the same tree twice. Compare a tree against one "
              "whose sentences differ.")
        return 2
    if not beh_same:
        for k in sorted(set(a["behaviour"]) | set(b["behaviour"])):
            if a["behaviour"].get(k) != b["behaviour"].get(k):
                print(f"BEHAVIOUR MOVED at {k!r}")
        return 1
    print("OK: sentences differ, behaviour is byte-identical.")
    return 0


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({"behaviour": behaviour(), "sentences": sentences()},
                     sort_keys=True, indent=1))