Packaged as a Home Assistant add-on, with a field guide
Turns the reference RS485 controller into something a technician can install at a client site: a typed config form instead of YAML, an ingress UI that names misconfiguration in words, and persistent state that cannot be broken by a timezone. Why an add-on rather than YAML packages or blueprints: - Blueprints cannot create helpers, and the maintenance cycle is a state machine whose phase and completion date must survive restarts. - YAML packages need filesystem access, a configuration.yaml edit and a restart - none of which belong in a client install. - Add-ons authenticate with SUPERVISOR_TOKEN, so there is no long-lived token to generate, store or leak on someone else's machine. - Requires HA OS/Supervised. Container and Core installs cannot run add-ons at all, which is a market decision, not an oversight. The control law and the maintenance machine are pure functions with no Home Assistant imports, and both ship with runnable checks (22 and 22 assertions). Every assertion corresponds to a rule whose absence caused an observed failure on hardware - the saturation duration term, the clamp-before-slew ordering, the deadband, the sign convention. One behaviour deliberately differs from the implementation it replaces: when its inputs go missing this commands 0 W rather than replaying the last setpoint. The reference version kept replaying, which the hardware watchdog cannot catch - from the ESP32's side, Home Assistant is still talking to it. Includes the ESPHome firmware (now parameterised: node name, inverter rating, watchdog timeout) and the optional RS485 e-stop. FIELD-GUIDE.md carries the commissioning gates, all judged on the wire rather than on how Home Assistant looks, plus the written statement a site without an e-stop needs signed. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016NckgXecasQb2eSsPYNSW6
This commit is contained in:
Claude Opus 5
@@ -0,0 +1,258 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Passive RS485 frame logger for the GoodWe bus.
|
||||
|
||||
The Pi takes over as witness during stage 1 (NOTES 18.4), when the T-CAN485
|
||||
sniffer board is reflashed as the master and stops listening.
|
||||
|
||||
python3 rs485_log.py --selftest # no hardware needed
|
||||
sudo python3 rs485_log.py --out bus.log # sudo: see tune_latency()
|
||||
|
||||
One line per frame: 21:00:29.412 ok F7 10 05 6F 00 01 02 05 DC D7 62
|
||||
Hex is space separated and CRC-suffixed exactly like master/frames.txt, so
|
||||
frame_check.py and the existing analysis take this file with an awk of $3-.
|
||||
|
||||
The log ROTATES (--max-mb / --keep): the Pi's SD card is finite and this file
|
||||
grows ~23 MB/day. History lands in bus.log.1 .. bus.log.N, oldest discarded.
|
||||
|
||||
With --panic it stops being purely passive: after --panic-after seconds of TOTAL
|
||||
bus silence it writes the 0 W frame itself. See panic_frame() for why that is
|
||||
safe and why 0 W is the only thing it may ever send.
|
||||
"""
|
||||
import argparse, itertools, logging, logging.handlers, pathlib, sys, time
|
||||
from datetime import datetime
|
||||
|
||||
BAUD = 9600
|
||||
GAP = 0.004 # 3.5 char times @ 9600 8N1 = 3.65 ms. Modbus RTU frame boundary.
|
||||
|
||||
# The atomic "grid = import, 0 W" write - NOTES §5.3/§5.5. The inverter has ACKed
|
||||
# this exact frame several hundred times. Byte-for-byte identical to
|
||||
# master/frame_check.py::setpoint_atomic(0); the selftest checks its CRC.
|
||||
ZERO_FRAME = bytes.fromhex("F710056E000204000300 00A6D0".replace(" ", ""))
|
||||
|
||||
|
||||
def crc16(data: bytes) -> int:
|
||||
"""Modbus RTU CRC16, little-endian on the wire. Same as master/frame_check.py."""
|
||||
crc = 0xFFFF
|
||||
for b in data:
|
||||
crc ^= b
|
||||
for _ in range(8):
|
||||
crc = (crc >> 1) ^ 0xA001 if crc & 1 else crc >> 1
|
||||
return crc
|
||||
|
||||
|
||||
def crc_ok(frame: bytes) -> bool:
|
||||
return len(frame) > 2 and crc16(frame[:-2]) == int.from_bytes(frame[-2:], "little")
|
||||
|
||||
|
||||
def frames(ser, idle=None):
|
||||
"""Split the byte stream on the inter-frame gap. ser.timeout must be GAP.
|
||||
|
||||
ponytail: byte-at-a-time. 960 B/s is nothing for a Pi 3, and it keeps the
|
||||
gap detection honest - a batched read cannot tell you where the silence was.
|
||||
|
||||
`idle` is called on every empty read, i.e. roughly every GAP while the bus is
|
||||
quiet. Without it the caller never regains control during silence - which is
|
||||
exactly the condition the panic writer has to notice.
|
||||
"""
|
||||
buf = bytearray()
|
||||
while True:
|
||||
b = ser.read(1) # blocks up to GAP, then returns b""
|
||||
if b:
|
||||
buf += b
|
||||
if len(buf) < 260: # 256 payload + header; runaway guard
|
||||
continue
|
||||
if buf:
|
||||
yield time.time(), bytes(buf)
|
||||
buf.clear()
|
||||
elif idle:
|
||||
idle()
|
||||
|
||||
|
||||
def tune_latency(port: str) -> str:
|
||||
"""USB serial chips buffer before handing bytes up; FTDI defaults to 16 ms,
|
||||
which smears the 3.65 ms gap that framing depends on. 1 ms is the minimum.
|
||||
Needs root, hence sudo. Not all chips expose it - CRC rate is the real check.
|
||||
"""
|
||||
p = pathlib.Path("/sys/bus/usb-serial/devices") / pathlib.Path(port).name / "latency_timer"
|
||||
try:
|
||||
now = p.read_text().strip()
|
||||
except OSError:
|
||||
return "latency_timer: chip does not expose it; watch the bad-CRC count"
|
||||
if now == "1":
|
||||
return "latency_timer = 1 ms"
|
||||
try: # needs root; the udev rule is the durable fix
|
||||
p.write_text("1")
|
||||
return "latency_timer = 1 ms (was %s)" % now
|
||||
except OSError:
|
||||
return ("latency_timer = %s ms and NOT writable as this user - frames will be "
|
||||
"glued together. Install the udev rule, or run under sudo." % now)
|
||||
|
||||
|
||||
def open_sink(path, max_mb, keep):
|
||||
"""Size-rotating log sink: bus.log, bus.log.1 .. bus.log.N, oldest dropped.
|
||||
|
||||
stdlib RotatingFileHandler, deliberately. logrotate would need root (a
|
||||
password someone has to type) plus a cron that actually ran - and a cron that
|
||||
silently stopped running is a full SD card and a dead witness.
|
||||
|
||||
ponytail: no gzip. Compressing 25 MB on a Pi 3 blocks this single-threaded
|
||||
loop for seconds and the tap would miss frames while it worked. Disk is the
|
||||
cheap resource here (25 GB free against 23 MB/day); the witness's continuity
|
||||
is not. Add compression only if space ever actually gets tight.
|
||||
"""
|
||||
h = logging.handlers.RotatingFileHandler(path, maxBytes=int(max_mb * 1_000_000),
|
||||
backupCount=keep, encoding="utf-8")
|
||||
h.setFormatter(logging.Formatter("%(message)s")) # no level/date prefix:
|
||||
log = logging.getLogger("bus") # the line format is the
|
||||
log.setLevel(logging.INFO) # documented interface
|
||||
log.addHandler(h)
|
||||
log.propagate = False
|
||||
return log
|
||||
|
||||
|
||||
class _FakeSerial:
|
||||
def __init__(self, chunks): self.q = list(chunks)
|
||||
def read(self, _n=1): return self.q.pop(0) if self.q else b""
|
||||
|
||||
|
||||
def selftest() -> int:
|
||||
real = [bytes.fromhex("F710056E0001020003947B"),
|
||||
bytes.fromhex("F710056F00010205DCD762")]
|
||||
assert all(crc_ok(f) for f in real), "known-good frame failed CRC"
|
||||
assert not crc_ok(real[1][:-3] + b"\x00" + real[1][-2:]), "corrupt frame passed CRC"
|
||||
|
||||
# the framer: two frames separated by one gap, split on silence not on length
|
||||
fake = _FakeSerial([bytes([b]) for b in real[0]] + [b""] +
|
||||
[bytes([b]) for b in real[1]] + [b""])
|
||||
got = [f for _, f in itertools.islice(frames(fake), 2)]
|
||||
assert got == real, [g.hex(" ") for g in got]
|
||||
assert crc_ok(ZERO_FRAME), "the panic frame's own CRC is wrong"
|
||||
assert ZERO_FRAME.hex(" ").upper() == "F7 10 05 6E 00 02 04 00 03 00 00 A6 D0"
|
||||
slave, fc, hi, lo, _, cnt, nbytes, dhi, dlo, mhi, mlo = ZERO_FRAME[:11]
|
||||
assert (slave, fc, (hi << 8) | lo, cnt, nbytes) == (0xF7, 0x10, 0x056E, 2, 4)
|
||||
assert (dhi << 8) | dlo == 3 and (mhi << 8) | mlo == 0, "panic frame is not 0 W"
|
||||
|
||||
# the idle callback must fire during silence, or the panic writer never runs
|
||||
ticks = []
|
||||
class _Stop(Exception): pass
|
||||
def _idle():
|
||||
ticks.append(1)
|
||||
if len(ticks) == 3:
|
||||
raise _Stop
|
||||
try:
|
||||
for _ in frames(_FakeSerial([]), idle=_idle):
|
||||
pass
|
||||
except _Stop:
|
||||
pass
|
||||
assert len(ticks) == 3, ticks
|
||||
|
||||
print("selftest ok: crc, corruption, gap framing, panic frame, idle tick")
|
||||
return 0
|
||||
|
||||
|
||||
def open_port(serial, port, rts=False, dtr=False):
|
||||
"""rts/dtr OFF: pyserial asserts RTS on open, and many USB-RS485 adapters use it
|
||||
as driver-enable - that would make a "passive" tap drive the bus. (Tested
|
||||
2026-08-19: not the cause of the frame corruption, but correct regardless.)"""
|
||||
s = serial.Serial(port, BAUD, bytesize=8, parity="N", stopbits=1, timeout=GAP,
|
||||
rtscts=False, dsrdtr=False)
|
||||
s.rts = rts
|
||||
s.dtr = dtr
|
||||
return s
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--port", default="/dev/ttyUSB0")
|
||||
ap.add_argument("--out", default="bus.log")
|
||||
ap.add_argument("--selftest", action="store_true")
|
||||
ap.add_argument("--rts", choices=["on", "off"], default="off",
|
||||
help="driver-enable polarity differs per adapter; test both")
|
||||
ap.add_argument("--seconds", type=float, default=0.0, help="stop after N s (0 = forever)")
|
||||
ap.add_argument("--max-mb", type=float, default=25.0, help="rotate at this size")
|
||||
ap.add_argument("--keep", type=int, default=30, help="how many rotated files to keep")
|
||||
ap.add_argument("--panic", action="store_true",
|
||||
help="EMERGENCY STOP: write 0 W if the bus goes totally silent")
|
||||
ap.add_argument("--panic-after", type=float, default=30.0,
|
||||
help="seconds of silence before the panic write (NOTES §4.6)")
|
||||
a = ap.parse_args()
|
||||
if a.selftest:
|
||||
return selftest()
|
||||
|
||||
import serial
|
||||
ser = open_port(serial, a.port, rts=(a.rts == "on"))
|
||||
print(tune_latency(a.port), file=sys.stderr)
|
||||
print(f"logging {a.port} -> {a.out}", file=sys.stderr)
|
||||
|
||||
sink = open_sink(a.out, a.max_mb, a.keep)
|
||||
print("rotating at %g MB, keeping %d (cap ~%g MB)"
|
||||
% (a.max_mb, a.keep, a.max_mb * (a.keep + 1)), file=sys.stderr)
|
||||
if a.panic:
|
||||
print("PANIC ARMED: 0 W after %gs of silence" % a.panic_after, file=sys.stderr)
|
||||
|
||||
good = bad = panics = 0
|
||||
t0 = time.time()
|
||||
last_frame = time.time()
|
||||
|
||||
def stamp_now():
|
||||
return datetime.now().strftime("%H:%M:%S.%f")[:-3]
|
||||
|
||||
def idle():
|
||||
"""Total bus silence means no master is alive - see NOTES §3.6/§4.6.
|
||||
|
||||
Silence is the discriminator precisely because a healthy master makes the
|
||||
bus busy every 5 s: if anything is driving, this never fires, so the tap
|
||||
cannot become a second master. And 0 W is the ONLY frame it may send, so
|
||||
even a false positive (say this adapter wedges while the bus is actually
|
||||
fine) costs at most one setpoint dip that the real master overwrites
|
||||
within its next 5 s cycle. A panic button that could send anything else
|
||||
would have to be as trustworthy as the master it replaces. This one does
|
||||
not.
|
||||
"""
|
||||
nonlocal last_frame, panics
|
||||
if not a.panic or time.time() - last_frame < a.panic_after:
|
||||
return
|
||||
ser.write(ZERO_FRAME)
|
||||
panics += 1
|
||||
sink.info("%s TX %s" % (stamp_now(), ZERO_FRAME.hex(" ").upper()))
|
||||
print("PANIC #%d: bus silent %gs, wrote 0 W" % (panics, a.panic_after),
|
||||
file=sys.stderr, flush=True)
|
||||
# The inverter's ACK lands in ~150 ms and resets this anyway; setting it
|
||||
# here makes the repeat cadence panic_after either way. Repeats matter:
|
||||
# §3.6 says never stop writing, a latched inverter is the failure mode.
|
||||
last_frame = time.time()
|
||||
|
||||
while True: # a replug re-enumerates the adapter and
|
||||
try: # invalidates the fd; do not just die
|
||||
for ts, fr in frames(ser, idle=idle):
|
||||
ok = crc_ok(fr)
|
||||
good, bad = good + ok, bad + (not ok)
|
||||
last_frame = ts # ANY traffic means a master is alive
|
||||
sink.info("%s %s %s" % (datetime.fromtimestamp(ts).strftime("%H:%M:%S.%f")[:-3],
|
||||
"ok " if ok else "BAD", fr.hex(" ").upper()))
|
||||
if (good + bad) % 100 == 0: # not every frame: often a file
|
||||
print(good, "frames,", bad, "bad CRC", file=sys.stderr, flush=True)
|
||||
if a.seconds and time.time() - t0 > a.seconds:
|
||||
print("RESULT rts=%s: %d ok, %d bad, %d panic" % (a.rts, good, bad, panics),
|
||||
file=sys.stderr, flush=True)
|
||||
return 0
|
||||
except serial.SerialException as e:
|
||||
print(f"lost {a.port}: {e}", file=sys.stderr, flush=True)
|
||||
try:
|
||||
ser.close()
|
||||
except Exception:
|
||||
pass
|
||||
while True: # udev has to reapply latency_timer too,
|
||||
time.sleep(2) # so there is no point hammering it
|
||||
try:
|
||||
ser = open_port(serial, a.port, rts=(a.rts == "on"))
|
||||
break
|
||||
except Exception:
|
||||
pass
|
||||
last_frame = time.time() # a 60 s replug is OUR blindness, not bus
|
||||
# silence - do not panic-write on reopen
|
||||
print("reopened", a.port, "|", tune_latency(a.port), file=sys.stderr, flush=True)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user