> ## Documentation Index
> Fetch the complete documentation index at: https://docs.zeroruntime.ai/llms.txt
> Use this file to discover all available pages before exploring further.
# Custom
> Connect your own hardware by extending Follower and Leader.
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.
```python follower.py highlight={12,15-16,26,29} theme={null}
from typing import Mapping
import zeroruntime.teleops as teleop
from zeroruntime.teleops 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](/robotics/api-reference/python/robot-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.
```python leader.py theme={null}
from zeroruntime.teleops 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.
```python theme={null}
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](/robotics/api-reference/python/camera-config).
## 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
```bash theme={null}
export ZERORUNTIME_AUTH_TOKEN="your-token-here"
export ZERORUNTIME_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](/robotics/integrations/ros2) and [LeRobot](/robotics/integrations/lerobot) already are
these subclasses.
## Related
Declaring joints, units and space, including machines that mix them.
Every constructor argument, method and event.
The checks a command passes before it reaches the motors.
This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.