Robotics — Cookbook

This page gives short answers that you can copy for common robotics tasks. Each example assumes a constructed client:

import { Robotics } from "@clutchcall/sdk/robotics";

const r = new Robotics({
  relayHost: "relay.clutchcall.dev",
  robotId:   "spot-12",
  token:     process.env.CLUTCHCALL_CREDENTIALS!,
});

Publish a telemetry stream

Open one track per topic and write raw CDR. Delivery is best-effort by default. This is good for high-rate pose / odometry, where the freshest sample wins.

const odom = await r.publishTelemetry({
  topic:    "odom",
  typeName: "nav_msgs/msg/Odometry",
});
setInterval(() => odom.write(serializeOdom()), 50); // 20 Hz

Send a teleop command

Commands go cloud → robot on robot/<id>/ctl. Use the reliable lane so the network cannot drop a stop or a goal.

const cmd = await r.publishCommand({
  topic:    "cmd_vel",
  typeName: "geometry_msgs/msg/Twist",
  qos:      { reliability: "reliable" },
});
cmd.write(twistCdr(0.4 /* m/s */, 0.0 /* rad/s */));

The callback receives the raw CDR and the wire type name. Check the type name before you decode.

await r.subscribeTelemetry({ topic: "odom" }, (cdr, typeName) => {
  if (typeName !== "nav_msgs/msg/Odometry") return;
  const { x, y } = decodeOdom(cdr);
  dashboard.plot(x, y);
});

Have the robot receive commands

On the robot (or the on-robot bridge), subscribe to the command track. Apply each message to the local actuator stack.

await r.subscribeCommand({ topic: "cmd_vel" }, (cdr) => applyTwist(cdr));

Pick the right QoS lane

Use reliable for data that must arrive: commands, maps, goals. Use best-effort for high-rate sensors, where low latency is more important than completeness.

// must-arrive map update → subgroup stream, ordered
await r.publishTelemetry({ topic: "map", typeName: "nav_msgs/msg/OccupancyGrid",
  qos: { reliability: "reliable" } });

// 30 Hz LIDAR scan → QUIC datagram, lossy, lowest latency
await r.publishTelemetry({ topic: "scan", typeName: "sensor_msgs/msg/LaserScan",
  qos: { reliability: "best_effort", depth: 1 } });

Latch a value for late subscribers

transient_local makes the relay retain the last group(s). A dashboard that connects later immediately gets the current value. This behavior is the same as a latched / retained message.

const state = await r.publishTelemetry({
  topic:    "robot_state",
  typeName: "std_msgs/msg/String",
  qos:      { durability: "transient_local", reliability: "reliable", depth: 1 },
});
state.write(encode("idle")); // a future subscriber sees "idle" on join

Override priority per message

The QoS reliability sets a default send priority. Override it on a single write to move an urgent frame ahead of the queue (0 highest, 255 lowest).

const cmd = await r.publishCommand({ topic: "cmd_vel", typeName: "geometry_msgs/msg/Twist" });
cmd.write(normalTwist);            // default priority for the lane
cmd.write(emergencyStop(), 0);     // jump the queue

Demux across languages

The type-name prefix lets a Python publisher and a TypeScript subscriber agree with no schema registry. Publish from Python:

from clutchcall.robotics import Robotics
r = Robotics(relay_host="relay.clutchcall.dev", robot_id="spot-12", token=tok)
pub = r.publish_telemetry(topic="odom", type_name="nav_msgs/msg/Odometry")
pub.write(serialize_message(msg))   # raw CDR

The TypeScript dashboard above receives ("nav_msgs/msg/Odometry", cdr) unchanged.

Fan multiple topics into one dashboard

One client can hold many subscriptions. They all use the single QUIC session.

for (const t of ["odom", "battery_state", "scan"]) {
  await r.subscribeTelemetry({ topic: t }, (cdr, typeName) => ingest(t, typeName, cdr));
}

Construct one client per robot. Each client opens its own session and namespace.

const fleet = ["spot-12", "spot-13", "spot-14"].map(id =>
  new Robotics({ relayHost: "relay.clutchcall.dev", robotId: id, token }));

await Promise.all(fleet.map(c =>
  c.subscribeTelemetry({ topic: "battery_state" }, (cdr) => trackBattery(cdr))));

Bridge a local ROS 2 topic onto the mesh

Subscribe to the local graph. Forward raw CDR to a telemetry track. The Python example below uses rclpy serialization:

from rclpy.serialization import serialize_message
pub = r.publish_telemetry(topic="odom", type_name="nav_msgs/msg/Odometry")
node.create_subscription(Odometry, "/odom", lambda m: pub.write(serialize_message(m)), 10)

Bridge a command back onto the local graph

This is the other direction: subscribe to the MoQT command track and re-publish onto local DDS. Check the type name so the bridge drops messages with schema drift and does not mis-decode them.

from rclpy.serialization import deserialize_message
cmd_pub = node.create_publisher(Twist, "/cmd_vel", 10)
def on_cmd(cdr, type_name):
    if type_name == "geometry_msgs/msg/Twist":
        cmd_pub.publish(deserialize_message(cdr, Twist))
r.subscribe_command(topic="cmd_vel", on_message=on_cmd)

Tune the late-join window

depth limits how many recent groups the relay holds for a late subscriber. Set it to match your ROS 2 keep-last history.

await r.publishTelemetry({ topic: "tf", typeName: "tf2_msgs/msg/TFMessage",
  qos: { reliability: "reliable", depth: 50 } });

Shut down cleanly

Close publications and subscriptions, then close the client. This closes the QUIC session.

odom.close();
r.close();

SDK Methods — full signatures.
Recipes — end-to-end examples.

​Publish a telemetry stream

​Send a teleop command

​Subscribe to a robot’s telemetry from the cloud

​Have the robot receive commands

​Pick the right QoS lane

​Latch a value for late subscribers

​Override priority per message

​Demux across languages

​Fan multiple topics into one dashboard

​Subscribe to many robots from one process

​Bridge a local ROS 2 topic onto the mesh

​Bridge a command back onto the local graph

​Tune the late-join window

​Shut down cleanly

​Related