goodwe-addon/goodwe_controller/tools/decode_bus_capture.py

#!/usr/bin/env python3
"""SAFETY-57 - turn a passive RS-485 capture into the three series the loop sees.

`python3 tools/decode_bus_capture.py scenarios/safety57-2026-09-21/bus-a-*.log.gz`

WHY THE RAW CAPTURE IS WHAT IS COMMITTED, AND NOT A DECODED TABLE. The other
scenario directories in this repository hold Home Assistant recorder exports -
a number per row, already interpreted. This one holds MODBUS FRAMES off the
wire, because the episode it records is one in which the add-on's own view and
the wire's disagree, and a decoded table would be this decoder's opinion with
the evidence thrown away. MUTATION.md habit 8: a committed transcript is a
check, and a check has to be re-derivable from something.

WHAT THE CAPTURE IS. `sniffer@192.168.2.167` is a passive listener on the
RS-485 pair between the ESP32 master and the GW5000S-BP. It writes nothing; it
has no address on the bus. Every line is one frame it heard:

    a|10:51:29.695 ok  F7 10 05 6E 00 02 04 00 03 04 24 ...
    ^ ^             ^  ^
    | |             |  the frame, hex
    | |             CRC verdict as the capture tool computed it
    | local wall clock, Europe/Brussels, no date - see `decode()`
    bus label

THE FOUR FRAME SHAPES THIS FILE UNDERSTANDS, and it deliberately understands no
others - an unrecognised frame is COUNTED and dropped, never guessed at:

    F7 03 <addr:2> <count:2> <crc:2>      a read REQUEST
    F7 03 <2*count> <data...> <crc:2>     its response, matched to the request
                                          immediately preceding it
    F7 10 05 6E 00 02 04 <data:4> <crc:2> a write of the two EMS registers
    F7 10 05 6E 00 02 <crc:2>             the inverter's echo of that write

⚠️ THE SETPOINT IS A MODE AND A MAGNITUDE, NOT A SIGNED NUMBER, and getting
that backwards is the one way this decoder could quietly invert the whole
episode. `0x056E` is the EMS mode and `0x056F` the magnitude in watts,
unsigned. Mode 2 is charge and mode 3 is discharge, and that is READ OFF THIS
CAPTURE rather than off a datasheet: in the committed window, every
`mode 2, 2170 W` write is followed within one 10 s publish by an 0x0518
reading near **-2287 W**, and `mode 3, 1060 W` by a POSITIVE-side idle. The
add-on's own sign convention (`Inverter.standing_w()`, + = discharge) is
therefore `+magnitude` for mode 3 and `-magnitude` for mode 2, and
`test_freshness.py` part 10a grades that correspondence on the capture itself
rather than trusting this paragraph.

WHAT THIS DECODER CANNOT SEE, largest hole first (habit 13/16):

  * **THE GRID METER.** It is not on this bus - it is a P1 telegram on a
    different wire - so nothing here says what the site was importing or
    exporting. The 2.4 kW export during the episode is from the operator's
    Home Assistant history and is NOT re-derivable from this file.
  * **WHAT THE ADD-ON BELIEVED.** The capture is the wire. It shows the
    commands that arrived and the readings that were answered; it does not show
    `Inverter.batt`, `inputs_ok`, or any log line. Everything about the add-on's
    internal state in this episode is INFERRED by replaying the shipped modules
    against these series, which is what part 10 does.
  * **DATE AND YEAR.** The capture tool stamps a time of day and nothing else.
    The date is supplied by the caller and is recorded in the scenario's
    `manifest.json` beside the sha256 of the file it came from.
  * **A FRAME THE CAPTURE TOOL SCORED BAD.** Lines whose CRC verdict is not
    `ok` are counted and dropped. In the committed window there are none.
  * **ANY OTHER REGISTER.** 0x0511 (a two-register read this master issues) is
    recognised as a read and its values are returned untouched under its own
    address; nothing here claims to know what it means.
"""

import gzip
import io
import re
import sys

# The registers this file will name. Anything else is returned by raw address.
REG_SETPOINT_MODE = 0x056E      # written as a pair with 0x056F
REG_BATT_W = 0x0518             # S16, watts, + = discharge (SAFETY-37)
REG_SOC_PCT = 0x050E            # U16, percent
MODE_CHARGE, MODE_DISCHARGE = 2, 3

_LINE = re.compile(r"^(\w+)\|(\d\d):(\d\d):(\d\d(?:\.\d+)?)\s+(\S+)\s+(.*)$")


def signed_setpoint_w(mode: int, magnitude: int) -> float:
    """The EMS pair as `Inverter.standing_w()` spells it: + = discharge.

    ⚠️ AN UNKNOWN MODE RAISES rather than defaulting. A default here would turn
    a firmware that started using a third mode into a silently mis-signed
    series, which is the failure this whole scenario is about one layer down.
    """
    if mode == MODE_DISCHARGE:
        return float(magnitude)
    if mode == MODE_CHARGE:
        return -float(magnitude)
    raise ValueError(f"unknown EMS mode {mode} at 0x{REG_SETPOINT_MODE:04X}; "
                     f"this decoder knows {MODE_CHARGE} (charge) and "
                     f"{MODE_DISCHARGE} (discharge) only")


