Write a driver

What a driver is

A driver is a separate executable the OpenVibe Node core starts. It can be written in any language. The two speak JSON lines over stdin/stdout: one JSON object per line, UTF-8, in each direction. The core never links driver code; the robot drivers are Python 3 because their hardware libraries are.

Nothing but protocol lines may go to stdout. Send diagnostics to stderr (the Python runtime points file descriptor 1 at stderr and keeps a private copy for the protocol, because hardware libraries print).

The Node core itself speaks a different protocol to openvibe.bot (bot.device-message@1, protocol.md); a driver never speaks it. The core is the one that pairs, holds the control link, applies the e-stop and the deadman, and forwards already-validated commands to a driver.

The three connection kinds

The robot's profile carries a kind, which says where the Node runs relative to the robot:

• on-board (kind: "onboard") — the Node runs on the robot itself, over its own GPIO/I²C bus. Example: adeept_adr036 on a Raspberry Pi.
• bridge (kind: "bridge") — the Node runs on a computer next to the robot and drives it over the robot's own link. Example: cozmo through PyCozmo, where the computer joins the robot's Wi-Fi on one interface and the internet on another.
• server-side (kind: "server") — there is no Node on the robot at all: Bot attaches the device itself from the server. Examples: the simulated rover and an ONVIF camera.

The pairing's device_kind is onboard or bridge; a server-side robot is not installed on a machine.

Lifecycle

1. The core starts the process and sends hello.
2. The driver answers describe before touching any hardware.
3. Unless hello.probe is true, the driver opens its hardware with every actuator stopped and sends ready (or fault if it cannot). Probe mode is used by openvibe-node pair and openvibe-node plugins: describe, then wait for EOF without opening anything.
4. The core sends heartbeat every 250 ms and commands as they arrive; the driver answers each command with ack or nack, and sends telemetry, event and video whenever it has them.
5. On shutdown the core sends stop and closes stdin; the driver stops everything and exits. After 2 s it is killed.

A driver that exits is restarted with exponential backoff (1 s doubling to 30 s, reset after a minute of uptime). It starts stopped; if the e-stop or the local kill switch is latched, the core sends estop right after hello. A driver that does not describe itself within 15 s, or stops reading stdin, is killed.

Messages

Every message is one JSON object on one line with an op field. Command value has already been validated and clamped to the owner's limits by the core, and deadline_ms is relative to receipt and already capped.

Core → driver

{ "op": "hello", "config": { "backend": "real" }, "probe": false }

{ "op": "command", "id": "cmd_01J8Z4M2Q0R7T9YV3K6N8P1W2X", "kind": "drive", "value": { "throttle": 0.5, "steer": 0 }, "deadline_ms": 300 }

{ "op": "heartbeat", "t": 1728384000000 }

{ "op": "stop" }

{ "op": "estop" }

{ "op": "resume" }

Driver → core

{ "op": "describe", "driver": "my_arm", "version": "0.1.0", "protocol": 1, "capabilities": { "actuator": { "names": ["shoulder", "gripper"] } }, "motion_kinds": ["drive", "actuator"] }

{ "op": "ready" }

{ "op": "fault", "fault_code": "hardware", "message": "gripper not responding" }

{ "op": "ack", "id": "cmd_01J8Z4M2Q0R7T9YV3K6N8P1W2X" }

{ "op": "nack", "id": "cmd_01J8Z4M2Q0R7T9YV3K6N8P1W2X", "fault_code": "unsupported", "message": "only actuators" }

{ "op": "telemetry", "battery": { "volts": 7.4, "percent": 82 }, "rssi": -54, "sensors": { "distance_cm": 34 } }

{ "op": "event", "name": "cliff", "fields": { "side": "left" } }

{ "op": "video", "format": "jpeg", "path": "/run/frame-1.jpg", "width": 640, "height": 360, "seq": 1 }

video writes each frame to a temporary name and renames it, so a slow reader never sees a partial file. Plugin events (cliff, picked_up, low_battery, …) travel as event. There is no separate event message type.

Safety rules (requirements, not options)

A driver must stop every actuator itself, because the core dying must never leave motors running:

• when its stdin closes (EOF), then exit;
• when a motion command's deadline passes without a newer command (its own timer to deadline_ms);
• when it has not heard a heartbeat for 1 s (and refuse motion with no_heartbeat until they return);
• on stop, estop and a halt command;
• when its own sensors say so (for example a cliff or a pick-up).

stop must be idempotent and safe at any time, including before setup. estop latches: keep refusing motion until the core sends resume. The e-stop is the owner's and is latched across reconnects by Bot; a driver never clears it.

