Receiving Control

A robot becomes useful when it does something with the input it receives. There are two sources of control:

operate.adamohq.com — gamepad and VR input from the hosted operator, published as ROS envelopes containing CDR-encoded payloads on control/joy and control/cdr/xr_tracking.
Your own custom operator — a teleop program you wrote yourself, publishing whatever topics and payload format you like (typically JSON). See Building Your Own Operator.

For both, the SDK primitive is the same: subscribe to a key expression and react to incoming samples.

Explicit control ownership

The hosted web operator can require a deliberate ownership step before teleoperation. An organization owner or admin enables Settings → Operator Control → Require explicit Take Control. Operators then enter robot pages in view-only mode and use Take Control and Release Control in the teleoperation header.

The policy is stored on the organization, so it follows users across browsers and applies to every operator in that organization.

Robot-facing ownership lease

While an operator holds explicit control, the web app publishes JSON on:

{robot}/control/json/operator_control

The SDK automatically adds the organization prefix. On the wire, the full key is adamo/{org}/{robot}/control/json/operator_control.

An acquisition looks like:

{
  "version": 1,
  "active": true,
  "event": "acquired",
  "operator": {
    "userId": "2b1d…",
    "sessionId": "b9f1…",
    "displayName": "Sam Considine"
  },
  "timestampMs": 1785270488243
}

The event field is one of:

Event	Meaning
`acquired`	This operator session has taken control
`heartbeat`	The same session still holds control; sent once per second
`released`	The session deliberately released control or left the robot page

Track ownership by operator.sessionId, not display name. A released event should only clear the matching session: during a takeover, the old holder’s release can race with the new holder’s acquisition.

Treat the stream as a lease. Stop accepting motion when either:

the matching released event arrives; or
no matching heartbeat arrives for your chosen timeout (three seconds is a reasonable starting point for the current one-second heartbeat).

The timeout is required because a crashed browser or broken network cannot reliably send released.

Inspecting the lifecycle with Python

The adamo-network repository includes adamo-py/examples/operator_control_listener.py. It listens to the ownership lease and browser keyboard topic together, printing acquired, heartbeats, keyboard edges, and released. Keyboard samples received without an active lease are labelled UNEXPECTED_WHILE_RELEASED.

Run it with:

ADAMO_API_KEY=ak_... \
python adamo-py/examples/operator_control_listener.py --robot my-arm

Use this listener while testing Take Control → input → Release Control. It is a diagnostic tool; the production receiver still needs to enforce the lease before applying commands.

Operator presence

The hosted web operator publishes a presence signal while a browser tab is viewing a robot. This is separate from explicit control ownership: a viewer can be present without holding control, and multiple viewers can be present at once.

Presence is a Zenoh liveliness token at:

{robot}/operator/presence/{userId}/{sessionId}/{displayNameBase64Url}

The SDK adds the organization prefix. On the wire, the full key is:

adamo/{org}/{robot}/operator/presence/{userId}/{sessionId}/{displayNameBase64Url}

Each browser tab gets its own sessionId. The last segment is the operator’s UTF-8 display name encoded as unpadded base64url. Use userId and sessionId as identity; display names are labels and are not unique.

Subscribe with history enabled so the callback first receives everyone already present and then receives changes:

import base64

import adamo


session = adamo.connect(api_key="ak_...")


def decode_display_name(segment: str) -> str:
    padding = "=" * (-len(segment) % 4)
    return base64.urlsafe_b64decode(segment + padding).decode("utf-8")


def on_operator_presence(key: str, is_present: bool) -> None:
    # SDK callbacks receive the org-prefix-stripped key:
    # robot/operator/presence/user/session/display
    robot, marker, presence, user_id, session_id, display = key.split("/", 5)
    if marker != "operator" or presence != "presence":
        return
    state = "joined" if is_present else "left"
    print(
        f"{decode_display_name(display)} {state} {robot} "
        f"(user={user_id}, session={session_id})"
    )


presence_sub = session.on_liveliness(
    "*/operator/presence/**",
    callback=on_operator_presence,
    history=True,
)

Keep presence_sub alive for as long as you want to receive events. The same primitive is available as Rust Session::on_liveliness, C adamo_liveliness_subscribe, and the TypeScript session’s liveliness().declareSubscriber. See Robot Discovery for cross-language examples.

A token appearing means that tab joined the robot view. It disappears when the tab leaves normally or when Zenoh detects that its session disconnected, so an unexpected browser or network failure does not depend on a final application message. Detection after an unclean disconnect is asynchronous and should not be used as a hard real-time actuator stop; use the short ownership lease above for control gating.

Viewing heartbeat

The web operator also publishes best-effort JSON every two seconds on:

{robot}/operator/viewing/{userId}/{sessionId}

{
  "displayName": "Sam Considine",
  "visible": true,
  "sentAt": 1785270488243
}

A clean exit publishes visible: false. If you consume this data topic instead of liveliness, expire a viewer after six seconds without a heartbeat because a crashed tab cannot reliably publish its final message. Prefer the liveliness token when you only need join/leave events.

There is currently no typed OperatorConnected / OperatorDisconnected SDK helper. These signals are readable through the public pub/sub and liveliness primitives, but applications must parse the key or JSON payload themselves.

Gamepad from operate.adamohq.com

The web app publishes a Joy message to {robot}/control/joy whenever the gamepad changes. The payload is a ROS envelope whose inner payload is a CDR-encoded sensor_msgs/msg/Joy.

Requires adamo>=0.4.54.

import time

import adamo
from adamo.xr import XRJoy, subscribe_xr_control


session = adamo.connect(api_key="ak_...")


