Leader

Leader is the base class for the machine that drives. Subclass it to read your own controller, and the SDK handles the session, the lease, video sync and recording.

from videosdk_teleop import Leader

class MyController(Leader):
    def descriptor(self):  return single_group_descriptor(...)
    def read_joints(self): return self.device.pose()

Constructor

Leader(meeting_id=None, *, token=None, publish_hz=50, heartbeat_hz=10,
       sync_buffer_ms=120.0, sync_video=True, max_misalignment=None,
       stats_hz=1.0, name="operator", transport=None)

str | None

The room the follower is in. Both ends must join the same room.

str

default:"VIDEOSDK_TOKEN"

VideoSDK token.

int

default:"50"

The rate run() paces at. Heartbeats flow at this rate whether or not the deadman is held.

int

default:"10"

Heartbeat rate. Must be comfortably faster than the follower’s watchdog_timeout_s. Heartbeats flow whether or not the deadman is held, so the follower can tell “operator let go” from “link died”.

float

default:"120.0"

How long each video frame is held so state samples can arrive either side of it. 0 means align on arrival, never “skip alignment”. See Sync.

bool

default:"True"

Receive the follower’s video. False sends and receives joints only, for a controller that never shows a frame, and saves the cost of decoding video. observation() still pairs joints with frames whenever there are frames, and returns images == {} when there are none.

float | None

default:"None"

Refuse to arm while any joint differs from the follower by more than this, in descriptor units. Set it on real hardware. See arm_when_aligned().

float

default:"1.0"

How often this leader publishes its own stats. 0 stops publishing them, and get_stats() still works. Mirrors the follower’s stats_hz.

str

default:"operator"

Display name in the meeting.

object

Use a transport of your own instead of joining a room. This is how the loopback harness runs a whole session with no network.

Subclasses call super().__init__() last, for the same reason the follower does: the base constructor calls your descriptor().

Running it

Call one of these from your own program. Both pace the loop for you.

leader.start()
for i in leader.ticks(hz=30):
    obs = leader.observation()      # your work each period
leader.stop()

leader.run(hz=30) is the same three steps in one call, for when there is nothing to read each period. On the leader there usually is, so ticks() is the one you want.

Already have a scheduler, such as a ROS 2 timer? Call leader.tick() from it instead and skip both.

Methods

You implement

@abstractmethod
def descriptor(self) -> RobotDescriptor: ...
@abstractmethod
def read_joints(self) -> Mapping[str, float]: ...   # the operator's arm

returns RobotDescriptor

The joint schema, matching the follower’s. Called once, at construction. See RobotDescriptor.

returns Mapping[str, float]

Where the operator’s controller is right now. Called every tick.

read_joints() is whatever the controller is: a backdrivable arm, a VR headset pose, a simulator, or a policy’s output.

You call

Provided by the base class.

Taking control

returns bool

Ask the follower for control and wait for its answer. True means you have it, False means you were refused or nobody answered.

returns None

Give control back. The follower stops applying your commands.

returns None

True starts sending commands, False stops. Releasing is never refused.With max_misalignment set, hold_deadman(True) raises NotAligned when the worst joint is further out than that, or when no follower pose has arrived yet. The check lives here as well as in arm_when_aligned().

returns None

Stop the follower now, on a channel of its own so it cannot queue behind video.

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

Control is first come and unauthenticated, with no handover queue. Anyone in the room can take it.

Starting to drive

returns None

Wait until the two arms match, hold still for hold_s, then start sending. Holding the pose is the trigger, since a leader arm falls the moment the operator lets go to press a key.Every argument is keyword only. on_status runs each poll if you want to show progress, and timeout raises NotAligned instead of waiting forever.

returns dict[str, float] | None

How far each joint is from the follower’s, leader minus follower, in descriptor units. None until the follower’s pose is known.

These compare positions, so they fit an arm driven by another arm, a VR hand or a wearable controller. For a base or a drone, which report speed rather than position, leave max_misalignment unset and call hold_deadman(True) directly.

Both raise NotAligned, so catch it if you want to tell the operator rather than exit:

from videosdk_teleop import NotAligned

try:
    leader.arm_when_aligned(hold_s=2.0, timeout=30.0)
except NotAligned as e:
    print(e)                       # names the joint and how far out it is

Observing

returns SyncedObservation

The latest frame from the follower with the joint positions from the instant that frame was taken. Until video is up you get joints and images == {}, which is how you know it is not ready yet. See Observation.

Recording

returns EpisodeRecorder

Attach a recorder. Does not open an episode.

returns str | None

Begin one recorded demonstration, ending any open episode first. None if there is no recorder.

returns None

Close the episode. success is True, False, or None for “not judged”.

returns None

Close the recorder and detach it.

Episodes do not open on their own here, so call start_episode() and end_episode() yourself. The follower does that from the deadman.

A leader recording is indexed by video frame, not by observation, so it is a different sampling of the session from the follower’s and must not be joined to it row for row. See Recording.

Monitoring

returns list[dict]

