Writing compatible drivers

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

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

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"),
        )

export CYBERWAVE_API_KEY=...
export CYBERWAVE_TWIN_UUID=<d455-twin-uuid>   # optional
cd cyberwave-python
PYTHONPATH=. python examples/fake_imu_driver.py

class FakeImu6dDriver(BaseDriver):
    ASSET_KEY = "intel/realsensed455"
    TICK_RATE_HZ = 50.0
    auto_register_interface = False  # demo calls set_schema() before run()

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"]),
  )

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)

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)

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()

git clone https://github.com/cyberwave-os/driver-skill ~/.claude/skills/cyberwave-driver

/cyberwave-driver

mqtt:
  twin:
    command:
      # ...
    webrtc-offer:
      direction: both
      payload_schema_ref: WebRTCOfferPayload
    webrtc-answer:
      direction: both
      payload_schema_ref: WebRTCAnswerPayload

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)

cw/{twin_uuid}/data/{channel}

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],
})

┌──────────────────┬──────────┬──────────┬─────────────────────┬─────────────────┐
│ header_len (u32) │ ts (f64) │ seq (i64)│ header JSON (UTF-8) │ payload (bytes) │
│   4 bytes, LE    │ 8 bytes  │ 8 bytes  │ variable length     │ variable length │
└──────────────────┴──────────┴──────────┴─────────────────────┴─────────────────┘

#include <zenoh.hxx>
#include <nlohmann/json.hpp>
#include <cstring>
#include <cstdint>

std::vector<uint8_t> 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<uint32_t>(json_str.size());
    std::vector<uint8_t> 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()));
    }
}

$CYBERWAVE_EDGE_CONFIG_DIR/data/{twin_uuid}/{channel}/{sensor_name}/

data/{twin_uuid}/frames/default/
├── ring/
│   ├── 000000.npy
│   ├── 000001.npy
│   └── ...         # numbered slots, wraps around
└── meta.json       # write pointer + format info

data/{twin_uuid}/joint_states/
└── latest.json     # overwritten each update

docker run -e CYBERWAVE_DATA_BACKEND=zenoh ...

# 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": [...]})

sub = cw.data.subscribe("joint_states", lambda data: print(data))
# Run your driver; you should see joint dicts printed