openvibers/openvibe.link Commits
Node: ROS 2 robots (a bundled ros2 plugin: cmd_vel, battery, scan/range, odometry and camera, over rclpy or rosbridge); plugins may retry setup (#37)
Changes
@@@ -90,8 +90,9 @@ jobs:- 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++ - run: python -m pytest -q plugins/ros2/tests- name: packages install-- run: pip install ./plugins/sdk ./plugins/dryrun ./plugins/adeept_adr036 ./plugins/cozmo ./plugins/relay++ run: pip install ./plugins/sdk ./plugins/dryrun ./plugins/adeept_adr036 ./plugins/cozmo ./plugins/relay ./plugins/ros2shell:name: shellcheck@@@ -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 relay; do python -m pytest -q plugins/$p/tests; done++ for p in sdk dryrun adeept_adr036 cozmo relay ros2; 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@@@ -8,7 +8,7 @@ for AI agents) can use the device. It connects out, needs no open ports, and keecurl -fsSL https://openvibe.bot/install | sh -s -- --robot rob_… --code ABCD-1234 --driver adeept```--openvibe.bot shows this command with the robot's id and a fresh code. `--driver` is `adeept`, `adeept-mecanum`, `cozmo`,++openvibe.bot shows this command with the robot's id and a fresh code. `--driver` is `adeept`, `adeept-mecanum`, `cozmo`, `ros2`,`looi` (the community's Generic LOOI driver, [docs/hardware-looi.md](docs/hardware-looi.md)) or `none` (dry run; the old `--robot adeept` form still works but is deprecated). A robot paired throughOpenVibe.Network gets `--network <URL> --pairing pair_… --code ABCD-1234` instead of `--robot`. Details, manualinstall, Windows and configuration: [docs/install.md](docs/install.md).@@@ -27,7 +27,8 @@ openvibe-node resumepublishing with pion, and the system service (systemd, launchd, Windows).- **Plugins** (separate processes, JSON lines over stdin/stdout): [`dryrun`](plugins/dryrun), the Adeept 4WD Smart CarKit [`adeept_adr036`](plugins/adeept_adr036/README.md), [`cozmo`](plugins/cozmo/README.md) through PyCozmo, and-- [`relay`](plugins/relay) for existing Live hardware clients. LOOI's driver is the community's++ [`relay`](plugins/relay) for existing Live hardware clients, and [`ros2`](plugins/ros2/README.md) for ROS 2 robots++ (`cmd_vel`, battery, scan/range, odometry and camera, as a node with rclpy or through rosbridge). LOOI's driver is the community's([docs/hardware-looi.md](docs/hardware-looi.md) is its contract). 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 —@@@ -13,10 +13,10 @@ ports, works behind any home router), and a local kill switch stops everything icurl -fsSL https://openvibe.bot/install | sh -s -- --robot rob_01J8Z4M2Q0R7T9YV3K6N8P1W2X --code ABCD-1234 --driver adeept```-- The shape is `--robot rob_… --code CODE [--driver adeept|adeept-mecanum|cozmo|looi|none]`. `--robot` is the robot's++ The shape is `--robot rob_… --code CODE [--driver adeept|adeept-mecanum|cozmo|looi|ros2|ros2-holonomic|none]`. `--robot` is the robot's`rob_…` id and `--code` the pairing code (both from openvibe.bot; the code may also be the first argument).-- `--driver` names the plugin: `adeept` (Adeept 4WD Smart Car, ordinary wheels), `adeept-mecanum`, `cozmo`, or-- `none` (the default: the dry-run plugin with a test-pattern camera, for trying the Node on a laptop). `--name NAME`++ `--driver` names the plugin: `adeept` (Adeept 4WD Smart Car, ordinary wheels), `adeept-mecanum`, `cozmo`, `ros2` (a++ ROS 2 robot through rosbridge, [plugins/ros2/README.md](../plugins/ros2/README.md)), `ros2-holonomic`, or `none` (the default: the dry-run plugin with a test-pattern camera, for trying the Node on a laptop). `--name NAME`names the device on openvibe.bot (default: the hostname).3. Confirm the new device on openvibe.bot.@@@ -5,7 +5,7 @@ object per line, UTF-8). Its stderr goes to the Node's log. The core never linksin 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), [`relay`](../plugins/relay). Python plugins use the runtime in [`plugins/sdk`](../plugins/sdk) (package++[`cozmo`](../plugins/cozmo/README.md), [`relay`](../plugins/relay), [`ros2`](../plugins/ros2/README.md). 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 itshardware.@@@ -95,6 +95,10 @@ Only `openvibe-node resume` on that machine releases it: the owner's remote cleareleases a latch. The Python runtime refuses motion from the moment it is called and ignores a `resume` that crosses iton the wire. `openvibe-node stop` sets the same latch from a terminal.++A plugin whose setup can fail only because something is not up yet (the robot's computer still booting) sets++`retry_setup_s` (Python): the runtime sends the fault once, sets up again that often, and sends `ready` when it works,++which ends the fault. Without it, a failed setup stays failed until the core restarts the plugin.++`stop` must be idempotent and safe at any time, including before setup. Handlers must not block: slow work (speechsynthesis, file I/O) goes on a thread. Nothing but protocol lines may go to stdout; the Python runtime enforces this bypointing file descriptor 1 at stderr and keeping a private copy for the protocol, because hardware libraries print@@@ -151,6 +155,17 @@ camera frames (320 × 240) as JPEG files. A cliff or pick-up stops the wheels inevent; while at a cliff only backing away is allowed, while picked up nothing drives. Firmware other than PyCozmo'ssupported version (2381, `allow_firmware`) is reported as `firmware_unsupported`.++### `ros2` — ROS 2 robots (rclpy or rosbridge)++++`drive` → `geometry_msgs/msg/Twist` on `cmd_vel` (`throttle`/`steer`, or `x`/`y`/`rotation` with `holonomic`), scaled by++`max_linear` m/s and `max_angular` rad/s, and a zero Twist at every stop; `say` → `std_msgs/msg/String` on `say_topic`.++Telemetry from `sensor_msgs/msg/BatteryState`, `sensor_msgs/msg/Range` or the front arc of a `sensor_msgs/msg/LaserScan`++(`distance_cm`) and `nav_msgs/msg/Odometry`; camera frames from a JPEG `sensor_msgs/msg/CompressedImage`. It reaches ROS 2++as a node of its own (`rclpy`, in a sourced ROS 2 environment) or through a rosbridge WebSocket server (no ROS 2++libraries where the Node runs; the installer's `--driver ros2` sets this up). When ROS is not up yet, or the link drops,++it reports `ros_unreachable` and tries again every `retry_s` seconds (`retry_setup_s` in the Python runtime). Topics,++speeds and setup: [plugins/ros2/README.md](../plugins/ros2/README.md).++### `relay` — existing OpenVibe.Live hardware clientsUse this when the owner already has a local script receiving Live stream controls. In `config.json`, replace the@@@ -1,7 +1,7 @@#!/bin/sh# OpenVibe Node installer.#--# curl -fsSL https://openvibe.bot/install | sh -s -- --robot <rob_…> --code <CODE> [--driver adeept|adeept-mecanum|cozmo|looi|none]++# curl -fsSL https://openvibe.bot/install | sh -s -- --robot <rob_…> --code <CODE> [--driver adeept|adeept-mecanum|cozmo|looi|ros2|ros2-holonomic|none]# curl -fsSL https://openvibe.bot/install | sh -s -- --network <URL> --pairing <pair_…> --code <CODE> [--driver …]## Detects the OS and CPU, downloads the openvibe-node binary and the plugin bundle (checking their SHA-256), creates@@@ -16,7 +16,8 @@# --network URL pair through OpenVibe.Network at URL (the machine gets a node credential, then Bot binds it).# --pairing ID the pair_… id of that Network pairing; a code given here instead (with no --code) is the code.# --name NAME the device's name on openvibe.bot (default: the hostname).--# --driver KIND adeept (ordinary wheels), adeept-mecanum, cozmo (bridge), looi (bridge, Bluetooth), none (dry run).++# --driver KIND adeept (ordinary wheels), adeept-mecanum, cozmo (bridge), looi (bridge, Bluetooth), ros2 (a ROS 2++# robot through rosbridge), ros2-holonomic (mecanum or omni wheels), none (dry run).# Default: none.# --no-service install and pair, but do not install the service.# --local DIR install from DIR (openvibe-node-<os>-<arch> and openvibe-node-plugins.tar.gz) instead of@@@ -47,7 +48,7 @@ TAG="latest"usage() {echo "usage: install.sh [--robot rob_… | --network URL --pairing pair_…] [--code CODE | CODE] [--name NAME]"-- echo " [--driver adeept|adeept-mecanum|cozmo|looi|none] [--no-service] [--local DIR] [--version TAG]"++ echo " [--driver adeept|adeept-mecanum|cozmo|looi|ros2|ros2-holonomic|none] [--no-service] [--local DIR] [--version TAG]"echo " curl -fsSL https://openvibe.bot/install | sh -s -- --robot rob_… --code ABCD-1234 --driver adeept"echo " curl -fsSL https://openvibe.bot/install | sh -s -- --network https://openvibe.network --pairing pair_… --code ABCD-1234"}@@@ -84,7 +85,7 @@ while [ $# -gt 0 ]; doesacdone--case "$DRIVER" in adeept|adeept-mecanum|cozmo|looi|none) ;; *) die "--driver must be adeept, adeept-mecanum, cozmo, looi or none" ;; esac++case "$DRIVER" in adeept|adeept-mecanum|cozmo|looi|ros2|ros2-holonomic|none) ;; *) die "--driver must be adeept, adeept-mecanum, cozmo, looi, ros2, ros2-holonomic or none" ;; esaccase "$ROBOT_ID" in ""|rob_*) ;; *) die "--robot must be a rob_… id" ;; esac# --pairing takes the pair_… id; with no --code, anything else there is the code itself.case "$PAIRING" in ""|pair_*) ;; *) [ -z "$CODE" ] || die "--pairing must be a pair_… id"; CODE="$PAIRING"; PAIRING="" ;; esac@@@ -316,6 +317,15 @@ cozmo)KIND=bridgecommand -v espeak-ng >/dev/null 2>&1 || say "tip: install espeak-ng so Cozmo can speak (sudo apt install espeak-ng)";;++ros2|ros2-holonomic)++ # A ROS 2 robot through rosbridge (plugins/ros2/README.md): no ROS 2 libraries are needed here, and the service++ # needs no sourced ROS environment. Topics, speeds and the rosbridge address are plugin config in config.json.++ $SUDO "$PIP" install --quiet "$STATE/plugins/ros2"++ HOLO=false; [ "$DRIVER" = ros2-holonomic ] && HOLO=true++ PLUGINS="[{\"name\": \"ros2\", \"config\": {\"transport\": \"rosbridge\", \"rosbridge_url\": \"ws://127.0.0.1:9090\", \"holonomic\": $HOLO}}]"++ say "tip: on the robot, run a rosbridge server (ros2 launch rosbridge_server rosbridge_websocket_launch.xml) and set"++ say " rosbridge_url, cmd_vel, max_linear and image_topic in $CONF/config.json if yours differ"++ ;;looi)# LOOI's camera is its phone, which streams on its own (docs/hardware-looi.md): the Node publishes no video.PLUGINS='[{"name": "looi"}]'@@@ -336,6 +336,7 @@ var Bundled = map[string]string{"adeept_adr036": "openvibe_adeept_adr036","cozmo": "openvibe_cozmo","relay": "openvibe_relay",++ "ros2": "openvibe_ros2",}// Default is the config written by `openvibe-node pair` when none exists: the dry-run plugin and its test pattern.nnew file mode 100644@@@ -0,0 +1,80 @@++# `ros2`: ROS 2 robots++++This plugin drives a ROS 2 robot from OpenVibe.Bot: TurtleBot 3 or 4, a Jackal, a home-built base, anything that takes++`geometry_msgs/msg/Twist` on a `cmd_vel` topic. The Node runs on the robot's computer, or on a computer on the same++network. The plugin turns Bot's commands into ROS 2 messages, and the robot's topics into telemetry and camera frames.++Bot's gate (roles, the owner's limits, cooldowns and the e-stop) has already judged every command before it gets here.++++## What it bridges++++| Bot | ROS 2 |++|---|---|++| `drive` `throttle`/`steer` | `geometry_msgs/msg/Twist` on `cmd_vel`: `linear.x = throttle × max_linear`, `angular.z = −steer × max_angular` (Bot's steer is positive to the right; ROS's z turns counter-clockwise) |++| `drive` `x`/`y`/`rotation` (`holonomic: true`) | `linear.x`, `linear.y`, `angular.z`. Bot's x is forward, y is left and rotation is counter-clockwise, as in REP 103 |++| any stop | a zero Twist: the last drive's deadline, a lost heartbeat, `stop`, `estop`, `halt`, stdin closing, before exit |++| `say` | `std_msgs/msg/String` on `say_topic`, when it is set |++| battery | `sensor_msgs/msg/BatteryState` on `battery_topic`: `percentage` (0–1) and `voltage` |++| `distance_cm` | `sensor_msgs/msg/Range` on `range_topic`, or else the nearest return within `front_arc_deg` of straight ahead on a `sensor_msgs/msg/LaserScan` (`scan_topic`) |++| `odom` | `nav_msgs/msg/Odometry` on `odom_topic`: x, y, yaw and speed |++| camera | a JPEG `sensor_msgs/msg/CompressedImage` on `image_topic`, at most `fps` frames a second; the Node publishes them as the robot's video |++++## Two ways to reach ROS 2++++- **`rclpy`**: the plugin is a ROS 2 node itself (named `node_name`). Run the Node where ROS 2 is installed, with its++ environment sourced (`source /opt/ros/jazzy/setup.bash`) so `rclpy` imports.++- **`rosbridge`**: JSON over WebSocket to a rosbridge server at `rosbridge_url`. Start one on the robot with++ `sudo apt install ros-$ROS_DISTRO-rosbridge-server` and `ros2 launch rosbridge_server rosbridge_websocket_launch.xml`.++ No ROS 2 libraries are needed where the Node runs, so a Raspberry Pi beside the robot, or a laptop, works. `wss://`++ works too.++- **`auto`** (the default) uses `rclpy` when it imports, otherwise `rosbridge`.++++If ROS cannot be reached, the panel shows the `ros_unreachable` fault and commands are refused with it. The plugin++tries again every `retry_s` seconds; when it reconnects, it publishes a zero Twist and says `ready`, which ends the++fault.++++## Setup++++1. On openvibe.bot, add a robot with the **ROS 2 robot** model (`ros2.diffdrive`; `ros2.holonomic` for mecanum or++ omni wheels) and keep its pairing code.++2. On the computer that will run the Node:++ `curl -fsSL https://openvibe.bot/install | sh -s -- --robot <rob_…> --code <CODE> --driver ros2`++3. Edit `config.json` (the installer prints where it is) if your topics differ from the defaults, then restart the++ Node's service:++++```json++{"plugins": [{"name": "ros2", "config": {++ "transport": "rosbridge",++ "rosbridge_url": "ws://turtlebot.local:9090",++ "cmd_vel": "/cmd_vel",++ "max_linear": 0.22,++ "max_angular": 2.0,++ "image_topic": "/camera/image_raw/compressed"++}}]}++```++++TurtleBot 4 publishes its camera as `/oakd/rgb/preview/image_raw/compressed` (with `image_transport` republishing),++and its base expects `cmd_vel` under the robot's namespace when one is set.++++## Configuration++++| key | default | meaning |++|---|---|---|++| `transport` | `auto` | `auto`, `rclpy` or `rosbridge` |++| `rosbridge_url` | `ws://127.0.0.1:9090` | the rosbridge server (`ws://` or `wss://`) |++| `node_name` | `openvibe_node` | the ROS 2 node's name with `rclpy` |++| `cmd_vel` | `/cmd_vel` | where Twists go |++| `max_linear` | `0.3` | m/s at full throttle (0–5) |++| `max_angular` | `1.5` | rad/s at full steer (0–10) |++| `holonomic` | `false` | `x`/`y`/`rotation` drive (mecanum, omni) instead of `throttle`/`steer` |++| `battery_topic` | `/battery_state` | `sensor_msgs/msg/BatteryState`; empty turns it off |++| `range_topic` | (empty) | `sensor_msgs/msg/Range`; replaces the scan for `distance_cm` |++| `scan_topic` | `/scan` | `sensor_msgs/msg/LaserScan`; empty turns it off |++| `front_arc_deg` | `30` | how wide "straight ahead" is on the scan |++| `odom_topic` | `/odom` | `nav_msgs/msg/Odometry`; empty turns it off |++| `image_topic` | (empty) | a JPEG `sensor_msgs/msg/CompressedImage`; empty means no camera from ROS |++| `fps` | `5` | camera frames a second (0.2–15) |++| `say_topic` | (empty) | `std_msgs/msg/String` for `say`; empty refuses `say` |++| `retry_s` | `3` | seconds between reconnection attempts |++++Bot's limits apply before the plugin's: the owner's `max_speed` and `max_turn` scale the command first, then++`max_linear` and `max_angular` turn it into m/s and rad/s. Set these two to what is safe for your robot indoors.nnew file mode 100644@@@ -0,0 +1,438 @@++"""OpenVibe Node bridge plugin for ROS 2 robots.++++The Node runs on the robot's computer (or next to it) and this plugin turns Bot's commands into ROS 2 messages and the++robot's topics into telemetry and camera frames:++++* ``drive`` → ``geometry_msgs/msg/Twist`` on ``cmd_vel`` (``throttle``/``steer``, or ``x``/``y``/``rotation`` on a++ holonomic base), scaled by ``max_linear`` m/s and ``max_angular`` rad/s. A zero Twist is published at every stop:++ the deadline of the last drive, a lost heartbeat, ``stop``, ``estop``, ``halt``, stdin closing, and before exit.++* ``say`` → ``std_msgs/msg/String`` on ``say_topic`` (when set).++* ``sensor_msgs/msg/BatteryState`` → battery percent and volts; ``sensor_msgs/msg/Range`` or the front arc of a++ ``sensor_msgs/msg/LaserScan`` → ``distance_cm``; ``nav_msgs/msg/Odometry`` → ``odom`` {x, y, yaw, speed}.++* ``sensor_msgs/msg/CompressedImage`` (JPEG) on ``image_topic`` → camera frames for the Node to publish.++++Two ways to reach ROS 2 (``transport``):++++* ``rclpy``: the plugin is a ROS 2 node itself. Needs a sourced ROS 2 environment (``rclpy`` importable).++* ``rosbridge``: JSON over WebSocket to a rosbridge server (``ros2 launch rosbridge_server++ rosbridge_websocket_launch.xml``), on this machine or the robot's. No ROS 2 libraries needed here.++* ``auto`` (default): ``rclpy`` when it imports, otherwise ``rosbridge``.++++If ROS cannot be reached the plugin reports the ``ros_unreachable`` fault, refuses commands with it, and tries again++every ``retry_s`` seconds; when it reconnects it says ``ready`` again, which ends the fault.++"""++++import base64++import importlib++import importlib.util++import json++import math++import os++import shutil++import tempfile++import threading++import time++++from openvibe_plugin import Fault, Plugin, clamp++++from .ws import WebSocket++++FRAME_RING = 4++++DEFAULTS = {++ "transport": "auto", # auto | rclpy | rosbridge++ "rosbridge_url": "ws://127.0.0.1:9090",++ "node_name": "openvibe_node",++ "cmd_vel": "/cmd_vel",++ "max_linear": 0.3, # m/s at full throttle++ "max_angular": 1.5, # rad/s at full steer++ "holonomic": False, # x/y/rotation drive (mecanum, omni) instead of throttle/steer++ "battery_topic": "/battery_state",++ "range_topic": "", # sensor_msgs/Range (an ultrasonic or ToF sensor)++ "scan_topic": "/scan", # sensor_msgs/LaserScan: the nearest return in the front arc++ "front_arc_deg": 30,++ "odom_topic": "/odom",++ "image_topic": "", # sensor_msgs/CompressedImage, JPEG++ "fps": 5,++ "say_topic": "", # std_msgs/String++ "retry_s": 3,++}++++TWIST = "geometry_msgs/msg/Twist"++++++def _num(v, default=0.0):++ try:++ f = float(v)++ except (TypeError, ValueError):++ return default++ return f if f == f else default++++++# ── Transports ─────────────────────────────────────────────────────────────────────────────────────────────────────++class RosbridgeBus:++ """rosbridge v2 over WebSocket: advertise + publish, subscribe; incoming publishes go to the topic's callbacks."""++++ def __init__(self, url, log, connect=WebSocket):++ self.log = log++ self.ws = connect(url)++ self.alive = True++ self.subs = {}++ self.advertised = set()++ self.thread = threading.Thread(target=self._read, name="rosbridge", daemon=True)++ self.thread.start()++++ def _send(self, obj):++ self.ws.send_text(json.dumps(obj, separators=(",", ":")))++++ def publish(self, topic, type_, msg):++ if topic not in self.advertised:++ self._send({"op": "advertise", "topic": topic, "type": type_})++ self.advertised.add(topic)++ self._send({"op": "publish", "topic": topic, "msg": msg})++++ def subscribe(self, topic, type_, callback, throttle_ms=0):++ self.subs.setdefault(topic, []).append(callback)++ self._send({"op": "subscribe", "topic": topic, "type": type_, "throttle_rate": int(throttle_ms), "queue_length": 1})++++ def _read(self):++ try:++ while True:++ text = self.ws.recv()++ if text is None:++ break++ try:++ m = json.loads(text)++ except ValueError:++ continue++ if isinstance(m, dict) and m.get("op") == "publish":++ for cb in self.subs.get(m.get("topic"), ()):++ try:++ cb(m.get("msg") or {})++ except Exception as e:++ self.log("ros2: %s callback failed: %r", m.get("topic"), e)++ finally:++ self.alive = False++++ def close(self):++ self.alive = False++ self.ws.close()++++++def _to_dict(m):++ """A ROS 2 message as plain data (fields by name; byte arrays as bytes)."""++ if hasattr(m, "get_fields_and_field_types"):++ return {k: _to_dict(getattr(m, k)) for k in m.get_fields_and_field_types()}++ if isinstance(m, (bytes, bytearray)):++ return bytes(m)++ if hasattr(m, "tobytes") and getattr(m, "typecode", None) in ("B", "b"): # array.array('B')++ return m.tobytes()++ if isinstance(m, (list, tuple)) or (hasattr(m, "__iter__") and hasattr(m, "typecode")):++ return [_to_dict(x) for x in m]++ return m++++++def _fill(m, d):++ for k, v in d.items():++ if isinstance(v, dict):++ _fill(getattr(m, k), v)++ else:++ setattr(m, k, v)++ return m++++++class RclpyBus:++ """The plugin as a ROS 2 node: publishers and subscriptions on its own executor thread."""++++ def __init__(self, node_name, log, rclpy=None):++ self.log = log++ self.rclpy = rclpy or importlib.import_module("rclpy")++ if not self.rclpy.ok():++ self.rclpy.init(args=None)++ self.node = self.rclpy.create_node(node_name)++ executors = importlib.import_module(self.rclpy.__name__ + ".executors") if not hasattr(self.rclpy, "executors") else self.rclpy.executors++ self.executor = executors.SingleThreadedExecutor()++ self.executor.add_node(self.node)++ self.pubs = {}++ self.alive = True++ self.thread = threading.Thread(target=self._spin, name="rclpy", daemon=True)++ self.thread.start()++++ def _spin(self):++ try:++ self.executor.spin()++ except Exception as e:++ self.log("ros2: executor stopped: %r", e)++ finally:++ self.alive = False++++ @staticmethod++ def _cls(type_):++ pkg, kind, name = type_.split("/")++ return getattr(importlib.import_module("%s.%s" % (pkg, kind)), name)++++ def publish(self, topic, type_, msg):++ pub = self.pubs.get(topic)++ if pub is None:++ pub = self.pubs[topic] = self.node.create_publisher(self._cls(type_), topic, 10)++ pub.publish(_fill(self._cls(type_)(), msg))++++ def subscribe(self, topic, type_, callback, throttle_ms=0):++ self.node.create_subscription(self._cls(type_), topic, lambda m: callback(_to_dict(m)), 1)++++ def close(self):++ self.alive = False++ try:++ self.executor.shutdown()++ self.node.destroy_node()++ except Exception:++ pass++ try:++ self.rclpy.try_shutdown()++ except Exception:++ pass++++++def default_bus(cfg, log):++ t = cfg["transport"]++ if t == "rclpy" or (t == "auto" and importlib.util.find_spec("rclpy") is not None):++ return RclpyBus(cfg["node_name"], log)++ return RosbridgeBus(cfg["rosbridge_url"], log)++++++# ── The plugin ─────────────────────────────────────────────────────────────────────────────────────────────────────++class Ros2(Plugin):++ driver, version = "ros2", "0.1.0"++ motion_kinds = frozenset({"drive"})++ retry_setup_s = DEFAULTS["retry_s"] # ROS not up yet at start: the runtime sets up again (setup below)++++ def __init__(self, bus_factory=default_bus, clock=time.monotonic):++ self.bus_factory = bus_factory++ self.clock = clock++ self.cfg = dict(DEFAULTS)++ self.bus = None++ self.faulted = False++ self.next_retry = 0.0++ self.lock = threading.Lock()++ self.battery = None++ self.distance_cm = None++ self.odom = None++ self.frame_dir = None++ self._own_frame_dir = False++ self.frame_seq = 0++ self.next_frame = 0.0++ self._warned = set()++++ @staticmethod++ def config_of(config):++ cfg = dict(DEFAULTS)++ cfg.update({k: v for k, v in (config or {}).items() if k in DEFAULTS or k == "frame_dir"})++ if cfg["transport"] not in ("auto", "rclpy", "rosbridge"):++ raise Fault("bad_config", "transport must be auto, rclpy or rosbridge")++ cfg["max_linear"] = clamp(cfg["max_linear"], 0.0, 5.0)++ cfg["max_angular"] = clamp(cfg["max_angular"], 0.0, 10.0)++ cfg["fps"] = clamp(cfg["fps"], 0.2, 15.0)++ return cfg++++ def describe(self, config):++ cfg = self.config_of(config)++ caps = {++ "drive": {"type": "mecanum" if cfg["holonomic"] else "differential", "max_linear_mps": cfg["max_linear"], "max_angular_rps": cfg["max_angular"]},++ "bridge": {"kind": "ros2", "transport": cfg["transport"], "cmd_vel": cfg["cmd_vel"]},++ }++ if cfg["battery_topic"]:++ caps["battery"] = {"topic": cfg["battery_topic"]}++ if cfg["range_topic"] or cfg["scan_topic"] or cfg["odom_topic"]:++ caps["sensors"] = {"distance_cm": bool(cfg["range_topic"] or cfg["scan_topic"]), "odom": bool(cfg["odom_topic"])}++ if cfg["image_topic"]:++ caps["camera"] = {"jpeg": True}++ if cfg["say_topic"]:++ caps["say"] = {"topic": cfg["say_topic"]}++ return caps++++ # Setup and reconnecting.++ def setup(self, config):++ self.cfg = self.config_of(config)++ self.retry_setup_s = max(1.0, _num(self.cfg["retry_s"], 3))++ if self.frame_dir:++ self.close()++ if self.cfg["image_topic"]:++ d = self.cfg.get("frame_dir")++ if d:++ os.makedirs(d, exist_ok=True)++ self.frame_dir, self._own_frame_dir = d, False++ else:++ self.frame_dir, self._own_frame_dir = tempfile.mkdtemp(prefix="openvibe-ros2-frames-"), True++ self._connect()++++ def _connect(self):++ try:++ bus = self.bus_factory(self.cfg, self.log)++ except Fault:++ raise++ except Exception as e:++ self.next_retry = self.clock() + _num(self.cfg["retry_s"], 3)++ raise Fault("ros_unreachable", "cannot reach ROS 2 (%s): %s" % (self.cfg["transport"], e))++ c = self.cfg++ if c["battery_topic"]:++ bus.subscribe(c["battery_topic"], "sensor_msgs/msg/BatteryState", self._on_battery, 1000)++ if c["range_topic"]:++ bus.subscribe(c["range_topic"], "sensor_msgs/msg/Range", self._on_range, 100)++ elif c["scan_topic"]:++ bus.subscribe(c["scan_topic"], "sensor_msgs/msg/LaserScan", self._on_scan, 200)++ if c["odom_topic"]:++ bus.subscribe(c["odom_topic"], "nav_msgs/msg/Odometry", self._on_odom, 200)++ if c["image_topic"]:++ bus.subscribe(c["image_topic"], "sensor_msgs/msg/CompressedImage", self._on_image, int(1000 / c["fps"]))++ self.bus = bus++ self.faulted = False++++ def _bus_alive(self):++ return self.bus is not None and self.bus.alive++++ def poll(self):++ if not self._bus_alive():++ if not self.faulted:++ self.faulted = True++ self.next_retry = self.clock() + _num(self.cfg["retry_s"], 3)++ if self.runtime is not None:++ self.runtime.send({"op": "fault", "fault_code": "ros_unreachable", "message": "the connection to ROS 2 was lost"})++ elif self.clock() >= self.next_retry:++ try:++ self._connect()++ self.publish_zero()++ if self.runtime is not None:++ self.runtime.send({"op": "ready"})++ except Fault:++ self.next_retry = self.clock() + _num(self.cfg["retry_s"], 3)++ return {"faults": ["ros_unreachable"]}++ out = {"sensors": {}}++ with self.lock:++ if self.battery:++ out["battery"] = dict(self.battery)++ if self.distance_cm is not None:++ out["sensors"]["distance_cm"] = self.distance_cm++ if self.odom:++ out["sensors"]["odom"] = dict(self.odom)++ return out++++ # Commands.++ def twist(self, lx=0.0, ly=0.0, az=0.0):++ return {"linear": {"x": float(lx), "y": float(ly), "z": 0.0}, "angular": {"x": 0.0, "y": 0.0, "z": float(az)}}++++ def publish_zero(self):++ if self._bus_alive():++ self.bus.publish(self.cfg["cmd_vel"], TWIST, self.twist())++++ def handle(self, cmd):++ if not self._bus_alive():++ raise Fault("ros_unreachable", "not connected to ROS 2")++ v = cmd.value if isinstance(cmd.value, dict) else {}++ c = self.cfg++ if cmd.kind == "drive":++ if c["holonomic"]:++ # Bot's x is forward, y is left, rotation is counter-clockwise: ROS's own axes (REP 103).++ t = self.twist(clamp(v.get("x", 0), -1, 1) * c["max_linear"], clamp(v.get("y", 0), -1, 1) * c["max_linear"],++ clamp(v.get("rotation", 0), -1, 1) * c["max_angular"])++ else:++ # Bot's steer is positive to the right; ROS's angular z is positive counter-clockwise (to the left).++ t = self.twist(clamp(v.get("throttle", 0), -1, 1) * c["max_linear"], 0.0, -clamp(v.get("steer", 0), -1, 1) * c["max_angular"])++ self.bus.publish(c["cmd_vel"], TWIST, t)++ return++ if cmd.kind == "say" and c["say_topic"]:++ text = str(v.get("text", ""))[:500]++ if not text:++ raise Fault("bad_value", "say needs text")++ self.bus.publish(c["say_topic"], "std_msgs/msg/String", {"data": text})++ return++ raise Fault("unsupported", "kind %s not supported" % cmd.kind)++++ def stop_motion(self):++ self.stop()++++ def stop(self):++ try:++ self.publish_zero()++ except Exception as e:++ self.log("ros2: could not publish the zero Twist: %r", e)++++ def close(self):++ bus, self.bus = self.bus, None++ if bus is not None:++ try:++ bus.close()++ except Exception:++ pass++ if self.frame_dir and self._own_frame_dir:++ shutil.rmtree(self.frame_dir, ignore_errors=True)++ self.frame_dir = None++++ # Topics.++ def _on_battery(self, m):++ pct = _num(m.get("percentage"), float("nan"))++ volts = _num(m.get("voltage"), float("nan"))++ b = {}++ if pct == pct and pct >= 0:++ b["percent"] = round(pct * 100 if pct <= 1.0 else pct, 1)++ if volts == volts and volts > 0:++ b["volts"] = round(volts, 2)++ with self.lock:++ self.battery = b or None++++ def _on_range(self, m):++ r = _num(m.get("range"), float("nan"))++ with self.lock:++ self.distance_cm = round(r * 100, 1) if r == r and math.isfinite(r) and r >= 0 else None++++ def _on_scan(self, m):++ ranges = m.get("ranges") or []++ a0, inc = _num(m.get("angle_min")), _num(m.get("angle_increment"))++ lo, hi = _num(m.get("range_min"), 0.0), _num(m.get("range_max"), float("inf"))++ half = math.radians(_num(self.cfg["front_arc_deg"], 30)) / 2++ best = None++ for i, r in enumerate(ranges):++ r = _num(r, float("nan"))++ if not (r == r) or not math.isfinite(r) or r < lo or r > hi:++ continue++ a = math.atan2(math.sin(a0 + i * inc), math.cos(a0 + i * inc)) # wrapped to -pi..pi, 0 is straight ahead++ if abs(a) <= half and (best is None or r < best):++ best = r++ with self.lock:++ self.distance_cm = round(best * 100, 1) if best is not None else None++++ def _on_odom(self, m):++ pose = ((m.get("pose") or {}).get("pose") or {})++ p, q = pose.get("position") or {}, pose.get("orientation") or {}++ qx, qy, qz, qw = (_num(q.get(k)) for k in ("x", "y", "z", "w"))++ yaw = math.atan2(2 * (qw * qz + qx * qy), 1 - 2 * (qy * qy + qz * qz))++ lin = (((m.get("twist") or {}).get("twist") or {}).get("linear") or {})++ with self.lock:++ self.odom = {"x": round(_num(p.get("x")), 3), "y": round(_num(p.get("y")), 3), "yaw": round(yaw, 3),++ "speed": round(math.hypot(_num(lin.get("x")), _num(lin.get("y"))), 3)}++++ def _on_image(self, m):++ fmt = str(m.get("format") or "").lower()++ if "jpeg" not in fmt and "jpg" not in fmt:++ if "format" not in self._warned:++ self._warned.add("format")++ self.log("ros2: %s is %r, not JPEG: no camera frames (use a JPEG CompressedImage topic)", self.cfg["image_topic"], fmt)++ return++ now = self.clock()++ if now < self.next_frame or self.frame_dir is None:++ return++ self.next_frame = now + 1.0 / self.cfg["fps"]++ data = m.get("data")++ if isinstance(data, str):++ try:++ data = base64.b64decode(data)++ except ValueError:++ return++ if not isinstance(data, (bytes, bytearray)) or len(data) < 4 or data[:2] != b"\xff\xd8":++ return++ self.frame_seq += 1++ path = os.path.join(self.frame_dir, "frame-%d.jpg" % (self.frame_seq % FRAME_RING))++ tmp = path + ".tmp"++ with open(tmp, "wb") as f:++ f.write(data)++ os.replace(tmp, path) # readers never see a partial file++ self.emit_video(path, format="jpeg", seq=self.frame_seq)nnew file mode 100644@@@ -0,0 +1,6 @@++import sys++++from openvibe_plugin import run++from openvibe_ros2 import Ros2++++sys.exit(run(Ros2()))nnew file mode 100644@@@ -0,0 +1,156 @@++"""A small WebSocket client (RFC 6455) on the standard library, enough for rosbridge's JSON protocol.++++Text frames only, client-masked, with ping/pong, close and continuation frames. ``ws://`` and ``wss://`` (TLS with the++system's trust store and hostname checks). A message larger than ``max_message`` closes the connection.++"""++++import base64++import hashlib++import os++import socket++import ssl++import struct++import threading++import urllib.parse++++GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"++++++class WebSocketError(Exception):++ pass++++++class WebSocket:++ def __init__(self, url, timeout=5.0, max_message=16 * 1024 * 1024):++ u = urllib.parse.urlsplit(url)++ if u.scheme not in ("ws", "wss"):++ raise WebSocketError("the rosbridge URL must be ws:// or wss://, not %r" % (url,))++ host = u.hostname or "127.0.0.1"++ port = u.port or (443 if u.scheme == "wss" else 80)++ path = (u.path or "/") + ("?" + u.query if u.query else "")++ self.max_message = max_message++ self._send_lock = threading.Lock()++ self._closed = False++ sock = socket.create_connection((host, port), timeout=timeout)++ try:++ if u.scheme == "wss":++ sock = ssl.create_default_context().wrap_socket(sock, server_hostname=host)++ key = base64.b64encode(os.urandom(16)).decode("ascii")++ hostport = host if port in (80, 443) else "%s:%d" % (host, port)++ if ":" in host and not host.startswith("["):++ hostport = "[%s]:%d" % (host, port)++ sock.sendall(("GET %s HTTP/1.1\r\nHost: %s\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n"++ "Sec-WebSocket-Key: %s\r\nSec-WebSocket-Version: 13\r\n\r\n" % (path, hostport, key)).encode("ascii"))++ raw = b""++ while b"\r\n\r\n" not in raw:++ chunk = sock.recv(4096)++ if not chunk:++ raise WebSocketError("the server closed the connection during the handshake")++ raw += chunk++ if len(raw) > 16384:++ raise WebSocketError("the handshake answer is too long")++ head, self._buf = raw.split(b"\r\n\r\n", 1)++ lines = head.decode("latin-1").split("\r\n")++ if len(lines[0].split()) < 2 or lines[0].split()[1] != "101":++ raise WebSocketError("the server did not switch to WebSocket: %s" % lines[0][:80])++ headers = {}++ for line in lines[1:]:++ k, _, v = line.partition(":")++ headers[k.strip().lower()] = v.strip()++ want = base64.b64encode(hashlib.sha1((key + GUID).encode("ascii")).digest()).decode("ascii")++ if headers.get("sec-websocket-accept") != want:++ raise WebSocketError("the server's handshake answer does not match")++ sock.settimeout(None)++ except Exception:++ sock.close()++ raise++ self.sock = sock++++ # Frames.++ def _send_frame(self, opcode, payload):++ header = bytearray([0x80 | opcode])++ n = len(payload)++ if n < 126:++ header.append(0x80 | n)++ elif n < 65536:++ header.append(0x80 | 126)++ header += struct.pack(">H", n)++ else:++ header.append(0x80 | 127)++ header += struct.pack(">Q", n)++ mask = os.urandom(4)++ header += mask++ body = bytes(b ^ mask[i % 4] for i, b in enumerate(payload)) if n < 4096 else _mask_fast(payload, mask)++ with self._send_lock:++ self.sock.sendall(bytes(header) + body)++++ def send_text(self, text):++ if self._closed:++ raise WebSocketError("closed")++ self._send_frame(0x1, text.encode("utf-8"))++++ def _read_exact(self, n):++ while len(self._buf) < n:++ chunk = self.sock.recv(max(65536, n - len(self._buf)))++ if not chunk:++ raise WebSocketError("the connection closed")++ self._buf += chunk++ out, self._buf = self._buf[:n], self._buf[n:]++ return out++++ def recv(self):++ """The next text message, or None when the connection closes."""++ parts = []++ size = 0++ try:++ while True:++ b1, b2 = self._read_exact(2)++ fin, opcode = b1 & 0x80, b1 & 0x0F++ n = b2 & 0x7F++ if n == 126:++ n = struct.unpack(">H", self._read_exact(2))[0]++ elif n == 127:++ n = struct.unpack(">Q", self._read_exact(8))[0]++ mask = self._read_exact(4) if b2 & 0x80 else None++ if size + n > self.max_message:++ raise WebSocketError("a message is larger than %d bytes" % self.max_message)++ payload = self._read_exact(n)++ if mask:++ payload = _mask_fast(payload, mask)++ if opcode == 0x8:++ self.close()++ return None++ if opcode == 0x9:++ self._send_frame(0xA, payload)++ continue++ if opcode == 0xA:++ continue++ if opcode in (0x1, 0x2, 0x0):++ parts.append(payload)++ size += n++ if fin:++ return b"".join(parts).decode("utf-8", "replace")++ except (OSError, WebSocketError, ValueError):++ self.close()++ return None++++ def close(self):++ if self._closed:++ return++ self._closed = True++ try:++ self._send_frame(0x8, b"")++ except Exception:++ pass++ try:++ self.sock.close()++ except Exception:++ pass++++++def _mask_fast(payload, mask):++ n = len(payload)++ if not n:++ return b""++ key = int.from_bytes((mask * (n // 4 + 1))[:n], "big")++ return (int.from_bytes(payload, "big") ^ key).to_bytes(n, "big")nnew file mode 100644@@@ -0,0 +1,14 @@++[build-system]++requires = ["setuptools>=61"]++build-backend = "setuptools.build_meta"++++[project]++name = "openvibe-ros2"++version = "0.1.0"++description = "OpenVibe Node bridge plugin for ROS 2 robots (rclpy or rosbridge)"++requires-python = ">=3.9"++license = { text = "AGPL-3.0-or-later" }++dependencies = ["openvibe-plugin"]++++[tool.setuptools]++packages = ["openvibe_ros2"]nnew file mode 100644@@@ -0,0 +1,441 @@++import base64++import hashlib++import io++import json++import math++import os++import socket++import struct++import sys++import tempfile++import threading++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 Runtime # noqa: E402++from openvibe_ros2 import DEFAULTS, RclpyBus, RosbridgeBus, Ros2, _fill, _to_dict # noqa: E402++from openvibe_ros2.ws import GUID, WebSocket, WebSocketError # noqa: E402++++JPEG = b"\xff\xd8\xff\xe0" + b"\x00" * 64 + b"\xff\xd9"++++++class Clock:++ t = 10.0++++ def __call__(self):++ return self.t++++++class FakeBus:++ """Records publishes and subscriptions; tests deliver topic messages with deliver()."""++++ def __init__(self):++ self.alive = True++ self.published = []++ self.subs = {}++ self.closed = False++++ def publish(self, topic, type_, msg):++ if not self.alive:++ raise OSError("closed")++ self.published.append((topic, type_, json.loads(json.dumps(msg))))++++ def subscribe(self, topic, type_, callback, throttle_ms=0):++ self.subs[topic] = (type_, callback)++++ def deliver(self, topic, msg):++ self.subs[topic][1](msg)++++ def close(self):++ self.closed = True++ self.alive = False++++++class Harness:++ def __init__(self, fail=0, **config):++ self.clock = Clock()++ self.buses = []++ self.fail = fail++++ def factory(cfg, log):++ if self.fail:++ self.fail -= 1++ raise OSError("connection refused")++ b = FakeBus()++ self.buses.append(b)++ return b++++ self.p = Ros2(bus_factory=factory, clock=self.clock)++ self.p.log = lambda *a: None++ self.out = io.StringIO()++ self.rt = Runtime(self.p, infile=io.StringIO(""), outfile=self.out, clock=self.clock)++ self.frames = tempfile.mkdtemp()++ self.config = dict(config)++++ def lines(self):++ return [json.loads(x) for x in self.out.getvalue().splitlines() if x.strip()]++++ def ops(self, op):++ return [m for m in self.lines() if m.get("op") == op]++++ def hello(self, probe=False):++ self.rt.dispatch({"op": "hello", "config": self.config, "probe": probe})++++ def cmd(self, cid, kind, value, deadline_ms=300):++ self.rt.dispatch({"op": "heartbeat", "t": 0})++ self.rt.dispatch({"op": "command", "id": cid, "kind": kind, "value": value, "deadline_ms": deadline_ms})++++ @property++ def bus(self):++ return self.buses[-1]++++ def twists(self):++ return [m for (t, ty, m) in self.bus.published if ty == "geometry_msgs/msg/Twist"]++++++def test_describe_touches_nothing_and_names_what_is_configured():++ h = Harness(image_topic="/camera/image_raw/compressed", say_topic="/say", holonomic=True)++ h.hello(probe=True)++ d = h.ops("describe")[0]++ assert d["driver"] == "ros2" and d["motion_kinds"] == ["drive"]++ caps = d["capabilities"]++ assert caps["drive"]["type"] == "mecanum"++ assert caps["camera"] == {"jpeg": True} and caps["say"]["topic"] == "/say"++ assert caps["sensors"]["distance_cm"] and caps["bridge"]["kind"] == "ros2"++ assert h.buses == [] and not h.ops("ready"), "probe mode never connects"++++++def test_setup_subscribes_publishes_a_zero_twist_and_says_ready():++ h = Harness()++ h.hello()++ assert h.ops("ready")++ assert set(h.bus.subs) == {"/battery_state", "/scan", "/odom"}++ assert h.bus.subs["/scan"][0] == "sensor_msgs/msg/LaserScan"++ assert h.twists()[-1] == {"linear": {"x": 0.0, "y": 0.0, "z": 0.0}, "angular": {"x": 0.0, "y": 0.0, "z": 0.0}}++++++def test_drive_scales_to_twist_with_ros_signs_and_stops_at_the_deadline():++ h = Harness(max_linear=0.5, max_angular=2.0)++ h.hello()++ h.cmd("c1", "drive", {"throttle": 0.5, "steer": 0.25})++ assert h.ops("ack")[-1]["id"] == "c1"++ t = h.twists()[-1]++ assert t["linear"]["x"] == pytest.approx(0.25) and t["angular"]["z"] == pytest.approx(-0.5), "steer right is clockwise"++ h.clock.t += 0.31++ h.rt.tick()++ assert h.twists()[-1]["linear"]["x"] == 0 and h.twists()[-1]["angular"]["z"] == 0++ assert "deadline" in h.rt.stop_reasons++ h.cmd("c2", "drive", {"throttle": 9, "steer": -9})++ t = h.twists()[-1]++ assert t["linear"]["x"] == pytest.approx(0.5) and t["angular"]["z"] == pytest.approx(2.0), "clamped to full scale"++++++def test_holonomic_drive_uses_x_forward_y_left_rotation_ccw():++ h = Harness(holonomic=True, max_linear=1.0, max_angular=1.0)++ h.hello()++ h.cmd("h1", "drive", {"x": 0.5, "y": 0.25, "rotation": -0.5})++ t = h.twists()[-1]++ assert (t["linear"]["x"], t["linear"]["y"], t["angular"]["z"]) == (0.5, 0.25, -0.5)++++++def test_estop_halt_heartbeat_and_eof_publish_zero_and_refuse_motion():++ h = Harness()++ h.hello()++ h.cmd("m1", "drive", {"throttle": 1})++ h.rt.dispatch({"op": "estop"})++ assert h.twists()[-1]["linear"]["x"] == 0++ h.cmd("m2", "drive", {"throttle": 1})++ assert h.ops("nack")[-1] == {"op": "nack", "id": "m2", "fault_code": "estopped"}++ h.rt.dispatch({"op": "resume"})++ h.cmd("m3", "drive", {"throttle": 1})++ h.clock.t += 1.5++ h.rt.tick()++ assert h.ops("event")[-1]["name"] == "heartbeat_lost" and h.twists()[-1]["linear"]["x"] == 0++ n = len(h.twists())++ h.rt.shutdown("eof")++ assert len(h.twists()) == n + 1 and h.twists()[-1]["linear"]["x"] == 0 and h.bus.closed++++++def test_say_publishes_a_string_only_when_configured():++ h = Harness(say_topic="/say")++ h.hello()++ h.cmd("s1", "say", {"text": "Hello from OpenVibe"})++ assert ("/say", "std_msgs/msg/String", {"data": "Hello from OpenVibe"}) in h.bus.published++ h2 = Harness()++ h2.hello()++ h2.cmd("s2", "say", {"text": "hi"})++ assert h2.ops("nack")[-1]["fault_code"] == "unsupported"++++++def test_telemetry_battery_scan_range_and_odom():++ h = Harness()++ h.hello()++ h.bus.deliver("/battery_state", {"percentage": 0.42, "voltage": 12.06})++ # 360 rays from -pi: the nearest return straight ahead (index 180) counts, a nearer one behind does not.++ ranges = [5.0] * 360++ ranges[180] = 0.8++ ranges[0] = 0.1++ ranges[181] = float("inf")++ h.bus.deliver("/scan", {"angle_min": -math.pi, "angle_increment": 2 * math.pi / 360, "range_min": 0.05, "range_max": 12, "ranges": ranges})++ yaw = 0.5++ h.bus.deliver("/odom", {"pose": {"pose": {"position": {"x": 1.234, "y": -2.0}, "orientation": {"x": 0, "y": 0, "z": math.sin(yaw / 2), "w": math.cos(yaw / 2)}}},++ "twist": {"twist": {"linear": {"x": 0.3, "y": 0.4}}}})++ t = h.p.poll()++ assert t["battery"] == {"percent": 42.0, "volts": 12.06}++ assert t["sensors"]["distance_cm"] == 80.0++ assert t["sensors"]["odom"] == {"x": 1.234, "y": -2.0, "yaw": 0.5, "speed": 0.5}++ hr = Harness(range_topic="/sonar")++ hr.hello()++ assert "/sonar" in hr.bus.subs and "/scan" not in hr.bus.subs, "a Range topic replaces the scan"++ hr.bus.deliver("/sonar", {"range": 0.355})++ assert hr.p.poll()["sensors"]["distance_cm"] == 35.5++++++def test_camera_frames_from_a_jpeg_compressed_image_rate_limited():++ h = Harness(image_topic="/cam/compressed", fps=2)++ h.config["frame_dir"] = h.frames++ h.hello()++ h.bus.deliver("/cam/compressed", {"format": "rgb8; jpeg compressed bgr8", "data": base64.b64encode(JPEG).decode()})++ v = h.ops("video")++ assert len(v) == 1 and v[0]["format"] == "jpeg"++ with open(v[0]["path"], "rb") as f:++ assert f.read() == JPEG++ h.bus.deliver("/cam/compressed", {"format": "jpeg", "data": JPEG}) # rclpy hands bytes++ assert len(h.ops("video")) == 1, "within 1/fps"++ h.clock.t += 0.6++ h.bus.deliver("/cam/compressed", {"format": "jpeg", "data": JPEG})++ assert len(h.ops("video")) == 2++ h.clock.t += 1++ h.bus.deliver("/cam/compressed", {"format": "png", "data": base64.b64encode(b"\x89PNG").decode()})++ assert len(h.ops("video")) == 2, "not JPEG: no frame"++++++def test_unreachable_ros_faults_and_the_runtime_sets_up_again_until_ready():++ h = Harness(fail=2, retry_s=3)++ h.hello()++ f = h.ops("fault")++ assert f and f[0]["fault_code"] == "ros_unreachable" and not h.ops("ready")++ h.cmd("x1", "drive", {"throttle": 1})++ assert h.ops("nack")[-1]["fault_code"] == "not_ready"++ h.clock.t += 3.1++ h.rt.tick() # second failure: still down, no second fault++ assert not h.ops("ready") and len(h.ops("fault")) == 1++ h.clock.t += 3.1++ h.rt.tick() # ROS is up++ assert h.ops("ready") and len(h.buses) == 1++ h.cmd("x2", "drive", {"throttle": 1})++ assert h.ops("ack")[-1]["id"] == "x2"++ # The link drops later: a fault, commands refused by the plugin, and ready again once it reconnects.++ h.bus.alive = False++ h.p.poll()++ assert h.ops("fault")[-1]["message"] == "the connection to ROS 2 was lost"++ h.cmd("x3", "drive", {"throttle": 1})++ assert h.ops("nack")[-1]["fault_code"] == "ros_unreachable"++ h.clock.t += 3.1++ h.p.poll()++ assert len(h.buses) == 2 and h.ops("ready")[-1] == {"op": "ready"} and len(h.ops("ready")) == 2++ assert h.twists()[-1]["linear"]["x"] == 0, "a zero Twist first on the new link"++++++def test_bad_transport_is_a_config_fault():++ h = Harness(transport="dds")++ h.hello()++ assert h.ops("fault")[0]["fault_code"] == "bad_config"++++++# ── rclpy helpers ─────────────────────────────────────────────────────────────────────────────────────────────────++class Vector3:++ def __init__(self):++ self.x = self.y = self.z = 0.0++++ @staticmethod++ def get_fields_and_field_types():++ return {"x": "double", "y": "double", "z": "double"}++++++class Twist:++ def __init__(self):++ self.linear, self.angular = Vector3(), Vector3()++++ @staticmethod++ def get_fields_and_field_types():++ return {"linear": "geometry_msgs/Vector3", "angular": "geometry_msgs/Vector3"}++++++def test_fill_and_to_dict_round_trip_a_nested_message():++ t = _fill(Twist(), {"linear": {"x": 0.2, "y": 0.0, "z": 0.0}, "angular": {"x": 0.0, "y": 0.0, "z": -1.0}})++ assert t.linear.x == 0.2 and t.angular.z == -1.0++ assert _to_dict(t) == {"linear": {"x": 0.2, "y": 0.0, "z": 0.0}, "angular": {"x": 0.0, "y": 0.0, "z": -1.0}}++ assert _to_dict(bytearray(b"\xff\xd8")) == b"\xff\xd8"++++++def test_rclpy_bus_with_a_fake_rclpy(monkeypatch):++ import types++ created = {"pubs": [], "subs": []}++++ class Pub:++ def __init__(self, cls, topic):++ self.topic, self.sent = topic, []++++ def publish(self, m):++ self.sent.append(m)++++ class Node:++ def create_publisher(self, cls, topic, qos):++ p = Pub(cls, topic)++ created["pubs"].append(p)++ return p++++ def create_subscription(self, cls, topic, cb, qos):++ created["subs"].append((cls, topic, cb))++++ def destroy_node(self):++ created["destroyed"] = True++++ class Executor:++ def __init__(self):++ self.stop = threading.Event()++++ def add_node(self, n):++ pass++++ def spin(self):++ self.stop.wait(5)++++ def shutdown(self):++ self.stop.set()++++ rclpy = types.SimpleNamespace(__name__="rclpy", ok=lambda: False, init=lambda args=None: created.setdefault("init", True),++ create_node=lambda name: Node(), executors=types.SimpleNamespace(SingleThreadedExecutor=Executor),++ try_shutdown=lambda: created.setdefault("shutdown", True))++ geometry = types.ModuleType("geometry_msgs.msg")++ geometry.Twist = Twist++ monkeypatch.setitem(sys.modules, "geometry_msgs", types.ModuleType("geometry_msgs"))++ monkeypatch.setitem(sys.modules, "geometry_msgs.msg", geometry)++ bus = RclpyBus("openvibe_node", lambda *a: None, rclpy=rclpy)++ bus.publish("/cmd_vel", "geometry_msgs/msg/Twist", {"linear": {"x": 0.1, "y": 0.0, "z": 0.0}, "angular": {"x": 0.0, "y": 0.0, "z": 0.3}})++ assert created["init"] and created["pubs"][0].topic == "/cmd_vel" and created["pubs"][0].sent[0].angular.z == 0.3++ got = []++ bus.subscribe("/cmd_vel_echo", "geometry_msgs/msg/Twist", got.append)++ created["subs"][0][2](_fill(Twist(), {"linear": {"x": 0.5}}))++ assert got == [{"linear": {"x": 0.5, "y": 0.0, "z": 0.0}, "angular": {"x": 0.0, "y": 0.0, "z": 0.0}}]++ bus.close()++ assert created["destroyed"] and created["shutdown"]++++++# ── The WebSocket client and rosbridge, against a small local server ──────────────────────────────────────────────++class TinyServer:++ """A one-connection WebSocket server: records text frames it receives; send() pushes a text frame."""++++ def __init__(self, bad_accept=False):++ self.srv = socket.socket()++ self.srv.bind(("127.0.0.1", 0))++ self.srv.listen(1)++ self.port = self.srv.getsockname()[1]++ self.got = []++ self.conn = None++ self.bad_accept = bad_accept++ self.ready = threading.Event()++ threading.Thread(target=self._serve, daemon=True).start()++++ def _serve(self):++ conn, _ = self.srv.accept()++ req = b""++ while b"\r\n\r\n" not in req:++ req += conn.recv(4096)++ key = [l.split(b":", 1)[1].strip() for l in req.split(b"\r\n") if l.lower().startswith(b"sec-websocket-key")][0]++ accept = base64.b64encode(hashlib.sha1(key + GUID.encode()).digest())++ if self.bad_accept:++ accept = b"nope"++ conn.sendall(b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: " + accept + b"\r\n\r\n")++ self.conn = conn++ self.ready.set()++ buf = b""++ while True:++ try:++ chunk = conn.recv(65536)++ except OSError:++ return++ if not chunk:++ return++ buf += chunk++ while len(buf) >= 2:++ n = buf[1] & 0x7F++ off = 2++ if n == 126:++ n = struct.unpack(">H", buf[2:4])[0]++ off = 4++ elif n == 127:++ n = struct.unpack(">Q", buf[2:10])[0]++ off = 10++ if len(buf) < off + 4 + n:++ break++ mask = buf[off:off + 4]++ payload = bytes(b ^ mask[i % 4] for i, b in enumerate(buf[off + 4:off + 4 + n]))++ op = buf[0] & 0x0F++ buf = buf[off + 4 + n:]++ if op == 0x1:++ self.got.append(json.loads(payload))++ elif op == 0x8:++ return++++ def send(self, obj, opcode=0x1):++ data = json.dumps(obj).encode() if opcode == 0x1 else obj++ n = len(data)++ head = bytes([0x80 | opcode]) + (bytes([n]) if n < 126 else bytes([126]) + struct.pack(">H", n) if n < 65536 else bytes([127]) + struct.pack(">Q", n))++ self.conn.sendall(head + data)++++ def wait_for(self, pred, timeout=3):++ end = time.time() + timeout++ while time.time() < end:++ if any(pred(m) for m in self.got):++ return True++ time.sleep(0.01)++ return False++++++def test_rosbridge_bus_advertises_publishes_subscribes_and_dispatches():++ s = TinyServer()++ bus = RosbridgeBus("ws://127.0.0.1:%d" % s.port, lambda *a: None)++ s.ready.wait(2)++ got = []++ bus.subscribe("/battery_state", "sensor_msgs/msg/BatteryState", got.append, 1000)++ bus.publish("/cmd_vel", "geometry_msgs/msg/Twist", {"linear": {"x": 0.1}})++ bus.publish("/cmd_vel", "geometry_msgs/msg/Twist", {"linear": {"x": 0.0}})++ assert s.wait_for(lambda m: m.get("op") == "publish" and m["msg"]["linear"]["x"] == 0.0)++ ops = [(m["op"], m["topic"]) for m in s.got]++ assert ops == [("subscribe", "/battery_state"), ("advertise", "/cmd_vel"), ("publish", "/cmd_vel"), ("publish", "/cmd_vel")], "advertised once"++ assert s.got[0]["type"] == "sensor_msgs/msg/BatteryState" and s.got[0]["throttle_rate"] == 1000++ s.send(b"", opcode=0x9) # a ping is answered and skipped++ big = "x" * 70000 # a 64-bit-length frame++ s.send({"op": "publish", "topic": "/battery_state", "msg": {"percentage": 0.9, "note": big}})++ end = time.time() + 3++ while not got and time.time() < end:++ time.sleep(0.01)++ assert got and got[0]["percentage"] == 0.9 and len(got[0]["note"]) == 70000++ s.conn.shutdown(socket.SHUT_RDWR) # FIN even while the server thread is in recv++ s.conn.close()++ end = time.time() + 3++ while bus.alive and time.time() < end:++ time.sleep(0.01)++ assert not bus.alive, "a closed connection marks the bus down"++ bus.close()++++++def test_websocket_refuses_a_bad_handshake_and_other_schemes():++ s = TinyServer(bad_accept=True)++ with pytest.raises(WebSocketError):++ WebSocket("ws://127.0.0.1:%d/" % s.port)++ with pytest.raises(WebSocketError):++ WebSocket("http://127.0.0.1:9090")++++++def test_defaults_document_every_config_key():++ readme = open(os.path.join(os.path.dirname(HERE), "README.md")).read()++ for k in DEFAULTS:++ assert "`%s`" % k in readme, k@@@ -79,6 +79,9 @@ class Plugin:version = "0.0.0"motion_kinds = MOTION_KINDStelemetry_interval_s = 0.5++ # Seconds between attempts to set up again after setup raised (the hardware or the robot's computer was not up++ # yet). None: a failed setup stays failed until the core restarts the plugin.++ retry_setup_s = Noneruntime = None # set by Runtime@@@ -156,6 +159,7 @@ class Runtime:self.last_heartbeat = Noneself.heartbeat_lost = Falseself.next_poll = 0.0++ self.setup_retry_at = Noneself.closed = Falseself.stop_reasons = [] # for tests: why stop() was called, in order@@@ -200,6 +204,9 @@ class Runtime:self.heartbeat_lost = Trueself.safe_stop("heartbeat")self.send({"op": "event", "name": "heartbeat_lost"})++ if not self.ready and self.setup_retry_at is not None and now >= self.setup_retry_at and not self.closed:++ self.setup_retry_at = None++ self._setup(retrying=True)if self.ready and now >= self.next_poll:self.next_poll = now + self.plugin.telemetry_interval_stry:@@@ -251,6 +258,11 @@ class Runtime:"motion_kinds": sorted(self.plugin.motion_kinds)})if self.probe:return++ self._setup()++++ def _setup(self, retrying=False):++ """Set the plugin up; ready on success, else a fault (sent once, not on every retry) and, when the plugin++ sets retry_setup_s, another attempt that long from now. A later success sends ready, which ends the fault."""try:with self.lock:self.plugin.setup(self.config)@@@ -258,10 +270,15 @@ class Runtime:self.ready = Trueself.last_heartbeat = self.clock()self.send({"op": "ready"})++ returnexcept Fault as f:-- self.send({"op": "fault", "fault_code": f.code, "message": f.message})++ fault = {"op": "fault", "fault_code": f.code, "message": f.message}except Exception as e:-- self.send({"op": "fault", "fault_code": "hardware", "message": str(e)})++ fault = {"op": "fault", "fault_code": "hardware", "message": str(e)}++ if not retrying:++ self.send(fault)++ if self.plugin.retry_setup_s:++ self.setup_retry_at = self.clock() + float(self.plugin.retry_setup_s)def on_command(self, msg, now=None):cid = msg.get("id")@@@ -162,6 +162,49 @@ class RuntimeTest(unittest.TestCase):rt.feed_line(json.dumps({"op": "command", "id": "c3", "kind": "drive"}))self.assertEqual(lines(out)[-1]["op"], "ack")++ def test_a_failed_setup_stays_failed_without_retry_setup_s(self):++ class Down(Recorder):++ def setup(self, config):++ raise Fault("unreachable", "not up yet")++ rt, p, out, clock = make(Down())++ self.hello(rt)++ clock.t += 60++ rt.tick()++ self.assertEqual([m["op"] for m in lines(out)], ["describe", "fault"])++ self.assertEqual(p.calls.count("describe"), 1)++++ def test_retry_setup_s_sets_up_again_and_ready_ends_the_fault(self):++ class Late(Recorder):++ retry_setup_s = 2++ tries = 0++++ def setup(self, config):++ Late.tries += 1++ if Late.tries < 3:++ raise Fault("unreachable", "not up yet")++ self.calls.append("setup")++ rt, p, out, clock = make(Late())++ self.hello(rt)++ rt.feed_line(json.dumps({"op": "command", "id": "c0", "kind": "drive", "value": {}}))++ self.assertEqual(lines(out)[-1]["fault_code"], "not_ready")++ clock.t += 1++ rt.tick()++ self.assertEqual(Late.tries, 1, "not before retry_setup_s")++ clock.t += 1.5++ rt.tick()++ self.assertEqual(Late.tries, 2)++ self.assertEqual([m["op"] for m in lines(out)].count("fault"), 1, "the fault is sent once, not on every retry")++ clock.t += 2.5++ rt.tick()++ self.assertEqual(Late.tries, 3)++ self.assertEqual(lines(out)[-1]["op"], "ready")++ self.assertTrue(rt.ready)++ rt.feed_line(json.dumps({"op": "command", "id": "c1", "kind": "drive", "value": {}}))++ self.assertEqual(lines(out)[-1], {"op": "ack", "id": "c1"})++ clock.t += 10++ rt.tick()++ self.assertEqual(Late.tries, 3, "no more attempts once ready")++def test_halt_always_accepted(self):rt, p, out, clock = make()self.hello(rt)@@@ -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 relay++ sdk dryrun adeept_adr036 cozmo relay ros2# 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
@@ -90,8 +90,9 @@ jobs:
@@ -90,8 +90,9 @@ jobs:
- run: python -m pytest -q plugins/adeept_adr036/tests
- run: python -m pytest -q plugins/adeept_adr036/tests
- run: python -m pytest -q plugins/cozmo/tests
- run: python -m pytest -q plugins/cozmo/tests
- run: python -m pytest -q plugins/relay/tests
- run: python -m pytest -q plugins/relay/tests
+ - run: python -m pytest -q plugins/ros2/tests
- name: packages install
- name: packages install
- run: pip install ./plugins/sdk ./plugins/dryrun ./plugins/adeept_adr036 ./plugins/cozmo ./plugins/relay
+ run: pip install ./plugins/sdk ./plugins/dryrun ./plugins/adeept_adr036 ./plugins/cozmo ./plugins/relay ./plugins/ros2
shell:
shell:
name: shellcheck
name: shellcheck
@@ -23,7 +23,7 @@ jobs:
@@ -23,7 +23,7 @@ jobs:
- run: sudo sysctl -w kernel.apparmor_restrict_unprivileged_userns=0 || true
- run: sudo sysctl -w kernel.apparmor_restrict_unprivileged_userns=0 || true
- run: go vet ./... && go test -count=1 ./...
- run: go vet ./... && go test -count=1 ./...
- run: |
- run: |
- for p in sdk dryrun adeept_adr036 cozmo relay; do python -m pytest -q plugins/$p/tests; done
+ for p in sdk dryrun adeept_adr036 cozmo relay ros2; do python -m pytest -q plugins/$p/tests; done
- run: scripts/dist.sh "${GITHUB_REF_NAME}"
- 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
# 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
# documented to serve (releases/latest/download/install.sh), the plugin bundle the installer downloads, and the
@@ -8,7 +8,7 @@ for AI agents) can use the device. It connects out, needs no open ports, and kee
@@ -8,7 +8,7 @@ for AI agents) can use the device. It connects out, needs no open ports, and kee
curl -fsSL https://openvibe.bot/install | sh -s -- --robot rob_… --code ABCD-1234 --driver adeept
curl -fsSL https://openvibe.bot/install | sh -s -- --robot rob_… --code ABCD-1234 --driver adeept
```
```
-openvibe.bot shows this command with the robot's id and a fresh code. `--driver` is `adeept`, `adeept-mecanum`, `cozmo`,
+openvibe.bot shows this command with the robot's id and a fresh code. `--driver` is `adeept`, `adeept-mecanum`, `cozmo`, `ros2`,
`looi` (the community's Generic LOOI driver, [docs/hardware-looi.md](docs/hardware-looi.md)) or `none` (dry run; the old `--robot adeept` form still works but is deprecated). A robot paired through
`looi` (the community's Generic LOOI driver, [docs/hardware-looi.md](docs/hardware-looi.md)) or `none` (dry run; the old `--robot adeept` form still works but is deprecated). A robot paired through
OpenVibe.Network gets `--network <URL> --pairing pair_… --code ABCD-1234` instead of `--robot`. Details, manual
OpenVibe.Network gets `--network <URL> --pairing pair_… --code ABCD-1234` instead of `--robot`. Details, manual
install, Windows and configuration: [docs/install.md](docs/install.md).
install, Windows and configuration: [docs/install.md](docs/install.md).
@@ -27,7 +27,8 @@ openvibe-node resume
@@ -27,7 +27,8 @@ openvibe-node resume
publishing with pion, and the system service (systemd, launchd, Windows).
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
- **Plugins** (separate processes, JSON lines over stdin/stdout): [`dryrun`](plugins/dryrun), the Adeept 4WD Smart Car
Kit [`adeept_adr036`](plugins/adeept_adr036/README.md), [`cozmo`](plugins/cozmo/README.md) through PyCozmo, and
Kit [`adeept_adr036`](plugins/adeept_adr036/README.md), [`cozmo`](plugins/cozmo/README.md) through PyCozmo, and
- [`relay`](plugins/relay) for existing Live hardware clients. LOOI's driver is the community's
+ [`relay`](plugins/relay) for existing Live hardware clients, and [`ros2`](plugins/ros2/README.md) for ROS 2 robots
+ (`cmd_vel`, battery, scan/range, odometry and camera, as a node with rclpy or through rosbridge). LOOI's driver is the community's
([docs/hardware-looi.md](docs/hardware-looi.md) is its contract). Each plugin stops its own actuators when stdin closes,
([docs/hardware-looi.md](docs/hardware-looi.md) is its contract). 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.
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 —
- **ESP32 library** ([`esp32/`](esp32/README.md), `OpenVibeNode`): the device side of the protocol for ESP32 boards —
@@ -13,10 +13,10 @@ ports, works behind any home router), and a local kill switch stops everything i
@@ -13,10 +13,10 @@ ports, works behind any home router), and a local kill switch stops everything i
curl -fsSL https://openvibe.bot/install | sh -s -- --robot rob_01J8Z4M2Q0R7T9YV3K6N8P1W2X --code ABCD-1234 --driver adeept
curl -fsSL https://openvibe.bot/install | sh -s -- --robot rob_01J8Z4M2Q0R7T9YV3K6N8P1W2X --code ABCD-1234 --driver adeept
```
```
- The shape is `--robot rob_… --code CODE [--driver adeept|adeept-mecanum|cozmo|looi|none]`. `--robot` is the robot's
+ The shape is `--robot rob_… --code CODE [--driver adeept|adeept-mecanum|cozmo|looi|ros2|ros2-holonomic|none]`. `--robot` is the robot's
`rob_…` id and `--code` the pairing code (both from openvibe.bot; the code may also be the first argument).
`rob_…` id and `--code` the pairing code (both from openvibe.bot; the code may also be the first argument).
- `--driver` names the plugin: `adeept` (Adeept 4WD Smart Car, ordinary wheels), `adeept-mecanum`, `cozmo`, or
- `none` (the default: the dry-run plugin with a test-pattern camera, for trying the Node on a laptop). `--name NAME`
+ `--driver` names the plugin: `adeept` (Adeept 4WD Smart Car, ordinary wheels), `adeept-mecanum`, `cozmo`, `ros2` (a
+ ROS 2 robot through rosbridge, [plugins/ros2/README.md](../plugins/ros2/README.md)), `ros2-holonomic`, or `none` (the default: the dry-run plugin with a test-pattern camera, for trying the Node on a laptop). `--name NAME`
names the device on openvibe.bot (default: the hostname).
names the device on openvibe.bot (default: the hostname).
3. Confirm the new device on openvibe.bot.
3. Confirm the new device on openvibe.bot.
@@ -5,7 +5,7 @@ object per line, UTF-8). Its stderr goes to the Node's log. The core never links
@@ -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.
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),
Bundled plugins: [`dryrun`](../plugins/dryrun), [`adeept_adr036`](../plugins/adeept_adr036/README.md),
-[`cozmo`](../plugins/cozmo/README.md), [`relay`](../plugins/relay). Python plugins use the runtime in [`plugins/sdk`](../plugins/sdk) (package
+[`cozmo`](../plugins/cozmo/README.md), [`relay`](../plugins/relay), [`ros2`](../plugins/ros2/README.md). 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
`openvibe_plugin`), which implements everything in "Safety rules" below, so a driver only says how to move its
hardware.
hardware.
@@ -95,6 +95,10 @@ Only `openvibe-node resume` on that machine releases it: the owner's remote clea
@@ -95,6 +95,10 @@ Only `openvibe-node resume` on that machine releases it: the owner's remote clea
releases a latch. The Python runtime refuses motion from the moment it is called and ignores a `resume` that crosses it
releases a latch. The Python runtime refuses motion from the moment it is called and ignores a `resume` that crosses it
on the wire. `openvibe-node stop` sets the same latch from a terminal.
on the wire. `openvibe-node stop` sets the same latch from a terminal.
+A plugin whose setup can fail only because something is not up yet (the robot's computer still booting) sets
+`retry_setup_s` (Python): the runtime sends the fault once, sets up again that often, and sends `ready` when it works,
+which ends the fault. Without it, a failed setup stays failed until the core restarts the plugin.
+
`stop` must be idempotent and safe at any time, including before setup. Handlers must not block: slow work (speech
`stop` must be idempotent and safe at any time, including before setup. Handlers must not block: slow work (speech
synthesis, file I/O) goes on a thread. Nothing but protocol lines may go to stdout; the Python runtime enforces this by
synthesis, file I/O) goes on a thread. Nothing but protocol lines may go to stdout; the Python runtime enforces this by
pointing file descriptor 1 at stderr and keeping a private copy for the protocol, because hardware libraries print
pointing file descriptor 1 at stderr and keeping a private copy for the protocol, because hardware libraries print
@@ -151,6 +155,17 @@ camera frames (320 × 240) as JPEG files. A cliff or pick-up stops the wheels in
@@ -151,6 +155,17 @@ 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
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`.
supported version (2381, `allow_firmware`) is reported as `firmware_unsupported`.
+### `ros2` — ROS 2 robots (rclpy or rosbridge)
+
+`drive` → `geometry_msgs/msg/Twist` on `cmd_vel` (`throttle`/`steer`, or `x`/`y`/`rotation` with `holonomic`), scaled by
+`max_linear` m/s and `max_angular` rad/s, and a zero Twist at every stop; `say` → `std_msgs/msg/String` on `say_topic`.
+Telemetry from `sensor_msgs/msg/BatteryState`, `sensor_msgs/msg/Range` or the front arc of a `sensor_msgs/msg/LaserScan`
+(`distance_cm`) and `nav_msgs/msg/Odometry`; camera frames from a JPEG `sensor_msgs/msg/CompressedImage`. It reaches ROS 2
+as a node of its own (`rclpy`, in a sourced ROS 2 environment) or through a rosbridge WebSocket server (no ROS 2
+libraries where the Node runs; the installer's `--driver ros2` sets this up). When ROS is not up yet, or the link drops,
+it reports `ros_unreachable` and tries again every `retry_s` seconds (`retry_setup_s` in the Python runtime). Topics,
+speeds and setup: [plugins/ros2/README.md](../plugins/ros2/README.md).
+
### `relay` — existing OpenVibe.Live hardware clients
### `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
Use this when the owner already has a local script receiving Live stream controls. In `config.json`, replace the
@@ -1,7 +1,7 @@
@@ -1,7 +1,7 @@
#!/bin/sh
#!/bin/sh
# OpenVibe Node installer.
# OpenVibe Node installer.
#
#
-# curl -fsSL https://openvibe.bot/install | sh -s -- --robot <rob_…> --code <CODE> [--driver adeept|adeept-mecanum|cozmo|looi|none]
+# curl -fsSL https://openvibe.bot/install | sh -s -- --robot <rob_…> --code <CODE> [--driver adeept|adeept-mecanum|cozmo|looi|ros2|ros2-holonomic|none]
# curl -fsSL https://openvibe.bot/install | sh -s -- --network <URL> --pairing <pair_…> --code <CODE> [--driver …]
# curl -fsSL https://openvibe.bot/install | sh -s -- --network <URL> --pairing <pair_…> --code <CODE> [--driver …]
#
#
# Detects the OS and CPU, downloads the openvibe-node binary and the plugin bundle (checking their SHA-256), creates
# Detects the OS and CPU, downloads the openvibe-node binary and the plugin bundle (checking their SHA-256), creates
@@ -16,7 +16,8 @@
@@ -16,7 +16,8 @@
# --network URL pair through OpenVibe.Network at URL (the machine gets a node credential, then Bot binds it).
# --network URL pair through OpenVibe.Network at URL (the machine gets a node credential, then Bot binds it).
# --pairing ID the pair_… id of that Network pairing; a code given here instead (with no --code) is the code.
# --pairing ID the pair_… id of that Network pairing; a code given here instead (with no --code) is the code.
# --name NAME the device's name on openvibe.bot (default: the hostname).
# --name NAME the device's name on openvibe.bot (default: the hostname).
-# --driver KIND adeept (ordinary wheels), adeept-mecanum, cozmo (bridge), looi (bridge, Bluetooth), none (dry run).
+# --driver KIND adeept (ordinary wheels), adeept-mecanum, cozmo (bridge), looi (bridge, Bluetooth), ros2 (a ROS 2
+# robot through rosbridge), ros2-holonomic (mecanum or omni wheels), none (dry run).
# Default: none.
# Default: none.
# --no-service install and pair, but do not install the service.
# --no-service install and pair, but do not install the service.
# --local DIR install from DIR (openvibe-node-<os>-<arch> and openvibe-node-plugins.tar.gz) instead of
# --local DIR install from DIR (openvibe-node-<os>-<arch> and openvibe-node-plugins.tar.gz) instead of
@@ -47,7 +48,7 @@ TAG="latest"
@@ -47,7 +48,7 @@ TAG="latest"
usage() {
usage() {
echo "usage: install.sh [--robot rob_… | --network URL --pairing pair_…] [--code CODE | CODE] [--name NAME]"
echo "usage: install.sh [--robot rob_… | --network URL --pairing pair_…] [--code CODE | CODE] [--name NAME]"
- echo " [--driver adeept|adeept-mecanum|cozmo|looi|none] [--no-service] [--local DIR] [--version TAG]"
+ echo " [--driver adeept|adeept-mecanum|cozmo|looi|ros2|ros2-holonomic|none] [--no-service] [--local DIR] [--version TAG]"
echo " curl -fsSL https://openvibe.bot/install | sh -s -- --robot rob_… --code ABCD-1234 --driver adeept"
echo " curl -fsSL https://openvibe.bot/install | sh -s -- --robot rob_… --code ABCD-1234 --driver adeept"
echo " curl -fsSL https://openvibe.bot/install | sh -s -- --network https://openvibe.network --pairing pair_… --code ABCD-1234"
echo " curl -fsSL https://openvibe.bot/install | sh -s -- --network https://openvibe.network --pairing pair_… --code ABCD-1234"
}
}
@@ -84,7 +85,7 @@ while [ $# -gt 0 ]; do
@@ -84,7 +85,7 @@ while [ $# -gt 0 ]; do
esac
esac
done
done
-case "$DRIVER" in adeept|adeept-mecanum|cozmo|looi|none) ;; *) die "--driver must be adeept, adeept-mecanum, cozmo, looi or none" ;; esac
+case "$DRIVER" in adeept|adeept-mecanum|cozmo|looi|ros2|ros2-holonomic|none) ;; *) die "--driver must be adeept, adeept-mecanum, cozmo, looi, ros2, ros2-holonomic or none" ;; esac
case "$ROBOT_ID" in ""|rob_*) ;; *) die "--robot must be a rob_… id" ;; esac
case "$ROBOT_ID" in ""|rob_*) ;; *) die "--robot must be a rob_… id" ;; esac
# --pairing takes the pair_… id; with no --code, anything else there is the code itself.
# --pairing takes the pair_… id; with no --code, anything else there is the code itself.
case "$PAIRING" in ""|pair_*) ;; *) [ -z "$CODE" ] || die "--pairing must be a pair_… id"; CODE="$PAIRING"; PAIRING="" ;; esac
case "$PAIRING" in ""|pair_*) ;; *) [ -z "$CODE" ] || die "--pairing must be a pair_… id"; CODE="$PAIRING"; PAIRING="" ;; esac
@@ -316,6 +317,15 @@ cozmo)
@@ -316,6 +317,15 @@ cozmo)
KIND=bridge
KIND=bridge
command -v espeak-ng >/dev/null 2>&1 || say "tip: install espeak-ng so Cozmo can speak (sudo apt install espeak-ng)"
command -v espeak-ng >/dev/null 2>&1 || say "tip: install espeak-ng so Cozmo can speak (sudo apt install espeak-ng)"
;;
;;
+ros2|ros2-holonomic)
+ # A ROS 2 robot through rosbridge (plugins/ros2/README.md): no ROS 2 libraries are needed here, and the service
+ # needs no sourced ROS environment. Topics, speeds and the rosbridge address are plugin config in config.json.
+ $SUDO "$PIP" install --quiet "$STATE/plugins/ros2"
+ HOLO=false; [ "$DRIVER" = ros2-holonomic ] && HOLO=true
+ PLUGINS="[{\"name\": \"ros2\", \"config\": {\"transport\": \"rosbridge\", \"rosbridge_url\": \"ws://127.0.0.1:9090\", \"holonomic\": $HOLO}}]"
+ say "tip: on the robot, run a rosbridge server (ros2 launch rosbridge_server rosbridge_websocket_launch.xml) and set"
+ say " rosbridge_url, cmd_vel, max_linear and image_topic in $CONF/config.json if yours differ"
+ ;;
looi)
looi)
# LOOI's camera is its phone, which streams on its own (docs/hardware-looi.md): the Node publishes no video.
# LOOI's camera is its phone, which streams on its own (docs/hardware-looi.md): the Node publishes no video.
PLUGINS='[{"name": "looi"}]'
PLUGINS='[{"name": "looi"}]'
@@ -336,6 +336,7 @@ var Bundled = map[string]string{
@@ -336,6 +336,7 @@ var Bundled = map[string]string{
"adeept_adr036": "openvibe_adeept_adr036",
"adeept_adr036": "openvibe_adeept_adr036",
"cozmo": "openvibe_cozmo",
"cozmo": "openvibe_cozmo",
"relay": "openvibe_relay",
"relay": "openvibe_relay",
+ "ros2": "openvibe_ros2",
}
}
// Default is the config written by `openvibe-node pair` when none exists: the dry-run plugin and its test pattern.
// Default is the config written by `openvibe-node pair` when none exists: the dry-run plugin and its test pattern.
new file mode 100644
new file mode 100644
@@ -0,0 +1,80 @@
@@ -0,0 +1,80 @@
+# `ros2`: ROS 2 robots
+
+This plugin drives a ROS 2 robot from OpenVibe.Bot: TurtleBot 3 or 4, a Jackal, a home-built base, anything that takes
+`geometry_msgs/msg/Twist` on a `cmd_vel` topic. The Node runs on the robot's computer, or on a computer on the same
+network. The plugin turns Bot's commands into ROS 2 messages, and the robot's topics into telemetry and camera frames.
+Bot's gate (roles, the owner's limits, cooldowns and the e-stop) has already judged every command before it gets here.
+
+## What it bridges
+
+| Bot | ROS 2 |
+|---|---|
+| `drive` `throttle`/`steer` | `geometry_msgs/msg/Twist` on `cmd_vel`: `linear.x = throttle × max_linear`, `angular.z = −steer × max_angular` (Bot's steer is positive to the right; ROS's z turns counter-clockwise) |
+| `drive` `x`/`y`/`rotation` (`holonomic: true`) | `linear.x`, `linear.y`, `angular.z`. Bot's x is forward, y is left and rotation is counter-clockwise, as in REP 103 |
+| any stop | a zero Twist: the last drive's deadline, a lost heartbeat, `stop`, `estop`, `halt`, stdin closing, before exit |
+| `say` | `std_msgs/msg/String` on `say_topic`, when it is set |
+| battery | `sensor_msgs/msg/BatteryState` on `battery_topic`: `percentage` (0–1) and `voltage` |
+| `distance_cm` | `sensor_msgs/msg/Range` on `range_topic`, or else the nearest return within `front_arc_deg` of straight ahead on a `sensor_msgs/msg/LaserScan` (`scan_topic`) |
+| `odom` | `nav_msgs/msg/Odometry` on `odom_topic`: x, y, yaw and speed |
+| camera | a JPEG `sensor_msgs/msg/CompressedImage` on `image_topic`, at most `fps` frames a second; the Node publishes them as the robot's video |
+
+## Two ways to reach ROS 2
+
+- **`rclpy`**: the plugin is a ROS 2 node itself (named `node_name`). Run the Node where ROS 2 is installed, with its
+ environment sourced (`source /opt/ros/jazzy/setup.bash`) so `rclpy` imports.
+- **`rosbridge`**: JSON over WebSocket to a rosbridge server at `rosbridge_url`. Start one on the robot with
+ `sudo apt install ros-$ROS_DISTRO-rosbridge-server` and `ros2 launch rosbridge_server rosbridge_websocket_launch.xml`.
+ No ROS 2 libraries are needed where the Node runs, so a Raspberry Pi beside the robot, or a laptop, works. `wss://`
+ works too.
+- **`auto`** (the default) uses `rclpy` when it imports, otherwise `rosbridge`.
+
+If ROS cannot be reached, the panel shows the `ros_unreachable` fault and commands are refused with it. The plugin
+tries again every `retry_s` seconds; when it reconnects, it publishes a zero Twist and says `ready`, which ends the
+fault.
+
+## Setup
+
+1. On openvibe.bot, add a robot with the **ROS 2 robot** model (`ros2.diffdrive`; `ros2.holonomic` for mecanum or
+ omni wheels) and keep its pairing code.
+2. On the computer that will run the Node:
+ `curl -fsSL https://openvibe.bot/install | sh -s -- --robot <rob_…> --code <CODE> --driver ros2`
+3. Edit `config.json` (the installer prints where it is) if your topics differ from the defaults, then restart the
+ Node's service:
+
+```json
+{"plugins": [{"name": "ros2", "config": {
+ "transport": "rosbridge",
+ "rosbridge_url": "ws://turtlebot.local:9090",
+ "cmd_vel": "/cmd_vel",
+ "max_linear": 0.22,
+ "max_angular": 2.0,
+ "image_topic": "/camera/image_raw/compressed"
+}}]}
+```
+
+TurtleBot 4 publishes its camera as `/oakd/rgb/preview/image_raw/compressed` (with `image_transport` republishing),
+and its base expects `cmd_vel` under the robot's namespace when one is set.
+
+## Configuration
+
+| key | default | meaning |
+|---|---|---|
+| `transport` | `auto` | `auto`, `rclpy` or `rosbridge` |
+| `rosbridge_url` | `ws://127.0.0.1:9090` | the rosbridge server (`ws://` or `wss://`) |
+| `node_name` | `openvibe_node` | the ROS 2 node's name with `rclpy` |
+| `cmd_vel` | `/cmd_vel` | where Twists go |
+| `max_linear` | `0.3` | m/s at full throttle (0–5) |
+| `max_angular` | `1.5` | rad/s at full steer (0–10) |
+| `holonomic` | `false` | `x`/`y`/`rotation` drive (mecanum, omni) instead of `throttle`/`steer` |
+| `battery_topic` | `/battery_state` | `sensor_msgs/msg/BatteryState`; empty turns it off |
+| `range_topic` | (empty) | `sensor_msgs/msg/Range`; replaces the scan for `distance_cm` |
+| `scan_topic` | `/scan` | `sensor_msgs/msg/LaserScan`; empty turns it off |
+| `front_arc_deg` | `30` | how wide "straight ahead" is on the scan |
+| `odom_topic` | `/odom` | `nav_msgs/msg/Odometry`; empty turns it off |
+| `image_topic` | (empty) | a JPEG `sensor_msgs/msg/CompressedImage`; empty means no camera from ROS |
+| `fps` | `5` | camera frames a second (0.2–15) |
+| `say_topic` | (empty) | `std_msgs/msg/String` for `say`; empty refuses `say` |
+| `retry_s` | `3` | seconds between reconnection attempts |
+
+Bot's limits apply before the plugin's: the owner's `max_speed` and `max_turn` scale the command first, then
+`max_linear` and `max_angular` turn it into m/s and rad/s. Set these two to what is safe for your robot indoors.
new file mode 100644
new file mode 100644
@@ -0,0 +1,438 @@
@@ -0,0 +1,438 @@
+"""OpenVibe Node bridge plugin for ROS 2 robots.
+
+The Node runs on the robot's computer (or next to it) and this plugin turns Bot's commands into ROS 2 messages and the
+robot's topics into telemetry and camera frames:
+
+* ``drive`` → ``geometry_msgs/msg/Twist`` on ``cmd_vel`` (``throttle``/``steer``, or ``x``/``y``/``rotation`` on a
+ holonomic base), scaled by ``max_linear`` m/s and ``max_angular`` rad/s. A zero Twist is published at every stop:
+ the deadline of the last drive, a lost heartbeat, ``stop``, ``estop``, ``halt``, stdin closing, and before exit.
+* ``say`` → ``std_msgs/msg/String`` on ``say_topic`` (when set).
+* ``sensor_msgs/msg/BatteryState`` → battery percent and volts; ``sensor_msgs/msg/Range`` or the front arc of a
+ ``sensor_msgs/msg/LaserScan`` → ``distance_cm``; ``nav_msgs/msg/Odometry`` → ``odom`` {x, y, yaw, speed}.
+* ``sensor_msgs/msg/CompressedImage`` (JPEG) on ``image_topic`` → camera frames for the Node to publish.
+
+Two ways to reach ROS 2 (``transport``):
+
+* ``rclpy``: the plugin is a ROS 2 node itself. Needs a sourced ROS 2 environment (``rclpy`` importable).
+* ``rosbridge``: JSON over WebSocket to a rosbridge server (``ros2 launch rosbridge_server
+ rosbridge_websocket_launch.xml``), on this machine or the robot's. No ROS 2 libraries needed here.
+* ``auto`` (default): ``rclpy`` when it imports, otherwise ``rosbridge``.
+
+If ROS cannot be reached the plugin reports the ``ros_unreachable`` fault, refuses commands with it, and tries again
+every ``retry_s`` seconds; when it reconnects it says ``ready`` again, which ends the fault.
+"""
+
+import base64
+import importlib
+import importlib.util
+import json
+import math
+import os
+import shutil
+import tempfile
+import threading
+import time
+
+from openvibe_plugin import Fault, Plugin, clamp
+
+from .ws import WebSocket
+
+FRAME_RING = 4
+
+DEFAULTS = {
+ "transport": "auto", # auto | rclpy | rosbridge
+ "rosbridge_url": "ws://127.0.0.1:9090",
+ "node_name": "openvibe_node",
+ "cmd_vel": "/cmd_vel",
+ "max_linear": 0.3, # m/s at full throttle
+ "max_angular": 1.5, # rad/s at full steer
+ "holonomic": False, # x/y/rotation drive (mecanum, omni) instead of throttle/steer
+ "battery_topic": "/battery_state",
+ "range_topic": "", # sensor_msgs/Range (an ultrasonic or ToF sensor)
+ "scan_topic": "/scan", # sensor_msgs/LaserScan: the nearest return in the front arc
+ "front_arc_deg": 30,
+ "odom_topic": "/odom",
+ "image_topic": "", # sensor_msgs/CompressedImage, JPEG
+ "fps": 5,
+ "say_topic": "", # std_msgs/String
+ "retry_s": 3,
+}
+
+TWIST = "geometry_msgs/msg/Twist"
+
+
+def _num(v, default=0.0):
+ try:
+ f = float(v)
+ except (TypeError, ValueError):
+ return default
+ return f if f == f else default
+
+
+# ── Transports ─────────────────────────────────────────────────────────────────────────────────────────────────────
+class RosbridgeBus:
+ """rosbridge v2 over WebSocket: advertise + publish, subscribe; incoming publishes go to the topic's callbacks."""
+
+ def __init__(self, url, log, connect=WebSocket):
+ self.log = log
+ self.ws = connect(url)
+ self.alive = True
+ self.subs = {}
+ self.advertised = set()
+ self.thread = threading.Thread(target=self._read, name="rosbridge", daemon=True)
+ self.thread.start()
+
+ def _send(self, obj):
+ self.ws.send_text(json.dumps(obj, separators=(",", ":")))
+
+ def publish(self, topic, type_, msg):
+ if topic not in self.advertised:
+ self._send({"op": "advertise", "topic": topic, "type": type_})
+ self.advertised.add(topic)
+ self._send({"op": "publish", "topic": topic, "msg": msg})
+
+ def subscribe(self, topic, type_, callback, throttle_ms=0):
+ self.subs.setdefault(topic, []).append(callback)
+ self._send({"op": "subscribe", "topic": topic, "type": type_, "throttle_rate": int(throttle_ms), "queue_length": 1})
+
+ def _read(self):
+ try:
+ while True:
+ text = self.ws.recv()
+ if text is None:
+ break
+ try:
+ m = json.loads(text)
+ except ValueError:
+ continue
+ if isinstance(m, dict) and m.get("op") == "publish":
+ for cb in self.subs.get(m.get("topic"), ()):
+ try:
+ cb(m.get("msg") or {})
+ except Exception as e:
+ self.log("ros2: %s callback failed: %r", m.get("topic"), e)
+ finally:
+ self.alive = False
+
+ def close(self):
+ self.alive = False
+ self.ws.close()
+
+
+def _to_dict(m):
+ """A ROS 2 message as plain data (fields by name; byte arrays as bytes)."""
+ if hasattr(m, "get_fields_and_field_types"):
+ return {k: _to_dict(getattr(m, k)) for k in m.get_fields_and_field_types()}
+ if isinstance(m, (bytes, bytearray)):
+ return bytes(m)
+ if hasattr(m, "tobytes") and getattr(m, "typecode", None) in ("B", "b"): # array.array('B')
+ return m.tobytes()
+ if isinstance(m, (list, tuple)) or (hasattr(m, "__iter__") and hasattr(m, "typecode")):
+ return [_to_dict(x) for x in m]
+ return m
+
+
+def _fill(m, d):
+ for k, v in d.items():
+ if isinstance(v, dict):
+ _fill(getattr(m, k), v)
+ else:
+ setattr(m, k, v)
+ return m
+
+
+class RclpyBus:
+ """The plugin as a ROS 2 node: publishers and subscriptions on its own executor thread."""
+
+ def __init__(self, node_name, log, rclpy=None):
+ self.log = log
+ self.rclpy = rclpy or importlib.import_module("rclpy")
+ if not self.rclpy.ok():
+ self.rclpy.init(args=None)
+ self.node = self.rclpy.create_node(node_name)
+ executors = importlib.import_module(self.rclpy.__name__ + ".executors") if not hasattr(self.rclpy, "executors") else self.rclpy.executors
+ self.executor = executors.SingleThreadedExecutor()
+ self.executor.add_node(self.node)
+ self.pubs = {}
+ self.alive = True
+ self.thread = threading.Thread(target=self._spin, name="rclpy", daemon=True)
+ self.thread.start()
+
+ def _spin(self):
+ try:
+ self.executor.spin()
+ except Exception as e:
+ self.log("ros2: executor stopped: %r", e)
+ finally:
+ self.alive = False
+
+ @staticmethod
+ def _cls(type_):
+ pkg, kind, name = type_.split("/")
+ return getattr(importlib.import_module("%s.%s" % (pkg, kind)), name)
+
+ def publish(self, topic, type_, msg):
+ pub = self.pubs.get(topic)
+ if pub is None:
+ pub = self.pubs[topic] = self.node.create_publisher(self._cls(type_), topic, 10)
+ pub.publish(_fill(self._cls(type_)(), msg))
+
+ def subscribe(self, topic, type_, callback, throttle_ms=0):
+ self.node.create_subscription(self._cls(type_), topic, lambda m: callback(_to_dict(m)), 1)
+
+ def close(self):
+ self.alive = False
+ try:
+ self.executor.shutdown()
+ self.node.destroy_node()
+ except Exception:
+ pass
+ try:
+ self.rclpy.try_shutdown()
+ except Exception:
+ pass
+
+
+def default_bus(cfg, log):
+ t = cfg["transport"]
+ if t == "rclpy" or (t == "auto" and importlib.util.find_spec("rclpy") is not None):
+ return RclpyBus(cfg["node_name"], log)
+ return RosbridgeBus(cfg["rosbridge_url"], log)
+
+
+# ── The plugin ─────────────────────────────────────────────────────────────────────────────────────────────────────
+class Ros2(Plugin):
+ driver, version = "ros2", "0.1.0"
+ motion_kinds = frozenset({"drive"})
+ retry_setup_s = DEFAULTS["retry_s"] # ROS not up yet at start: the runtime sets up again (setup below)
+
+ def __init__(self, bus_factory=default_bus, clock=time.monotonic):
+ self.bus_factory = bus_factory
+ self.clock = clock
+ self.cfg = dict(DEFAULTS)
+ self.bus = None
+ self.faulted = False
+ self.next_retry = 0.0
+ self.lock = threading.Lock()
+ self.battery = None
+ self.distance_cm = None
+ self.odom = None
+ self.frame_dir = None
+ self._own_frame_dir = False
+ self.frame_seq = 0
+ self.next_frame = 0.0
+ self._warned = set()
+
+ @staticmethod
+ def config_of(config):
+ cfg = dict(DEFAULTS)
+ cfg.update({k: v for k, v in (config or {}).items() if k in DEFAULTS or k == "frame_dir"})
+ if cfg["transport"] not in ("auto", "rclpy", "rosbridge"):
+ raise Fault("bad_config", "transport must be auto, rclpy or rosbridge")
+ cfg["max_linear"] = clamp(cfg["max_linear"], 0.0, 5.0)
+ cfg["max_angular"] = clamp(cfg["max_angular"], 0.0, 10.0)
+ cfg["fps"] = clamp(cfg["fps"], 0.2, 15.0)
+ return cfg
+
+ def describe(self, config):
+ cfg = self.config_of(config)
+ caps = {
+ "drive": {"type": "mecanum" if cfg["holonomic"] else "differential", "max_linear_mps": cfg["max_linear"], "max_angular_rps": cfg["max_angular"]},
+ "bridge": {"kind": "ros2", "transport": cfg["transport"], "cmd_vel": cfg["cmd_vel"]},
+ }
+ if cfg["battery_topic"]:
+ caps["battery"] = {"topic": cfg["battery_topic"]}
+ if cfg["range_topic"] or cfg["scan_topic"] or cfg["odom_topic"]:
+ caps["sensors"] = {"distance_cm": bool(cfg["range_topic"] or cfg["scan_topic"]), "odom": bool(cfg["odom_topic"])}
+ if cfg["image_topic"]:
+ caps["camera"] = {"jpeg": True}
+ if cfg["say_topic"]:
+ caps["say"] = {"topic": cfg["say_topic"]}
+ return caps
+
+ # Setup and reconnecting.
+ def setup(self, config):
+ self.cfg = self.config_of(config)
+ self.retry_setup_s = max(1.0, _num(self.cfg["retry_s"], 3))
+ if self.frame_dir:
+ self.close()
+ if self.cfg["image_topic"]:
+ d = self.cfg.get("frame_dir")
+ if d:
+ os.makedirs(d, exist_ok=True)
+ self.frame_dir, self._own_frame_dir = d, False
+ else:
+ self.frame_dir, self._own_frame_dir = tempfile.mkdtemp(prefix="openvibe-ros2-frames-"), True
+ self._connect()
+
+ def _connect(self):
+ try:
+ bus = self.bus_factory(self.cfg, self.log)
+ except Fault:
+ raise
+ except Exception as e:
+ self.next_retry = self.clock() + _num(self.cfg["retry_s"], 3)
+ raise Fault("ros_unreachable", "cannot reach ROS 2 (%s): %s" % (self.cfg["transport"], e))
+ c = self.cfg
+ if c["battery_topic"]:
+ bus.subscribe(c["battery_topic"], "sensor_msgs/msg/BatteryState", self._on_battery, 1000)
+ if c["range_topic"]:
+ bus.subscribe(c["range_topic"], "sensor_msgs/msg/Range", self._on_range, 100)
+ elif c["scan_topic"]:
+ bus.subscribe(c["scan_topic"], "sensor_msgs/msg/LaserScan", self._on_scan, 200)
+ if c["odom_topic"]:
+ bus.subscribe(c["odom_topic"], "nav_msgs/msg/Odometry", self._on_odom, 200)
+ if c["image_topic"]:
+ bus.subscribe(c["image_topic"], "sensor_msgs/msg/CompressedImage", self._on_image, int(1000 / c["fps"]))
+ self.bus = bus
+ self.faulted = False
+
+ def _bus_alive(self):
+ return self.bus is not None and self.bus.alive
+
+ def poll(self):
+ if not self._bus_alive():
+ if not self.faulted:
+ self.faulted = True
+ self.next_retry = self.clock() + _num(self.cfg["retry_s"], 3)
+ if self.runtime is not None:
+ self.runtime.send({"op": "fault", "fault_code": "ros_unreachable", "message": "the connection to ROS 2 was lost"})
+ elif self.clock() >= self.next_retry:
+ try:
+ self._connect()
+ self.publish_zero()
+ if self.runtime is not None:
+ self.runtime.send({"op": "ready"})
+ except Fault:
+ self.next_retry = self.clock() + _num(self.cfg["retry_s"], 3)
+ return {"faults": ["ros_unreachable"]}
+ out = {"sensors": {}}
+ with self.lock:
+ if self.battery:
+ out["battery"] = dict(self.battery)
+ if self.distance_cm is not None:
+ out["sensors"]["distance_cm"] = self.distance_cm
+ if self.odom:
+ out["sensors"]["odom"] = dict(self.odom)
+ return out
+
+ # Commands.
+ def twist(self, lx=0.0, ly=0.0, az=0.0):
+ return {"linear": {"x": float(lx), "y": float(ly), "z": 0.0}, "angular": {"x": 0.0, "y": 0.0, "z": float(az)}}
+
+ def publish_zero(self):
+ if self._bus_alive():
+ self.bus.publish(self.cfg["cmd_vel"], TWIST, self.twist())
+
+ def handle(self, cmd):
+ if not self._bus_alive():
+ raise Fault("ros_unreachable", "not connected to ROS 2")
+ v = cmd.value if isinstance(cmd.value, dict) else {}
+ c = self.cfg
+ if cmd.kind == "drive":
+ if c["holonomic"]:
+ # Bot's x is forward, y is left, rotation is counter-clockwise: ROS's own axes (REP 103).
+ t = self.twist(clamp(v.get("x", 0), -1, 1) * c["max_linear"], clamp(v.get("y", 0), -1, 1) * c["max_linear"],
+ clamp(v.get("rotation", 0), -1, 1) * c["max_angular"])
+ else:
+ # Bot's steer is positive to the right; ROS's angular z is positive counter-clockwise (to the left).
+ t = self.twist(clamp(v.get("throttle", 0), -1, 1) * c["max_linear"], 0.0, -clamp(v.get("steer", 0), -1, 1) * c["max_angular"])
+ self.bus.publish(c["cmd_vel"], TWIST, t)
+ return
+ if cmd.kind == "say" and c["say_topic"]:
+ text = str(v.get("text", ""))[:500]
+ if not text:
+ raise Fault("bad_value", "say needs text")
+ self.bus.publish(c["say_topic"], "std_msgs/msg/String", {"data": text})
+ return
+ raise Fault("unsupported", "kind %s not supported" % cmd.kind)
+
+ def stop_motion(self):
+ self.stop()
+
+ def stop(self):
+ try:
+ self.publish_zero()
+ except Exception as e:
+ self.log("ros2: could not publish the zero Twist: %r", e)
+
+ def close(self):
+ bus, self.bus = self.bus, None
+ if bus is not None:
+ try:
+ bus.close()
+ except Exception:
+ pass
+ if self.frame_dir and self._own_frame_dir:
+ shutil.rmtree(self.frame_dir, ignore_errors=True)
+ self.frame_dir = None
+
+ # Topics.
+ def _on_battery(self, m):
+ pct = _num(m.get("percentage"), float("nan"))
+ volts = _num(m.get("voltage"), float("nan"))
+ b = {}
+ if pct == pct and pct >= 0:
+ b["percent"] = round(pct * 100 if pct <= 1.0 else pct, 1)
+ if volts == volts and volts > 0:
+ b["volts"] = round(volts, 2)
+ with self.lock:
+ self.battery = b or None
+
+ def _on_range(self, m):
+ r = _num(m.get("range"), float("nan"))
+ with self.lock:
+ self.distance_cm = round(r * 100, 1) if r == r and math.isfinite(r) and r >= 0 else None
+
+ def _on_scan(self, m):
+ ranges = m.get("ranges") or []
+ a0, inc = _num(m.get("angle_min")), _num(m.get("angle_increment"))
+ lo, hi = _num(m.get("range_min"), 0.0), _num(m.get("range_max"), float("inf"))
+ half = math.radians(_num(self.cfg["front_arc_deg"], 30)) / 2
+ best = None
+ for i, r in enumerate(ranges):
+ r = _num(r, float("nan"))
+ if not (r == r) or not math.isfinite(r) or r < lo or r > hi:
+ continue
+ a = math.atan2(math.sin(a0 + i * inc), math.cos(a0 + i * inc)) # wrapped to -pi..pi, 0 is straight ahead
+ if abs(a) <= half and (best is None or r < best):
+ best = r
+ with self.lock:
+ self.distance_cm = round(best * 100, 1) if best is not None else None
+
+ def _on_odom(self, m):
+ pose = ((m.get("pose") or {}).get("pose") or {})
+ p, q = pose.get("position") or {}, pose.get("orientation") or {}
+ qx, qy, qz, qw = (_num(q.get(k)) for k in ("x", "y", "z", "w"))
+ yaw = math.atan2(2 * (qw * qz + qx * qy), 1 - 2 * (qy * qy + qz * qz))
+ lin = (((m.get("twist") or {}).get("twist") or {}).get("linear") or {})
+ with self.lock:
+ self.odom = {"x": round(_num(p.get("x")), 3), "y": round(_num(p.get("y")), 3), "yaw": round(yaw, 3),
+ "speed": round(math.hypot(_num(lin.get("x")), _num(lin.get("y"))), 3)}
+
+ def _on_image(self, m):
+ fmt = str(m.get("format") or "").lower()
+ if "jpeg" not in fmt and "jpg" not in fmt:
+ if "format" not in self._warned:
+ self._warned.add("format")
+ self.log("ros2: %s is %r, not JPEG: no camera frames (use a JPEG CompressedImage topic)", self.cfg["image_topic"], fmt)
+ return
+ now = self.clock()
+ if now < self.next_frame or self.frame_dir is None:
+ return
+ self.next_frame = now + 1.0 / self.cfg["fps"]
+ data = m.get("data")
+ if isinstance(data, str):
+ try:
+ data = base64.b64decode(data)
+ except ValueError:
+ return
+ if not isinstance(data, (bytes, bytearray)) or len(data) < 4 or data[:2] != b"\xff\xd8":
+ return
+ self.frame_seq += 1
+ path = os.path.join(self.frame_dir, "frame-%d.jpg" % (self.frame_seq % FRAME_RING))
+ tmp = path + ".tmp"
+ with open(tmp, "wb") as f:
+ f.write(data)
+ os.replace(tmp, path) # readers never see a partial file
+ self.emit_video(path, format="jpeg", seq=self.frame_seq)
new file mode 100644
new file mode 100644
@@ -0,0 +1,6 @@
@@ -0,0 +1,6 @@
+import sys
+
+from openvibe_plugin import run
+from openvibe_ros2 import Ros2
+
+sys.exit(run(Ros2()))
new file mode 100644
new file mode 100644
@@ -0,0 +1,156 @@
@@ -0,0 +1,156 @@
+"""A small WebSocket client (RFC 6455) on the standard library, enough for rosbridge's JSON protocol.
+
+Text frames only, client-masked, with ping/pong, close and continuation frames. ``ws://`` and ``wss://`` (TLS with the
+system's trust store and hostname checks). A message larger than ``max_message`` closes the connection.
+"""
+
+import base64
+import hashlib
+import os
+import socket
+import ssl
+import struct
+import threading
+import urllib.parse
+
+GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
+
+
+class WebSocketError(Exception):
+ pass
+
+
+class WebSocket:
+ def __init__(self, url, timeout=5.0, max_message=16 * 1024 * 1024):
+ u = urllib.parse.urlsplit(url)
+ if u.scheme not in ("ws", "wss"):
+ raise WebSocketError("the rosbridge URL must be ws:// or wss://, not %r" % (url,))
+ host = u.hostname or "127.0.0.1"
+ port = u.port or (443 if u.scheme == "wss" else 80)
+ path = (u.path or "/") + ("?" + u.query if u.query else "")
+ self.max_message = max_message
+ self._send_lock = threading.Lock()
+ self._closed = False
+ sock = socket.create_connection((host, port), timeout=timeout)
+ try:
+ if u.scheme == "wss":
+ sock = ssl.create_default_context().wrap_socket(sock, server_hostname=host)
+ key = base64.b64encode(os.urandom(16)).decode("ascii")
+ hostport = host if port in (80, 443) else "%s:%d" % (host, port)
+ if ":" in host and not host.startswith("["):
+ hostport = "[%s]:%d" % (host, port)
+ sock.sendall(("GET %s HTTP/1.1\r\nHost: %s\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n"
+ "Sec-WebSocket-Key: %s\r\nSec-WebSocket-Version: 13\r\n\r\n" % (path, hostport, key)).encode("ascii"))
+ raw = b""
+ while b"\r\n\r\n" not in raw:
+ chunk = sock.recv(4096)
+ if not chunk:
+ raise WebSocketError("the server closed the connection during the handshake")
+ raw += chunk
+ if len(raw) > 16384:
+ raise WebSocketError("the handshake answer is too long")
+ head, self._buf = raw.split(b"\r\n\r\n", 1)
+ lines = head.decode("latin-1").split("\r\n")
+ if len(lines[0].split()) < 2 or lines[0].split()[1] != "101":
+ raise WebSocketError("the server did not switch to WebSocket: %s" % lines[0][:80])
+ headers = {}
+ for line in lines[1:]:
+ k, _, v = line.partition(":")
+ headers[k.strip().lower()] = v.strip()
+ want = base64.b64encode(hashlib.sha1((key + GUID).encode("ascii")).digest()).decode("ascii")
+ if headers.get("sec-websocket-accept") != want:
+ raise WebSocketError("the server's handshake answer does not match")
+ sock.settimeout(None)
+ except Exception:
+ sock.close()
+ raise
+ self.sock = sock
+
+ # Frames.
+ def _send_frame(self, opcode, payload):
+ header = bytearray([0x80 | opcode])
+ n = len(payload)
+ if n < 126:
+ header.append(0x80 | n)
+ elif n < 65536:
+ header.append(0x80 | 126)
+ header += struct.pack(">H", n)
+ else:
+ header.append(0x80 | 127)
+ header += struct.pack(">Q", n)
+ mask = os.urandom(4)
+ header += mask
+ body = bytes(b ^ mask[i % 4] for i, b in enumerate(payload)) if n < 4096 else _mask_fast(payload, mask)
+ with self._send_lock:
+ self.sock.sendall(bytes(header) + body)
+
+ def send_text(self, text):
+ if self._closed:
+ raise WebSocketError("closed")
+ self._send_frame(0x1, text.encode("utf-8"))
+
+ def _read_exact(self, n):
+ while len(self._buf) < n:
+ chunk = self.sock.recv(max(65536, n - len(self._buf)))
+ if not chunk:
+ raise WebSocketError("the connection closed")
+ self._buf += chunk
+ out, self._buf = self._buf[:n], self._buf[n:]
+ return out
+
+ def recv(self):
+ """The next text message, or None when the connection closes."""
+ parts = []
+ size = 0
+ try:
+ while True:
+ b1, b2 = self._read_exact(2)
+ fin, opcode = b1 & 0x80, b1 & 0x0F
+ n = b2 & 0x7F
+ if n == 126:
+ n = struct.unpack(">H", self._read_exact(2))[0]
+ elif n == 127:
+ n = struct.unpack(">Q", self._read_exact(8))[0]
+ mask = self._read_exact(4) if b2 & 0x80 else None
+ if size + n > self.max_message:
+ raise WebSocketError("a message is larger than %d bytes" % self.max_message)
+ payload = self._read_exact(n)
+ if mask:
+ payload = _mask_fast(payload, mask)
+ if opcode == 0x8:
+ self.close()
+ return None
+ if opcode == 0x9:
+ self._send_frame(0xA, payload)
+ continue
+ if opcode == 0xA:
+ continue
+ if opcode in (0x1, 0x2, 0x0):
+ parts.append(payload)
+ size += n
+ if fin:
+ return b"".join(parts).decode("utf-8", "replace")
+ except (OSError, WebSocketError, ValueError):
+ self.close()
+ return None
+
+ def close(self):
+ if self._closed:
+ return
+ self._closed = True
+ try:
+ self._send_frame(0x8, b"")
+ except Exception:
+ pass
+ try:
+ self.sock.close()
+ except Exception:
+ pass
+
+
+def _mask_fast(payload, mask):
+ n = len(payload)
+ if not n:
+ return b""
+ key = int.from_bytes((mask * (n // 4 + 1))[:n], "big")
+ return (int.from_bytes(payload, "big") ^ key).to_bytes(n, "big")
new file mode 100644
new file mode 100644
@@ -0,0 +1,14 @@
@@ -0,0 +1,14 @@
+[build-system]
+requires = ["setuptools>=61"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "openvibe-ros2"
+version = "0.1.0"
+description = "OpenVibe Node bridge plugin for ROS 2 robots (rclpy or rosbridge)"
+requires-python = ">=3.9"
+license = { text = "AGPL-3.0-or-later" }
+dependencies = ["openvibe-plugin"]
+
+[tool.setuptools]
+packages = ["openvibe_ros2"]
new file mode 100644
new file mode 100644
@@ -0,0 +1,441 @@
@@ -0,0 +1,441 @@
+import base64
+import hashlib
+import io
+import json
+import math
+import os
+import socket
+import struct
+import sys
+import tempfile
+import threading
+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 Runtime # noqa: E402
+from openvibe_ros2 import DEFAULTS, RclpyBus, RosbridgeBus, Ros2, _fill, _to_dict # noqa: E402
+from openvibe_ros2.ws import GUID, WebSocket, WebSocketError # noqa: E402
+
+JPEG = b"\xff\xd8\xff\xe0" + b"\x00" * 64 + b"\xff\xd9"
+
+
+class Clock:
+ t = 10.0
+
+ def __call__(self):
+ return self.t
+
+
+class FakeBus:
+ """Records publishes and subscriptions; tests deliver topic messages with deliver()."""
+
+ def __init__(self):
+ self.alive = True
+ self.published = []
+ self.subs = {}
+ self.closed = False
+
+ def publish(self, topic, type_, msg):
+ if not self.alive:
+ raise OSError("closed")
+ self.published.append((topic, type_, json.loads(json.dumps(msg))))
+
+ def subscribe(self, topic, type_, callback, throttle_ms=0):
+ self.subs[topic] = (type_, callback)
+
+ def deliver(self, topic, msg):
+ self.subs[topic][1](msg)
+
+ def close(self):
+ self.closed = True
+ self.alive = False
+
+
+class Harness:
+ def __init__(self, fail=0, **config):
+ self.clock = Clock()
+ self.buses = []
+ self.fail = fail
+
+ def factory(cfg, log):
+ if self.fail:
+ self.fail -= 1
+ raise OSError("connection refused")
+ b = FakeBus()
+ self.buses.append(b)
+ return b
+
+ self.p = Ros2(bus_factory=factory, clock=self.clock)
+ self.p.log = lambda *a: None
+ self.out = io.StringIO()
+ self.rt = Runtime(self.p, infile=io.StringIO(""), outfile=self.out, clock=self.clock)
+ self.frames = tempfile.mkdtemp()
+ self.config = dict(config)
+
+ def lines(self):
+ return [json.loads(x) for x in self.out.getvalue().splitlines() if x.strip()]
+
+ def ops(self, op):
+ return [m for m in self.lines() if m.get("op") == op]
+
+ def hello(self, probe=False):
+ self.rt.dispatch({"op": "hello", "config": self.config, "probe": probe})
+
+ def cmd(self, cid, kind, value, deadline_ms=300):
+ self.rt.dispatch({"op": "heartbeat", "t": 0})
+ self.rt.dispatch({"op": "command", "id": cid, "kind": kind, "value": value, "deadline_ms": deadline_ms})
+
+ @property
+ def bus(self):
+ return self.buses[-1]
+
+ def twists(self):
+ return [m for (t, ty, m) in self.bus.published if ty == "geometry_msgs/msg/Twist"]
+
+
+def test_describe_touches_nothing_and_names_what_is_configured():
+ h = Harness(image_topic="/camera/image_raw/compressed", say_topic="/say", holonomic=True)
+ h.hello(probe=True)
+ d = h.ops("describe")[0]
+ assert d["driver"] == "ros2" and d["motion_kinds"] == ["drive"]
+ caps = d["capabilities"]
+ assert caps["drive"]["type"] == "mecanum"
+ assert caps["camera"] == {"jpeg": True} and caps["say"]["topic"] == "/say"
+ assert caps["sensors"]["distance_cm"] and caps["bridge"]["kind"] == "ros2"
+ assert h.buses == [] and not h.ops("ready"), "probe mode never connects"
+
+
+def test_setup_subscribes_publishes_a_zero_twist_and_says_ready():
+ h = Harness()
+ h.hello()
+ assert h.ops("ready")
+ assert set(h.bus.subs) == {"/battery_state", "/scan", "/odom"}
+ assert h.bus.subs["/scan"][0] == "sensor_msgs/msg/LaserScan"
+ assert h.twists()[-1] == {"linear": {"x": 0.0, "y": 0.0, "z": 0.0}, "angular": {"x": 0.0, "y": 0.0, "z": 0.0}}
+
+
+def test_drive_scales_to_twist_with_ros_signs_and_stops_at_the_deadline():
+ h = Harness(max_linear=0.5, max_angular=2.0)
+ h.hello()
+ h.cmd("c1", "drive", {"throttle": 0.5, "steer": 0.25})
+ assert h.ops("ack")[-1]["id"] == "c1"
+ t = h.twists()[-1]
+ assert t["linear"]["x"] == pytest.approx(0.25) and t["angular"]["z"] == pytest.approx(-0.5), "steer right is clockwise"
+ h.clock.t += 0.31
+ h.rt.tick()
+ assert h.twists()[-1]["linear"]["x"] == 0 and h.twists()[-1]["angular"]["z"] == 0
+ assert "deadline" in h.rt.stop_reasons
+ h.cmd("c2", "drive", {"throttle": 9, "steer": -9})
+ t = h.twists()[-1]
+ assert t["linear"]["x"] == pytest.approx(0.5) and t["angular"]["z"] == pytest.approx(2.0), "clamped to full scale"
+
+
+def test_holonomic_drive_uses_x_forward_y_left_rotation_ccw():
+ h = Harness(holonomic=True, max_linear=1.0, max_angular=1.0)
+ h.hello()
+ h.cmd("h1", "drive", {"x": 0.5, "y": 0.25, "rotation": -0.5})
+ t = h.twists()[-1]
+ assert (t["linear"]["x"], t["linear"]["y"], t["angular"]["z"]) == (0.5, 0.25, -0.5)
+
+
+def test_estop_halt_heartbeat_and_eof_publish_zero_and_refuse_motion():
+ h = Harness()
+ h.hello()
+ h.cmd("m1", "drive", {"throttle": 1})
+ h.rt.dispatch({"op": "estop"})
+ assert h.twists()[-1]["linear"]["x"] == 0
+ h.cmd("m2", "drive", {"throttle": 1})
+ assert h.ops("nack")[-1] == {"op": "nack", "id": "m2", "fault_code": "estopped"}
+ h.rt.dispatch({"op": "resume"})
+ h.cmd("m3", "drive", {"throttle": 1})
+ h.clock.t += 1.5
+ h.rt.tick()
+ assert h.ops("event")[-1]["name"] == "heartbeat_lost" and h.twists()[-1]["linear"]["x"] == 0
+ n = len(h.twists())
+ h.rt.shutdown("eof")
+ assert len(h.twists()) == n + 1 and h.twists()[-1]["linear"]["x"] == 0 and h.bus.closed
+
+
+def test_say_publishes_a_string_only_when_configured():
+ h = Harness(say_topic="/say")
+ h.hello()
+ h.cmd("s1", "say", {"text": "Hello from OpenVibe"})
+ assert ("/say", "std_msgs/msg/String", {"data": "Hello from OpenVibe"}) in h.bus.published
+ h2 = Harness()
+ h2.hello()
+ h2.cmd("s2", "say", {"text": "hi"})
+ assert h2.ops("nack")[-1]["fault_code"] == "unsupported"
+
+
+def test_telemetry_battery_scan_range_and_odom():
+ h = Harness()
+ h.hello()
+ h.bus.deliver("/battery_state", {"percentage": 0.42, "voltage": 12.06})
+ # 360 rays from -pi: the nearest return straight ahead (index 180) counts, a nearer one behind does not.
+ ranges = [5.0] * 360
+ ranges[180] = 0.8
+ ranges[0] = 0.1
+ ranges[181] = float("inf")
+ h.bus.deliver("/scan", {"angle_min": -math.pi, "angle_increment": 2 * math.pi / 360, "range_min": 0.05, "range_max": 12, "ranges": ranges})
+ yaw = 0.5
+ h.bus.deliver("/odom", {"pose": {"pose": {"position": {"x": 1.234, "y": -2.0}, "orientation": {"x": 0, "y": 0, "z": math.sin(yaw / 2), "w": math.cos(yaw / 2)}}},
+ "twist": {"twist": {"linear": {"x": 0.3, "y": 0.4}}}})
+ t = h.p.poll()
+ assert t["battery"] == {"percent": 42.0, "volts": 12.06}
+ assert t["sensors"]["distance_cm"] == 80.0
+ assert t["sensors"]["odom"] == {"x": 1.234, "y": -2.0, "yaw": 0.5, "speed": 0.5}
+ hr = Harness(range_topic="/sonar")
+ hr.hello()
+ assert "/sonar" in hr.bus.subs and "/scan" not in hr.bus.subs, "a Range topic replaces the scan"
+ hr.bus.deliver("/sonar", {"range": 0.355})
+ assert hr.p.poll()["sensors"]["distance_cm"] == 35.5
+
+
+def test_camera_frames_from_a_jpeg_compressed_image_rate_limited():
+ h = Harness(image_topic="/cam/compressed", fps=2)
+ h.config["frame_dir"] = h.frames
+ h.hello()
+ h.bus.deliver("/cam/compressed", {"format": "rgb8; jpeg compressed bgr8", "data": base64.b64encode(JPEG).decode()})
+ v = h.ops("video")
+ assert len(v) == 1 and v[0]["format"] == "jpeg"
+ with open(v[0]["path"], "rb") as f:
+ assert f.read() == JPEG
+ h.bus.deliver("/cam/compressed", {"format": "jpeg", "data": JPEG}) # rclpy hands bytes
+ assert len(h.ops("video")) == 1, "within 1/fps"
+ h.clock.t += 0.6
+ h.bus.deliver("/cam/compressed", {"format": "jpeg", "data": JPEG})
+ assert len(h.ops("video")) == 2
+ h.clock.t += 1
+ h.bus.deliver("/cam/compressed", {"format": "png", "data": base64.b64encode(b"\x89PNG").decode()})
+ assert len(h.ops("video")) == 2, "not JPEG: no frame"
+
+
+def test_unreachable_ros_faults_and_the_runtime_sets_up_again_until_ready():
+ h = Harness(fail=2, retry_s=3)
+ h.hello()
+ f = h.ops("fault")
+ assert f and f[0]["fault_code"] == "ros_unreachable" and not h.ops("ready")
+ h.cmd("x1", "drive", {"throttle": 1})
+ assert h.ops("nack")[-1]["fault_code"] == "not_ready"
+ h.clock.t += 3.1
+ h.rt.tick() # second failure: still down, no second fault
+ assert not h.ops("ready") and len(h.ops("fault")) == 1
+ h.clock.t += 3.1
+ h.rt.tick() # ROS is up
+ assert h.ops("ready") and len(h.buses) == 1
+ h.cmd("x2", "drive", {"throttle": 1})
+ assert h.ops("ack")[-1]["id"] == "x2"
+ # The link drops later: a fault, commands refused by the plugin, and ready again once it reconnects.
+ h.bus.alive = False
+ h.p.poll()
+ assert h.ops("fault")[-1]["message"] == "the connection to ROS 2 was lost"
+ h.cmd("x3", "drive", {"throttle": 1})
+ assert h.ops("nack")[-1]["fault_code"] == "ros_unreachable"
+ h.clock.t += 3.1
+ h.p.poll()
+ assert len(h.buses) == 2 and h.ops("ready")[-1] == {"op": "ready"} and len(h.ops("ready")) == 2
+ assert h.twists()[-1]["linear"]["x"] == 0, "a zero Twist first on the new link"
+
+
+def test_bad_transport_is_a_config_fault():
+ h = Harness(transport="dds")
+ h.hello()
+ assert h.ops("fault")[0]["fault_code"] == "bad_config"
+
+
+# ── rclpy helpers ─────────────────────────────────────────────────────────────────────────────────────────────────
+class Vector3:
+ def __init__(self):
+ self.x = self.y = self.z = 0.0
+
+ @staticmethod
+ def get_fields_and_field_types():
+ return {"x": "double", "y": "double", "z": "double"}
+
+
+class Twist:
+ def __init__(self):
+ self.linear, self.angular = Vector3(), Vector3()
+
+ @staticmethod
+ def get_fields_and_field_types():
+ return {"linear": "geometry_msgs/Vector3", "angular": "geometry_msgs/Vector3"}
+
+
+def test_fill_and_to_dict_round_trip_a_nested_message():
+ t = _fill(Twist(), {"linear": {"x": 0.2, "y": 0.0, "z": 0.0}, "angular": {"x": 0.0, "y": 0.0, "z": -1.0}})
+ assert t.linear.x == 0.2 and t.angular.z == -1.0
+ assert _to_dict(t) == {"linear": {"x": 0.2, "y": 0.0, "z": 0.0}, "angular": {"x": 0.0, "y": 0.0, "z": -1.0}}
+ assert _to_dict(bytearray(b"\xff\xd8")) == b"\xff\xd8"
+
+
+def test_rclpy_bus_with_a_fake_rclpy(monkeypatch):
+ import types
+ created = {"pubs": [], "subs": []}
+
+ class Pub:
+ def __init__(self, cls, topic):
+ self.topic, self.sent = topic, []
+
+ def publish(self, m):
+ self.sent.append(m)
+
+ class Node:
+ def create_publisher(self, cls, topic, qos):
+ p = Pub(cls, topic)
+ created["pubs"].append(p)
+ return p
+
+ def create_subscription(self, cls, topic, cb, qos):
+ created["subs"].append((cls, topic, cb))
+
+ def destroy_node(self):
+ created["destroyed"] = True
+
+ class Executor:
+ def __init__(self):
+ self.stop = threading.Event()
+
+ def add_node(self, n):
+ pass
+
+ def spin(self):
+ self.stop.wait(5)
+
+ def shutdown(self):
+ self.stop.set()
+
+ rclpy = types.SimpleNamespace(__name__="rclpy", ok=lambda: False, init=lambda args=None: created.setdefault("init", True),
+ create_node=lambda name: Node(), executors=types.SimpleNamespace(SingleThreadedExecutor=Executor),
+ try_shutdown=lambda: created.setdefault("shutdown", True))
+ geometry = types.ModuleType("geometry_msgs.msg")
+ geometry.Twist = Twist
+ monkeypatch.setitem(sys.modules, "geometry_msgs", types.ModuleType("geometry_msgs"))
+ monkeypatch.setitem(sys.modules, "geometry_msgs.msg", geometry)
+ bus = RclpyBus("openvibe_node", lambda *a: None, rclpy=rclpy)
+ bus.publish("/cmd_vel", "geometry_msgs/msg/Twist", {"linear": {"x": 0.1, "y": 0.0, "z": 0.0}, "angular": {"x": 0.0, "y": 0.0, "z": 0.3}})
+ assert created["init"] and created["pubs"][0].topic == "/cmd_vel" and created["pubs"][0].sent[0].angular.z == 0.3
+ got = []
+ bus.subscribe("/cmd_vel_echo", "geometry_msgs/msg/Twist", got.append)
+ created["subs"][0][2](_fill(Twist(), {"linear": {"x": 0.5}}))
+ assert got == [{"linear": {"x": 0.5, "y": 0.0, "z": 0.0}, "angular": {"x": 0.0, "y": 0.0, "z": 0.0}}]
+ bus.close()
+ assert created["destroyed"] and created["shutdown"]
+
+
+# ── The WebSocket client and rosbridge, against a small local server ──────────────────────────────────────────────
+class TinyServer:
+ """A one-connection WebSocket server: records text frames it receives; send() pushes a text frame."""
+
+ def __init__(self, bad_accept=False):
+ self.srv = socket.socket()
+ self.srv.bind(("127.0.0.1", 0))
+ self.srv.listen(1)
+ self.port = self.srv.getsockname()[1]
+ self.got = []
+ self.conn = None
+ self.bad_accept = bad_accept
+ self.ready = threading.Event()
+ threading.Thread(target=self._serve, daemon=True).start()
+
+ def _serve(self):
+ conn, _ = self.srv.accept()
+ req = b""
+ while b"\r\n\r\n" not in req:
+ req += conn.recv(4096)
+ key = [l.split(b":", 1)[1].strip() for l in req.split(b"\r\n") if l.lower().startswith(b"sec-websocket-key")][0]
+ accept = base64.b64encode(hashlib.sha1(key + GUID.encode()).digest())
+ if self.bad_accept:
+ accept = b"nope"
+ conn.sendall(b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: " + accept + b"\r\n\r\n")
+ self.conn = conn
+ self.ready.set()
+ buf = b""
+ while True:
+ try:
+ chunk = conn.recv(65536)
+ except OSError:
+ return
+ if not chunk:
+ return
+ buf += chunk
+ while len(buf) >= 2:
+ n = buf[1] & 0x7F
+ off = 2
+ if n == 126:
+ n = struct.unpack(">H", buf[2:4])[0]
+ off = 4
+ elif n == 127:
+ n = struct.unpack(">Q", buf[2:10])[0]
+ off = 10
+ if len(buf) < off + 4 + n:
+ break
+ mask = buf[off:off + 4]
+ payload = bytes(b ^ mask[i % 4] for i, b in enumerate(buf[off + 4:off + 4 + n]))
+ op = buf[0] & 0x0F
+ buf = buf[off + 4 + n:]
+ if op == 0x1:
+ self.got.append(json.loads(payload))
+ elif op == 0x8:
+ return
+
+ def send(self, obj, opcode=0x1):
+ data = json.dumps(obj).encode() if opcode == 0x1 else obj
+ n = len(data)
+ head = bytes([0x80 | opcode]) + (bytes([n]) if n < 126 else bytes([126]) + struct.pack(">H", n) if n < 65536 else bytes([127]) + struct.pack(">Q", n))
+ self.conn.sendall(head + data)
+
+ def wait_for(self, pred, timeout=3):
+ end = time.time() + timeout
+ while time.time() < end:
+ if any(pred(m) for m in self.got):
+ return True
+ time.sleep(0.01)
+ return False
+
+
+def test_rosbridge_bus_advertises_publishes_subscribes_and_dispatches():
+ s = TinyServer()
+ bus = RosbridgeBus("ws://127.0.0.1:%d" % s.port, lambda *a: None)
+ s.ready.wait(2)
+ got = []
+ bus.subscribe("/battery_state", "sensor_msgs/msg/BatteryState", got.append, 1000)
+ bus.publish("/cmd_vel", "geometry_msgs/msg/Twist", {"linear": {"x": 0.1}})
+ bus.publish("/cmd_vel", "geometry_msgs/msg/Twist", {"linear": {"x": 0.0}})
+ assert s.wait_for(lambda m: m.get("op") == "publish" and m["msg"]["linear"]["x"] == 0.0)
+ ops = [(m["op"], m["topic"]) for m in s.got]
+ assert ops == [("subscribe", "/battery_state"), ("advertise", "/cmd_vel"), ("publish", "/cmd_vel"), ("publish", "/cmd_vel")], "advertised once"
+ assert s.got[0]["type"] == "sensor_msgs/msg/BatteryState" and s.got[0]["throttle_rate"] == 1000
+ s.send(b"", opcode=0x9) # a ping is answered and skipped
+ big = "x" * 70000 # a 64-bit-length frame
+ s.send({"op": "publish", "topic": "/battery_state", "msg": {"percentage": 0.9, "note": big}})
+ end = time.time() + 3
+ while not got and time.time() < end:
+ time.sleep(0.01)
+ assert got and got[0]["percentage"] == 0.9 and len(got[0]["note"]) == 70000
+ s.conn.shutdown(socket.SHUT_RDWR) # FIN even while the server thread is in recv
+ s.conn.close()
+ end = time.time() + 3
+ while bus.alive and time.time() < end:
+ time.sleep(0.01)
+ assert not bus.alive, "a closed connection marks the bus down"
+ bus.close()
+
+
+def test_websocket_refuses_a_bad_handshake_and_other_schemes():
+ s = TinyServer(bad_accept=True)
+ with pytest.raises(WebSocketError):
+ WebSocket("ws://127.0.0.1:%d/" % s.port)
+ with pytest.raises(WebSocketError):
+ WebSocket("http://127.0.0.1:9090")
+
+
+def test_defaults_document_every_config_key():
+ readme = open(os.path.join(os.path.dirname(HERE), "README.md")).read()
+ for k in DEFAULTS:
+ assert "`%s`" % k in readme, k
@@ -79,6 +79,9 @@ class Plugin:
@@ -79,6 +79,9 @@ class Plugin:
version = "0.0.0"
version = "0.0.0"
motion_kinds = MOTION_KINDS
motion_kinds = MOTION_KINDS
telemetry_interval_s = 0.5
telemetry_interval_s = 0.5
+ # Seconds between attempts to set up again after setup raised (the hardware or the robot's computer was not up
+ # yet). None: a failed setup stays failed until the core restarts the plugin.
+ retry_setup_s = None
runtime = None # set by Runtime
runtime = None # set by Runtime
@@ -156,6 +159,7 @@ class Runtime:
@@ -156,6 +159,7 @@ class Runtime:
self.last_heartbeat = None
self.last_heartbeat = None
self.heartbeat_lost = False
self.heartbeat_lost = False
self.next_poll = 0.0
self.next_poll = 0.0
+ self.setup_retry_at = None
self.closed = False
self.closed = False
self.stop_reasons = [] # for tests: why stop() was called, in order
self.stop_reasons = [] # for tests: why stop() was called, in order
@@ -200,6 +204,9 @@ class Runtime:
@@ -200,6 +204,9 @@ class Runtime:
self.heartbeat_lost = True
self.heartbeat_lost = True
self.safe_stop("heartbeat")
self.safe_stop("heartbeat")
self.send({"op": "event", "name": "heartbeat_lost"})
self.send({"op": "event", "name": "heartbeat_lost"})
+ if not self.ready and self.setup_retry_at is not None and now >= self.setup_retry_at and not self.closed:
+ self.setup_retry_at = None
+ self._setup(retrying=True)
if self.ready and now >= self.next_poll:
if self.ready and now >= self.next_poll:
self.next_poll = now + self.plugin.telemetry_interval_s
self.next_poll = now + self.plugin.telemetry_interval_s
try:
try:
@@ -251,6 +258,11 @@ class Runtime:
@@ -251,6 +258,11 @@ class Runtime:
"motion_kinds": sorted(self.plugin.motion_kinds)})
"motion_kinds": sorted(self.plugin.motion_kinds)})
if self.probe:
if self.probe:
return
return
+ self._setup()
+
+ def _setup(self, retrying=False):
+ """Set the plugin up; ready on success, else a fault (sent once, not on every retry) and, when the plugin
+ sets retry_setup_s, another attempt that long from now. A later success sends ready, which ends the fault."""
try:
try:
with self.lock:
with self.lock:
self.plugin.setup(self.config)
self.plugin.setup(self.config)
@@ -258,10 +270,15 @@ class Runtime:
@@ -258,10 +270,15 @@ class Runtime:
self.ready = True
self.ready = True
self.last_heartbeat = self.clock()
self.last_heartbeat = self.clock()
self.send({"op": "ready"})
self.send({"op": "ready"})
+ return
except Fault as f:
except Fault as f:
- self.send({"op": "fault", "fault_code": f.code, "message": f.message})
+ fault = {"op": "fault", "fault_code": f.code, "message": f.message}
except Exception as e:
except Exception as e:
- self.send({"op": "fault", "fault_code": "hardware", "message": str(e)})
+ fault = {"op": "fault", "fault_code": "hardware", "message": str(e)}
+ if not retrying:
+ self.send(fault)
+ if self.plugin.retry_setup_s:
+ self.setup_retry_at = self.clock() + float(self.plugin.retry_setup_s)
def on_command(self, msg, now=None):
def on_command(self, msg, now=None):
cid = msg.get("id")
cid = msg.get("id")
@@ -162,6 +162,49 @@ class RuntimeTest(unittest.TestCase):
@@ -162,6 +162,49 @@ class RuntimeTest(unittest.TestCase):
rt.feed_line(json.dumps({"op": "command", "id": "c3", "kind": "drive"}))
rt.feed_line(json.dumps({"op": "command", "id": "c3", "kind": "drive"}))
self.assertEqual(lines(out)[-1]["op"], "ack")
self.assertEqual(lines(out)[-1]["op"], "ack")
+ def test_a_failed_setup_stays_failed_without_retry_setup_s(self):
+ class Down(Recorder):
+ def setup(self, config):
+ raise Fault("unreachable", "not up yet")
+ rt, p, out, clock = make(Down())
+ self.hello(rt)
+ clock.t += 60
+ rt.tick()
+ self.assertEqual([m["op"] for m in lines(out)], ["describe", "fault"])
+ self.assertEqual(p.calls.count("describe"), 1)
+
+ def test_retry_setup_s_sets_up_again_and_ready_ends_the_fault(self):
+ class Late(Recorder):
+ retry_setup_s = 2
+ tries = 0
+
+ def setup(self, config):
+ Late.tries += 1
+ if Late.tries < 3:
+ raise Fault("unreachable", "not up yet")
+ self.calls.append("setup")
+ rt, p, out, clock = make(Late())
+ self.hello(rt)
+ rt.feed_line(json.dumps({"op": "command", "id": "c0", "kind": "drive", "value": {}}))
+ self.assertEqual(lines(out)[-1]["fault_code"], "not_ready")
+ clock.t += 1
+ rt.tick()
+ self.assertEqual(Late.tries, 1, "not before retry_setup_s")
+ clock.t += 1.5
+ rt.tick()
+ self.assertEqual(Late.tries, 2)
+ self.assertEqual([m["op"] for m in lines(out)].count("fault"), 1, "the fault is sent once, not on every retry")
+ clock.t += 2.5
+ rt.tick()
+ self.assertEqual(Late.tries, 3)
+ self.assertEqual(lines(out)[-1]["op"], "ready")
+ self.assertTrue(rt.ready)
+ rt.feed_line(json.dumps({"op": "command", "id": "c1", "kind": "drive", "value": {}}))
+ self.assertEqual(lines(out)[-1], {"op": "ack", "id": "c1"})
+ clock.t += 10
+ rt.tick()
+ self.assertEqual(Late.tries, 3, "no more attempts once ready")
+
def test_halt_always_accepted(self):
def test_halt_always_accepted(self):
rt, p, out, clock = make()
rt, p, out, clock = make()
self.hello(rt)
self.hello(rt)
@@ -24,7 +24,7 @@ done
@@ -24,7 +24,7 @@ done
# The plugin bundle: the Python packages without their tests.
# The plugin bundle: the Python packages without their tests.
tar -czf dist/openvibe-node-plugins.tar.gz -C plugins \
tar -czf dist/openvibe-node-plugins.tar.gz -C plugins \
--exclude=tests --exclude=__pycache__ --exclude='*.egg-info' --exclude=.pytest_cache --exclude=build \
--exclude=tests --exclude=__pycache__ --exclude='*.egg-info' --exclude=.pytest_cache --exclude=build \
- sdk dryrun adeept_adr036 cozmo relay
+ sdk dryrun adeept_adr036 cozmo relay ros2
# OpenVibe.Bot's GET /install route and BOT_INSTALLER_SOURCE_URL are documented to serve this asset name:
# 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.
# https://github.com/OpenVibers/OpenVibe.Node/releases/latest/download/install.sh — do not rename it.
cp install/install.sh dist/install.sh
cp install/install.sh dist/install.sh