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,
       name="operator")

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().

str

default:"operator"

Display name in the meeting.

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.

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 bool

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.on_status runs each poll if you want to show progress, and timeout gives up 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.

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.

Properties

leader.has_control        # True while you hold control
leader.cameras            # list[str], the labels the follower announced
leader.commands_sent      # int
leader.observations_seen  # int

Events

once, on the first descriptor

The follower announced itself, so its joints are known.

on grant

You have control.

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.

Get started

Classes

Adapters

Constructor

Running it

Methods

You implement

You call

Taking control

Starting to drive

Observing

Recording

Properties

Events

Example

Follower

Observation

Adapters

Sync

​Constructor

​Running it

​Methods

​You implement

​You call

​Taking control

​Starting to drive

​Observing

​Recording

​Properties

​Events

​Example

​Related

Follower

Observation

Adapters

Sync

Constructor

Running it

Methods

You implement

You call

Taking control

Starting to drive

Observing

Recording

Properties

Events

Example

Related