Quickstart

Two scripts, one per machine, joined by a VideoSDK room. The follower runs on the robot. The leader drives it. Already using LeRobot or ROS 2? See the integration quickstarts instead.

Install

pip install videosdk-teleop

Credentials

Both machines need a token and the same room id.

export VIDEOSDK_TOKEN="your-token-here"
export VIDEOSDK_MEETING_ID="abcd-efgh-ijkl"

Get a token from app.videosdk.live. See Installation for the full list.

Follower

Runs on the machine with the robot. Three methods: what the joints are called, how to read them, and how to write them.

follower.py

import videosdk_teleop as teleop
from videosdk_teleop import Follower, SafetyConfig, single_group_descriptor

JOINTS = ("shoulder", "elbow", "wrist", "gripper")


class MyArm(Follower):
    def __init__(self, **kw):
        self.robot_id = "my-arm"
        self.position = {j: 0.0 for j in JOINTS}
        super().__init__(**kw)              # LAST

    def descriptor(self):
        return single_group_descriptor(self.robot_id, "my-arm", JOINTS,
                                       control_hz=30)

    def limits(self):
        return {j: (-90.0, 90.0) for j in JOINTS}

    def read_joints(self):
        return dict(self.position)         # replace with your driver

    def write_joints(self, joints):
        self.position.update(joints)       # replace with your driver


follower = MyArm(
    safety=SafetyConfig(slew=2.0, watchdog_timeout_s=0.5),
    cameras=[teleop.CameraConfig("platform-3610000.usb-usb-0:2.2:1.0-video-index0", label="wrist")],
    control_hz=30,
    observation_hz=10)

@follower.on_applied
def _(action):
    print("applied", action.command_id, action.action)

follower.run(hz=30)

run() joins the room and paces the loop for you. On every pass it applies whatever command arrived, and holds position if the leader goes quiet. Ten times a second it sends the joint positions along with a camera frame. control_hz is the loop rate and observation_hz the publish rate, so an observation goes out every third tick here. slew caps how far a joint may move in one tick. See Configuration for the rest.

Swap the two dictionary lines for your own driver calls and nothing else changes. write_joints() receives values that have already passed the limits and speed cap in SafetyConfig.

Every constructor argument and callback is in the Follower reference, and the safety fields in Configuration.

Leader

Runs on the machine with the controller. Two methods, and no writes.

leader.py

from videosdk_teleop import Leader, single_group_descriptor

JOINTS = ("shoulder", "elbow", "wrist", "gripper")


class MyController(Leader):
    def __init__(self, **kw):
        self.robot_id = "my-controller"
        self.target = {j: 0.0 for j in JOINTS}
        super().__init__(**kw)             # LAST

    def descriptor(self):
        return single_group_descriptor(self.robot_id, "my-arm", JOINTS,
                                       control_hz=30)

    def read_joints(self):
        return dict(self.target)           # replace with your input device


leader = MyController(sync_buffer_ms=120.0)

leader.start()
if not leader.claim_control(timeout_s=10):
    raise SystemExit("no control: denied, or the follower never answered")

leader.hold_deadman(True)

for i in leader.ticks(hz=30):
    obs = leader.observation()             # joints and camera, one instant
    if i % 30 == 0:
        print(obs.joints, list(obs.images))

    leader.target["elbow"] = 20.0          # your input device goes here

ticks() paces the loop and sends a command on every pass, using whatever read_joints() returns. observation() gives you the latest from the follower: a camera frame, and the joint positions from the moment that frame was taken.

The follower applies nothing until claim_control() succeeds and the deadman is held. On real hardware, set max_misalignment so the arm refuses to arm while the two poses disagree.

Control, arming and the fields on a synced observation are in the Leader reference.

Run it

Two terminals, both with the token and the room id exported.

python follower.py     # on the robot machine
python leader.py       # on the controller machine

Output

state: holding (starvation)
follower ready: my-arm, joints ['shoulder', 'elbow', 'wrist', 'gripper']
control granted to 42mv3omt
state: active
applied 40 <- observation 302 {'shoulder': 0.0, 'elbow': 11.8716, 'wrist': 0.0, 'gripper': 0.0}
applied 76 <- observation 309 {'shoulder': 0.0, 'elbow': 18.8893, 'wrist': 0.0, 'gripper': 0.0}
applied 120 <- observation 317 {'shoulder': 0.0, 'elbow': 18.2238, 'wrist': 0.0, 'gripper': 0.0}
applied 155 <- observation 324 {'shoulder': 0.0, 'elbow': 10.6515, 'wrist': 0.0, 'gripper': 0.0}
applied 195 <- observation 332 {'shoulder': 0.0, 'elbow': -2.16, 'wrist': 0.0, 'gripper': 0.0}

robot online: my-controller, joints ['shoulder', 'elbow', 'wrist', 'gripper']
control granted
tick    0  joints={}  cameras=[]  sent=1
tick   30  joints={'shoulder': 0.0, 'elbow': 7.8358, 'wrist': 0.0, 'gripper': 0.0}  sent=31
tick   60  joints={'shoulder': 0.0, 'elbow': 15.6952, 'wrist': 0.0, 'gripper': 0.0}  sent=61
tick   90  joints={'shoulder': 0.0, 'elbow': 19.7172, 'wrist': 0.0, 'gripper': 0.0}  sent=91
tick  120  joints={'shoulder': 0.0, 'elbow': 18.4872, 'wrist': 0.0, 'gripper': 0.0}  sent=121

holding (starvation) is the starting state. The follower applies nothing until the leader claims control.
applied 40 <- observation 302 pairs each command with the observation the operator was acting on. That pairing is what a recording keeps.
joints={} on the first tick means nothing has arrived from the follower yet. It fills in on the next observation.
cameras=[] because this run had no camera. With one configured you see its label here, and obs.images carries a frame for each.

Seeing holding (starvation) again mid-run is the watchdog. Commands stopped arriving for longer than watchdog_timeout_s, so the arm held position until they resumed. Stop the leader entirely and it stays there.

New to the terms used here? Concepts explains how a session works, and the Python SDK reference lists every class and argument.

Record a dataset

Add a path and the session is written to disk as you drive.

follower = MyArm(record="./sessions", ...)   # starts recording automatically

follower.start_episode("pick up the cube")
# ... drive ...
follower.end_episode(success=True)

record= attaches a recorder when the session starts, and stop() closes it. To control that yourself, leave it out and call start_recording() and stop_recording() instead. See Recording for the format and the LeRobot export.

Integration quickstarts

If your robot is already on one of these, start there instead.

LeRobot

Pass your existing LeRobot robot and teleoperator to the adapters. They work out the joints and units for you, so there is no class to write.

ROS 2

Run a bridge node on each host, beside the drivers you already have. Your drivers keep the hardware.

Next steps

Follower

Every constructor argument, callback and method on the robot side.

Leader

Control, arming, and the synced observation.

Safety

The checks a command passes before it reaches the motors.

Custom robot

The full version of what this page showed, including cameras and telemetry.

Get started

Quickstarts

Concepts

Guides

Install

Credentials

Follower

Leader

Run it

Output

Record a dataset

Integration quickstarts

LeRobot

ROS 2

Next steps

Follower

Leader

Safety

Custom robot

​Install

​Credentials

​Follower

​Leader

​Run it

​Output

​Record a dataset

​Integration quickstarts

LeRobot

ROS 2

​Next steps

Follower

Leader

Safety

Custom robot

Install

Credentials

Follower

Leader

Run it

Output

Record a dataset

Integration quickstarts

Next steps