A stop on the robot itself. If your robot or bridge has its own stop (a button, a hotkey, a sensor you trust), send {"op": "local_stop", "reason": "…"} (Python: self.local_stop("reason")), or run openvibe-node stop from a terminal. That sets the Node's local latch: every actuator stops within a quarter second, the panel shows the robot as stopped on the robot, and the owner's remote clear cannot lift it. Only openvibe-node resume on that machine releases it, and the owner then clears the panel's e-stop. It works even when the Node is not answering and survives restarts. Handlers must not block — slow work (speech synthesis, file I/O) goes on a thread.

A minimal driver in Python

The Python runtime (openvibe_plugin, plugins/sdk) implements every safety rule above, so a driver only says how to move its hardware. motion_kinds is the set of kinds the runtime stops at their deadline.

from openvibe_plugin import Fault, Plugin, clamp, run

class Arm(Plugin):
    driver, version = "my_arm", "0.1.0"
    motion_kinds = frozenset({"drive", "actuator"})   # stop these at their deadline

    def describe(self, config):
        return {"actuator": {"names": ["shoulder", "gripper"]}}

    def setup(self, config):
        self.bus = open_serial(config["port"])           # every actuator stopped when this returns

    def handle(self, cmd):
        if cmd.kind != "actuator":
            raise Fault("unsupported")
        self.bus.speed(cmd.value["name"], clamp(cmd.value.get("value"), -1, 1))

    def stop(self):
        if getattr(self, "bus", None):
            self.bus.stop_all()

if __name__ == "__main__":
    raise SystemExit(run(Arm()))

Test it with openvibe_plugin.Runtime(plugin, infile, outfile, clock) and a fake clock, as the bundled drivers do.

The same driver in plain stdin/stdout

Any language: read one JSON object per line, write one per line, and keep your own stop timers. This is the same arm, without the SDK.

#!/usr/bin/env python3
import json, sys, threading, time

last = time.monotonic()          # when the last heartbeat (or fresh motion command) arrived
latched = False

def out(message):
    sys.stdout.write(json.dumps(message) + "\n")
    sys.stdout.flush()

def stop():
    pass                         # stop every actuator here: idempotent, safe even before setup

def watchdog():
    while True:
        time.sleep(0.1)
        if time.monotonic() - last > 1.0:   # the deadman: no heartbeat for 1 s → stop
            stop()

threading.Thread(target=watchdog, daemon=True).start()

for line in sys.stdin:                    # EOF means the core is gone: stop and exit
    m = json.loads(line)
    if m["op"] == "hello":
        out({"op": "describe", "driver": "my_arm", "version": "0.1.0", "protocol": 1,
             "capabilities": {"actuator": {"names": ["shoulder", "gripper"]}},
             "motion_kinds": ["drive", "actuator"]})
        if not m.get("probe"):
            out({"op": "ready"})          # open the bus with every actuator stopped, then say ready
    elif m["op"] == "command":
        if latched or m["kind"] != "actuator":
            out({"op": "nack", "id": m["id"], "fault_code": "unsupported", "message": "only actuators"})
            continue
        move(m["value"]["name"], m["value"].get("value"))   # clamp and set your own deadline_ms timer
        last = time.monotonic()
        out({"op": "ack", "id": m["id"]})
    elif m["op"] == "heartbeat":
        last = time.monotonic()
    elif m["op"] == "stop":
        stop()
    elif m["op"] == "estop":
        latched = True
        stop()
    elif m["op"] == "resume":
        latched = False
stop()

Test it

With the dryrun driver (plugins/dryrun): it accepts every kind, logs it, reports simulated battery and distance, and asks the core for the test-pattern camera. Add it to your config.json plugins and set config.record: "<path>" to append one JSON line per setup, command and stop — end-to-end tests read that file to check what reached the "robot".

{ "plugins": [ { "name": "dryrun", "config": { "record": "/tmp/dryrun.jsonl" } } ] }

With the simulated rover on this site: add a robot and pick the Simulated rover profile (Your robots). Bot runs it in-process, so the panel renders from the profile and drives with no hardware and no Node. Use it to see exactly what your profile shows before wiring a driver.

Contribute it

Open a pull request to OpenVibers/OpenVibe.Node adding your driver under plugins/, with the robot profile it pairs with. The profile is validated by Bot's loader (bot.robot-profile@1 plus the capability, driver, widget and command registries), and the driver is reviewed before it runs on anyone's Node. A genuinely new capability, driver or widget name is one line in Bot's registries — never a schema change.