def on_joy(sample):
    if sample.topic != "/joy" or not isinstance(sample.message, XRJoy):
        return
    axes = sample.message.axes
    buttons = sample.message.buttons
    # axes[0..1] = left stick X/Y, axes[2..3] = right stick X/Y, etc.
    drive(axes[0], axes[1])

# The ROS Header timestamp is preserved. Missing timestamps and samples that
# were already buffered for more than one second are dropped before on_joy.
sub = subscribe_xr_control(
    session,
    "my-arm",
    on_joy,
    channel="joy",
    max_age_seconds=1.0,
)

try:
    while True:
        time.sleep(1)
finally:
    sub.close()
    session.close()

The Rust SDK does not yet expose the typed ROS-envelope/CDR decoder. Decode the envelope and CDR manually (or with a ROS/CDR crate), and preserve the Header timestamp so delayed input can be rejected before it reaches the robot.

use adamo::Session;

fn main() -> adamo::Result<()> {
    let session = Session::open_default("ak_...")?;
    let _sub = session.subscribe_with("my-arm/control/joy", |sample| {
        // sample.payload is a ROS envelope. Strip it before decoding CDR.
        let Ok((topic, type_name, cdr_payload)) = decode_ros_envelope(&sample.payload) else {
            return;
        };
        if topic != "/joy" || type_name != "sensor_msgs/msg/Joy" {
            return;
        }

        // The helper below decodes the Joy fields used here.
        if let Ok(joy) = decode_joy_cdr(cdr_payload) {
            let Some(source_us) = joy.source_timestamp_us() else {
                return; // an unset timestamp is unsafe for robot commands
            };
            if adamo::fabric_now_us().saturating_sub(source_us) > 1_000_000 {
                return; // buffered for more than one second
            }
            drive(joy.axes[0], joy.axes[1]);
        }
    })?;
    std::thread::park();
    Ok(())
}

fn decode_ros_envelope(data: &[u8]) -> Result<(&str, &str, &[u8]), &'static str> {
    if data.len() < 8 {
        return Err("payload too short");
    }

    let mut offset = 0;
    let topic_len = u32::from_be_bytes(data[offset..offset + 4].try_into().unwrap()) as usize;
    offset += 4;
    if offset + topic_len + 4 > data.len() {
        return Err("bad topic length");
    }
    let topic = std::str::from_utf8(&data[offset..offset + topic_len]).map_err(|_| "bad topic")?;
    offset += topic_len;

    let type_len = u32::from_be_bytes(data[offset..offset + 4].try_into().unwrap()) as usize;
    offset += 4;
    if offset + type_len > data.len() {
        return Err("bad type length");
    }
    let type_name = std::str::from_utf8(&data[offset..offset + type_len]).map_err(|_| "bad type")?;
    offset += type_len;

    Ok((topic, type_name, &data[offset..]))
}

struct JoyMsg {
    stamp_sec: i32,
    stamp_nanosec: u32,
    axes: Vec<f32>,
    buttons: Vec<i32>,
}

impl JoyMsg {
    fn source_timestamp_us(&self) -> Option<u64> {
        if self.stamp_sec < 0
            || (self.stamp_sec == 0 && self.stamp_nanosec == 0)
            || self.stamp_nanosec >= 1_000_000_000
        {
            return None;
        }
        Some(self.stamp_sec as u64 * 1_000_000 + self.stamp_nanosec as u64 / 1_000)
    }
}

struct MinimalCdrReader<'a> {
    data: &'a [u8],
    offset: usize,
}

impl<'a> MinimalCdrReader<'a> {
    fn new(data: &'a [u8]) -> Result<Self, &'static str> {
        if data.len() < 4 || data[0] != 0x00 || data[1] != 0x01 {
            return Err("expected little-endian CDR payload");
        }
        Ok(Self { data, offset: 4 })
    }

    fn align(&mut self, size: usize) {
        let relative = self.offset - 4;
        self.offset += (size - (relative % size)) % size;
    }

    fn take<const N: usize>(&mut self) -> Result<[u8; N], &'static str> {
        if self.offset + N > self.data.len() {
            return Err("CDR payload ended early");
        }
        let bytes = self.data[self.offset..self.offset + N].try_into().unwrap();
        self.offset += N;
        Ok(bytes)
    }

    fn read_i32(&mut self) -> Result<i32, &'static str> {
        self.align(4);
        Ok(i32::from_le_bytes(self.take()?))
    }

    fn read_u32(&mut self) -> Result<u32, &'static str> {
        self.align(4);
        Ok(u32::from_le_bytes(self.take()?))
    }

    fn read_f32(&mut self) -> Result<f32, &'static str> {
        self.align(4);
        Ok(f32::from_le_bytes(self.take()?))
    }

    fn read_string(&mut self) -> Result<&'a str, &'static str> {
        let len = self.read_u32()? as usize;
        if len == 0 {
            return Ok("");
        }
        if self.offset + len > self.data.len() {
            return Err("CDR string ended early");
        }
        let value = std::str::from_utf8(&self.data[self.offset..self.offset + len - 1])
            .map_err(|_| "bad CDR string")?;
        self.offset += len;
        Ok(value)
    }
}

fn decode_joy_cdr(data: &[u8]) -> Result<JoyMsg, &'static str> {
    let mut reader = MinimalCdrReader::new(data)?;
    let stamp_sec = reader.read_i32()?;
    let stamp_nanosec = reader.read_u32()?;
    if stamp_nanosec >= 1_000_000_000 {
        return Err("invalid ROS timestamp");
    }
    let _frame_id = reader.read_string()?;

    let axes_len = reader.read_u32()? as usize;
    let mut axes = Vec::with_capacity(axes_len);
    for _ in 0..axes_len {
        axes.push(reader.read_f32()?);
    }

    let buttons_len = reader.read_u32()? as usize;
    let mut buttons = Vec::with_capacity(buttons_len);
    for _ in 0..buttons_len {
        buttons.push(reader.read_i32()?);
    }

    Ok(JoyMsg { stamp_sec, stamp_nanosec, axes, buttons })
}