def _s16(v: int) -> int:
    return v - 65536 if v >= 32768 else v


def decode(path):
    """Decode one capture file. Returns `(events, stats)`.

    `events` is a list of `(t, kind, value)` in capture order, where `t` is
    SECONDS SINCE LOCAL MIDNIGHT (the capture carries no date) and `kind` is
    one of `"setpoint_w"`, `"batt_w"`, `"soc_pct"`, or `"reg_0xNNNN"`.

    `stats` counts every line by what became of it, so a caller can assert that
    the frames it cares about were actually present rather than that the parse
    produced no error.
    """
    opener = gzip.open if str(path).endswith(".gz") else io.open
    events = []
    stats = {"lines": 0, "unparsed": 0, "bad_crc": 0, "requests": 0,
             "responses": 0, "writes": 0, "write_echoes": 0, "unknown": 0}
    pending = None                      # the read request awaiting its response
    with opener(path, "rt") as fh:
        for line in fh:
            line = line.strip()
            if not line:
                continue
            stats["lines"] += 1
            m = _LINE.match(line)
            if not m:
                stats["unparsed"] += 1
                continue
            _bus, hh, mm, ss, verdict, hexs = m.groups()
            if verdict != "ok":
                stats["bad_crc"] += 1
                continue
            t = int(hh) * 3600 + int(mm) * 60 + float(ss)
            try:
                b = [int(x, 16) for x in hexs.split()]
            except ValueError:
                stats["unparsed"] += 1
                continue
            # --- a read request: remember it so its response can be addressed
            if len(b) == 8 and b[1] == 0x03:
                pending = ((b[2] << 8) | b[3], (b[4] << 8) | b[5])
                stats["requests"] += 1
                continue
            # --- a read response, addressed by the request before it
            if (b[1] == 0x03 and pending is not None and len(b) >= 5
                    and b[2] == 2 * pending[1] and len(b) == b[2] + 5):
                data = b[3:3 + b[2]]
                regs = [(data[i] << 8) | data[i + 1]
                        for i in range(0, len(data), 2)]
                addr = pending[0]
                pending = None
                stats["responses"] += 1
                if addr == REG_BATT_W:
                    events.append((t, "batt_w", float(_s16(regs[0]))))
                elif addr == REG_SOC_PCT:
                    events.append((t, "soc_pct", float(regs[0])))
                else:
                    events.append((t, f"reg_0x{addr:04X}", tuple(regs)))
                continue
            # --- a write of the EMS pair
            if (len(b) > 8 and b[1] == 0x10 and b[2] == 0x05 and b[3] == 0x6E
                    and b[6] == 4):
                data = b[7:11]
                mode = (data[0] << 8) | data[1]
                mag = (data[2] << 8) | data[3]
                events.append((t, "setpoint_w", signed_setpoint_w(mode, mag)))
                stats["writes"] += 1
                continue
            if len(b) == 8 and b[1] == 0x10:
                stats["write_echoes"] += 1
                continue
            stats["unknown"] += 1
    return events, stats


def series(events, kind):
    """`[(t, value)]` for one kind, in capture order."""
    return [(t, v) for t, k, v in events if k == kind]


def held_at(rows, t):
    """The last value at or before `t`, or None before the first row.

    This is how the add-on sees an entity: Home Assistant answers with whatever
    the publisher last put there, and a value that has not been re-published is
    still the value. See `freshness.py` on why that is the whole problem.
    """
    out = None
    for ts, v in rows:
        if ts > t:
            break
        out = v
    return out


def _hms(t):
    return f"{int(t // 3600):02d}:{int(t % 3600 // 60):02d}:{t % 60:06.3f}"


def unreadable(stats):
    """Why a capture decoded to too little to trust, or None: most lines must be frames."""
    framed = sum(stats[k] for k in ("requests", "responses", "writes",
                                     "write_echoes"))
    if framed > 0 and stats["lines"] - framed <= framed:
        return None
    return (f"{framed} of {stats['lines']} line(s) decoded as frames "
            f"(unparsed {stats['unparsed']}, bad_crc {stats['bad_crc']}, "
            f"unknown {stats['unknown']}) - not a capture this decoder can read")


def main(argv):
    """Decode each capture named; exit 1 if any is unreadable, 2 on no argument."""
    if len(argv) < 2:
        print(__doc__.splitlines()[2].strip())
        return 2
    rc = 0
    for path in argv[1:]:
        rc |= _report(path)
    return rc


def _report(path):
    events, stats = decode(path)
    print(f"== {path}")
    for k, v in stats.items():
        print(f"   {k:14} {v}")
    bad = unreadable(stats)
    if bad:
        print(f"REFUSED: {path}: {bad}")
        return 1
    for kind in ("setpoint_w", "batt_w", "soc_pct"):
        rows = series(events, kind)
        if not rows:
            continue
        changes = [r for i, r in enumerate(rows)
                   if i == 0 or r[1] != rows[i - 1][1]]
        print(f"== {kind}: {len(rows)} frames, {len(changes)} changes, "
              f"{_hms(rows[0][0])} .. {_hms(rows[-1][0])}")
        for t, v in changes:
            print(f"   {_hms(t)}  {v:+10.1f}")
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv))