Follower

Follower is the base class for the machine that moves. Subclass it to bind your own hardware, and the SDK handles the session, safety, correlation and recording.

from videosdk_teleop import Follower, SafetyConfig, single_group_descriptor

class MyArm(Follower):
    def descriptor(self):           return single_group_descriptor(...)
    def read_joints(self):          return self.bus.read()
    def write_joints(self, joints): self.bus.write(joints)

MyArm(MEETING_ID, safety=SafetyConfig(slew=5.0)).run(hz=50)

Constructor

Follower(meeting_id=None, *, token=None, safety=None, cameras=None,
         control_hz=50, observation_hz=10, telemetry_hz=1.0,
         announce_hz=0.5, stats_hz=1.0, record=None, video_bitrate=None,
         name="robot", transport=None, cloud_recording=False,
         cloud_recording_timeout_s=90.0)

str | None

A VideoSDK room id shaped xxxx-xxxx-xxxx. None reads the environment, and creates a room if that is empty too.

str

default:"VIDEOSDK_TOKEN"

VideoSDK token. Falls back to the VIDEOSDK_TOKEN environment variable, and raises ValueError if neither is present.

SafetyConfig

default:"SafetyConfig()"

Watchdog, staleness, slew and failsafe policy, from SafetyConfig. Position limits come from limits(), not from here.

int

default:"50"

The rate run() paces at, and the rate the descriptor announces. The slew clamp is per tick, so this changes the effective speed limit.

int

default:"10"

How often the follower reads and sends. This is also the state frame publish rate and the dataset row rate.Rounded to a whole number of ticks, so you get control_hz divided by an integer. Asking for 20 Hz at control_hz=50 gives you 25.

float

default:"1.0"

How often telemetry() is polled, and cached in between. Not a publish rate: every observation sends a state frame regardless. Capped at observation_hz.

float

default:"0.5"

Descriptor re-announce, for late joiners.

float

default:"1.0"

How often stats are sent. 0 stops sending them, and stats() still works.

str

A path to attach an EpisodeRecorder immediately. It does not open an episode.

int

Target bitrate for the published camera tracks, in bits per second. Leave it out for the transport’s own default.

str

default:"robot"

Display name in the meeting.

see Cloud recording

Also upload the episodes. Both are documented in Cloud recording.

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.

A subclass must call super().__init__() last. The base constructor calls your descriptor() and limits(), so every attribute they read has to be set before it runs. Reversing the order produces an AttributeError inside the base constructor that reads like a base-class bug and is not one.

Running it

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

follower.run(hz=50)

That one call is the same as writing this out:

follower.start()
for i in follower.ticks(hz=50):
    ...                             # your work each period, if any
follower.stop()

Use run() when there is nothing to do each period. Use ticks() when there is, and call start() and stop() yourself.

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

Methods

You implement

@abstractmethod
def descriptor(self) -> RobotDescriptor: ...
@abstractmethod
def read_joints(self) -> Mapping[str, float]: ...
@abstractmethod
def write_joints(self, joints: Mapping[str, float]) -> None: ...

# optional
def limits(self) -> Mapping[str, tuple[float, float]]: ...
def telemetry(self) -> Mapping[str, Mapping[str, float]]: ...
def connect(self) -> None: ...
def disconnect(self) -> None: ...
def read_images(self) -> Mapping[str, Any]: ...
def push_camera_configs(self) -> list | None: ...
def torque_enable(self, on: bool) -> None: ...

returns RobotDescriptor

Your machine’s joint schema: what the joints are called, in what order, in what units. Called once, at construction. See RobotDescriptor.

returns Mapping[str, float]

The current position of every joint, in the descriptor’s units. Called every observation.

returns None

Send those values to the motors. They are already clamped by the time you get them, so write them as they are.

returns Mapping[str, tuple[float, float]]

Optional. How far each joint may travel, as {joint: (min, max)} in the descriptor’s units. Every command is held inside this before write_joints() is called.The default returns {}, which means no position limits at all. Set it on real hardware.

returns Mapping[str, Mapping[str, float]]

Optional. Slow diagnostics per joint, as {"load": {...}, "temp_c": {...}}. Polled at telemetry_hz and cached in between, so keep it cheap.

returns None

Optional. Open your bus here rather than in __init__.

returns None

Optional. Close it again. Always runs, even if the session never came up.

returns Mapping[str, Any]

Optional. Camera frames from the same call as read_joints(), for hardware that returns both together. The default returns {}, which means the SDK opens the cameras itself from cameras=.

returns list | None

Optional, and the pair to read_images(). Declare the cameras your own code owns, so each pushed frame gets a published track and the same timestamps a captured frame would. Return None, the default, to let the SDK open the cameras from cameras=.

returns None

Optional. Cut power to the motors, and restore it. The default does nothing, since not every machine can release its motors.Required if you set on_starvation=TORQUE_OFF. Without it the constructor raises TypeError, because the failsafe would report TORQUE_OFF while the motors stayed powered.

write_joints() must raise on failure. A lie about position is far more dangerous than an exception. The base turns a raise into a ROBOT_FAULT failsafe. read_joints() raising does the same.

You call

Provided by the base class. None of these are needed for a session to run.

Safety

returns None

Stop the arm now. It stays stopped until you call clear_estop().

returns None

Release that stop, so the arm can move again. If the motors were released before the stop landed, this puts the power back too.

returns None

Let the arm move again after a safety hold, and put the power back if the motors were released. It does not release an estop().It goes to HOLDING, not ACTIVE: this says the arm may resume, and the next command that passes the gates is what actually moves it.