The plain C API does not expose a typed ROS-envelope/CDR decoder. Parse the outer envelope, then use a CDR library and retain header.stamp for freshness checks.

adamo_session_t *sess = adamo_open_default("ak_...");

void on_joy(const adamo_sample_t *s, void *user) {
    /*
     * s->payload is a ROS envelope:
     *   u32be topic_len, topic bytes, u32be type_len, type bytes, CDR payload.
     * Strip the envelope, verify topic="/joy" and type="sensor_msgs/msg/Joy",
     * then decode the remaining CDR bytes as sensor_msgs/msg/Joy. Preserve
     * Header.stamp and reject unset or old source timestamps against
     * adamo_fabric_now_us() before issuing a robot command.
     */
}

adamo_cb_sub_t *sub = adamo_subscribe_cb(
    sess, "my-arm/control/joy", on_joy, /* user */ NULL);

For C/C++ robots, eProsima Fast CDR is a small library that handles the wire format.

The standard W3C / Xbox button and axis mapping is documented in the TypeScript SDK reference.

VR Headset from operate.adamohq.com

When a viewer enters immersive VR mode on a stereo track, the headset publishes pose data continuously:

Inner topic	Payload	Description
`/head_pose`	`geometry_msgs/msg/PoseStamped`	Headset pose
`/controller/left`	`geometry_msgs/msg/PoseStamped`	Left controller grip (wrist/hold pose)
`/controller/right`	`geometry_msgs/msg/PoseStamped`	Right controller grip (wrist/hold pose)
`/controller/{handedness}/tip`	`geometry_msgs/msg/PoseStamped`	Controller tip — the aim pose (WebXR `targetRaySpace`)
`/controller/{handedness}/joy`	`sensor_msgs/msg/Joy`	XR controller axes and buttons

The hosted UI sends XR data on {robot}/control/cdr/xr_tracking. Each payload is a ROS envelope whose inner topic is /head_pose, /controller/{handedness}, /controller/{handedness}/tip, or /controller/{handedness}/joy.

Save this as adamo_xr_to_ros.py:

#!/usr/bin/env python3
import argparse
from functools import partial

import adamo
import rclpy
from adamo.xr import XRDecodeError, decode_ros_envelope
from geometry_msgs.msg import PoseStamped
from rclpy.serialization import deserialize_message
from sensor_msgs.msg import Joy


ROS_TYPES = {
    "geometry_msgs/msg/PoseStamped": PoseStamped,
    "sensor_msgs/msg/Joy": Joy,
}


def publish_xr_sample(sample, *, node, publishers):
    try:
        envelope = decode_ros_envelope(bytes(sample.payload))
    except XRDecodeError as error:
        node.get_logger().warn(f"Ignoring malformed XR payload: {error}")
        return

    msg_type = ROS_TYPES.get(envelope.type_name)
    if msg_type is None:
        node.get_logger().warn(f"Ignoring unsupported XR type: {envelope.type_name}")
        return

    ros_topic = f"/adamo/xr{envelope.topic}"
    if ros_topic not in publishers:
        publishers[ros_topic] = node.create_publisher(msg_type, ros_topic, 10)
        node.get_logger().info(f"Publishing XR {envelope.topic} -> {ros_topic}")

    publishers[ros_topic].publish(deserialize_message(envelope.cdr, msg_type))


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--api-key", required=True)
    parser.add_argument("--robot", required=True)
    args = parser.parse_args()

    rclpy.init()
    node = rclpy.create_node("adamo_xr_bridge")
    publishers = {}
    session = adamo.connect(api_key=args.api_key)

    sub = session.subscribe(
        f"{args.robot}/control/cdr/xr_tracking",
        callback=partial(publish_xr_sample, node=node, publishers=publishers),
    )
    node.get_logger().info(f"Bridging {args.robot}/control/cdr/xr_tracking")

    try:
        rclpy.spin(node)
    finally:
        sub.close()
        session.close()
        node.destroy_node()
        rclpy.shutdown()


if __name__ == "__main__":
    main()

Run it from a ROS-sourced shell:

python3 adamo_xr_to_ros.py --api-key ak_... --robot my-arm

It republishes to ROS topics such as /adamo/xr/head_pose, /adamo/xr/controller/left, and /adamo/xr/controller/left/joy. If you do not run ROS, use subscribe_xr_control from adamo.xr. It returns a typed PoseStamped or XRJoy, preserves the source Header timestamp and frame, and can reject stale source input before your callback runs. Quaternion order is always ROS/WebXR scalar-last: x, y, z, w.

Custom Control Topics

When you write your own operator (XR app, mobile, GELLO leader, …), you choose the topic name and the payload format. JSON is the default in the Python SDK and works across all three languages.

The example below is the robot-side receiver that pairs with the bimanual XR operator on the Building Your Own Operator page.

The @robot.on(...) decorator subscribes and decodes JSON in one step. Brace-wrapped path segments capture the matched value as a keyword argument.

import adamo

robot = adamo.Robot(api_key="ak_...", name="my-arm")

# Cameras streamed back to the operator
robot.attach_video("wrist_left",  device="/dev/video0")
robot.attach_video("wrist_right", device="/dev/video1")
robot.attach_video("head",        shm="head_cam")

# Bimanual hand controllers — {side} captures "left" or "right"
@robot.on("xr-operator", "control/xr/hand/{side}", priority=250)
def hand(msg, side):
    move_arm(side, msg["pos"], msg["quat"])
    set_gripper(side, msg["trigger"])

