openvibers/openvibe.link Commits

Plugin runtime (Python) and the dry-run plugin

OP OpenVibers committed 29 Sept 2026 · eb77aa16e5 · 8 files, +723 −0

openvibe_plugin handles the JSON-lines protocol and the safety rules every
driver shares: describe before touching hardware, probe mode, stop at the
motion deadline, stop after 1 s without a heartbeat, stop on stdin EOF,
latched estop until resume. dryrun accepts every command and reports
simulated telemetry.

Changes

ddiff --git a/.gitignore b/.gitignore
nnew file mode 100644
iindex 0000000..6c4bbdb
---- /dev/null
++++ b/.gitignore
@@@ -0,0 +1,8 @@
++/dist/
++/openvibe-node
++*.exe
++__pycache__/
++*.pyc
++.pytest_cache/
++*.egg-info/
++/build/
ddiff --git a/plugins/dryrun/openvibe_dryrun/__init__.py b/plugins/dryrun/openvibe_dryrun/__init__.py
nnew file mode 100644
iindex 0000000..4197bc0
---- /dev/null
++++ b/plugins/dryrun/openvibe_dryrun/__init__.py
@@@ -0,0 +1,80 @@
++"""Dry-run driver: a pretend differential robot with a pan-tilt head, a speaker and a display.
++
++It accepts every command, logs it to stderr, keeps a simulated state (speeds, pan/tilt, battery) and reports it as
++telemetry. The camera is the core's built-in test pattern. Used by CI, by people trying the Node on a laptop, and by
++OpenVibe.Bot's end-to-end tests.
++"""
++
++import json
++import time
++
++from openvibe_plugin import Fault, Plugin, clamp
++
++__version__ = "0.1.0"
++
++
++class DryRun(Plugin):
++ driver = "dryrun"
++ version = __version__
++ motion_kinds = frozenset({"drive"})
++
++ def __init__(self):
++ self.state = {"throttle": 0.0, "steer": 0.0, "pan": 0.0, "tilt": 0.0, "actuators": {}}
++ self.battery = 100.0
++ self.started = time.monotonic()
++ self.log_commands = True
++
++ def describe(self, config):
++ return {
++ "drive": {"type": "differential"},
++ "ptz": {"pan": {"min": -1, "max": 1}, "tilt": {"min": -1, "max": 1}},
++ "actuator": {"names": ["any"]},
++ "say": {},
++ "display": {"text": True},
++ "battery": {},
++ "sensors": ["distance_cm"],
++ "camera": {"source": "test_pattern"},
++ }
++
++ def setup(self, config):
++ self.log_commands = bool(config.get("log_commands", True))
++ self.log("dryrun: ready (no hardware)")
++
++ def handle(self, cmd):
++ v = cmd.value if isinstance(cmd.value, dict) else {}
++ if cmd.kind == "drive":
++ if "x" in v or "y" in v:
++ self.state["throttle"] = clamp(v.get("x", 0), -1, 1)
++ self.state["steer"] = clamp(v.get("rotation", 0), -1, 1)
++ else:
++ self.state["throttle"] = clamp(v.get("throttle", 0), -1, 1)
++ self.state["steer"] = clamp(v.get("steer", 0), -1, 1)
++ elif cmd.kind == "ptz":
++ if "pan" in v:
++ self.state["pan"] = clamp(v["pan"], -1, 1)
++ if "tilt" in v:
++ self.state["tilt"] = clamp(v["tilt"], -1, 1)
++ elif cmd.kind == "actuator":
++ name = v.get("name")
++ if not isinstance(name, str) or not name:
++ raise Fault("bad_value", "actuator needs a name")
++ self.state["actuators"][name] = v.get("value")
++ elif cmd.kind in ("say", "display"):
++ pass
++ else:
++ raise Fault("unsupported", "kind %s" % cmd.kind)
++ if self.log_commands:
++ self.log("dryrun: %s %s (deadline %d ms)", cmd.kind, json.dumps(v, sort_keys=True), cmd.deadline_ms)
++
++ def stop(self):
++ if self.state["throttle"] or self.state["steer"]:
++ self.log("dryrun: stop")
++ self.state["throttle"] = 0.0
++ self.state["steer"] = 0.0
++
++ def poll(self):
++ self.battery = max(0.0, 100.0 - (time.monotonic() - self.started) / 360.0)
++ return {
++ "battery": {"volts": round(6.0 + 2.4 * self.battery / 100.0, 2), "percent": int(self.battery)},
++ "sensors": {"distance_cm": 123.4, "moving": bool(self.state["throttle"] or self.state["steer"])},
++ }
ddiff --git a/plugins/dryrun/openvibe_dryrun/__main__.py b/plugins/dryrun/openvibe_dryrun/__main__.py
nnew file mode 100644
iindex 0000000..619e4b3
---- /dev/null
++++ b/plugins/dryrun/openvibe_dryrun/__main__.py
@@@ -0,0 +1,6 @@
++import sys
++
++from openvibe_dryrun import DryRun
++from openvibe_plugin import run
++
++sys.exit(run(DryRun()))
ddiff --git a/plugins/dryrun/pyproject.toml b/plugins/dryrun/pyproject.toml
nnew file mode 100644
iindex 0000000..7ebd128
---- /dev/null
++++ b/plugins/dryrun/pyproject.toml
@@@ -0,0 +1,14 @@
++[build-system]
++requires = ["setuptools>=61"]
++build-backend = "setuptools.build_meta"
++
++[project]
++name = "openvibe-dryrun"
++version = "0.1.0"
++description = "OpenVibe Node dry-run plugin: accepts every command, logs it, reports simulated telemetry"
++requires-python = ">=3.9"
++license = { text = "AGPL-3.0-or-later" }
++dependencies = ["openvibe-plugin"]
++
++[tool.setuptools]
++packages = ["openvibe_dryrun"]
ddiff --git a/plugins/dryrun/tests/test_dryrun.py b/plugins/dryrun/tests/test_dryrun.py
nnew file mode 100644
iindex 0000000..7448719
---- /dev/null
++++ b/plugins/dryrun/tests/test_dryrun.py
@@@ -0,0 +1,77 @@
++import io
++import json
++import os
++import sys
++import unittest
++
++HERE = os.path.dirname(os.path.abspath(__file__))
++sys.path.insert(0, os.path.dirname(HERE))
++sys.path.insert(0, os.path.join(os.path.dirname(os.path.dirname(HERE)), "sdk"))
++
++from openvibe_dryrun import DryRun # noqa: E402
++from openvibe_plugin import Runtime # noqa: E402
++
++
++class Clock:
++ t = 10.0
++
++ def __call__(self):
++ return self.t
++
++
++class DryRunTest(unittest.TestCase):
++ def setUp(self):
++ self.out = io.StringIO()
++ self.clock = Clock()
++ self.p = DryRun()
++ self.p.log = lambda *a: None
++ self.rt = Runtime(self.p, infile=io.StringIO(""), outfile=self.out, clock=self.clock)
++ self.send({"op": "hello", "config": {}})
++
++ def send(self, m):
++ self.rt.feed_line(json.dumps(m))
++
++ def last(self):
++ return json.loads(self.out.getvalue().splitlines()[-1])
++
++ def test_accepts_every_kind(self):
++ for i, (kind, value) in enumerate([
++ ("drive", {"throttle": 0.5, "steer": -0.2}),
++ ("drive", {"x": 0.1, "y": 0.2, "rotation": 0.3}),
++ ("ptz", {"pan": 0.4, "tilt": -0.4}),
++ ("actuator", {"name": "lift", "value": 0.5}),
++ ("say", {"text": "hello"}),
++ ("display", {"text": "hi"}),
++ ("halt", {}),
++ ]):
++ self.send({"op": "command", "id": "c%d" % i, "kind": kind, "value": value, "deadline_ms": 300})
++ self.assertEqual(self.last(), {"op": "ack", "id": "c%d" % i}, kind)
++
++ def test_deadline_stop(self):
++ self.send({"op": "command", "id": "d", "kind": "drive", "value": {"throttle": 1}, "deadline_ms": 300})
++ self.assertEqual(self.p.state["throttle"], 1.0)
++ self.clock.t += 0.31
++ self.send({"op": "heartbeat", "t": 1})
++ self.rt.tick()
++ self.assertEqual(self.p.state["throttle"], 0.0)
++
++ def test_bad_value(self):
++ self.send({"op": "command", "id": "b", "kind": "drive", "value": {"throttle": "fast"}})
++ self.assertEqual(self.last()["fault_code"], "bad_value")
++
++ def test_telemetry(self):
++ self.clock.t += 1
++ self.send({"op": "heartbeat", "t": 1})
++ self.rt.tick()
++ t = self.last()
++ self.assertEqual(t["op"], "telemetry")
++ self.assertIn("percent", t["battery"])
++
++ def test_describe(self):
++ d = json.loads(self.out.getvalue().splitlines()[0])
++ self.assertEqual(d["driver"], "dryrun")
++ self.assertEqual(d["capabilities"]["camera"]["source"], "test_pattern")
++
++
++if __name__ == "__main__":
++ unittest.main()
ddiff --git a/plugins/sdk/openvibe_plugin/__init__.py b/plugins/sdk/openvibe_plugin/__init__.py
nnew file mode 100644
iindex 0000000..fefbf74
---- /dev/null
++++ b/plugins/sdk/openvibe_plugin/__init__.py
@@@ -0,0 +1,337 @@
++"""OpenVibe Node plugin runtime.
++
++A plugin is an executable the Node core starts. It reads JSON lines from stdin and writes JSON lines to stdout (one
++object per line; logs go to stderr). This module implements the parts every driver must get right, so a driver only
++says how to move its hardware:
++
++* describe first, touch hardware only after ``hello`` (and never in probe mode);
++* stop every actuator when stdin closes, when a motion command's deadline passes without a newer command, when no
++ heartbeat has arrived from the core for ``HEARTBEAT_TIMEOUT_S``, and on ``stop`` / ``estop``;
++* keep an e-stop latched until the core sends ``resume``.
++
++See docs/plugins.md for the wire format.
++"""
++
++import json
++import queue
++import sys
++import threading
++import time
++
++__all__ = ["Plugin", "Fault", "Command", "Runtime", "run", "clamp", "MOTION_KINDS", "HEARTBEAT_TIMEOUT_S",
++ "DEFAULT_DEADLINE_MS", "PROTOCOL_VERSION"]
++
++PROTOCOL_VERSION = 1
++HEARTBEAT_TIMEOUT_S = 1.0
++DEFAULT_DEADLINE_MS = 300
++MAX_DEADLINE_MS = 5000
++# Kinds that move something continuously and therefore stop at their deadline. A plugin may widen this set
++# (Plugin.motion_kinds) for actuators it drives by speed rather than by position.
++MOTION_KINDS = frozenset({"drive"})
++# Kinds refused while an e-stop is latched or the heartbeat is lost. halt is always accepted.
++GUARDED_KINDS = frozenset({"drive", "ptz", "actuator"})
++
++
++class Fault(Exception):
++ """Raised by a driver to answer a command with a nack."""
++
++ def __init__(self, code, message=""):
++ super().__init__(message or code)
++ self.code = code
++ self.message = message
++
++
++class Command:
++ __slots__ = ("id", "kind", "value", "deadline_ms", "deadline", "target")
++
++ def __init__(self, id, kind, value, deadline_ms, now, target=None):
++ self.id = id
++ self.kind = kind
++ self.value = value if value is not None else {}
++ self.deadline_ms = deadline_ms
++ self.deadline = now + deadline_ms / 1000.0
++ self.target = target
++
++ def __repr__(self):
++ return "Command(id=%r, kind=%r, value=%r, deadline_ms=%r)" % (self.id, self.kind, self.value, self.deadline_ms)
++
++
++def clamp(v, lo, hi):
++ try:
++ v = float(v)
++ except (TypeError, ValueError):
++ raise Fault("bad_value", "not a number: %r" % (v,))
++ if v != v: # NaN
++ raise Fault("bad_value", "NaN")
++ return max(lo, min(hi, v))
++
++
++class Plugin:
++ """Base class for drivers. Override the methods you need.
++
++ All methods run on the runtime's main thread, except that ``stop`` may also be called from ``emit_event`` callers
++ (it takes the runtime lock). Methods must not block for long: do slow work (speech synthesis, file I/O) on a
++ thread of your own.
++ """
++
++ driver = "plugin"
++ version = "0.0.0"
++ motion_kinds = MOTION_KINDS
++ telemetry_interval_s = 0.5
++
++ runtime = None # set by Runtime
++
++ def describe(self, config):
++ """Return the capabilities dict. Called before setup and in probe mode; must not touch hardware."""
++ return {}
++
++ def setup(self, config):
++ """Open the hardware. Every actuator must be stopped when this returns."""
++
++ def handle(self, cmd):
++ """Execute a command. Return normally for ack, raise Fault for nack."""
++ raise Fault("unsupported", "kind %s not supported" % cmd.kind)
++
++ def stop_motion(self):
++ """Stop what moves on its own (wheels). Called at a motion deadline. Defaults to stop()."""
++ self.stop()
++
++ def stop(self):
++ """Stop every actuator. Must be idempotent and safe to call at any time, including before setup."""
++
++ def poll(self):
++ """Return a telemetry dict (battery, rssi, sensors) or None. Called every telemetry_interval_s."""
++ return None
++
++ def close(self):
++ """Release the hardware. stop() has already been called."""
++
++ # Helpers for drivers.
++ def emit_event(self, name, **fields):
++ if self.runtime is not None:
++ self.runtime.send(dict(op="event", name=name, **fields))
++
++ def emit_video(self, path, format="jpeg", **fields):
++ if self.runtime is not None:
++ self.runtime.send(dict(op="video", format=format, path=path, **fields))
++
++ def emit_telemetry(self, **fields):
++ if self.runtime is not None:
++ self.runtime.send(dict(op="telemetry", **fields))
++
++ def log(self, msg, *args):
++ sys.stderr.write((msg % args if args else msg) + "\n")
++ sys.stderr.flush()
++
++
++class Runtime:
++ """Drives one Plugin from a line-oriented input and output. Tests construct it with in-memory streams."""
++
++ def __init__(self, plugin, infile=None, outfile=None, clock=time.monotonic,
++ heartbeat_timeout_s=HEARTBEAT_TIMEOUT_S):
++ self.plugin = plugin
++ plugin.runtime = self
++ self.infile = infile if infile is not None else sys.stdin
++ self.outfile = outfile if outfile is not None else sys.stdout
++ self.clock = clock
++ self.heartbeat_timeout_s = heartbeat_timeout_s
++ self.lock = threading.RLock()
++ self.out_lock = threading.Lock()
++ self.inbox = queue.Queue()
++ self.config = {}
++ self.probe = False
++ self.ready = False
++ self.estopped = False
++ self.motion_deadline = None
++ self.last_heartbeat = None
++ self.heartbeat_lost = False
++ self.next_poll = 0.0
++ self.closed = False
++ self.stop_reasons = [] # for tests: why stop() was called, in order
++
++ # Output.
++ def send(self, obj):
++ line = json.dumps(obj, separators=(",", ":"), default=str) + "\n"
++ with self.out_lock:
++ try:
++ self.outfile.write(line)
++ self.outfile.flush()
++ except (BrokenPipeError, ValueError, OSError):
++ pass
++
++ # Safety.
++ def safe_stop(self, reason, motion_only=False):
++ with self.lock:
++ self.stop_reasons.append(reason)
++ self.motion_deadline = None
++ try:
++ if motion_only:
++ self.plugin.stop_motion()
++ else:
++ self.plugin.stop()
++ except Exception as e: # a failing stop is the worst case; report it loudly
++ self.plugin.log("stop failed (%s): %r", reason, e)
++ self.send({"op": "event", "name": "stop_failed", "reason": reason, "error": str(e)})
++
++ def tick(self):
++ """Run time-based safety checks. Called by the loop at least every 20 ms."""
++ now = self.clock()
++ if self.motion_deadline is not None and now >= self.motion_deadline:
++ self.safe_stop("deadline", motion_only=True)
++ if self.ready and self.last_heartbeat is not None and not self.heartbeat_lost \
++ and now - self.last_heartbeat > self.heartbeat_timeout_s:
++ self.heartbeat_lost = True
++ self.safe_stop("heartbeat")
++ self.send({"op": "event", "name": "heartbeat_lost"})
++ if self.ready and now >= self.next_poll:
++ self.next_poll = now + self.plugin.telemetry_interval_s
++ try:
++ t = self.plugin.poll()
++ except Exception as e:
++ t = None
++ self.plugin.log("poll failed: %r", e)
++ if t:
++ self.send(dict(op="telemetry", **t))
++
++ # Input.
++ def dispatch(self, msg):
++ op = msg.get("op")
++ if op == "hello":
++ self.on_hello(msg)
++ elif op == "heartbeat":
++ self.last_heartbeat = self.clock()
++ if self.heartbeat_lost:
++ self.heartbeat_lost = False
++ elif op == "command":
++ self.on_command(msg)
++ elif op == "stop":
++ self.safe_stop("stop")
++ elif op == "estop":
++ self.estopped = True
++ self.safe_stop("estop")
++ elif op == "resume":
++ self.estopped = False
++ else:
++ self.plugin.log("unknown op %r", op)
++
++ def on_hello(self, msg):
++ self.config = msg.get("config") or {}
++ self.probe = bool(msg.get("probe"))
++ try:
++ caps = self.plugin.describe(self.config)
++ except Exception as e:
++ caps = {}
++ self.plugin.log("describe failed: %r", e)
++ self.send({"op": "describe", "driver": self.plugin.driver, "version": self.plugin.version,
++ "protocol": PROTOCOL_VERSION, "capabilities": caps,
++ "motion_kinds": sorted(self.plugin.motion_kinds)})
++ if self.probe:
++ return
++ try:
++ with self.lock:
++ self.plugin.setup(self.config)
++ self.plugin.stop()
++ self.ready = True
++ self.last_heartbeat = self.clock()
++ self.send({"op": "ready"})
++ except Fault as f:
++ self.send({"op": "fault", "fault_code": f.code, "message": f.message})
++ except Exception as e:
++ self.send({"op": "fault", "fault_code": "hardware", "message": str(e)})
++
++ def on_command(self, msg):
++ cid = msg.get("id")
++ kind = msg.get("kind")
++ now = self.clock()
++ try:
++ dl = int(msg.get("deadline_ms") or DEFAULT_DEADLINE_MS)
++ except (TypeError, ValueError):
++ dl = DEFAULT_DEADLINE_MS
++ dl = max(1, min(dl, MAX_DEADLINE_MS))
++ cmd = Command(cid, kind, msg.get("value"), dl, now, msg.get("target"))
++ if kind == "halt":
++ self.safe_stop("halt")
++ self.send({"op": "ack", "id": cid})
++ return
++ if not self.ready:
++ self.send({"op": "nack", "id": cid, "fault_code": "not_ready"})
++ return
++ if kind in GUARDED_KINDS or kind in self.plugin.motion_kinds:
++ if self.estopped:
++ self.send({"op": "nack", "id": cid, "fault_code": "estopped"})
++ return
++ if self.heartbeat_lost:
++ self.send({"op": "nack", "id": cid, "fault_code": "no_heartbeat"})
++ return
++ try:
++ with self.lock:
++ self.plugin.handle(cmd)
++ if kind in self.plugin.motion_kinds:
++ self.motion_deadline = cmd.deadline
++ self.send({"op": "ack", "id": cid})
++ except Fault as f:
++ out = {"op": "nack", "id": cid, "fault_code": f.code}
++ if f.message:
++ out["message"] = f.message
++ self.send(out)
++ except Exception as e:
++ self.safe_stop("handler_error")
++ self.send({"op": "nack", "id": cid, "fault_code": "hardware", "message": str(e)})
++
++ def feed_line(self, line):
++ line = line.strip()
++ if not line:
++ return
++ try:
++ msg = json.loads(line)
++ except ValueError:
++ self.plugin.log("bad json from core: %r", line[:200])
++ return
++ if isinstance(msg, dict):
++ self.dispatch(msg)
++
++ def shutdown(self, reason="eof"):
++ if self.closed:
++ return
++ self.closed = True
++ self.safe_stop(reason)
++ try:
++ self.plugin.close()
++ except Exception as e:
++ self.plugin.log("close failed: %r", e)
++
++ # Loop.
++ def _reader(self):
++ try:
++ for line in self.infile:
++ self.inbox.put(line)
++ except Exception:
++ pass
++ self.inbox.put(None)
++
++ def run(self):
++ t = threading.Thread(target=self._reader, name="stdin", daemon=True)
++ t.start()
++ try:
++ while True:
++ try:
++ line = self.inbox.get(timeout=0.02)
++ except queue.Empty:
++ line = ""
++ if line is None:
++ break
++ if line:
++ self.feed_line(line)
++ self.tick()
++ finally:
++ self.shutdown("eof")
++ return 0
++
++
++def run(plugin):
++ """Entry point for a plugin's __main__."""
++ try:
++ sys.stdout.reconfigure(line_buffering=True)
++ except AttributeError:
++ pass
++ return Runtime(plugin).run()
ddiff --git a/plugins/sdk/pyproject.toml b/plugins/sdk/pyproject.toml
nnew file mode 100644
iindex 0000000..e6f5b99
---- /dev/null
++++ b/plugins/sdk/pyproject.toml
@@@ -0,0 +1,13 @@
++[build-system]
++requires = ["setuptools>=61"]
++build-backend = "setuptools.build_meta"
++
++[project]
++name = "openvibe-plugin"
++version = "0.1.0"
++description = "Runtime for OpenVibe Node plugins (JSON lines over stdin/stdout, deadlines, heartbeat, e-stop)"
++requires-python = ">=3.9"
++license = { text = "AGPL-3.0-or-later" }
++
++[tool.setuptools]
++packages = ["openvibe_plugin"]
ddiff --git a/plugins/sdk/tests/test_runtime.py b/plugins/sdk/tests/test_runtime.py
nnew file mode 100644
iindex 0000000..afdb4c8
---- /dev/null
++++ b/plugins/sdk/tests/test_runtime.py
@@@ -0,0 +1,188 @@
++import io
++import json
++import os
++import subprocess
++import sys
++import time
++import unittest
++
++HERE = os.path.dirname(os.path.abspath(__file__))
++sys.path.insert(0, os.path.dirname(HERE))
++
++from openvibe_plugin import Fault, Plugin, Runtime # noqa: E402
++
++
++class FakeClock:
++ def __init__(self):
++ self.t = 100.0
++
++ def __call__(self):
++ return self.t
++
++
++class Recorder(Plugin):
++ driver = "recorder"
++ version = "1.0"
++
++ def __init__(self):
++ self.calls = []
++ self.moving = False
++
++ def describe(self, config):
++ self.calls.append("describe")
++ return {"drive": {"type": "differential"}}
++
++ def setup(self, config):
++ self.calls.append("setup")
++
++ def handle(self, cmd):
++ if cmd.kind == "drive":
++ self.moving = True
++ return
++ if cmd.kind == "say":
++ return
++ raise Fault("unsupported")
++
++ def stop(self):
++ self.calls.append("stop")
++ self.moving = False
++
++
++def make(plugin=None):
++ out = io.StringIO()
++ clock = FakeClock()
++ p = plugin or Recorder()
++ rt = Runtime(p, infile=io.StringIO(""), outfile=out, clock=clock)
++ return rt, p, out, clock
++
++
++def lines(out):
++ return [json.loads(x) for x in out.getvalue().splitlines() if x]
++
++
++class RuntimeTest(unittest.TestCase):
++ def hello(self, rt):
++ rt.feed_line(json.dumps({"op": "hello", "config": {}}))
++
++ def test_describe_then_setup_stopped(self):
++ rt, p, out, _ = make()
++ self.hello(rt)
++ msgs = lines(out)
++ self.assertEqual(msgs[0]["op"], "describe")
++ self.assertEqual(msgs[0]["driver"], "recorder")
++ self.assertEqual(msgs[1]["op"], "ready")
++ self.assertEqual(p.calls[:3], ["describe", "setup", "stop"])
++
++ def test_probe_never_touches_hardware(self):
++ rt, p, out, _ = make()
++ rt.feed_line(json.dumps({"op": "hello", "config": {}, "probe": True}))
++ self.assertEqual(p.calls, ["describe"])
++ self.assertEqual(lines(out)[-1]["op"], "describe")
++
++ def test_deadline_stops(self):
++ rt, p, out, clock = make()
++ self.hello(rt)
++ rt.feed_line(json.dumps({"op": "command", "id": "c1", "kind": "drive", "value": {}, "deadline_ms": 300}))
++ self.assertTrue(p.moving)
++ clock.t += 0.29
++ rt.feed_line(json.dumps({"op": "heartbeat", "t": 1}))
++ rt.tick()
++ self.assertTrue(p.moving)
++ clock.t += 0.02
++ rt.tick()
++ self.assertFalse(p.moving)
++ self.assertIn("deadline", rt.stop_reasons)
++
++ def test_newer_command_extends_deadline(self):
++ rt, p, out, clock = make()
++ self.hello(rt)
++ rt.feed_line(json.dumps({"op": "command", "id": "c1", "kind": "drive", "deadline_ms": 300}))
++ clock.t += 0.15
++ rt.feed_line(json.dumps({"op": "command", "id": "c2", "kind": "drive", "deadline_ms": 300}))
++ clock.t += 0.2
++ rt.feed_line(json.dumps({"op": "heartbeat", "t": 1}))
++ rt.tick()
++ self.assertTrue(p.moving)
++ clock.t += 0.2
++ rt.tick()
++ self.assertFalse(p.moving)
++
++ def test_heartbeat_loss_stops_and_refuses(self):
++ rt, p, out, clock = make()
++ self.hello(rt)
++ rt.feed_line(json.dumps({"op": "command", "id": "c1", "kind": "drive", "deadline_ms": 5000}))
++ clock.t += 1.01
++ rt.tick()
++ self.assertFalse(p.moving)
++ self.assertIn("heartbeat", rt.stop_reasons)
++ rt.feed_line(json.dumps({"op": "command", "id": "c2", "kind": "drive"}))
++ self.assertEqual(lines(out)[-1], {"op": "nack", "id": "c2", "fault_code": "no_heartbeat"})
++ rt.feed_line(json.dumps({"op": "heartbeat", "t": 2}))
++ rt.feed_line(json.dumps({"op": "command", "id": "c3", "kind": "drive"}))
++ self.assertEqual(lines(out)[-1], {"op": "ack", "id": "c3"})
++
++ def test_estop_latches_until_resume(self):
++ rt, p, out, clock = make()
++ self.hello(rt)
++ rt.feed_line(json.dumps({"op": "command", "id": "c1", "kind": "drive"}))
++ rt.feed_line(json.dumps({"op": "estop"}))
++ self.assertFalse(p.moving)
++ rt.feed_line(json.dumps({"op": "command", "id": "c2", "kind": "drive"}))
++ self.assertEqual(lines(out)[-1]["fault_code"], "estopped")
++ rt.feed_line(json.dumps({"op": "command", "id": "c3", "kind": "say", "value": {"text": "hi"}}))
++ self.assertEqual(lines(out)[-1]["op"], "ack")
++ rt.feed_line(json.dumps({"op": "resume"}))
++ rt.feed_line(json.dumps({"op": "command", "id": "c4", "kind": "drive"}))
++ self.assertEqual(lines(out)[-1]["op"], "ack")
++
++ def test_halt_always_accepted(self):
++ rt, p, out, clock = make()
++ self.hello(rt)
++ rt.feed_line(json.dumps({"op": "estop"}))
++ rt.feed_line(json.dumps({"op": "command", "id": "h", "kind": "halt"}))
++ self.assertEqual(lines(out)[-1], {"op": "ack", "id": "h"})
++
++ def test_unsupported_nack(self):
++ rt, p, out, clock = make()
++ self.hello(rt)
++ rt.feed_line(json.dumps({"op": "command", "id": "x", "kind": "display"}))
++ self.assertEqual(lines(out)[-1]["fault_code"], "unsupported")
++
++ def test_eof_stops(self):
++ p = Recorder()
++ out = io.StringIO()
++ inp = io.StringIO(json.dumps({"op": "hello", "config": {}}) + "\n" +
++ json.dumps({"op": "command", "id": "c", "kind": "drive", "deadline_ms": 5000}) + "\n")
++ Runtime(p, infile=inp, outfile=out).run()
++ self.assertFalse(p.moving)
++ self.assertEqual(p.calls[-1], "stop")
++
++
++class ProcessTest(unittest.TestCase):
++ """The real thing: a child process whose stdin closes must stop and exit."""
++
++ def test_child_exits_on_eof(self):
++ code = ("import sys; sys.path.insert(0, %r)\n"
++ "from openvibe_plugin import Plugin, run\n"
++ "class P(Plugin):\n"
++ " driver='t'\n"
++ " def handle(self, cmd): pass\n"
++ " def stop(self): sys.stderr.write('STOPPED\\n'); sys.stderr.flush()\n"
++ "sys.exit(run(P()))\n") % os.path.dirname(HERE)
++ proc = subprocess.Popen([sys.executable, "-c", code], stdin=subprocess.PIPE, stdout=subprocess.PIPE,
++ stderr=subprocess.PIPE, text=True)
++ proc.stdin.write(json.dumps({"op": "hello", "config": {}}) + "\n")
++ proc.stdin.write(json.dumps({"op": "command", "id": "a", "kind": "drive", "deadline_ms": 5000}) + "\n")
++ proc.stdin.flush()
++ time.sleep(0.2)
++ proc.stdin.close()
++ rc = proc.wait(timeout=5)
++ err = proc.stderr.read()
++ out = proc.stdout.read()
++ self.assertEqual(rc, 0)
++ self.assertIn('"op":"ack"', out)
++ self.assertIn("STOPPED", err)
++
++
++if __name__ == "__main__":
++ unittest.main()