Both read the joints before re-energising, so the arm holds where it has come to rest rather than travelling back to the old target.If the power cannot be restored, both leave the follower in TORQUE_OFF rather than reporting HOLDING over an arm that is still released.

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.

With auto_episode=True, the default, episodes open and close on the deadman. Squeeze to start, release to end.

Monitoring

returns list[dict]

The full health report, one entry per subject: transport, control, latency and session, plus one per joint, one per camera, and the recorder while recording.Every entry carries type, id and timestamp and is shaped like WebRTC’s getStats(), so a collector you already run can scrape it unchanged. Counters are cumulative and never reset, so take two snapshots and subtract to get a rate.

returns dict

A quick health check in one flat mapping: tick_jitter_ms, tick_overruns, state, stale_rejected, applied_seq, unresolved_observations and the transport counters.Use this for a log line and stats() when you want the breakdown.

control = next(r for r in follower.stats() if r["type"] == "teleop-control")
print(control["tickOverruns"], control["tickJitterMsCurrent"])

print(follower.metrics()["tick_overruns"])

Properties

follower.state          # SafeState: HOLDING, ACTIVE, ESTOPPED, TORQUE_OFF
follower.reason         # why it is in that state, or None
follower.lease_holder   # who holds control, or None
follower.last_applied   # AppliedAction | None, what was last written

Events

Callbacks register as decorators. Each can be registered many times, and all registrations fire in order.

@follower.on_safe_state
def _(state, reason):
    print(state.value, reason.value if reason else "")

once, at the end of start()

The session is up and the descriptor has been announced.

every observation

The follower read the joints and cameras and sent them.

every observation

The joint values that went with that observation, on the same tick.

every tick while ACTIVE

A command reached the motors. Fires at your tick rate, not once per command, since the loop re-applies the current target every tick.

once per clamped joint

A joint was held back by a limit or the speed cap, carrying a LimitEvent.

per dropped command

A command was thrown away. reason is stale or schema.

every transition

The follower changed state, for example ACTIVE back to HOLDING.

on grant

An operator took control.

on release

Control was dropped. why is released, holder_left, estop, or a string beginning transport:.

per episode

A recorded episode opened or closed. phase is start or end.

on each report from the leader

The leader’s own stats, as it measured them. Pairs with stats() to see both ends of the link.

once, on cloud ready

The cloud recorder is live. See Cloud recording.

on every change

The cloud recorder changed state, with a reason when there is one.

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

Example

A complete follower for a serial bus, driven by its own paced loop.

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

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


class MyArm(Follower):
    def __init__(self, meeting_id, port, **kw):
        self.robot_id = "my-arm"
        self.port = port
        self._bus = None
        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,
                                       capabilities=("load",))

    def limits(self) -> Mapping[str, tuple[float, float]]:
        return {"shoulder": (-90.0, 90.0), "elbow": (-120.0, 120.0),
                "wrist": (-180.0, 180.0), "gripper": (0.0, 40.0)}

    def connect(self):    self._bus = open_my_bus(self.port)
    def disconnect(self):
        if self._bus is not None:
            self._bus.close()
            self._bus = None

    def read_joints(self):          return self._bus.read_all()
    def write_joints(self, joints): self._bus.write_all(joints)
    def telemetry(self):            return {"load": self._bus.read_load()}


follower = MyArm(
    "abcd-efgh-ijkl", port="/dev/ttyACM0",
    safety=teleop.SafetyConfig(watchdog_timeout_s=0.5, max_staleness_ms=150.0,
                               slew=4.0),
    control_hz=50, observation_hz=10,
    cameras=[teleop.CameraConfig("platform-3610000.usb-usb-0:2.2:1.0-video-index0", label="wrist"),
             teleop.CameraConfig("platform-3610000.usb-usb-0:2.4:1.0-video-index0", label="front")],
    record="./sessions")

clamps = {"n": 0, "worst": 0.0, "joint": ""}

@follower.on_limited
def _(event):
    clamps["n"] += 1
    if event.excess > clamps["worst"]:
        clamps.update(worst=event.excess, joint=event.joint)

@follower.on_state
def _(state):
    if follower.timer.ticks % 100:
        return
    print(f"{state.safe_state:<9} worst clamp {clamps['joint']} "
          f"{clamps['worst']:.1f} over {clamps['n']} hits "
          f"jitter {follower.timer.jitter_ms:.1f} ms")
    clamps.update(n=0, worst=0.0, joint="")

follower.run(hz=50)

Three lines in there matter more than the rest:

super().__init__() goes last. The base constructor calls descriptor() and limits(), so every attribute they read has to be set before it runs.
limits() is what keeps the arm inside its travel. Return {} and there is no position limit at all.
write_joints() receives values that have already passed those limits and the speed cap. Write them exactly as they arrive.

Aggregate clamp events per window rather than printing each one. On a six joint arm printing every event buries the state changes that matter.

SafetyConfig

Every safety field, its default, and what changing it does.

RobotDescriptor

Declaring the joints, units and space your machine reports.

Adapters

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

Leader

The other half of a session.

Python SDK

Adapters

Classes

Constructor

Running it

Methods

You implement

You call

Safety

Recording

Monitoring

Properties

Events

Example

SafetyConfig

RobotDescriptor

Adapters

Leader

​Constructor

​Running it

​Methods

​You implement

​You call

​Safety

​Recording

​Monitoring

​Properties

​Events

​Example

​Related

SafetyConfig

RobotDescriptor

Adapters

Leader

Constructor

Running it

Methods

You implement

You call

Safety

Recording

Monitoring

Properties

Events

Example

Related