openvibers/openvibe.link Commits

Node: the relay plugin forwards button and point commands to an owner's script in Live's hardware-client messages (plan T15 R9 step 4) (#33)

OP OpenVibers committed 7 Oct 2026 · feba14a7e4 · 16 files, +634 −17

## Changes (file:line)

- `plugins/relay/openvibe_relay/__init__.py` — new plugin: `live_message()` mapping (:20), stdlib WebSocket server bound to 127.0.0.1 (:63–165), `Relay` lifecycle, child supervisor and held-button release (:167–303).
- `plugins/relay/openvibe_relay/__main__.py:1` — `python3 -m openvibe_relay` entry point.
- `plugins/relay/pyproject.toml:1` — `openvibe-relay` package, depends on `openvibe-plugin` (same as dryrun).
- `plugins/relay/tests/test_relay.py` — 10 tests (:76,80,89,110,125,140,169).
- `internal/protocol/protocol.go:156–164` — `KindButton`/`KindPoint` in `Kinds` and `GuardedKinds`.
- `internal/safety/limits.go:146–161` — core clamp/validate: button needs name, state `down`/`up`; point x/y numbers in 0..1.
- `internal/safety/safety_test.go:109–135` — `TestClamp` now covers button/point (valid + bad name/state/range).
- `internal/plugins/manager.go:18–19` — `capabilityFor` routes `button`/`point` to the declaring plugin.
- `internal/config/config.go:338` — `Bundled["relay"] = "openvibe_relay"`.
- `install/install.sh:301` — installer pip-installs the relay package.
- `scripts/dist.sh:27` — plugin bundle includes `relay`.
- `README.md:30,44` — plugin list and test loop.
- `docs/plugins.md:146–178` — "relay" section (both outputs, shapes, one-line change).
- `docs/protocol.md:198–199,207,222–227` — kinds table and command rules.
- `.github/workflows/ci.yml:92,94` — relay tests + package install in the Python matrix.
- `.github/workflows/release.yml:26` — relay tests in the release job.

## Live's exact message shapes (and where I read them)

- Press: `{"type":"command","command":name}` — `server/controls/control-server.js:360`, dispatched by `control-bridge-example-cozmo.py:148`.
- Hold: `{"type":"key_down","command":name}` / `{"type":"key_up","command":name}` — `control-server.js:475`, dispatched at `control-bridge-example-cozmo.py:150,153`.
- Point: `{"type":"video_click","x":…,"y":…}` rounded to 4 dp — `control-server.js:527` (+`:528–529` rounding), dispatched at `control-bridge-example-cozmo.py:156`.
- Extra fields `control_id`/`from_user`/`timestamp` are omitted: unavailable in Bot's command, and the example bridge reads `from_user` only for logging (defaults it).

## Tests

- `~/.local/bin/pytest -q plugins/relay/tests` → **10 passed** (the WebSocket test really ran — it bound 127.0.0.1, not skipped).
- `go vet` + `go test` on the packages my change reaches: `internal/protocol`, `internal/safety`, `internal/plugins`, `internal/config`, `internal/node`, `internal/link`, `cmd/openvibe-node` → **all ok**.

Changes

ddiff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
iindex fc52c3a..e6ec239 100644
---- a/.github/workflows/ci.yml
++++ b/.github/workflows/ci.yml
@@@ -89,8 +89,9 @@ jobs:
- run: python -m pytest -q plugins/dryrun/tests
- run: python -m pytest -q plugins/adeept_adr036/tests
- run: python -m pytest -q plugins/cozmo/tests
++ - run: python -m pytest -q plugins/relay/tests
- name: packages install
-- run: pip install ./plugins/sdk ./plugins/dryrun ./plugins/adeept_adr036 ./plugins/cozmo
++ run: pip install ./plugins/sdk ./plugins/dryrun ./plugins/adeept_adr036 ./plugins/cozmo ./plugins/relay
shell:
name: shellcheck
ddiff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
iindex 2121d27..ee05ffd 100644
---- a/.github/workflows/release.yml
++++ b/.github/workflows/release.yml
@@@ -23,7 +23,7 @@ jobs:
- run: sudo sysctl -w kernel.apparmor_restrict_unprivileged_userns=0 || true
- run: go vet ./... && go test -count=1 ./...
- run: |
-- for p in sdk dryrun adeept_adr036 cozmo; do python -m pytest -q plugins/$p/tests; done
++ for p in sdk dryrun adeept_adr036 cozmo relay; do python -m pytest -q plugins/$p/tests; done
- run: scripts/dist.sh "${GITHUB_REF_NAME}"
# Every release must carry the installer OpenVibe.Bot's GET /install route and BOT_INSTALLER_SOURCE_URL are
# documented to serve (releases/latest/download/install.sh), the plugin bundle the installer downloads, and the
ddiff --git a/README.md b/README.md
iindex c22815f..657efd2 100644
---- a/README.md
++++ b/README.md
@@@ -26,9 +26,9 @@ openvibe-node resume
clamped before any driver sees a value, a persisted e-stop, the local kill switch), the plugin supervisor, WHIP video
publishing with pion, and the system service (systemd, launchd, Windows).
- **Plugins** (separate processes, JSON lines over stdin/stdout): [`dryrun`](plugins/dryrun), the Adeept 4WD Smart Car
-- Kit [`adeept_adr036`](plugins/adeept_adr036/README.md), and [`cozmo`](plugins/cozmo/README.md) through PyCozmo. Each
-- plugin stops its own actuators when stdin closes, at a command's deadline and after 1 s without a heartbeat, so the
-- core dying never leaves motors running.
++ Kit [`adeept_adr036`](plugins/adeept_adr036/README.md), [`cozmo`](plugins/cozmo/README.md) through PyCozmo, and
++ [`relay`](plugins/relay) for existing Live hardware clients. Each plugin stops its own actuators when stdin closes,
++ at a command's deadline and after 1 s without a heartbeat, so the core dying never leaves motors running.
- **ESP32 library** ([`esp32/`](esp32/README.md), `OpenVibeNode`): the device side of the protocol for ESP32 boards —
pairing, the control link, drive/actuator commands, telemetry, heartbeat, deadman and e-stop — but no jobs or video.
@@@ -41,7 +41,7 @@ Decisions: [ADR-043](docs/ADR-043-bot-devices-and-control.md).
```sh
go vet ./... && go test -race ./... # the end-to-end tests run the Python dry-run plugin
pip install pytest pillow numpy
--for p in sdk dryrun adeept_adr036 cozmo; do python3 -m pytest -q plugins/$p/tests; done
++for p in sdk dryrun adeept_adr036 cozmo relay; do python3 -m pytest -q plugins/$p/tests; done
scripts/dist.sh v0.0.0-dev # every binary, the plugin bundle and SHA256SUMS in dist/
```
ddiff --git a/docs/plugins.md b/docs/plugins.md
iindex da93877..afc5e34 100644
---- a/docs/plugins.md
++++ b/docs/plugins.md
@@@ -5,7 +5,7 @@ object per line, UTF-8). Its stderr goes to the Node's log. The core never links
in any language; the robot drivers are Python 3 because the kit libraries and PyCozmo are.
Bundled plugins: [`dryrun`](../plugins/dryrun), [`adeept_adr036`](../plugins/adeept_adr036/README.md),
--[`cozmo`](../plugins/cozmo/README.md). Python plugins use the runtime in [`plugins/sdk`](../plugins/sdk) (package
++[`cozmo`](../plugins/cozmo/README.md), [`relay`](../plugins/relay). Python plugins use the runtime in [`plugins/sdk`](../plugins/sdk) (package
`openvibe_plugin`), which implements everything in "Safety rules" below, so a driver only says how to move its
hardware.
@@@ -104,6 +104,8 @@ the panel (ADR-043 decision 7: names are open strings checked against a registry
| `actuator` | `{"names": ["lift", …] \| ["any"], "ranges": {…}}` | `actuator` |
| `say` | `{…}` | `say` |
| `display` | `{"width", "height", "text", "faces", "image"}` | `display` |
++| `button` | `{…}` | `button` |
++| `point` | `{…}` | `point` |
| `camera` | `{"h264_command": [argv]}` (the core runs it), `{"jpeg": true}` (frames by `video`), or `{"source": "test_pattern"}` | video |
| `battery`, `sensors`, `lights`, `bridge` | informational | — |
@@@ -141,6 +143,42 @@ camera frames (320 × 240) as JPEG files. A cliff or pick-up stops the wheels in
event; while at a cliff only backing away is allowed, while picked up nothing drives. Firmware other than PyCozmo's
supported version (2381, `allow_firmware`) is reported as `firmware_unsupported`.
++### `relay` — existing OpenVibe.Live hardware clients
++
++Use this when the owner already has a local script receiving Live stream controls. In `config.json`, replace the
++`dryrun` plugin entry with one of these (the release bundle and installer include `relay`):
++
++```json
++{"name": "relay", "config": {"output": "child", "argv": ["python3", "/home/pi/controls.py", "--stdin"]}}
++```
++
++The child receives one JSON object per line on **its stdin**. `argv` is passed directly to the OS, never a shell
++command. If it exits, the relay restarts it with backoff. On Node shutdown, the relay closes its stdin and stops
++the process. A script that already uses Live's WebSocket protocol may instead use:
++
++```json
++{"name": "relay", "config": {"output": "websocket", "port": 8765}}
++```
++
++It listens only on `127.0.0.1` (default port 8765) and broadcasts to every connected client. The one-line change for
++an existing Pi script is its Live socket URL — for Live's own example
++[`control-bridge-example-cozmo.py`](https://github.com/OpenVibers/OpenVibe.Live/blob/main/control-bridge-example-cozmo.py)
++that is line 27, `WS_URL = "wss://openvibe.live/ws/control?mode=hardware&stream_key=YOUR_STREAM_KEY"` →
++`ws://127.0.0.1:8765/ws/control?mode=hardware`; no stream key is needed locally, and the relay ignores the path and
++query. The WebSocket server uses only Python's standard library. Clients may send nothing; text messages are ignored
++and protocol pings receive pongs.
++
++Both outputs carry the hardware-facing Live fields exactly as Live's hardware socket sends them
++(`server/controls/control-server.js:360,475,527`, dispatched by `control-bridge-example-cozmo.py:148-156`):
++`{"type":"command","command":"forward"}` for a press, `{"type":"key_down","command":"forward"}` /
++`{"type":"key_up","command":"forward"}` for a hold, and `{"type":"video_click","x":0.25,"y":0.75}` for a point
++(rounded to four decimal places, as Live does). The Bot button `name` is the old Live button's **command string**,
++not its label. Live's `key_held` / `key_released` names are viewer status broadcasts on its control socket, never the
++hardware messages; the hardware socket, and so this relay, sends `key_down` / `key_up`. The old `control_id`,
++`from_user` and `timestamp` fields are unavailable in Bot's Node → plugin command; Live's example bridge reads
++`from_user` only for logging and defaults it, so hardware clients treat all three as optional. The SDK releases all
++held buttons on deadline, missed heartbeat, stop, e-stop, halt and stdin close.
++
## Writing a plugin in Python
```python
ddiff --git a/docs/protocol.md b/docs/protocol.md
iindex 3f3616d..b9b245e 100644
---- a/docs/protocol.md
++++ b/docs/protocol.md
@@@ -195,14 +195,16 @@ at most every 500 ms (≤ 2 Hz, Bot's cap). There is no separate `event` message
| `actuator` | `{"name": "lift", "value": …}` — number, object or bool per actuator ([plugins.md](plugins.md)) | guarded |
| `say` | `{"text": "…"}` (1–1000 bytes; plugins may cap lower) | no |
| `display` | `{"text": "…"}`, `{"face": "happy"}` or `{"image_png_b64": "…"}` | no |
++| `button` | `{"name": "forward"}` press; `{"name": …, "state": "down"\|"up"}` hold (names from the robot's profile) | guarded |
++| `point` | `{"x": 0..1, "y": 0..1}` normalised click on the video, for the [`relay`](plugins.md) plugin | guarded |
| `halt` | `{}` — stop every actuator now (not latched); always accepted | — |
Sign conventions: `throttle`/`x` positive = forward; `steer` positive = turn right; `y` positive = strafe left;
`rotation` positive = counter-clockwise (left); `pan` positive = right, `tilt` positive = up (plugins can invert a
servo in their config to make the hardware match).
--`deadline_ms` is an **absolute** instant in Unix milliseconds on the server's clock, present on `drive`, `actuator` and
--`ptz`. A device's clock may be minutes off, so the Node never compares it with its own clock: it estimates the
++`deadline_ms` is an **absolute** instant in Unix milliseconds on the server's clock, present on `drive`, `actuator`,
++`ptz` and `button` (a held button is re-sent, so its deadline releases the hold — see the [`relay`](plugins.md) plugin). A device's clock may be minutes off, so the Node never compares it with its own clock: it estimates the
server's clock from the `ts` of every frame on the connection (the largest `ts` − receive time: the skew less the
fastest transit, kept for the whole connection so a run of delayed frames cannot reopen an expired deadline) and
compares `deadline_ms` with that at receipt. A command that arrives at or after its deadline →
@@@ -217,11 +219,13 @@ What the Node does with a command, in order:
2. Unknown kind → `nack unsupported`.
3. `halt` → every plugin stops, `ack`.
4. No `config` yet on this connection → `nack not_ready`. Kind not in `config.allowed_commands` → `nack not_allowed`.
--5. `drive`, `ptz`, `actuator` while the local kill switch is latched → `nack local_stop`; while the remote e-stop is
-- latched → `nack estopped`; past its `deadline_ms` → `nack expired`.
++5. `drive`, `ptz`, `actuator`, `button` and `point` while the local kill switch is latched → `nack local_stop`; while
++ the remote e-stop is latched → `nack estopped`; past its `deadline_ms` → `nack expired`.
6. The value is clamped in the core, before any plugin sees it: `throttle`, `x`, `y` to ±`max_speed`; `steer`,
-- `rotation` to ±`max_turn`; `pan`, `tilt`, `zoom` and a numeric actuator `value` to ±1. The limits are the stricter of
-- the server's `config.limits` and the local config's `limits`. Non-numbers → `nack bad_value`.
++ `rotation` to ±`max_turn`; `pan`, `tilt`, `zoom` and a numeric actuator `value` to ±1; a `point`'s `x`/`y` must be
++ numbers in 0..1. The limits are the stricter of the server's `config.limits` and the local config's `limits`.
++ Non-numbers, a `button` without a `name` or with a `state` other than `down`/`up`, or a `point` out of range →
++ `nack bad_value`.
7. It goes to the plugin named by `target`, or the first plugin whose `describe` declares the kind (for actuators, the
first listing the name). None → `nack unsupported`.
8. The plugin's `ack`/`nack` is relayed. No answer within max(deadline, 1 s) → the plugin is told to stop and the
ddiff --git a/install/install.sh b/install/install.sh
iindex bb9335d..70e1493 100755
---- a/install/install.sh
++++ b/install/install.sh
@@@ -298,7 +298,7 @@ if [ ! -x "$STATE/venv/bin/python" ]; then
fi
PIP="$STATE/venv/bin/pip"
$SUDO "$PIP" install --quiet --upgrade pip >/dev/null 2>&1 || true
--$SUDO "$PIP" install --quiet "$STATE/plugins/sdk" "$STATE/plugins/dryrun"
++$SUDO "$PIP" install --quiet "$STATE/plugins/sdk" "$STATE/plugins/dryrun" "$STATE/plugins/relay"
PLUGINS='[{"name": "dryrun"}]'
KIND=onboard
VIDEO='"source": "auto"'
ddiff --git a/internal/config/config.go b/internal/config/config.go
iindex 5255d6a..11b6e6d 100644
---- a/internal/config/config.go
++++ b/internal/config/config.go
@@@ -335,6 +335,7 @@ var Bundled = map[string]string{
"dryrun": "openvibe_dryrun",
"adeept_adr036": "openvibe_adeept_adr036",
"cozmo": "openvibe_cozmo",
++ "relay": "openvibe_relay",
}
// Default is the config written by `openvibe-node pair` when none exists: the dry-run plugin and its test pattern.
ddiff --git a/internal/plugins/manager.go b/internal/plugins/manager.go
iindex 0f516e2..2144199 100644
---- a/internal/plugins/manager.go
++++ b/internal/plugins/manager.go
@@@ -15,6 +15,8 @@ var capabilityFor = map[string]string{
protocol.KindActuator: "actuator",
protocol.KindSay: "say",
protocol.KindDisplay: "display",
++ protocol.KindButton: "button",
++ protocol.KindPoint: "point",
}
// Manager supervises every configured plugin.
ddiff --git a/internal/protocol/protocol.go b/internal/protocol/protocol.go
iindex 73b60b2..4572c02 100644
---- a/internal/protocol/protocol.go
++++ b/internal/protocol/protocol.go
@@@ -153,13 +153,15 @@ const (
KindSay = "say"
KindDisplay = "display"
KindHalt = "halt"
++ KindButton = "button"
++ KindPoint = "point"
)
// Kinds lists every command kind the Node understands.
--var Kinds = []string{KindDrive, KindActuator, KindPTZ, KindSay, KindDisplay, KindHalt}
++var Kinds = []string{KindDrive, KindActuator, KindPTZ, KindSay, KindDisplay, KindHalt, KindButton, KindPoint}
// GuardedKinds move something and are refused while an e-stop or the local kill switch is latched.
--var GuardedKinds = map[string]bool{KindDrive: true, KindActuator: true, KindPTZ: true}
++var GuardedKinds = map[string]bool{KindDrive: true, KindActuator: true, KindPTZ: true, KindButton: true, KindPoint: true}
// Fault codes carried in nack.fault_code and status.faults[].code.
const (
ddiff --git a/internal/safety/limits.go b/internal/safety/limits.go
iindex 5c14582..cdba81c 100644
---- a/internal/safety/limits.go
++++ b/internal/safety/limits.go
@@@ -143,6 +143,22 @@ func (l Limits) Clamp(kind string, value json.RawMessage) (json.RawMessage, erro
if s, ok := v["text"].(string); !ok || s == "" || len(s) > 1000 {
return nil, badValue("say needs text (1 to 1000 bytes)")
}
++ case protocol.KindButton:
++ name, ok := v["name"].(string)
++ if !ok || name == "" {
++ return nil, badValue("button needs a name")
++ }
++ if state, present := v["state"]; present && state != "down" && state != "up" {
++ return nil, badValue("button state must be down or up")
++ }
++ case protocol.KindPoint:
++ for _, key := range []string{"x", "y"} {
++ f, err := num(key, v[key])
++ if err != nil || f < 0 || f > 1 {
++ return nil, badValue("point %s must be a number from 0 to 1", key)
++ }
++ v[key] = f
++ }
}
out, err := json.Marshal(v)
if err != nil {
ddiff --git a/internal/safety/safety_test.go b/internal/safety/safety_test.go
iindex c8d92d6..2d407fe 100644
---- a/internal/safety/safety_test.go
++++ b/internal/safety/safety_test.go
@@@ -106,16 +106,32 @@ func TestClamp(t *testing.T) {
if v["pan"] != 1 || v["tilt"] != -0.5 {
t.Fatalf("%s", out)
}
++ out, _ = l.Clamp("button", json.RawMessage(`{"name":"forward","state":"down"}`))
++ var b map[string]any
++ json.Unmarshal(out, &b)
++ if b["name"] != "forward" || b["state"] != "down" {
++ t.Fatalf("%s", out)
++ }
++ out, _ = l.Clamp("point", json.RawMessage(`{"x":0.25,"y":1}`))
++ json.Unmarshal(out, &v)
++ if v["x"] != 0.25 || v["y"] != 1 {
++ t.Fatalf("%s", out)
++ }
for kind, bad := range map[string]string{
"drive": `{"throttle":"fast"}`,
"ptz": `{"pan":null}`,
"actuator": `{"value":1}`,
"say": `{}`,
++ "button": `{"state":"down"}`,
++ "point": `{"x":2,"y":0}`,
} {
if _, err := l.Clamp(kind, json.RawMessage(bad)); err == nil {
t.Errorf("%s %s accepted", kind, bad)
}
}
++ if _, err := l.Clamp("button", json.RawMessage(`{"name":"forward","state":"sideways"}`)); err == nil {
++ t.Error("bad button state accepted")
++ }
if _, err := l.Clamp("drive", json.RawMessage(`[1]`)); err == nil {
t.Error("array accepted")
}
ddiff --git a/plugins/relay/openvibe_relay/__init__.py b/plugins/relay/openvibe_relay/__init__.py
nnew file mode 100644
iindex 0000000..b55b3a9
---- /dev/null
++++ b/plugins/relay/openvibe_relay/__init__.py
@@@ -0,0 +1,310 @@
++"""Relay Bot buttons and points to a local OpenVibe.Live hardware client."""
++
++import base64
++import hashlib
++import json
++import queue
++import socket
++import socketserver
++import subprocess
++import sys
++import threading
++import time
++
++from openvibe_plugin import Fault, Plugin
++
++__version__ = "0.1.0"
++_GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
++
++
++def live_message(kind, value):
++ """Return the hardware-facing Live fields available in a Bot command.
++
++ Live's hardware socket sends ``{type:'command', command}`` for a press, ``key_down`` / ``key_up`` for a hold and
++ ``video_click`` with ``x`` / ``y`` (server/controls/control-server.js:360,475,527), and its example Pi bridge
++ dispatches on exactly those names (control-bridge-example-cozmo.py:148-156). ``key_held`` / ``key_released`` are
++ the names Live broadcasts to *viewers*, never to the hardware; using them here would break existing scripts.
++ """
++ if not isinstance(value, dict):
++ raise Fault("bad_value", "value must be an object")
++ if kind == "button":
++ name = value.get("name")
++ if not isinstance(name, str) or not name:
++ raise Fault("bad_value", "button needs a name")
++ if "state" not in value:
++ return {"type": "command", "command": name}
++ state = value["state"]
++ if state == "down":
++ return {"type": "key_down", "command": name}
++ if state == "up":
++ return {"type": "key_up", "command": name}
++ raise Fault("bad_value", "button state must be down or up")
++ if kind == "point":
++ coords = {}
++ for key in ("x", "y"):
++ number = value.get(key)
++ if isinstance(number, bool) or not isinstance(number, (int, float)) or not 0 <= number <= 1:
++ raise Fault("bad_value", "point %s must be from 0 to 1" % key)
++ coords[key] = round(number, 4)
++ return {"type": "video_click", **coords}
++ raise Fault("unsupported", "kind %s not supported" % kind)
++
++
++def _read_exact(sock, size):
++ result = bytearray()
++ while len(result) < size:
++ chunk = sock.recv(size - len(result))
++ if not chunk:
++ raise EOFError
++ result.extend(chunk)
++ return bytes(result)
++
++
++class _WebSocketHandler(socketserver.BaseRequestHandler):
++ def handle(self):
++ sock = self.request
++ sock.settimeout(1)
++ try:
++ request = bytearray()
++ while b"\r\n\r\n" not in request and len(request) < 8192:
++ request.extend(_read_exact(sock, 1))
++ lines = request.decode("latin1").split("\r\n")
++ headers = {}
++ for line in lines[1:]:
++ if ":" in line:
++ key, val = line.split(":", 1)
++ headers[key.lower()] = val.strip()
++ key = headers.get("sec-websocket-key", "")
++ if not lines[0].startswith("GET ") or headers.get("sec-websocket-version") != "13" or not key:
++ return
++ accept = base64.b64encode(hashlib.sha1((key + _GUID).encode("ascii")).digest()).decode("ascii")
++ sock.sendall(("HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\n"
++ "Connection: Upgrade\r\nSec-WebSocket-Accept: %s\r\n\r\n" % accept).encode("ascii"))
++ with self.server.clients_lock:
++ self.server.clients[sock] = threading.Lock()
++ while not self.server.closing.is_set():
++ try:
++ first, second = _read_exact(sock, 2)
++ except (TimeoutError, socket.timeout):
++ # socket.timeout is not TimeoutError before Python 3.10; a client that only receives (the
++ # normal Pi case) sits idle between commands, so this must not drop the connection.
++ continue
++ opcode = first & 0x0f
++ length = second & 0x7f
++ if not second & 0x80 or opcode not in (0x1, 0x2, 0x8, 0x9, 0xA):
++ break
++ if length == 126:
++ length = int.from_bytes(_read_exact(sock, 2), "big")
++ elif length == 127:
++ length = int.from_bytes(_read_exact(sock, 8), "big")
++ if length > 65536:
++ break
++ mask = _read_exact(sock, 4)
++ payload = _read_exact(sock, length)
++ if opcode == 0x8:
++ break
++ if opcode == 0x9:
++ payload = bytes(b ^ mask[i % 4] for i, b in enumerate(payload))
++ with self.server.clients_lock:
++ lock = self.server.clients.get(sock)
++ if lock:
++ with lock:
++ sock.sendall(_frame(0xA, payload))
++ except (EOFError, OSError, ValueError, UnicodeError):
++ pass
++ finally:
++ with self.server.clients_lock:
++ self.server.clients.pop(sock, None)
++
++
++def _frame(opcode, payload):
++ size = len(payload)
++ if size < 126:
++ prefix = bytes((0x80 | opcode, size))
++ elif size < 65536:
++ prefix = bytes((0x80 | opcode, 126)) + size.to_bytes(2, "big")
++ else:
++ prefix = bytes((0x80 | opcode, 127)) + size.to_bytes(8, "big")
++ return prefix + payload
++
++
++class _WebSocketServer(socketserver.ThreadingTCPServer):
++ allow_reuse_address = True
++ daemon_threads = True
++
++ def __init__(self, port):
++ self.clients = {}
++ self.clients_lock = threading.Lock()
++ self.closing = threading.Event()
++ super().__init__(("127.0.0.1", port), _WebSocketHandler)
++
++ def broadcast(self, message):
++ frame = _frame(0x1, json.dumps(message, separators=(",", ":")).encode("utf-8"))
++ with self.clients_lock:
++ clients = list(self.clients.items())
++ for sock, lock in clients:
++ try:
++ with lock:
++ sock.sendall(frame)
++ except OSError:
++ with self.clients_lock:
++ self.clients.pop(sock, None)
++
++ def close(self):
++ self.closing.set()
++ self.shutdown()
++ self.server_close()
++ with self.clients_lock:
++ clients = list(self.clients)
++ self.clients.clear()
++ for sock in clients:
++ try:
++ sock.shutdown(2)
++ except OSError:
++ pass
++
++
++class Relay(Plugin):
++ driver = "relay"
++ version = __version__
++ # A held button must be released when its command deadline expires.
++ motion_kinds = frozenset({"button"})
++
++ def __init__(self):
++ self.held = set()
++ self.held_lock = threading.Lock()
++ self.server = None
++ self.server_thread = None
++ self.child = None
++ self.worker = None
++ self.queue = queue.Queue(maxsize=256)
++ self.closing = threading.Event()
++ self.argv = None
++
++ def describe(self, config):
++ return {"button": {}, "point": {}}
++
++ def setup(self, config):
++ output = config.get("output", "websocket")
++ if output == "websocket":
++ port = config.get("port", 8765)
++ if isinstance(port, bool) or not isinstance(port, int) or not 1 <= port <= 65535:
++ raise Fault("bad_value", "port must be from 1 to 65535")
++ try:
++ self.server = _WebSocketServer(port)
++ except OSError as exc:
++ raise Fault("not_connected", str(exc)) from exc
++ self.server_thread = threading.Thread(target=self.server.serve_forever, daemon=True)
++ self.server_thread.start()
++ elif output == "child":
++ self.argv = config.get("argv")
++ if not isinstance(self.argv, list) or not self.argv or not all(isinstance(a, str) and a for a in self.argv):
++ raise Fault("bad_value", "child output needs a nonempty argv array")
++ self._start_child()
++ self.worker = threading.Thread(target=self._run_child, daemon=True)
++ self.worker.start()
++ else:
++ raise Fault("bad_value", "output must be websocket or child")
++
++ def _start_child(self):
++ try:
++ self.child = subprocess.Popen(self.argv, stdin=subprocess.PIPE, stdout=sys.stderr, stderr=sys.stderr)
++ except OSError as exc:
++ raise Fault("not_connected", str(exc)) from exc
++
++ def _run_child(self):
++ delay = 0.1
++ while not self.closing.is_set():
++ if self.child.poll() is not None:
++ self.log("relay child exited (%s); restarting", self.child.returncode)
++ try:
++ self.child.stdin.close()
++ except OSError:
++ pass
++ # Do not replay commands buffered while the child was absent.
++ while True:
++ try:
++ self.queue.get_nowait()
++ except queue.Empty:
++ break
++ if self.closing.wait(delay):
++ break
++ try:
++ self._start_child()
++ delay = min(delay * 2, 5.0)
++ except Fault as exc:
++ self.log("relay child restart failed: %s", exc)
++ continue
++ try:
++ message = self.queue.get(timeout=0.05)
++ except queue.Empty:
++ continue
++ try:
++ self.child.stdin.write((json.dumps(message, separators=(",", ":")) + "\n").encode("utf-8"))
++ self.child.stdin.flush()
++ except (BrokenPipeError, OSError, ValueError):
++ pass
++
++ def _send(self, message):
++ if self.server is not None:
++ self.server.broadcast(message)
++ elif self.child is not None and self.child.poll() is None and not self.closing.is_set():
++ try:
++ self.queue.put_nowait(message)
++ except queue.Full as exc:
++ raise Fault("not_connected", "relay child is not reading") from exc
++ else:
++ raise Fault("not_connected", "relay output is unavailable")
++
++ def handle(self, cmd):
++ message = live_message(cmd.kind, cmd.value)
++ if cmd.kind == "button" and "state" in cmd.value:
++ name = cmd.value["name"]
++ with self.held_lock:
++ if cmd.value["state"] == "down":
++ self._send(message)
++ self.held.add(name)
++ else:
++ self._send(message)
++ self.held.discard(name)
++ else:
++ self._send(message)
++
++ def stop_motion(self):
++ self.stop()
++
++ def stop(self):
++ with self.held_lock:
++ names = sorted(self.held)
++ self.held.clear()
++ for name in names:
++ try:
++ self._send({"type": "key_up", "command": name})
++ except Fault:
++ pass
++
++ def close(self):
++ if self.server is not None:
++ self.server.close()
++ self.server = None
++ if self.worker is not None:
++ # Give queued safety releases a brief chance to reach the child.
++ end = time.monotonic() + 0.5
++ while not self.queue.empty() and time.monotonic() < end:
++ time.sleep(0.01)
++ self.closing.set()
++ self.worker.join(timeout=0.5)
++ if self.child is not None:
++ try:
++ self.child.stdin.close()
++ except OSError:
++ pass
++ try:
++ self.child.wait(timeout=0.5)
++ except subprocess.TimeoutExpired:
++ self.child.terminate()
++ try:
++ self.child.wait(timeout=0.5)
++ except subprocess.TimeoutExpired:
++ self.child.kill()
++ self.child.wait()
ddiff --git a/plugins/relay/openvibe_relay/__main__.py b/plugins/relay/openvibe_relay/__main__.py
nnew file mode 100644
iindex 0000000..ae65951
---- /dev/null
++++ b/plugins/relay/openvibe_relay/__main__.py
@@@ -0,0 +1,6 @@
++import sys
++
++from openvibe_plugin import run
++from openvibe_relay import Relay
++
++sys.exit(run(Relay()))
ddiff --git a/plugins/relay/pyproject.toml b/plugins/relay/pyproject.toml
nnew file mode 100644
iindex 0000000..970b95b
---- /dev/null
++++ b/plugins/relay/pyproject.toml
@@@ -0,0 +1,14 @@
++[build-system]
++requires = ["setuptools>=61"]
++build-backend = "setuptools.build_meta"
++
++[project]
++name = "openvibe-relay"
++version = "0.1.0"
++description = "OpenVibe Node relay for existing Live hardware clients"
++requires-python = ">=3.9"
++license = { text = "AGPL-3.0-or-later" }
++dependencies = ["openvibe-plugin"]
++
++[tool.setuptools]
++packages = ["openvibe_relay"]
ddiff --git a/plugins/relay/tests/test_relay.py b/plugins/relay/tests/test_relay.py
nnew file mode 100644
iindex 0000000..fd6f7f2
---- /dev/null
++++ b/plugins/relay/tests/test_relay.py
@@@ -0,0 +1,207 @@
++import base64
++import io
++import json
++import os
++import socket
++import sys
++import time
++
++import pytest
++
++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_plugin import Fault, Runtime # noqa: E402
++from openvibe_relay import Relay, live_message # noqa: E402
++
++
++class Clock:
++ now = 10.0
++
++ def __call__(self):
++ return self.now
++
++
++def wait_for(predicate, timeout=3):
++ end = time.monotonic() + timeout
++ while time.monotonic() < end:
++ if predicate():
++ return
++ time.sleep(0.01)
++ assert predicate()
++
++
++def start(config):
++ p = Relay()
++ out = io.StringIO()
++ clock = Clock()
++ rt = Runtime(p, infile=io.StringIO(""), outfile=out, clock=clock)
++ rt.dispatch({"op": "hello", "config": config})
++ assert json.loads(out.getvalue().splitlines()[-1]) == {"op": "ready"}
++ return p, rt, out, clock
++
++
++def command(rt, out, kind, value, cid="c", deadline=300):
++ rt.dispatch({"op": "command", "id": cid, "kind": kind, "value": value, "deadline_ms": deadline})
++ return json.loads(out.getvalue().splitlines()[-1])
++
++
++def read_lines(path):
++ try:
++ with open(path) as f:
++ return [json.loads(line) for line in f]
++ except FileNotFoundError:
++ return []
++
++
++def child_config(path, restart=False):
++ code = ("import sys,json,os\n"
++ "p=sys.argv[1]\n"
++ "for line in sys.stdin:\n"
++ " with open(p,'a') as f: f.write(line)\n"
++ " if len(open(p).readlines()) == 1 and len(sys.argv)>2: break\n")
++ argv = [sys.executable, "-c", code, str(path)]
++ if restart:
++ argv.append("once")
++ return {"output": "child", "argv": argv}
++
++
+[email protected]("kind,value,expected", [
++ ("button", {"name": "forward"}, {"type": "command", "command": "forward"}),
++ ("button", {"name": "forward", "state": "down"}, {"type": "key_down", "command": "forward"}),
++ ("button", {"name": "forward", "state": "up"}, {"type": "key_up", "command": "forward"}),
++ ("point", {"x": 0.123456, "y": 1}, {"type": "video_click", "x": 0.1235, "y": 1}),
++])
++def test_live_shapes(kind, value, expected):
++ assert live_message(kind, value) == expected
++
++
++def test_refuses_other_kinds_and_bad_values():
++ with pytest.raises(Fault) as error:
++ live_message("drive", {"throttle": 1})
++ assert error.value.code == "unsupported"
++ with pytest.raises(Fault) as error:
++ live_message("point", {"x": 2, "y": 0})
++ assert error.value.code == "bad_value"
++
++
++def test_child_output_and_eof_release(tmp_path):
++ path = tmp_path / "messages.jsonl"
++ p, rt, out, _ = start(child_config(path))
++ try:
++ assert command(rt, out, "button", {"name": "lights"}) == {"op": "ack", "id": "c"}
++ assert command(rt, out, "point", {"x": 0.25, "y": 0.75}) == {"op": "ack", "id": "c"}
++ assert command(rt, out, "button", {"name": "forward", "state": "down"}) == {"op": "ack", "id": "c"}
++ assert command(rt, out, "drive", {"throttle": 1})["fault_code"] == "unsupported"
++ rt.shutdown("eof")
++ wait_for(lambda: len(read_lines(path)) == 4)
++ assert read_lines(path) == [
++ {"type": "command", "command": "lights"},
++ {"type": "video_click", "x": 0.25, "y": 0.75},
++ {"type": "key_down", "command": "forward"},
++ {"type": "key_up", "command": "forward"},
++ ]
++ finally:
++ rt.shutdown()
++ p.close()
++
++
++def test_child_restarts_after_exit(tmp_path):
++ path = tmp_path / "messages.jsonl"
++ p, rt, out, _ = start(child_config(path, restart=True))
++ try:
++ first_pid = p.child.pid
++ assert command(rt, out, "button", {"name": "first"})["op"] == "ack"
++ wait_for(lambda: len(read_lines(path)) == 1)
++ wait_for(lambda: p.child is not None and p.child.pid != first_pid and p.child.poll() is None)
++ assert command(rt, out, "button", {"name": "second"})["op"] == "ack"
++ wait_for(lambda: len(read_lines(path)) == 2)
++ assert [x["command"] for x in read_lines(path)] == ["first", "second"]
++ finally:
++ rt.shutdown()
++
++
++def test_missed_heartbeat_releases_hold(tmp_path):
++ path = tmp_path / "messages.jsonl"
++ p, rt, out, clock = start(child_config(path))
++ try:
++ assert command(rt, out, "button", {"name": "forward", "state": "down"}, deadline=5000)["op"] == "ack"
++ wait_for(lambda: len(read_lines(path)) == 1)
++ clock.now += 1.1
++ rt.tick()
++ wait_for(lambda: len(read_lines(path)) == 2)
++ assert read_lines(path)[-1] == {"type": "key_up", "command": "forward"}
++ assert command(rt, out, "button", {"name": "forward", "state": "down"})["fault_code"] == "no_heartbeat"
++ finally:
++ rt.shutdown()
++
++
++def test_deadline_releases_hold(tmp_path):
++ path = tmp_path / "messages.jsonl"
++ p, rt, out, clock = start(child_config(path))
++ try:
++ assert command(rt, out, "button", {"name": "forward", "state": "down"})["op"] == "ack"
++ wait_for(lambda: len(read_lines(path)) == 1)
++ clock.now += 0.31
++ rt.tick()
++ wait_for(lambda: len(read_lines(path)) == 2)
++ assert read_lines(path)[-1] == {"type": "key_up", "command": "forward"}
++ finally:
++ rt.shutdown()
++
++
++def _recv_exact(sock, n):
++ data = b""
++ while len(data) < n:
++ data += sock.recv(n - len(data))
++ return data
++
++
++def _recv_ws(sock):
++ first, size = _recv_exact(sock, 2)
++ assert first == 0x81
++ if size == 126:
++ size = int.from_bytes(_recv_exact(sock, 2), "big")
++ return json.loads(_recv_exact(sock, size))
++
++
++def test_websocket_broadcast():
++ with socket.socket() as probe:
++ try:
++ probe.bind(("127.0.0.1", 0))
++ except OSError as exc:
++ pytest.skip("loopback bind unavailable: %s" % exc)
++ port = probe.getsockname()[1]
++ p, rt, out, _ = start({"output": "websocket", "port": port})
++ clients = []
++ try:
++ for _ in range(2):
++ sock = socket.create_connection(("127.0.0.1", port), timeout=2)
++ sock.settimeout(2)
++ key = base64.b64encode(os.urandom(16)).decode()
++ sock.sendall(("GET /ws/control?mode=hardware HTTP/1.1\r\nHost: localhost\r\n"
++ "Upgrade: websocket\r\nConnection: Upgrade\r\n"
++ "Sec-WebSocket-Version: 13\r\nSec-WebSocket-Key: %s\r\n\r\n" % key).encode())
++ response = b""
++ while b"\r\n\r\n" not in response:
++ response += sock.recv(1024)
++ assert b"101 Switching Protocols" in response
++ clients.append(sock)
++ wait_for(lambda: len(p.server.clients) == 2)
++ assert command(rt, out, "button", {"name": "horn"})["op"] == "ack"
++ assert [_recv_ws(c) for c in clients] == [
++ {"type": "command", "command": "horn"},
++ {"type": "command", "command": "horn"},
++ ]
++ assert command(rt, out, "button", {"name": "forward", "state": "down"})["op"] == "ack"
++ assert [_recv_ws(c)["type"] for c in clients] == ["key_down", "key_down"]
++ rt.shutdown("eof")
++ assert [_recv_ws(c) for c in clients] == [
++ {"type": "key_up", "command": "forward"},
++ {"type": "key_up", "command": "forward"},
++ ]
++ finally:
++ rt.shutdown()
++ for c in clients:
++ c.close()
ddiff --git a/scripts/dist.sh b/scripts/dist.sh
iindex a190ceb..367c07c 100755
---- a/scripts/dist.sh
++++ b/scripts/dist.sh
@@@ -24,7 +24,7 @@ done
# The plugin bundle: the Python packages without their tests.
tar -czf dist/openvibe-node-plugins.tar.gz -C plugins \
--exclude=tests --exclude=__pycache__ --exclude='*.egg-info' --exclude=.pytest_cache --exclude=build \
-- sdk dryrun adeept_adr036 cozmo
++ sdk dryrun adeept_adr036 cozmo relay
# OpenVibe.Bot's GET /install route and BOT_INSTALLER_SOURCE_URL are documented to serve this asset name:
# https://github.com/OpenVibers/OpenVibe.Node/releases/latest/download/install.sh — do not rename it.
cp install/install.sh dist/install.sh