qemu: rs485 silence|restore takes a simulated device off the bus mid-run

ui-drive.sh --rs485-devices fixed the roster for the whole boot, so no
script could show the guest noticing a device go quiet. mbsim.py now
serves a control socket (flare-edge --control); ui-drive.sh opens it next
to the pty and hands its path to qmp.py drive (--rs485-control), whose new
verb `rs485 silence|restore ADDR` sends one command and judges the reply.
A run without a simulated bus records the step as fatal rather than a
silent pass. Rig: unit 5 silenced reads online=false after 65 s in the
status json, restored reads online=true after 64 s (flare-edge #184).

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01N3G6m9Aw5RyVY4ZowtKzEj
This commit is contained in:
Noah
2026-09-09 06:55:07 -06:00
co-authored by Claude Fable 5.1
parent cab6c25c7e
commit 48be35dc7b
3 changed files with 66 additions and 5 deletions
+40 -2
View File
@@ -37,6 +37,10 @@ a real panel, so these verbs mean the same thing on the rig and on hardware:
eats the touch that ends either); every compiled
flow script opens with it
page | stats | hit X Y print the reply, judge nothing
rs485 silence|restore ADDR
take a simulated bus device off the bus (holds its
address, answers nothing) or put it back; needs
--rs485-control (ui-drive.sh --rs485-devices)
ctl WORDS... raw passthrough for anything the channel grows
assert_page MENU/TAB the active page is exactly this
assert_hit X Y CLASS [TEXT...]
@@ -486,7 +490,18 @@ def write_png(img, path):
_PILImage.frombytes("RGB", (w, h), data).save(path)
def drive(s, f, script_path, outdir, size, ctl_path=None, console_path=None, refs_path=None):
def rs485_send(control_path, line):
"""One command to mbsim.py's control socket (flare-edge tools/modbus-sim,
--control): -> its one-line reply, `ok ...` or `error ...`."""
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as c:
c.settimeout(5)
c.connect(control_path)
c.sendall((line + "\n").encode())
return c.makefile("r").readline().strip()
def drive(s, f, script_path, outdir, size, ctl_path=None, console_path=None, refs_path=None,
rs485_control=None):
os.makedirs(outdir, exist_ok=True)
with open(script_path) as fh:
lines = fh.readlines()
@@ -594,6 +609,26 @@ def drive(s, f, script_path, outdir, size, ctl_path=None, console_path=None, ref
need_ctl(lineno, cmd)
reply = ctl.send(line if cmd != "ctl" else " ".join(args))
record(lineno, line, "ok", reply)
elif cmd == "rs485":
# rs485 silence|restore ADDR: take a simulated device off the bus
# (it holds its address and answers nothing, exactly an absent
# unit) or put it back, while the guest keeps polling. This is
# the only runtime lever on the roster ui-drive.sh --rs485-devices
# fixed at boot, and it exists so a flow can prove a screen
# noticing an adopted device go quiet (flare-edge #184). No
# simulated bus in this run is fatal: the step could not check
# anything, and a silent skip would read as a pass.
if rs485_control is None:
record(lineno, line, "fatal", "no simulated RS485 bus in this run "
"(ui-drive.sh --rs485-devices)")
elif len(args) != 2 or args[0] not in ("silence", "restore"):
record(lineno, line, "fatal", "expected: rs485 silence|restore ADDR")
else:
try:
reply = rs485_send(rs485_control, f"{args[0]} {args[1]}")
except OSError as e:
reply = f"error control socket: {e}"
record(lineno, line, "ok" if reply.startswith("ok") else "fail", reply)
elif cmd == "assert_page":
need_ctl(lineno, cmd)
want = " ".join(args)
@@ -815,6 +850,8 @@ def main():
ctl_path = sys.argv[sys.argv.index("--ctl") + 1] if "--ctl" in sys.argv else None
console_path = sys.argv[sys.argv.index("--console") + 1] if "--console" in sys.argv else None
refs_path = sys.argv[sys.argv.index("--refs") + 1] if "--refs" in sys.argv else None
rs485_control = (sys.argv[sys.argv.index("--rs485-control") + 1]
if "--rs485-control" in sys.argv else None)
s = socket.socket(socket.AF_UNIX)
s.connect(path)
@@ -827,7 +864,8 @@ def main():
elif cmd == "tap":
do_tap(s, f, int(sys.argv[3]), int(sys.argv[4]))
elif cmd == "drive":
drive(s, f, sys.argv[3], sys.argv[4], size, ctl_path, console_path, refs_path)
drive(s, f, sys.argv[3], sys.argv[4], size, ctl_path, console_path, refs_path,
rs485_control)
elif cmd == "quit":
s.sendall(b'{"execute":"quit"}\n')