# Head pose — separate handler
@robot.on("xr-operator", "control/xr/head", priority=250)
def on_head(msg):
    update_head_tracking(msg["pos"], msg["quat"])

robot.run()

The first argument to @robot.on(...) is the broadcaster name — the operator publishing those topics. The decorator subscribes to {broadcaster}/{track} under the hood.

use adamo::Session;
use serde::Deserialize;

#[derive(Deserialize)]
struct HandMsg { pos: [f32; 3], quat: [f32; 4], trigger: f32 }

fn main() -> adamo::Result<()> {
    let session = Session::open_default("ak_...")?;

    let _left = session.subscribe_with("xr-operator/control/xr/hand/left", |s| {
        let msg: HandMsg = serde_json::from_slice(&s.payload).unwrap();
        move_arm("left", msg.pos, msg.quat);
        set_gripper("left", msg.trigger);
    })?;

    let _right = session.subscribe_with("xr-operator/control/xr/hand/right", |s| {
        let msg: HandMsg = serde_json::from_slice(&s.payload).unwrap();
        move_arm("right", msg.pos, msg.quat);
        set_gripper("right", msg.trigger);
    })?;

    std::thread::park();
    Ok(())
}

The Rust SDK doesn’t have brace-capture pattern matching like Python — declare a subscriber per side, or subscribe to xr-operator/control/xr/hand/* and pull the side off the key string yourself.

Use one callback subscriber per topic, or subscribe to a wildcard and switch on the key:

void on_hand(const adamo_sample_t *s, void *user) {
    const char *side = strrchr(s->key, '/');
    if (side) side++;  /* "left" or "right" */
    /* decode JSON from s->payload, s->payload_len */
    /* move_arm(side, ...); */
}

adamo_subscribe_cb(sess, "xr-operator/control/xr/hand/*", on_hand, NULL);

Task Signals

For labelling data collection, the hosted UI can emit task start/stop signals — the operator picks a named task and marks when it begins and ends. This is opt-in (Settings → Tasks → “Show task list” in the web app) and is a standalone signal: it does not touch the recording pipeline. Use it to segment and tag what your robot is doing.

The signal is published on:

{robot}/tasks/signal

reliably (DATA priority), as UTF-8 JSON:

{
  "action": "start",
  "task_id": "d69c5a80-9f03-4d8a-b32a-3819cd00a7c9",
  "task_name": "Solder my own hand",
  "ts": 1780686715804
}

action is "start" or "stop". task_id is the task’s UUID (matches the task in the web app); task_name is the human-readable label (may be absent); ts is Date.now() in milliseconds. Subscribe to */tasks/signal to catch every robot in the org — the robot is the first key segment.

import adamo, json

session = adamo.connect(api_key="ak_...")

def on_task(sample):
    sig = json.loads(sample.payload)
    robot = sample.key.split("/")[0]
    print(robot, sig["action"], sig["task_id"], sig.get("task_name"))

sub = session.subscribe("*/tasks/signal", callback=on_task)

use adamo::Session;
use serde::Deserialize;

#[derive(Deserialize)]
struct TaskSignal { action: String, task_id: String, task_name: Option<String> }

fn main() -> adamo::Result<()> {
    let session = Session::open_default("ak_...")?;
    let _sub = session.subscribe_with("*/tasks/signal", |s| {
        let sig: TaskSignal = serde_json::from_slice(&s.payload).unwrap();
        println!("{} {} {:?}", sig.action, sig.task_id, sig.task_name);
    })?;
    std::thread::park();
    Ok(())
}

void on_task(const adamo_sample_t *s, void *user) {
    /* s->key is e.g. "arm-01/tasks/signal"; decode JSON from
       s->payload, s->payload_len for action / task_id / task_name */
}

adamo_subscribe_cb(sess, "*/tasks/signal", on_task, NULL);

Managing tasks programmatically

Tasks and sets are normally created in the web app, but you can also manage them over a small REST API — handy for pre-creating a set before a collection run so the task_id is known up front. Authenticate with your org API key (the same ak_… key the SDKs use): exchange it for a short-lived bearer token, then call the task endpoints. The token response also gives you your org_id.