This leader’s own health report: transport, latency and session. Same shape as the follower’s, so a collector scrapes both unchanged.Named get_stats() here and stats() on the Follower. Two spellings, one contract. leader.stats() does not exist.

Tracking error, clamps and loop jitter are not in here. They belong to the machine with the motors and arrive on remote_stats from its own report, rather than being guessed at from this end.

Properties

leader.has_control        # True while you hold control
leader.lease              # str | None, the lease you hold
leader.commands_sent      # int
leader.observations_seen  # int

What the follower reports about itself:

Property	What it is
`robot_safe_state`	the arm’s own word for its state, as a `str`. Empty before its first state frame. This is the one for an operator screen
`robot_lease_holder`	who the arm believes holds the lease. Compare with `lease` to catch a silent handover
`remote_stats`	the arm’s newest stats report, or `[]` if it has sent none
`remote_stats_age_ms`	how old that report is, or `None`. A number that keeps climbing is the arm going quiet
`cameras`	the camera labels the follower announced
`camera_sizes`	`{label: (height, width)}` for each of them
`schema`	the schema hash the session is running on

Acknowledgement and sync quality:

Property	What it is
`seq`	the sequence number of the last command you sent
`applied_seq`	the last one the arm reported applying, or `-1` before its first state frame
`sync_quality`	live counters for the current episode: how many frames were bracketed by real samples, and how far the rest were interpolated

seq - applied_seq is how many commands are in flight or were dropped. Both numbers come from this side, so the gap needs no clock sync.

cameras stays empty on a follower that has no cameras, so it is a “the far machine spoke” signal only when there are cameras to announce. on_observation is the unconditional one.

Events

once, at the end of start()

The transport is up. descriptor is this leader’s own, not the follower’s, so it says nothing about the far machine yet.For that, watch on_observation for the first state frame, or read cameras, which stays empty until the follower’s first announce arrives.

on grant

You have control.

on each report from the follower

The arm published its stats. The same list remote_stats holds.

per frame joined to state

A video frame was matched to the joint values from its own capture instant.This one runs on the media thread, so stash the frame and return. Decoding or writing to disk here holds up all video. Unlike observation(), it does not stamp the next command.

on refusal

Your claim was refused, or the joints did not match. It does not fire when claim_control() simply times out, so a False return covers both refused and never answered.

per new observation

A newer observation arrived from the follower.

Keep callbacks short. Anything slow in one holds up the session, so hand the work off rather than doing it here.

Example

A controller on a generic device, driving one follower.

from typing import Mapping
import videosdk_teleop as teleop
from videosdk_teleop import Leader, Units, single_group_descriptor

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


class MyController(Leader):
    def __init__(self, meeting_id, port, **kw):
        self.robot_id = "my-controller"
        self.port = port
        self._device = open_my_device(port)      # no connect() hook here
        super().__init__(meeting_id, **kw)       # LAST

    def descriptor(self):
        return single_group_descriptor(self.robot_id, "my-arm", JOINTS,
                                       control_hz=50, units=Units.DEG)

    def read_joints(self) -> Mapping[str, float]:
        return self._device.pose()               # raises if the device fails


leader = MyController("abcd-efgh-ijkl", port="/dev/ttyACM1",
                      sync_buffer_ms=120.0, max_misalignment=20.0)

@leader.on_control_denied
def _(reason):
    print("denied:", reason)

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

print("Hold the controller matching the follower to arm.")
leader.arm_when_aligned(hold_s=2.0)

try:
    for i in leader.ticks(hz=30):
        obs = leader.observation()
        if i % 60 == 0:
            print(f"sent={leader.commands_sent} "
                  f"acting_on={obs.observation_id} "
                  f"interp={obs.interpolated_ms:.1f}ms")
        render(obs.images.get("wrist"))
except KeyboardInterrupt:
    leader.hold_deadman(False)
finally:
    leader.stop()

Four lines in there matter more than the rest:

Open your device before super().__init__(). The leader has no connect() hook, so this is the only place to do it.
max_misalignment stops the follower moving suddenly to meet a controller that is somewhere else.
Check what claim_control() returns. False means you do not have control, and commands go nowhere.
arm_when_aligned() is what starts the commands. Nothing moves before it.

Follower

The other half of a session.

Observation

What observation() hands back, field by field.

Adapters

The LeRobot and ROS 2 subclasses that ship with the SDK.

Sync

How joints and frames are matched to the same instant.

Python SDK

Adapters

Classes

Constructor

Running it

Methods

You implement

You call

Taking control

Starting to drive

Observing

Recording

Monitoring

Properties

Events

Example

Follower

Observation

Adapters

Sync

​Constructor

​Running it

​Methods

​You implement

​You call

​Taking control

​Starting to drive

​Observing

​Recording

​Monitoring

​Properties

​Events

​Example

​Related

Follower

Observation

Adapters

Sync

Constructor

Running it

Methods

You implement

You call

Taking control

Starting to drive

Observing

Recording

Monitoring

Properties

Events

Example

Related