#!/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 a read REQUEST F7 03 <2*count> its response, matched to the request immediately preceding it F7 10 05 6E 00 02 04 a write of the two EMS registers F7 10 05 6E 00 02 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))