Writing compatible drivers

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