RESP=$(curl -s -X POST https://api.adamohq.com/api/keys/token -H "X-API-Key: ak_...")
TOKEN=$(jq -r .token   <<<"$RESP")
ORG=$(jq -r .org_id <<<"$RESP")

All endpoints are under https://api.adamohq.com/api/orgs/{org_id} and take/return JSON:

Method	Path	Body	Description
`GET`	`/task-sets`	—	List sets, each with its tasks (ordered by `position`)
`POST`	`/task-sets`	`{name, position?}`	Create a set
`GET`	`/task-sets/{set_id}`	—	Get one set with its tasks
`DELETE`	`/task-sets/{set_id}`	—	Delete a set (and its tasks)
`POST`	`/task-sets/{set_id}/tasks`	`{name, position?}`	Add a task
`PATCH`	`/task-sets/{set_id}/tasks/{task_id}`	`{name?, position?}`	Rename / reorder a task
`DELETE`	`/task-sets/{set_id}/tasks/{task_id}`	—	Remove a task

# Create a set, then add a task to it
SET=$(curl -s -X POST https://api.adamohq.com/api/orgs/$ORG/task-sets \
  -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" \
  -d '{"name":"Pick and place"}' | jq -r .id)

curl -X POST https://api.adamohq.com/api/orgs/$ORG/task-sets/$SET/tasks \
  -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" \
  -d '{"name":"Grasp the cube"}'

The task_id returned here is the same UUID that appears in the task signal above, so a script can lay out a set and the operator just selects and records.

Bilateral Force Feedback

A bilateral teleop setup sends measured follower joint efforts back to the leader on a real-time control topic. The leader keeps only the newest sample, scales it negatively, then applies it in the leader arm’s external-effort mode.

The example below isolates the feedback path. Replace the two hardware-driver functions with calls to your robot API:

follower side: read measured external efforts from the follower arm
leader side: apply external efforts to the leader arm

The payload is eight big-endian f64 values: [timestamp_seconds, effort_0, ..., effort_6].

import os
import struct
import sys
import time

import adamo

N_JOINTS = 7
RATE_HZ = 100.0
GAIN = 0.1
ROBOT = os.getenv("ADAMO_ROBOT_NAME", "my-arm")
EFFORT_TOPIC = f"{ROBOT}/control/force_feedback/follower_effort"


def now_seconds(session):
    return session.fabric_now_us() / 1_000_000.0


def pack_efforts(timestamp, efforts):
    if len(efforts) != N_JOINTS:
        raise ValueError(f"expected {N_JOINTS} joint efforts")
    return struct.pack("!" + "d" * (1 + N_JOINTS), timestamp, *efforts)


def unpack_efforts(payload):
    if len(payload) != 8 * (1 + N_JOINTS):
        raise ValueError(f"bad effort payload size: {len(payload)}")
    values = struct.unpack("!" + "d" * (1 + N_JOINTS), payload)
    return values[0], list(values[1:])


def read_follower_external_efforts():
    raise NotImplementedError("read joint efforts from your follower driver")


def apply_leader_external_efforts(efforts):
    raise NotImplementedError("send joint efforts to your leader driver")


def run_follower(session):
    with session.publisher(
        EFFORT_TOPIC,
        priority=250,
        express=True,
        reliable=False,
    ) as pub:
        while True:
            efforts = read_follower_external_efforts()
            pub.put(pack_efforts(now_seconds(session), efforts))
            time.sleep(1.0 / RATE_HZ)


def run_leader(session):
    teleop_started_at = now_seconds(session)

    with session.subscribe(EFFORT_TOPIC) as sub:
        while True:
            latest = None
            while True:
                sample = sub.try_recv()
                if sample is None:
                    break
                latest = sample

            if latest is not None:
                timestamp, measured = unpack_efforts(latest.payload)
                if timestamp >= teleop_started_at:
                    applied = [-GAIN * effort for effort in measured]
                    apply_leader_external_efforts(applied)

            time.sleep(1.0 / RATE_HZ)


session = adamo.connect(
    api_key=os.environ["ADAMO_API_KEY"],
)

mode = sys.argv[1] if len(sys.argv) > 1 else "leader"
if mode == "follower":
    run_follower(session)
else:
    run_leader(session)

use adamo::{PublisherOptions, Session};
use std::{env, thread, time::Duration};

const N_JOINTS: usize = 7;
const EFFORT_BYTES: usize = (1 + N_JOINTS) * 8;
const RATE_HZ: f64 = 100.0;
const GAIN: f64 = 0.1;

fn effort_topic() -> String {
    let robot = env::var("ADAMO_ROBOT_NAME").unwrap_or_else(|_| "my-arm".to_string());
    format!("{robot}/control/force_feedback/follower_effort")
}

fn now_seconds() -> f64 {
    adamo::fabric_now_us() as f64 / 1_000_000.0
}

fn pack_efforts(timestamp: f64, efforts: [f64; N_JOINTS]) -> [u8; EFFORT_BYTES] {
    let mut out = [0_u8; EFFORT_BYTES];
    for (i, value) in std::iter::once(timestamp)
        .chain(efforts.iter().copied())
        .enumerate()
    {
        let start = i * 8;
        out[start..start + 8].copy_from_slice(&value.to_be_bytes());
    }
    out
}

fn unpack_efforts(payload: &[u8]) -> Option<(f64, [f64; N_JOINTS])> {
    if payload.len() != EFFORT_BYTES {
        return None;
    }

    let mut timestamp_bytes = [0_u8; 8];
    timestamp_bytes.copy_from_slice(&payload[0..8]);

    let mut efforts = [0.0_f64; N_JOINTS];
    for (i, effort) in efforts.iter_mut().enumerate() {
        let start = (i + 1) * 8;
        let mut bytes = [0_u8; 8];
        bytes.copy_from_slice(&payload[start..start + 8]);
        *effort = f64::from_be_bytes(bytes);
    }

    Some((f64::from_be_bytes(timestamp_bytes), efforts))
}

fn realtime_options() -> PublisherOptions {
    PublisherOptions {
        priority: 250,
        express: true,
        reliable: false,
    }
}

fn read_follower_external_efforts() -> [f64; N_JOINTS] {
    todo!("read joint efforts from your follower driver")
}

fn apply_leader_external_efforts(_efforts: [f64; N_JOINTS]) {
    todo!("send joint efforts to your leader driver")
}

fn run_follower(session: &Session) -> adamo::Result<()> {
    let topic = effort_topic();
    let publisher = session.publisher(&topic, realtime_options())?;

    loop {
        let efforts = read_follower_external_efforts();
        let payload = pack_efforts(now_seconds(), efforts);
        publisher.put(&payload)?;
        thread::sleep(Duration::from_secs_f64(1.0 / RATE_HZ));
    }
}

fn run_leader(session: &Session) -> adamo::Result<()> {
    let topic = effort_topic();
    let subscriber = session.subscribe(&topic)?;
    let teleop_started_at = now_seconds();

    loop {
        let mut latest = None;
        while let Some(sample) = subscriber.try_recv()? {
            latest = Some(sample);
        }

        if let Some(sample) = latest {
            if let Some((timestamp, measured)) = unpack_efforts(&sample.payload) {
                if timestamp >= teleop_started_at {
                    let mut applied = [0.0_f64; N_JOINTS];
                    for (out, effort) in applied.iter_mut().zip(measured.iter()) {
                        *out = -GAIN * *effort;
                    }
                    apply_leader_external_efforts(applied);
                }
            }
        }

        thread::sleep(Duration::from_secs_f64(1.0 / RATE_HZ));
    }
}

fn main() -> adamo::Result<()> {
    let api_key = env::var("ADAMO_API_KEY").expect("set ADAMO_API_KEY");
    let session = Session::open_default(&api_key)?;
    let mode = env::args().nth(1).unwrap_or_else(|| "leader".to_string());

    if mode == "follower" {
        run_follower(&session)
    } else {
        run_leader(&session)
    }
}

#define _POSIX_C_SOURCE 199309L

#include <adamo/adamo.h>

#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>

#define N_JOINTS 7
#define EFFORT_BYTES ((1 + N_JOINTS) * 8)
#define RATE_HZ 100.0
#define GAIN 0.1

static int host_is_little_endian(void) {
    const uint16_t probe = 1;
    return *(const uint8_t *)&probe == 1;
}

static void pack_be_double(double value, uint8_t *out) {
    uint8_t bytes[8];
    memcpy(bytes, &value, 8);
    if (host_is_little_endian()) {
        for (int i = 0; i < 8; i++) out[i] = bytes[7 - i];
    } else {
        memcpy(out, bytes, 8);
    }
}

static double unpack_be_double(const uint8_t *in) {
    uint8_t bytes[8];
    if (host_is_little_endian()) {
        for (int i = 0; i < 8; i++) bytes[i] = in[7 - i];
    } else {
        memcpy(bytes, in, 8);
    }
    double value = 0.0;
    memcpy(&value, bytes, 8);
    return value;
}

static void pack_efforts(double timestamp,
                         const double efforts[N_JOINTS],
                         uint8_t out[EFFORT_BYTES]) {
    pack_be_double(timestamp, out);
    for (int i = 0; i < N_JOINTS; i++) {
        pack_be_double(efforts[i], out + (i + 1) * 8);
    }
}

static int unpack_efforts(const uint8_t *payload,
                          size_t len,
                          double *timestamp,
                          double efforts[N_JOINTS]) {
    if (len != EFFORT_BYTES) return -1;
    *timestamp = unpack_be_double(payload);
    for (int i = 0; i < N_JOINTS; i++) {
        efforts[i] = unpack_be_double(payload + (i + 1) * 8);
    }
    return 0;
}

static double now_seconds(void) {
    return (double)adamo_fabric_now_us() / 1000000.0;
}

static void sleep_tick(void) {
    struct timespec ts;
    ts.tv_sec = 0;
    ts.tv_nsec = (long)(1000000000.0 / RATE_HZ);
    nanosleep(&ts, NULL);
}

static void read_follower_external_efforts(double efforts[N_JOINTS]) {
    /* Fill from your follower driver, e.g. get_all_external_efforts(). */
    for (int i = 0; i < N_JOINTS; i++) efforts[i] = 0.0;
}

static void apply_leader_external_efforts(const double efforts[N_JOINTS]) {
    /* Send to your leader driver, e.g. set_all_external_efforts(...). */
    (void)efforts;
}

static int run_follower(adamo_session_t *session, const char *topic) {
    adamo_publisher_t *pub = adamo_publisher(
        session, topic, /* priority */ 250, /* express */ 1, /* reliable */ 0);
    if (!pub) return -1;

    for (;;) {
        double efforts[N_JOINTS];
        uint8_t payload[EFFORT_BYTES];
        read_follower_external_efforts(efforts);
        pack_efforts(now_seconds(), efforts, payload);
        if (adamo_publisher_put(pub, payload, sizeof(payload)) != 0) {
            adamo_publisher_free(pub);
            return -1;
        }
        sleep_tick();
    }
}

static int run_leader(adamo_session_t *session, const char *topic) {
    adamo_subscriber_t *sub = adamo_subscribe(session, topic);
    if (!sub) return -1;

    const double teleop_started_at = now_seconds();
    for (;;) {
        adamo_sample_t *latest = NULL;
        for (;;) {
            adamo_sample_t *sample = adamo_sub_try_recv(sub);
            if (!sample) {
                const char *err = adamo_last_error();
                if (err && *err) {
                    if (latest) adamo_sample_free(latest);
                    adamo_sub_free(sub);
                    return -1;
                }
                break;
            }
            if (latest) adamo_sample_free(latest);
            latest = sample;
        }

        if (latest) {
            double timestamp = 0.0;
            double measured[N_JOINTS];
            if (unpack_efforts(latest->payload, latest->payload_len,
                               &timestamp, measured) == 0 &&
                timestamp >= teleop_started_at) {
                double applied[N_JOINTS];
                for (int i = 0; i < N_JOINTS; i++) {
                    applied[i] = -GAIN * measured[i];
                }
                apply_leader_external_efforts(applied);
            }
            adamo_sample_free(latest);
        }

        sleep_tick();
    }
}

int main(int argc, char **argv) {
    const char *api_key = getenv("ADAMO_API_KEY");
    const char *robot = getenv("ADAMO_ROBOT_NAME");
    if (!api_key) {
        fprintf(stderr, "set ADAMO_API_KEY\n");
        return 1;
    }
    if (!robot) robot = "my-arm";

    char topic[256];
    snprintf(topic, sizeof(topic), "%s/control/force_feedback/follower_effort", robot);

    adamo_session_t *session = adamo_open_default(api_key);
    if (!session) {
        fprintf(stderr, "adamo_open: %s\n", adamo_last_error());
        return 1;
    }

    const char *mode = argc > 1 ? argv[1] : "leader";
    int rc = strcmp(mode, "follower") == 0
        ? run_follower(session, topic)
        : run_leader(session, topic);

    fprintf(stderr, "adamo error: %s\n", adamo_last_error());
    adamo_session_free(session);
    return rc == 0 ? 0 : 1;
}

Real-Time Priority

Control topics should be published with REAL_TIME priority and dropped on congestion — a command that arrives 200 ms late is worse than no command at all. The robot side doesn’t choose priority on subscribe (the router uses the publisher’s choice), but it’s worth knowing the convention so your own operator programs follow it.

robot.publish("control/joy", priority=250, express=True)
# 0–255 mapped to 8 priority classes; ≥240 is REAL_TIME.

session.publisher("control/joy", PublisherOptions {
    priority: 250,
    express: true,
    reliable: false,
})?;

adamo_publisher(sess, "control/joy",
    /* priority */ 250, /* express */ 1, /* reliable */ 0);

Link Safety Gate

Robots publish a 1 Hz heartbeat on {robot}/heartbeat carrying latency stats — the network regime classified by the robot’s congestion forecaster, jitter, loss rate, and queuing delay. Use it as a safety gate before applying teleop commands: stop the robot when heartbeats go stale or the regime degrades.

import time
from adamo.stats import Regime

last_beat = time.monotonic()

def on_stats(stats):
    global last_beat
    last_beat = time.monotonic()
    if stats.regime is not Regime.STABLE:
        slow_down()

sub = session.watch_latency("my-arm", on_stats)

# In the control loop: two missed heartbeats = stale link
if time.monotonic() - last_beat > 2.5:
    stop_robot()

import { watchLatest } from "adamo/fleet";

let lastBeat = performance.now();
const hb = await watchLatest(session, `adamo/${org}/my-arm/heartbeat`);
hb.onMessage(() => { lastBeat = performance.now(); });

// In the control loop: two missed heartbeats = stale link
if (performance.now() - lastBeat > 2500) stopRobot();

For dashboards, the parsed stats expose regime, jitter_ms, garch_sigma_ms, target_bitrate_kbps, loss_rate, and queuing_delay_ms — see the per-SDK LatencyStats reference.

See Building Your Own Operator for the publisher side end-to-end.

Hosted UI Payload Reference

The current adamo-ts/examples/web frontend publishes operator input as real-time, best-effort messages. The exact key and payload depend on the input mode.

Browser keyboard

Raw keyboard state is published on:

{robot}/control/json/keyboard

The payload is UTF-8 JSON:

{
  "key": "w",
  "code": "KeyW",
  "action": "down",
  "stamp": 1770210000000
}

key is the browser KeyboardEvent.key value, so it reflects the active keyboard layout and modifiers. code is the browser KeyboardEvent.code value, so it identifies the physical key position. action is "down" or "up", and stamp is Date.now() in milliseconds.

The hosted UI suppresses browser auto-repeat events. While a physical key remains held, it republishes a "down" heartbeat about every 100 ms. Receivers should treat the heartbeat as a dead-man signal and stop the associated motion if it expires, rather than relying only on a single "up" edge.

CDR ROS envelope

When the UI is using CDR mode, the bytes on the Adamo key are a small ROS envelope followed by the CDR-encoded ROS message:

u32 topic_length_be
utf8 topic
u32 type_length_be
utf8 type
bytes cdr_payload

The topic and type fields describe the inner ROS message. The cdr_payload is the serialized message named by type.

Python applications should normally use decode_ros_envelope, decode_xr_control, or subscribe_xr_control from adamo.xr instead of unpacking these fields manually. The SDK validates name lengths, UTF-8, CDR alignment and endianness, string bounds, sequence bounds, and ROS timestamps. The wire layout remains documented here for non-Python SDKs and protocol debugging.

Browser gamepad / Xbox controller

Gamepad input is published on:

adamo/{org}/{robot}/control/joy

By default the payload is a ROS envelope with:

Inner field	Value
`topic`	`/joy`
`type`	`sensor_msgs/msg/Joy`
`payload`	CDR `sensor_msgs/msg/Joy`

The Joy message uses header.frame_id = "joy" and contains six axes plus 21 button slots. Standard W3C / Xbox controllers populate buttons 0..16; buttons 17..20 are reserved and normally remain 0.

Axis	Meaning	Range
`axes[0]`	Left stick X	`-1` left, `+1` right
`axes[1]`	Left stick Y	`-1` up, `+1` down
`axes[2]`	Right stick X	`-1` left, `+1` right
`axes[3]`	Right stick Y	`-1` up, `+1` down
`axes[4]`	Left trigger analog	`0` released, `1` pressed
`axes[5]`	Right trigger analog	`0` released, `1` pressed

Button	Xbox / W3C control
`buttons[0]`	A
`buttons[1]`	B
`buttons[2]`	X
`buttons[3]`	Y
`buttons[4]`	LB
`buttons[5]`	RB
`buttons[6]`	LT pressed
`buttons[7]`	RT pressed
`buttons[8]`	Back / Select
`buttons[9]`	Start
`buttons[10]`	Left stick click
`buttons[11]`	Right stick click
`buttons[12]`	D-pad up
`buttons[13]`	D-pad down
`buttons[14]`	D-pad left
`buttons[15]`	D-pad right
`buttons[16]`	Guide / Xbox

If joystick serialization is switched to JSON in the UI, the same key carries:

{
  "type": "JoystickCommand",
  "sequence_id": 42,
  "stamp": 1710000000.123,
  "axes": [0, 0, 0, 0, 0, 0],
  "buttons": [0, 0, 0]
}

Logitech Extreme 3D Pro

The Logitech Extreme 3D Pro uses the same control/joy key and payload format, but the axes are mapped as a joystick:

Axis	Meaning
`axes[0]`	Stick X
`axes[1]`	Stick Y
`axes[2]`	Twist / rudder
`axes[3]`	Throttle, with forward usually negative
`axes[4]`	Unused, normally `0`
`axes[5]`	Unused, normally `0`

Buttons 0..11 are copied from the physical buttons. Buttons 12..15 are the hat switch when the browser reports it as buttons.

XR head, controllers, and hands

XR tracking is published on one Adamo key:

adamo/{org}/{robot}/control/cdr/xr_tracking

Each publish contains one ROS envelope. Consumers should decode the envelope and dispatch by the inner topic:

Inner topic	Type	Contents
`/head_pose`	`geometry_msgs/msg/PoseStamped`	Head pose
`/controller/{handedness}`	`geometry_msgs/msg/PoseStamped`	Physical controller grip pose (the wrist/hold point), or hand wrist pose when hand tracking is active without a physical controller
`/controller/{handedness}/tip`	`geometry_msgs/msg/PoseStamped`	Physical controller tip — the aim pose from WebXR `targetRaySpace`. Published alongside the grip pose above: grip is where the hand holds the controller, tip is where it points. Omitted for a frame if the runtime reports no target-ray pose.
`/controller/{handedness}/joy`	`sensor_msgs/msg/Joy`	XR controller axes and buttons
`/hand/{handedness}`	`geometry_msgs/msg/PoseArray`	25 hand joint poses when hand tracking is enabled

{handedness} is normally left or right. Pose headers use frame_id = "xr_origin". Positions are in meters in the WebXR local-floor reference space. Orientations are published as ROS quaternions {x, y, z, w}; the frontend converts from WebXR’s internal [w, x, y, z] order before encoding the ROS message.

In Python, adamo.xr decodes PoseStamped and Joy into immutable typed objects and retains the complete ROS Header. Compare XRControlSample.source_timestamp with fabric time via is_stale(...), or pass max_age_seconds to subscribe_xr_control, before using controller input to issue commands. PoseArray hand-joint payloads are not currently part of the typed helper and still require ROS deserialization or a general CDR decoder.

XR controller Joy messages use:

const rawAxisCount = xrGamepad.axes.length;
axes = [
  ...xrGamepad.axes,
  ...xrGamepad.buttons.map((button) => button.value),
];
buttons = xrGamepad.buttons.map((button) => button.pressed ? 1 : 0);

The raw axis and button order is the order reported by the WebXR runtime for that controller. The UI does not remap XR controller buttons to Xbox-style button indices.

For controllers using the WebXR xr-standard mapping, the Joy values are:

Joy value	WebXR value	Meaning
`axes[0]`	`xrGamepad.axes[0]`	Primary touchpad X, or placeholder `0`
`axes[1]`	`xrGamepad.axes[1]`	Primary touchpad Y, or placeholder `0`
`axes[2]`	`xrGamepad.axes[2]`	Primary thumbstick X
`axes[3]`	`xrGamepad.axes[3]`	Primary thumbstick Y
`buttons[0]`	`xrGamepad.buttons[0].pressed`	Primary trigger pressed
`axes[rawAxisCount + 0]`	`xrGamepad.buttons[0].value`	Primary trigger analog value
`buttons[1]`	`xrGamepad.buttons[1].pressed`	Grip / squeeze pressed
`axes[rawAxisCount + 1]`	`xrGamepad.buttons[1].value`	Grip / squeeze analog value
`buttons[2]`	`xrGamepad.buttons[2].pressed`	Primary touchpad pressed, if present
`axes[rawAxisCount + 2]`	`xrGamepad.buttons[2].value`	Primary touchpad button value
`buttons[3]`	`xrGamepad.buttons[3].pressed`	Primary thumbstick pressed, if present
`axes[rawAxisCount + 3]`	`xrGamepad.buttons[3].value`	Primary thumbstick button value
`buttons[4]`	`xrGamepad.buttons[4].pressed`	First extra button. On current Quest/Pico-style controllers this is usually X on left hand and A on right hand.
`axes[rawAxisCount + 4]`	`xrGamepad.buttons[4].value`	First extra button value
`buttons[5]`	`xrGamepad.buttons[5].pressed`	Second extra button. On current Quest/Pico-style controllers this is usually Y on left hand and B on right hand.
`axes[rawAxisCount + 5]`	`xrGamepad.buttons[5].value`	Second extra button value

rawAxisCount is commonly 4 for thumbstick controllers, so the trigger analog value is commonly axes[4], grip is axes[5], thumbstick button value is axes[7], and the first extra face button value is axes[8]. Check axes.length - buttons.length if you want to derive the offset at runtime.

Hand tracking publishes PoseArray joints in this fixed order:

wrist,
thumb-metacarpal, thumb-phalanx-proximal, thumb-phalanx-distal, thumb-tip,
index-finger-metacarpal, index-finger-phalanx-proximal, index-finger-phalanx-intermediate, index-finger-phalanx-distal, index-finger-tip,
middle-finger-metacarpal, middle-finger-phalanx-proximal, middle-finger-phalanx-intermediate, middle-finger-phalanx-distal, middle-finger-tip,
ring-finger-metacarpal, ring-finger-phalanx-proximal, ring-finger-phalanx-intermediate, ring-finger-phalanx-distal, ring-finger-tip,
pinky-finger-metacarpal, pinky-finger-phalanx-proximal, pinky-finger-phalanx-intermediate, pinky-finger-phalanx-distal, pinky-finger-tip