Custom - Torq

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.

If a read or a write fails, 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.

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 two descriptors are compared when the leader connects, and a follower and leader that disagree cannot run together.

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.

Writing your own

The same thing as reference, with every method and rule.

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.

​Follower

​Leader

​Run it

​Before hardware

​Related

Writing your own

Descriptor

Follower

Safety

Follower

Leader

Run it

Before hardware

Related