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

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.

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.

str

default:"robot"

Display name in the meeting.

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: ...

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.

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.

returns None

Let the arm move again after a safety hold. It does not release an estop().

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.

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.

Get started

Classes

Adapters

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