Custom

One file per machine. Extend Follower on the machine that moves and Leader on the one that drives, and everything else, the session, safety, recording and video sync, comes with them.

Follower

Three methods are required: what the joints are called, how to read them, and how to write them. The rest are there when your hardware needs them.

follower.py

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)

    def read_joints(self) -> Mapping[str, float]:
        return self._bus.read_all()                 # raise if the bus fails

    def write_joints(self, joints: Mapping[str, float]) -> None:
        self._bus.write_all(joints)                 # already within your limits

    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) -> None:
        self._bus = open_my_bus(self.port)          # your driver

    def disconnect(self) -> None:
        if self._bus is not None:
            self._bus.close()
            self._bus = None


arm = MyArm("abcd-efgh-ijkl", port="/dev/ttyACM0",
            safety=teleop.SafetyConfig(slew=4.0))
arm.run(hz=50)

super().__init__() goes last. It runs your descriptor() and limits(), so anything they read has to exist first.
descriptor() declares your units. Everything else, slew, limits() and every recorded row, is read in them. See Descriptor.
limits() is what keeps the arm inside its travel. Leave it out and there are no position limits at all.
connect() is where you open your bus.

Leader

Two methods, and nothing to write. Whatever read_joints() returns is where the operator is: a second arm, a VR controller, a simulator, or a policy.

leader.py

from videosdk_teleop import Leader, Units, single_group_descriptor

from joints import JOINTS          # the same module the follower imports


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

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

    def read_joints(self) -> Mapping[str, float]:
        return self._device.pose()


leader = MyController("abcd-efgh-ijkl", port="/dev/ttyACM1")

leader.start()
if not leader.claim_control(timeout_s=10):
    raise SystemExit("no control: another leader may already be connected")

leader.hold_deadman(True)

for i in leader.ticks(hz=50):
    obs = leader.observation()

Keep JOINTS and the units in one module both machines import. The leader compares the two descriptors when control is granted and declines if they disagree, so claim_control() returns False and on_control_denied gives you both hashes.

Cameras

The SDK opens cameras for you when you pass cameras=. If your driver already returns frames alongside the joints, hand them over instead.

class MyArm(Follower):
    def read_images(self):
        return self._bus.last_frames()          # {label: ndarray}

    def push_camera_configs(self):
        return [teleop.CameraConfig(handle="wrist", label="wrist"),
                teleop.CameraConfig(handle="front", label="front")]

The two go together. read_images() supplies the pixels and push_camera_configs() declares the cameras, so each frame gets a published track and the same timestamps a captured frame would. Declare nothing and your frames reach the observation and stop there.

Leave both out on hardware that does not carry its own cameras. The default opens whatever you passed in cameras=. See CameraConfig.

Rules

Four that are worth getting right before you run it on hardware.

Call super().__init__() last. It runs your descriptor() and limits(), so anything they read has to exist by then.

Let failures raise. If the bus does not answer, raise. The follower stops the arm, which is safe. Returning the last value you saw instead leaves it moving on numbers that are no longer true.

Add torque_enable() before setting on_starvation=TORQUE_OFF. Without it, construction fails rather than promising to cut power it cannot cut.

Keep your reads and writes quick. Both have to finish inside one tick, 20 ms at 50 Hz. Anything slower shows up as overruns in stats().

Run it

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

python follower.py     # on the machine with the hardware
python leader.py       # on the machine with the controller

Before hardware

Run your subclass against a simulated bus first, then real hardware with the network removed, then hardware over the network. Each step removes one source of failure, so a break is never ambiguous.

Writing a subclass is not the route to a ROS 2 or LeRobot arm. ROS 2 and LeRobot already are these subclasses.

Descriptor

Declaring joints, units and space, including machines that mix them.

Follower

Every constructor argument, method and event.

Safety

The checks a command passes before it reaches the motors.

Get started

Quickstarts

Concepts

Guides

Integrations

Follower

Leader

Cameras

Rules

Run it

Before hardware

Descriptor

Follower

Safety

​Follower

​Leader

​Cameras

​Rules

​Run it

​Before hardware

​Related

Descriptor

Follower

Safety

Follower

Leader

Cameras

Rules

Run it

Before hardware

Related