> ## Documentation Index > Fetch the complete documentation index at: https://docs.cyberwave.com/llms.txt > Use this file to discover all available pages before exploring further. # Writing compatible drivers > Build a platform-compliant edge driver for your own asset: container contract, cw-driver.yml interface manifest, MQTT catalog, and sensor output. ## What is a compatible driver? A compatible driver connects **your hardware's native API** to a **Cyberwave asset and twin**: it translates the device's protocol into MQTT topics, command envelopes, and (optionally) edge data channels that the UI, workflows, and SDK understand. This guide is for **integrators building a driver from scratch** for a custom or third-party asset in your workspace. Cyberwave-maintained drivers use the same rules; their `cw-driver.yml` files are linked later as examples only. Drivers can run on edge hardware, on the robot, in the cloud, or on a developer laptop. In production they usually run beside the device under **Edge Core**. Each driver ships as a **Docker image**. Edge Core pulls and runs it with the environment variables below, so local development matches production deployment. *** ## Choose a starting path End-to-end Docker lab + `ursim_driver.py`. SDK concepts: [Python SDK drivers](#python-sdk-drivers) below. Synthetic IMU on a D455 twin — manifest, commands, MQTT + optional Zenoh before real sensors. **BaseDriver** from scratch — serial, SDK, WebRTC, or hand-written MQTT. Deep references: **[BaseDriver](/feature-reference/edge/drivers/base-driver-class)** · **[BaseROS2Driver](/feature-reference/edge/drivers/ros2-base-driver)** *** ## Quickstart: Python SDK Use the **[`cyberwave`](https://pypi.org/project/cyberwave/)** package and **`BaseDriver`** in `cyberwave.driver`. You declare topics and commands in Python; the SDK generates the MQTT catalog and wires runtime subscriptions. Lifecycle, hooks, manifest export, alerts, and what the SDK runs for you. Dual lifecycle, `from_ros` forwarding, hooks — often under 100 LOC for ROS → twin streaming. ```python theme={null} from cyberwave import Cyberwave from my_driver import MyDriver cw = Cyberwave() twin = cw.twin("acme-corp/my-device", twin_id="...") # 1. Register catalog on the twin (no running driver required) twin.driver.set_schema(MyDriver.get_manifest()) # 2. Optional: export cw-driver.yml for Docker / git MyDriver.get_manifest(path="cw-driver.yml") # 3. Run — driver wires MQTT (+ Zenoh when declared) MyDriver(twin).run() # Introspection vs invocation print(twin.commands.get_supported_commands()) # command catalog introspection twin.commands.get_schema() # MQTT catalog (topics, commands.specs) twin.commands.stop() # bound methods from commands.supported ``` `get_manifest()` builds the compiled **MQTT** catalog from `define_interface()` only — no twin connect, no `run()`. Use `get_manifest_bundles()` when you also declare Zenoh channels (see [Zenoh edge catalog](#zenoh-edge-catalog)). | Handle | Role | | --------------- | ----------------------------------------------------------------------------------------------------------------- | | `twin.driver` | Full catalogs (`get_schemas`, Zenoh/topics/transports getters) and **`set_schema`** (persist + re-bind commands) | | `twin.commands` | **`get_schema()`**, **`get_supported_commands()`**, and **`twin.commands.(...)`** from `commands.supported` | **ROS 2 integrators:** start with the **[UR Sim tutorial](/tutorials/ur-sim-cyberwave-driver)**, then read [Python SDK drivers](#python-sdk-drivers) below for what the SDK handles vs what you implement. Otherwise see the [fake IMU tutorial](#tutorial-build-your-first-driver-fake-imu), then [reference](#reference) for `cw-driver.yml`, REST/MQTT, and sensor buses. (C++ drivers use the same platform contract via `cw-driver.yml`; see the [C++ SDK](/tools/cpp-sdk).) ***

Python SDK drivers

The [`cyberwave`](https://pypi.org/project/cyberwave/) package provides **base classes** so you declare the platform contract once in Python instead of hand-wiring MQTT and catalog JSON. | Base class | Use when | | ------------------------------------------------------------------------ | ------------------------------------------------------------------------------------- | | **[`BaseDriver`](/feature-reference/edge/drivers/base-driver-class)** | Serial, vendor SDK, WebRTC, cameras, custom polling loops | | **[`BaseROS2Driver`](/feature-reference/edge/drivers/ros2-base-driver)** | ROS 2 Humble+ — subscribe on ROS topics and forward to Cyberwave with **`from_ros=`** | ### What you declare vs what the SDK runs | You provide | SDK provides automatically | | ------------------------------------------------------------ | ---------------------------------------------------------------------------------------------------------------------------------------- | | **`ASSET_KEY`** on the driver subclass (required) | `Cyberwave(source_type="edge")` — API/MQTT from env / CLI | | **`define_interface(iface)`** — topics, commands, `from_ros` | Compiles `metadata["mqtt"]` / optional `metadata["zenoh"]`; `twin.driver.set_schema` | | Lifecycle hooks (`on_*` or ROS `configure` / `activate`) | MQTT subscribe/publish wiring, tick publishers, session telemetry | | Optional publisher/command callbacks | Default management commands (`stop`, teleop modes), `driver_info` on `twin/telemetry` | | Optional `operation_modes=` per interface entry | **Operation mode derived from the twin's attached controller** at startup (`NO_OP` when none) and re-wired on controller/teleop commands | | — | Lifecycle **twin alerts** (*Driver starting* → *Driver active*) after connect | | **`from_ros=Ros2TopicSpec(...)`** only on ROS drivers | ROS subscribe + `JointState` → `cyberwave/joint/{uuid}/update` serialization | **Source-type convention.** A driver **publishes** its state as **`edge`** (or **`sim`** when driving a simulator) and **listens** for teleop commands — **`tele`**, `edit`, `sim_tele`. It must **never act on `edge*`** messages (that's its own feedback), and inbound messages with **no** `source_type` are accepted leniently. Full rules: [BaseDriver → Source-type convention](/feature-reference/edge/drivers/base-driver-class#source-type-convention). A driver with **no controller assigned starts in `NO_OP`** — control commands stay idle while edge telemetry/`from_ros` forwarders keep streaming. Declare always-on publishers with `operation_modes=frozenset(DriverOperationMode)`. See [Operation modes](/feature-reference/edge/drivers/base-driver-class#operation-modes-driveroperationmode). High-rate streams should throttle with `TopicSpec.rate_hz` on `from_ros` publishers (or `acquire_stream_publish_slot()` in custom callbacks; default cap 30 Hz). ROS drivers can also have the SDK spawn a vendor `ros2 launch` (`managed_launch`) and export a combined ROS+MQTT `manifest.yaml` (`write-manifest`) — see the [BaseROS2Driver reference](/feature-reference/edge/drivers/ros2-base-driver#managed-vendor-launch-managed_launch).

Driver mixins (arm + joints)

For manipulator arms, compose SDK mixins instead of hand-wiring MQTT joint teleop, arm commands, and ROS feedback. See **[Driver mixins and capabilities](/feature-reference/edge/drivers/driver-mixins)** for the full inventory and a **Piper walkthrough**. | Concern | Mixin / pattern | | --------------------------------------- | --------------------------------------------------------------------------------------- | | MQTT `joint/update` teleop in | `JointControllerMixin` (`use_joint_controller=True`) — **do not** add your own listener | | `ee_move`, `grip`, `home`, … | `ArmCapabilityMixin` (`use_base_commands=True`) | | ROS → local joint state + MQTT feedback | `Ros2JointFeedbackMixin` + `add_publisher(..., from_ros=...)` in `define_interface` | | Async MQTT → ROS executor hand-off | `JointCommandBufferMixin` — `_publish_joint_command`, `tick()` | **Bidirectional `joint/update`:** call `super().define_interface(iface)` first (mixins register the **teleop listener**); then add your **edge publisher** with `from_ros`. Clients use `twin.joints.set()` for commands and `twin.joints.get()` to read feedback — drivers use the registry, not `twin.joints.set()`. #### Integration example: `PiperDriver` Reference: [`cyberwave-edge-nodes/piper/piper_driver.py`](https://github.com/cyberwave-os/cyberwave/blob/main/cyberwave-edge-nodes/piper/piper_driver.py). ```python theme={null} class PiperDriver( JointControllerMixin, ArmCapabilityMixin, Ros2JointFeedbackMixin, JointCommandBufferMixin, BaseROS2Driver, ): use_joint_controller = True use_joint_feedback = True use_base_commands = True def __init__(self, node_name, **kwargs): super().__init__(node_name, **kwargs) self._init_joint_feedback() self._init_joint_command_buffer() def define_interface(self, iface): super().define_interface(iface) # listeners: joint/update + twin/command iface.add_publisher( # publisher: joint/update @ edge only TopicSpec(namespace="joint", leaf="update", rate_hz=30.0, ...), CallbackGroup(), protocol=ProtocolArgs(source_types=[SOURCE_TYPE_EDGE]), from_ros=Ros2TopicSpec(topic="joint_states_single"), ) def joint_controller_motion(self): return None # passthrough — Piper sets vel/effort on the ROS JointState def joint_controller_after_process(self): return lambda cmd: self.submit_joint_targets(cmd.positions) def tick(self): self.pump_joint_commands() # inbox → _publish_joint_command → ros_publish ``` | Step | Responsibility | | ----------------- | ------------------------------------------------------- | | MQTT teleop in | `JointControllerMixin` → `submit_joint_targets` | | Arm commands | `ArmCapabilityMixin` → IK → same controller path | | ROS cmd out | `JointCommandBufferMixin._publish_joint_command` | | ROS feedback in | `Ros2JointFeedbackMixin` updates `_joint_states` for IK | | MQTT feedback out | `from_ros` forwarder (30 Hz, `source_type=edge`) | Motion shaping (trapezoidal vs passthrough): [Joint motion shaping](/feature-reference/edge/drivers/joint-motion-shaping). MRO, seams, Piper-style ROS arm pattern. `joint_controller_motion()` — shaped vs passthrough. ### ROS 2 minimal example (`ursim_driver.py`) Flagship sample: [`examples/ursim_driver.py`](https://github.com/cyberwave-os/cyberwave/blob/main/cyberwave-sdks/cyberwave-python/examples/ursim_driver.py) (\~75 lines). ```python theme={null} from cyberwave.driver import CallbackGroup, TopicSpec from cyberwave.driver.ros2 import BaseROS2Driver, Ros2TopicSpec class UrSimDriver(BaseROS2Driver): ASSET_KEY = "universal_robots/UR7" def define_interface(self, iface): iface.add_publisher( TopicSpec( topic_slug="cyberwave/joint/{twin_uuid}/update", payload_schema_ref="JointStatesPayload", ), CallbackGroup(), # JointState auto-serialized — no manual MQTT loop from_ros=Ros2TopicSpec(topic="/joint_states"), ) ``` Run after ROS and UR Sim are up (full lab): **[UR Sim tutorial](/tutorials/ur-sim-cyberwave-driver)**. Docker, pendant program, `ur_robot_driver`, run `ursim_driver.py`, GIFs. TCP, ROS DDS, MQTT — how the lab stack fits together. Dual lifecycle, env vars, `ros_publish()` for commands. Non-ROS drivers, manifest export, hooks. Arm + joint mixin composition, bidirectional joint/update. *** ## Tutorial: build your first driver (fake IMU) Reference implementation: **`cyberwave-python/examples/fake_imu_driver.py`** — hardware-free, useful to validate manifest + MQTT + `twin.imu.get()` before real sensors. **ROS 2 arms:** use the **[UR Sim tutorial](/tutorials/ur-sim-cyberwave-driver)** and [Python SDK drivers](#python-sdk-drivers). ```bash theme={null} export CYBERWAVE_API_KEY=... export CYBERWAVE_TWIN_UUID= # optional cd cyberwave-python PYTHONPATH=. python examples/fake_imu_driver.py ``` ### Step 1 — Class skeleton and identity ```python theme={null} class FakeImu6dDriver(BaseDriver): ASSET_KEY = "intel/realsensed455" TICK_RATE_HZ = 50.0 auto_register_interface = False # demo calls set_schema() before run() ``` `ASSET_KEY` ties the driver to a catalog asset. `TICK_RATE_HZ` drives both `on_tick` and registry publisher scheduling. ### Step 2 — `define_interface` (manifest + runtime contract) ```python theme={null} def define_interface(self, iface): iface.add_listener( TopicSpec(namespace="twin", leaf="command", payload_schema_ref="TwinCommandPayload"), CallbackGroup(callback=self._on_rotate), command=CommandArgs(name="rotate"), protocol=ProtocolArgs(source_types=["tele", "live", "edge"]), ) iface.add_publisher( TopicSpec( topic_slug=TWIN_IMU_TOPIC_SLUG, payload_schema_ref="ImuPayload", enable_zenoh=True, zenoh_channel="imu", ), CallbackGroup(callback=self._publish_imu_sample), publisher=PublisherArgs(rate_hz=self._imu_params.imu_rate_hz), protocol=ProtocolArgs(source_types=["live"]), ) ``` This single method feeds `get_manifest()` / `driver.manifest`, optional **`metadata["zenoh"]`**, and `_wire_interface_from_registry()` (MQTT subscribe/publish plus edge **DataBus** when dual specs are present). **`twin/command` stays MQTT-only** — do not set `enable_zenoh=True` on command listeners. ### Step 3 — Lifecycle hooks (map to `run_async` order) | Hook | Fake IMU behavior | | ----------------------- | ---------------------------------------- | | `on_configure` | Log params (`sensor_id`, rate) | | `on_connect_to_device` | No hardware — log ready | | `on_register_callbacks` | `pass` — no device SDK callbacks | | `on_activate` | Log IMU topic active; `emit_driver_info` | | `on_shutdown` | `pass` | For a real camera driver, `on_connect_to_device` opens the device SDK and `on_register_callbacks` attaches frame callbacks. ### Step 4 — Publisher callback (MQTT stream) ```python theme={null} def _publish_imu_sample(self) -> ImuPayload: gyro, accel = random_imu_vectors(gyro_base=self._imu.gyro(), ...) return build_imu_payload(gyro=gyro, accel=accel, sensor_id=self._imu_params.sensor_id) ``` `BaseDriver` calls this from `_run_registry_publishers()` at 50 Hz and publishes to `cyberwave/twin/{uuid}/imu`. You do **not** call `mqtt.publish` yourself for registry publishers. ### Step 5 — Command callback (`rotate`) ```python theme={null} def _on_rotate(self, envelope: dict[str, Any]) -> None: if envelope.get("command") == "status": return data = envelope.get("data") or {} axis = str(data.get("axis", "")).lower() rate_deg_s = float(data.get("amount_deg", 0.0)) logger.info("command received, rotating sensor ...") self._imu.set_gyro_rate_deg_s(axis, rate_deg_s) ``` `twin.commands.rotate(...)` publishes to `cyberwave/twin/{uuid}/command`; the registry routes it here while the driver is **ACTIVE**. ### Step 6 — Twin bring-up script (recommended order) | Order | Action | | ----- | ----------------------------------------------------------------------------------------------------------------------------------------------- | | 1 | Create / resolve twin; add `d455_imu` to `capabilities.sensors` if missing (D455 has no hardware IMU) | | 2 | `twin.driver.set_schema(FakeImu6dDriver.get_manifest())` — persists `metadata["mqtt"]` / `metadata["zenoh"]` and **re-binds** `twin.commands.*` | | 3 | `driver.run()` or `asyncio.create_task(driver.run_async())` on the **main** thread | | 4 | `twin.imu.get(timeout=5)` from a consumer client while ACTIVE | | 5 | `twin.commands.rotate(axis="yaw", amount_deg=30, source_type="live")` | **Expected logs:** `inactive → active`, `IMU stream active`, then `command received, rotating sensor` on the driver logger; `imu.get()` shows higher `gyro.z` after yaw rotate (\~0.52 rad/s + noise). `rotate` is a **sensor command** (not locomotion) — no teleop policy required on camera-only twins. ### Minimal skeleton (any driver) ```python theme={null} from cyberwave.driver import BaseDriver, CallbackGroup, CommandArgs, TopicSpec class MyDriver(BaseDriver): ASSET_KEY = "acme-corp/my-arm-v1" def define_interface(self, iface): iface.add_listener( TopicSpec(namespace="twin", leaf="command", payload_schema_ref="TwinCommandPayload"), CallbackGroup(callback=self._on_stop), command=CommandArgs(name="stop"), ) async def on_configure(self): ... async def on_connect_to_device(self): ... async def on_register_callbacks(self): ... async def on_activate(self): ... async def on_shutdown(self): ... @classmethod def create(cls): return cls.from_env() ``` ### Registry API reference | API | Role | | ------------------------------------- | ------------------------------------------------------------------------------- | | `define_interface(iface)` | **Source of truth** — topics, commands, callbacks | | `CustomDriver.get_manifest()` | Classmethod → compiled **MQTT** bundle for `twin.driver.set_schema` | | `CustomDriver.get_manifest_bundles()` | `{ "mqtt": ..., "zenoh": ... \| null }` — both catalogs from `define_interface` | | `CustomDriver.get_manifest(path=...)` | Also write `cw-driver.yml` | | `driver.manifest` | Compiled MQTT bundle on a live instance | | `driver.register_interface_on_twin()` | `twin.driver.set_schema(manifest)` on bound twin | | `DriverOperationMode` | Gates actuation-related listeners | | `BaseTelemetry` | Debounced `driver_info` on `twin/telemetry` | Session markers (`telemetry_start`, `connected`, …) use `client.mqtt` directly. Queue operational fields with `emit_driver_info()`; the registry telemetry publisher flushes them. *** ## Twin SDK: `twin.driver` vs `twin.commands` After you register a catalog with **`twin.driver.set_schema(...)`**: The fastest way to get started is the **Cyberwave Driver skill** for [Claude Code](https://claude.ai/claude-code). It asks you a few questions about your hardware and scaffolds a complete, production-ready driver project — including the Dockerfile, local dev setup, and a working twin connection. **Install the skill:** ```bash theme={null} git clone https://github.com/cyberwave-os/driver-skill ~/.claude/skills/cyberwave-driver ``` Then in any Claude Code session: ``` /cyberwave-driver ``` Claude will generate the full project tree and walk you through connecting it to a real twin locally. The skill source is open source at [cyberwave-os/driver-skill](https://github.com/cyberwave-os/driver-skill). *** ## Quickstart: use the SDK The fastest way to write a compatible driver is to use one of the official SDKs: * **Python SDK** — [`cyberwave-sdk`](https://pypi.org/project/cyberwave/) * **C++ SDK** — see the [C++ SDK docs](/tools/cpp-sdk) The SDKs handle twin synchronization, file I/O, reconnection logic, and more, so you can focus on the hardware integration. ## Environment variables When Edge Core starts a driver container it injects the following environment variables. You can develop your driver assuming these are always set to valid values — no need to handle the case where they are absent. | Variable | Description | | ---------------------------- | ----------------------------------------------------------------------------------------------------------------- | | `CYBERWAVE_TWIN_UUID` | UUID of the twin instance this driver manages | | `CYBERWAVE_API_KEY` | API key scoped to this driver for authenticating platform calls | | `CYBERWAVE_TWIN_JSON_FILE` | Absolute path to a writable JSON file containing the twin's current state (see [Twin JSON file](#twin-json-file)) | | `CYBERWAVE_CHILD_TWIN_UUIDS` | *(optional)* Comma-separated UUIDs of child twins (e.g. cameras) attached to this driver | `CYBERWAVE_CHILD_TWIN_UUIDS` is set when child twins are attached to the driver twin. Drivers can use this to coordinate child devices (for example, multiple cameras) without additional configuration. ### Restart behavior tuning The following optional variables let you override Edge Core's restart defaults: | Variable | Default | Description | | ---------------------------------------------- | ---------------------------- | ---------------------------------------------------------- | | `CYBERWAVE_DRIVER_RESTART_LOOP_THRESHOLD` | `4` | Number of restarts before the driver is marked as flapping | | `CYBERWAVE_DRIVER_RESTART_LOOP_WINDOW_SECONDS` | `60` | Time window (seconds) used to count restarts | | `CYBERWAVE_DRIVER_TROUBLESHOOTING_URL` | `https://docs.cyberwave.com` | URL surfaced in platform alerts for operator guidance | ## Driver failure handling Drivers must exit with a **non-zero** code when they cannot access required hardware (for example, a missing `/dev/video*` device or a disconnected peripheral). This allows Edge Core to detect startup failures and trigger restart logic. Edge Core raises the following alerts: * `driver_start_failure` — raised when a driver container cannot reach a stable running state. * `driver_restart_loop` — raised when a driver exceeds the restart threshold within the window. The container is stopped and marked as flapping. ## Twin JSON file `CYBERWAVE_TWIN_JSON_FILE` points to a JSON file on disk that contains the digital twin instance (including its `metadata`) and the associated catalog twin data, matching the `TwinSchema` and `AssetSchema` API schemas. Drivers may read and modify this file. Edge Core syncs any changes back to the backend when connectivity is available. ## Runtime configuration Drivers should treat `metadata["edge_configs"]` as the source of truth for per-device runtime configuration, and `metadata["edge_fingerprint"]` as the edge identity (not duplicated inside `edge_configs`). Read `edge_configs` from `CYBERWAVE_TWIN_JSON_FILE` at startup to obtain per-device settings without hardcoding them in the image. ## Declare your driver interface (`cw-driver.yml`) To make a **custom asset** work with Cyberwave, your driver project should include a **`cw-driver.yml`** at the repository root (next to your `Dockerfile`). The file is the **contract** between your hardware bridge, the platform APIs, the web UI, and the SDK: it lists which MQTT topics you use and which command strings you accept. Cyberwave’s own edge drivers (Go2, SO-101, DJI Mini, and others) use the same format — treat them as **reference implementations**, not as something you must fork. Copy the patterns that match your robot (locomotion, joint bus, camera stream, etc.) and trim what you do not implement. ### Compliance checklist | Step | What you do | | ------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | 1. **Asset** | Create an asset in your workspace for the hardware type you are integrating. Note its `registry_id` (or `slug`) — your manifest must target that id. | | 2. **Manifest** | Add `cw-driver.yml` to your driver repo. Declare every topic segment and `commands.supported` entry your code actually handles. | | 3. **Implement** | Subscribe/publish on the MQTT topics from your catalog (e.g. `cyberwave/twin/{uuid}/command`); see [Runtime messaging (MQTT)](#runtime-messaging-mqtt) and [MQTT API Reference](/api-reference/mqtt/main). | | 4. **Register interface** | `twin.commands.set_schema("./cw-driver.yml")` → `PUT /api/v1/twins/{uuid}` (see [Apply the manifest](#apply-the-manifest-to-your-twin) and [Platform API reference](#platform-api-reference)). | | 5. **Deploy** | Run the image with Edge Core (or locally with the same env vars). New twins spawned from that asset inherit `metadata["mqtt"]`. | If the manifest and the running driver disagree (undeclared command, wrong topic), teleop, agents, and SDK helpers will not line up with what the hardware actually does. ### Why you need it Without a declared interface catalog, the platform cannot know which MQTT topics your driver uses, which command strings are valid on `cyberwave/twin/{uuid}/command`, or whether a command is **continuous** (stick-held) vs **discrete** (one-shot). That metadata drives: | Consumer | What it uses from the catalog | | --------------- | -------------------------------------------------------------------------------------------------------------------------------------------------- | | **Web UI** | Dynamic control surfaces — e.g. the keyboard teleop overlay lists only `commands.supported` entries and respects `commands.specs[name].continuous` | | **Python SDK** | `twin.commands.get_schema()` and catalog methods such as `twin.commands.move_forward(...)` bound from `commands.supported` | | **Your driver** | A single source of truth for reviews, tests, and parity with production behavior | **Planned** — topic entries will also reference dedicated payload schemas (protobuf / robot-native layouts) for typed robot data on the wire. Today, `payload_schema_ref` names the logical schema; strict validation against those refs is not enforced yet. ### What `cw-driver.yml` contains Top-level fields: | Field | Purpose | | --------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------- | | `registry_ids` | One or more asset `registry_id` values this manifest describes — use the id of **your** workspace asset (e.g. `acme-corp/my-arm-v1`) | | `mqtt.schema_version` | Catalog format version (currently `1`) | | `mqtt.driver_family` | Implementation hint (`ros_cpp`, `python`, `android`, …) | | `mqtt..` | Per-topic contract — see [YAML path → MQTT topic](#yaml-path--mqtt-topic) below | | `mqtt.commands.supported` | Command names the driver accepts on the twin command topic — plain strings (discrete) or `{name, continuous, rate_hz, description}` objects | | `mqtt.commands.routers` | Optional routing hints for the edge bridge (e.g. `actuation`, `get_status`) | | `mqtt.commands.constraints` | Human-readable rules (source types, safety notes) surfaced to agents and docs | ### YAML path → MQTT topic Topic entries are nested as **`mqtt..`**. The leaf key is the **MQTT path segment** (use hyphens in YAML, e.g. `webrtc-offer`). Each entry becomes a flat key in `metadata["mqtt"]["topics"]` by joining a namespace prefix with the leaf: | YAML path | Compiled slug | Example live topic (`twin_uuid = abc`) | | ---------------------------- | ------------------------------------------------- | -------------------------------------- | | `mqtt.joint.update` | `cyberwave/joint/{twin_uuid}/update` | `cyberwave/joint/abc/update` | | `mqtt.joint.+` | `cyberwave/joint/{twin_uuid}/+` | `cyberwave/joint/abc/+` (wildcard) | | `mqtt.twin.command` | `cyberwave/twin/{twin_uuid}/command` | `cyberwave/twin/abc/command` | | `mqtt.twin.position` | `cyberwave/twin/{twin_uuid}/position` | `cyberwave/twin/abc/position` | | `mqtt.twin.telemetry` | `cyberwave/twin/{twin_uuid}/telemetry` | `cyberwave/twin/abc/telemetry` | | `mqtt.twin.webrtc-offer` | `cyberwave/twin/{twin_uuid}/webrtc-offer` | `cyberwave/twin/abc/webrtc-offer` | | `mqtt.twin.webrtc-answer` | `cyberwave/twin/{twin_uuid}/webrtc-answer` | `cyberwave/twin/abc/webrtc-answer` | | `mqtt.twin.webrtc-candidate` | `cyberwave/twin/{twin_uuid}/webrtc-candidate` | `cyberwave/twin/abc/webrtc-candidate` | | `mqtt.environment.` | `cyberwave/environment/{environment_uuid}/` | environment-scoped | **Namespace prefixes:** | `mqtt` namespace | MQTT prefix | | ---------------- | ------------------------------------------ | | `joint` | `cyberwave/joint/{twin_uuid}` | | `twin` | `cyberwave/twin/{twin_uuid}` | | `environment` | `cyberwave/environment/{environment_uuid}` | Each leaf entry sets `direction`, `payload_schema_ref`, `description`, and optional `source_types`. **WebRTC** signaling is always on the **twin** prefix (`cyberwave/twin/{uuid}/webrtc-offer`, `webrtc-answer`, `webrtc-candidate`, and `webrtc-command` for media-service commands). Declare them under **`mqtt.twin`** with hyphenated leaf names — not as a separate top-level namespace: ```yaml theme={null} mqtt: twin: command: # ... webrtc-offer: direction: both payload_schema_ref: WebRTCOfferPayload webrtc-answer: direction: both payload_schema_ref: WebRTCAnswerPayload ``` Only list WebRTC leaves when **this twin** participates in signaling (e.g. onboard camera or gimbal stream). Arm-only or joint-bus drivers (such as SO-101) should omit them; camera video uses **child camera twins** and the media service, not WebRTC topics on the arm twin. In some first-party Cyberwave profiles you may see a legacy `mqtt.webrtc.offer` shorthand (`offer` → `webrtc-offer` under the twin prefix). For **new** drivers, use `mqtt.twin.webrtc-*` only. ### Compiled catalog shape On the platform, your asset (and twins created from it) store a JSON bundle at **`metadata["mqtt"]`**: * `topics` — map of canonical slug → `{ direction, payload_schema_ref, description, … }` * `commands.supported` — list of command name strings * `commands.specs` — per-command flags such as `continuous`, `rate_hz`, and `description` (defaults to `""`) * `schema_version`, `driver_family` — format and implementation hints stub: expand with a worked example once the description field has shipped on more than one driver family. **`metadata["zenoh"]`** (optional, edge workers on the same host): * `channels` — map of channel name → `{ direction, payload_schema_ref, description, … }` * `locality` — typically `edge_colocated` * `schema_version`, `driver_family` — same versioning pattern as MQTT Author in YAML (or Python `define_interface`); the SDK compiles into metadata when you call **`twin.driver.set_schema`**. That call also **re-binds** `twin.commands.` from the updated `commands.supported` list. ### Reference implementations (Cyberwave drivers) Use these open-source manifests as templates — copy the namespaces and command style that match your hardware, then delete what you do not implement: | Profile | Manifest | Good template for | | ---------- | ----------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------ | | UGV Beast | [cw-driver.yml](https://github.com/cyberwave-os/cyberwave-edge-ros-ugv/blob/main/cw-driver.yml) | Mobile base, continuous locomotion, discrete utilities | | SO-101 arm | [cw-driver.yml](https://github.com/cyberwave-os/cyberwave-edge-so101/blob/main/cw-driver.yml) | `joint` bus + `twin.command` script commands (no WebRTC on the arm twin) | ### Apply the manifest to your twin After you author `cw-driver.yml`, register it on a **twin you own** with the Python SDK. You do not need to hand-build JSON or call low-level asset APIs — point `set_schema` at your manifest file and the platform updates that twin's `metadata["mqtt"]`, refreshes the catalog cache, and binds **catalog-derived command methods** on `twin.commands` (for example `twin.commands.stop()`, `twin.commands.move_forward(...)` when those names appear in `commands.supported`). ```python theme={null} from cyberwave import Cyberwave cw = Cyberwave(api_key="...") twin = cw.twin("your-twin-uuid") # or resolve from environment # Compile cw-driver.yml, persist on this twin, re-bind catalog methods twin.commands.set_schema("./cw-driver.yml") # Inspect the catalog the UI and SDK now use print(twin.commands.get_schema()["commands"]["supported"]) # Call any command you declared — derived from the manifest twin.commands.stop() twin.commands.move_forward(linear_x=0.3, duration=0.5, rate_hz=10) ``` Set `registry_ids` in the YAML to match the asset your twin was created from. Re-run `set_schema` whenever you change the manifest so teleop, agents, and SDK callers stay aligned with your driver. #### `set_schema` — REST and SDK | | | | ---------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | **Python SDK** | `twin.commands.set_schema(path_or_dict, merge=True)` — compiles `cw-driver.yml`, calls the twin update API, re-binds `twin.commands.` | | **REST** | `PUT /api/v1/twins/{uuid}` | | **Auth** | `Authorization: Bearer ` (same as `CYBERWAVE_API_KEY`) | | **Request body** | `{ "metadata": { "mqtt": { "schema_version": 1, "topics": { ... }, "commands": { "supported": [...], "specs": { ... } }, ... } } }` — only fields you want to change need to be sent; `set_schema` sends the full `metadata` object with `mqtt` set or merged | | **Response** | `TwinSchema` including updated `metadata` | | **Read back** | `GET /api/v1/twins/{uuid}` → `metadata.mqtt`; SDK: `twin.commands.get_schema()` | Full OpenAPI entry: [Update Twin](/api-reference/rest/DefaultApi#src_app_api_twins_update_twin). `merge=True` (default) deep-merges the new `metadata.mqtt` into existing twin metadata. `merge=False` replaces the entire `mqtt` block. Invoke catalog commands after `set_schema` — they publish over **MQTT**, not REST: | SDK | Wire | | ----------------------------------- | --------------------------------------------------------------------------------------------------------- | | `twin.commands.(**kwargs)` | Publish to `cyberwave/twin/{twin_uuid}/command` with envelope `{ source_type, command, data, timestamp }` | ### Platform API reference Use the **Python SDK** for day-to-day driver work; use **REST** when integrating from another language or CI. All REST paths are under `/api/v1` and require a bearer token unless noted otherwise. See the [REST API reference](/api-reference/rest/DefaultApi) for request/response schemas. #### Bring-up (workspace) | Step | REST | Python SDK | | ----------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------- | | Create catalog asset | `POST /api/v1/assets` — [Create Asset](/api-reference/rest/DefaultApi#src_app_api_assets_create_asset) | `client.assets.create(...)` | | Check `registry_id` free | `GET /api/v1/assets/check-registry-id/{registry_id}` — [Check Registry Id](/api-reference/rest/DefaultApi#src_app_api_assets_check_registry_id_availability) | — | | Update asset (optional shared catalog on asset) | `PUT /api/v1/assets/{uuid}` — [Update Asset](/api-reference/rest/DefaultApi#src_app_api_assets_update_asset) — set `metadata.mqtt` on the asset if every new twin should inherit the same manifest | `client.assets.update(asset_uuid, {"metadata": {"mqtt": ...}})` | | Create twin | `POST /api/v1/twins` — [Create Twin](/api-reference/rest/DefaultApi#src_app_api_twins_create_twin) | `client.twins.create(...)` | | List twins in environment | `GET /api/v1/environments/{uuid}/twins` — [Get Environment Twins](/api-reference/rest/DefaultApi#src_app_api_environments_get_environment_twins) | environment helpers / twin resolution | | Register driver manifest on **one twin** | `PUT /api/v1/twins/{uuid}` — [Update Twin](/api-reference/rest/DefaultApi#src_app_api_twins_update_twin) | `twin.commands.set_schema("./cw-driver.yml")` | | Read twin + catalog | `GET /api/v1/twins/{uuid}` — [Get Twin](/api-reference/rest/DefaultApi#src_app_api_twins_get_twin) | `twin.commands.get_schema()` or `twin.refresh()` | #### Runtime state (REST alternatives to MQTT) Drivers normally stream state over **MQTT** (see below). These REST endpoints are available for tools, simulators, or HTTP-only bridges: | Data | REST | Python SDK | | ---------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------- | | Pose (position / rotation / scale) | `PATCH /api/v1/twins/{uuid}/state` — [Update Twin State](/api-reference/rest/DefaultApi#src_app_api_twins_update_twin_state) | `client.twins.update_state(twin_uuid, {...})` | | Joint states (HTTP) | `PUT /api/v1/twins/{uuid}/joint_states` — [Update Twin Joint States](/api-reference/rest/DefaultApi#src_app_api_urdf_update_twin_joint_states) | `client.twins.update_joint_state(...)` / URDF helpers | | Single joint | `PUT /api/v1/twins/{uuid}/joints/{joint_name}/state` — [Update Twin Joint State](/api-reference/rest/DefaultApi#src_app_api_urdf_update_twin_joint_state) | — | | Latest camera frame | `GET /api/v1/twins/{uuid}/latest-frame` — [Get Twin Latest Frame](/api-reference/rest/DefaultApi#src_app_api_twins_get_twin_latest_frame) | `client.twins.get_latest_frame(...)` | | Driver logs | `GET /api/v1/twins/{uuid}/logs` — [Get Twin Driver Logs](/api-reference/rest/DefaultApi#src_app_api_twins_get_twin_driver_logs) | — | #### Runtime messaging (MQTT) Declared in `cw-driver.yml` → compiled into `metadata.mqtt.topics`. At runtime the driver (or SDK teleop) uses the **MQTT broker** (WebSocket URL in `NEXT_PUBLIC_MQTT_URL` for frontends; port `1883` / `9001` locally for edge). Topic prefix may include an environment segment in deployed stacks; slugs below are canonical. | Catalog slug (`metadata.mqtt.topics` key) | Example topic | Typical driver role | | --------------------------------------------- | ------------------------------------- | -------------------------------------------------------------------------------- | | `cyberwave/twin/{twin_uuid}/command` | `cyberwave/twin/abc/command` | **Subscribe** for `source_type: tele` commands; map `command` string to hardware | | `cyberwave/joint/{twin_uuid}/update` | `cyberwave/joint/abc/update` | **Subscribe** for tele joint targets; **publish** follower/sim observations | | `cyberwave/twin/{twin_uuid}/position` | `cyberwave/twin/abc/position` | **Publish** pose updates | | `cyberwave/twin/{twin_uuid}/telemetry` | `cyberwave/twin/abc/telemetry` | **Publish** lifecycle / status JSON | | `cyberwave/twin/{twin_uuid}/webrtc-offer` | `cyberwave/twin/abc/webrtc-offer` | **Subscribe/publish** signaling (camera twins) | | `cyberwave/twin/{twin_uuid}/webrtc-answer` | `cyberwave/twin/abc/webrtc-answer` | Signaling | | `cyberwave/twin/{twin_uuid}/webrtc-candidate` | `cyberwave/twin/abc/webrtc-candidate` | ICE candidates | | SDK publish helper | MQTT topic pattern | Typical caller | | --------------------------------------- | -------------------------------- | ------------------------------------------- | | `twin.commands.(...)` | `cyberwave/twin/{uuid}/command` | SDK clients, UI | | `twin.joints.set(...)` | `cyberwave/joint/{uuid}/update` | **Application teleop** (commands) | | `define_interface` + `from_ros` | `cyberwave/joint/{uuid}/update` | **Edge drivers** (measured `edge` feedback) | | `client.mqtt.update_joints_state(...)` | `cyberwave/joint/{uuid}/update` | Scripts, sims, non-registry publishers | | `client.mqtt.update_twin_position(...)` | `cyberwave/twin/{uuid}/position` | Drivers, sims | | `client.mqtt.update_twin_rotation(...)` | `cyberwave/twin/{uuid}/rotation` | Drivers, sims | Payload shapes: [MQTT API Reference](/api-reference/mqtt/main). #### Edge data bus (local, not REST) | SDK | Key expression | | -------------------------------------------- | ------------------------------------ | | `cw.data.publish("joint_states", {...})` | `cw/{twin_uuid}/data/joint_states` | | `cw.data.publish("frames/default", ndarray)` | `cw/{twin_uuid}/data/frames/default` | See [Data Wire Format](/edge/drivers/data-wire-format) for encoding rules. ### Assets, twins, and public catalog entries | Layer | Who controls it | Behavior | | --------------------------- | --------------- | -------------------------------------------------------------------------------------------------------------------------------- | | **Your asset** | Your workspace | Defines the hardware type; twins usually start with asset metadata. | | **Your twin** | Your workspace | `set_schema` updates **this twin's** `metadata["mqtt"]` so you can iterate without changing a shared asset definition. | | **Public Cyberwave assets** | Cyberwave | You can spawn twins from them; use `set_schema` on **your** twin if you need a custom command surface for a private integration. | First-party Cyberwave drivers ship a `cw-driver.yml` in their repositories for reference — you follow the same file format, then apply it to **your** twin with `set_schema` during bring-up. ## Sensor data output If your driver produces sensor data (video frames, depth maps, audio, joint states, etc.), publish it to the **edge data bus** so worker containers and ML models can consume it locally with zero network overhead. There are two options: the **Zenoh data bus** (recommended) and the **filesystem convention** (fallback for constrained environments). Both use the same channel names — a driver can switch between them by changing one env var. ### Option A: Zenoh data bus (recommended) The Zenoh data bus provides zero-copy shared memory between driver and worker containers. Data is consumed directly by worker hooks and `cw.data.latest()`. #### Key expression convention ``` cw/{twin_uuid}/data/{channel} ``` | Segment | Value | Example | | ----------- | ---------------------- | ---------------- | | `cw` | Fixed prefix | `cw` | | `twin_uuid` | UUID of the twin | `a1b2c3d4-...` | | `data` | Fixed namespace | `data` | | `channel` | Canonical channel name | `frames/default` | The `DataBus` handles key composition automatically via `CYBERWAVE_TWIN_UUID`. #### Canonical channels | Channel | Encoding | Pattern | Wire payload | | -------------------- | ------------------ | ------------ | ------------------------------------------------------------ | | `frames/default` | `numpy/ndarray` | Stream | SDK header + raw BGR/RGB uint8 | | `depth/default` | `numpy/ndarray` | Stream | SDK header + raw uint16 depth (mm) | | `joint_states` | `application/json` | Latest value | `{ts, names, positions, velocities?, efforts?, source_type}` | | `position` | `application/json` | Latest value | `{ts, x, y, z, qx?, qy?, qz?, qw?}` | | `audio/default` | `numpy/ndarray` | Stream | SDK header + float32 PCM | | `pointcloud/default` | `numpy/ndarray` | Stream | SDK header + Nx3 float32 | | `imu` | `application/json` | Stream | `{ts, accel: {x,y,z}, gyro: {x,y,z}}` | | `battery` | `application/json` | Latest value | `{ts, voltage_v, current_a, charge_pct}` | | `telemetry` | `application/json` | Latest value | Free-form `{ts, ...}` | You can define custom channels by picking any channel name. #### Python SDK example ```python theme={null} from cyberwave import Cyberwave import numpy as np import os, time cw = Cyberwave(api_key=os.environ["CYBERWAVE_API_KEY"], source_type="edge") # Binary stream: numpy array published with SDK header frame = np.zeros((480, 640, 3), dtype=np.uint8) cw.data.publish("frames/default", frame) # JSON latest-value: dict published as application/json cw.data.publish("joint_states", { "ts": time.time(), "names": ["shoulder_pan", "elbow_flex"], "positions": [0.1, -0.5], }) ``` `CYBERWAVE_TWIN_UUID` is read automatically from the environment. `CYBERWAVE_DATA_BACKEND` selects the transport (`zenoh` or `filesystem`). #### Wire format reference (for native language publishers) For C++, Rust, or any language that needs to publish without the Python SDK: ``` ┌──────────────────┬──────────┬──────────┬─────────────────────┬─────────────────┐ │ header_len (u32) │ ts (f64) │ seq (i64)│ header JSON (UTF-8) │ payload (bytes) │ │ 4 bytes, LE │ 8 bytes │ 8 bytes │ variable length │ variable length │ └──────────────────┴──────────┴──────────┴─────────────────────┴─────────────────┘ ``` Required JSON fields: * `content_type`: `"numpy/ndarray"` | `"application/json"` | `"application/octet-stream"` * `shape`: `[H, W, C]` (for ndarray; omit for JSON/bytes) * `dtype`: `"uint8"` | `"uint16"` | `"float32"` etc. (for ndarray; omit for JSON/bytes) #### C++ native publish example Minimal `zenoh-cpp` snippet that publishes frames with the correct header: ```cpp theme={null} #include #include #include #include std::vector pack_frame( const uint8_t* pixels, size_t pixel_bytes, int height, int width, int channels, double ts, int64_t seq ) { nlohmann::json meta; meta["content_type"] = "numpy/ndarray"; meta["shape"] = {height, width, channels}; meta["dtype"] = "uint8"; std::string json_str = meta.dump(); uint32_t header_len = 16 + static_cast(json_str.size()); std::vector buf(4 + header_len + pixel_bytes); size_t off = 0; memcpy(buf.data() + off, &header_len, 4); off += 4; memcpy(buf.data() + off, &ts, 8); off += 8; memcpy(buf.data() + off, &seq, 8); off += 8; memcpy(buf.data() + off, json_str.data(), json_str.size()); off += json_str.size(); memcpy(buf.data() + off, pixels, pixel_bytes); return buf; } int main() { auto config = zenoh::Config::default_config(); auto session = zenoh::Session::open(std::move(config)); std::string twin_uuid = std::getenv("CYBERWAVE_TWIN_UUID"); std::string key = "cw/" + twin_uuid + "/data/frames/default"; auto pub = session.declare_publisher(key); int64_t seq = 0; while (true) { // ... capture frame into pixels[] ... auto wire = pack_frame(pixels, pixel_bytes, 480, 640, 3, wall_clock_seconds(), seq++); pub.put(zenoh::Bytes(wire.data(), wire.size())); } } ``` The Python SDK's `DataBus.subscribe()` automatically decodes this payload — no adapter code needed. ### Option B: Filesystem convention (fallback) The filesystem convention is the **fallback** for environments where `eclipse-zenoh` cannot be installed. For most drivers, use `cw.data.publish()` (Zenoh data bus) instead — it provides zero-copy shared memory and is consumed directly by worker hooks. Both conventions use the same channel names. Write sensor data to a subfolder of the config directory that Edge Core mounts into your container: ``` $CYBERWAVE_EDGE_CONFIG_DIR/data/{twin_uuid}/{channel}/{sensor_name}/ ``` `CYBERWAVE_EDGE_CONFIG_DIR` is always set by Edge Core (defaults to `/app/.cyberwave`). #### Ring buffer (for stream data) ``` data/{twin_uuid}/frames/default/ ├── ring/ │ ├── 000000.npy │ ├── 000001.npy │ └── ... # numbered slots, wraps around └── meta.json # write pointer + format info ``` **Rules:** * Write `.npy` files to numbered slots: `{slot:06d}.npy` * Slot index = `write_count % buffer_size` (default: 120) * **Atomic writes**: write to `{slot}.npy.tmp`, then `rename()` to `{slot}.npy` * Update `meta.json` after each write #### Latest value (for state data) ``` data/{twin_uuid}/joint_states/ └── latest.json # overwritten each update ``` **Rules:** * Write a single JSON file: `latest.json` * **Atomic writes**: write to `latest.json.tmp`, then `rename()` * Include a `timestamp` field This is a **filesystem convention**, not a Python API. C++, Rust, or any other language can write `.npy` files and JSON to the same paths. ## MQTT topics and payloads If you publish data over MQTT directly (rather than through the SDK's `cw.data.publish`), see the [MQTT API Reference](/api-reference/mqtt/main) for the complete list of topics and payload schemas supported by the platform. That page covers: * Twin transform: position, rotation, scale * Joint state updates (single-joint, flat multi-joint, and aggregated formats) * Navigation commands and status reporting * Locomotion commands (`move_forward`, `turn_left`, etc.) * Telemetry lifecycle events (`connected`, `telemetry_start`, `telemetry_end`) * Sensor data: depth frames, point clouds, metrics * Edge health reporting * WebRTC signalling * Health check ping/pong ## Migrating from MQTT-only drivers If your driver currently publishes sensor data over MQTT, you can add Zenoh publishing without removing the MQTT path. The two paths serve different consumers: * **MQTT** → cloud backend (telemetry, frontend, workflows) * **Zenoh** → local worker containers (zero-copy inference, fusion) ### Step 1: Set CYBERWAVE\_DATA\_BACKEND Ensure `CYBERWAVE_DATA_BACKEND=zenoh` is set in the driver container. Edge Core sets this automatically for managed drivers. For manual testing: ```bash theme={null} docker run -e CYBERWAVE_DATA_BACKEND=zenoh ... ``` ### Step 2: Add cw\.data.publish alongside the MQTT call ```python theme={null} # Before (MQTT only) twin.client.mqtt.update_joints_state(twin_uuid=twin_uuid, ...) # After (dual-publish) twin.client.mqtt.update_joints_state(twin_uuid=twin_uuid, ...) # unchanged cw.data.publish("joint_states", {"ts": ts, "names": [...], "positions": [...]}) ``` Zenoh publish errors are caught and logged — they do not affect the MQTT path. ### Step 3: Verify with a subscriber ```python theme={null} sub = cw.data.subscribe("joint_states", lambda data: print(data)) # Run your driver; you should see joint dicts printed ``` ### Controlling which paths are active Set `CYBERWAVE_PUBLISH_MODE` to choose: | Value | Effect | | ------------ | ------------------------------------- | | `dual` | Both MQTT and Zenoh publish (default) | | `zenoh_only` | Only Zenoh (local-only drivers) | | `mqtt_only` | Only MQTT (legacy mode) | ## Licensing your driver You own your driver code. There are two common paths: * **Open source** — publish your driver as a public repository on GitHub under the [Apache 2.0 license](https://www.apache.org/licenses/LICENSE-2.0). This is our recommended default and makes it easier for the community to contribute and reuse your work. * **Closed source** — keep your driver proprietary. In this case, we recommend obfuscating your code before distributing the image and including a clear license file that reflects your distribution terms. Interested in writing a closed-source driver? [Reach out to us](mailto:info@cyberwave.com). ## Example driver repositories Fork or read these repositories when building your own compliant driver — each includes a `Dockerfile`, Edge Core env contract, and (where applicable) a `cw-driver.yml`: * **Camera** — [cyberwave-edge-camera-driver](https://github.com/cyberwave-os/cyberwave-edge-camera-driver): USB/RTSP camera as a twin (child of a parent robot twin). * **SO-101 arm** — [cyberwave-edge-so101](https://github.com/cyberwave-os/cyberwave-edge-so101): joint bus + command topic pattern for a manipulator. * **ROS UGV** — [cyberwave-edge-ros-ugv](https://github.com/cyberwave-os/cyberwave-edge-nodes/tree/main/cyberwave-edge-ros-ugv): locomotion and onboard camera profiles under `cw-driver-profiles/`. ## Advanced topics Once you have a working driver, these guides cover the platform features your driver can leverage: Hook-based worker modules for on-device ML inference and event-driven processing. SDK header encoding, key expressions, and the on-wire contract for edge data channels. Time-aware sensor fusion: interpolated point reads and time-window queries. Approximate time synchronizer that fires when samples from all listed channels arrive within tolerance. Capture live edge data to disk and replay it for deterministic debugging. Complete list of MQTT topics and payload schemas: telemetry, commands, navigation, joint states, and more. ## ROS 2 drivers Use **`BaseROS2Driver`** with **`from_ros=`** instead of hand-written MQTT bridges. Full end-to-end lab and success GIFs. What BaseDriver / BaseROS2Driver automate. API, troubleshooting, `ros_publish()` for commands. Protocol stack (TCP, ROS DDS, MQTT) for the UR Sim lab. ## Cartesian pose & dual-arm drivers Set `use_cartesian_pose = True` to expose an absolute `ee_pose` command. By default the raw pose is forwarded to your `_apply_ee_pose(pose, *, arm=None)` seam (implement it to send your robot's native cartesian target). Pass `process_joints_ik=True` to solve IK locally and apply joint targets instead. Set `use_dual_arm = True` and return a `DualArmConfig(left=..., right=...)` from `dual_arm_config()` for a single twin with two arms. Commands are exposed per arm (`left_*` / `right_*`) plus a conjunct `home`; telemetry stays a single combined payload.