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

# Follower

> The class you extend to drive a machine: constructor, properties, methods, events and a worked example.

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

```python theme={null}
from zeroruntime.teleops 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

```python theme={null}
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)
```

<ParamField path="meeting_id" type="str | None">
  A room id shaped `xxxx-xxxx-xxxx`. `None` reads the environment, and
  creates a room if that is empty too.
</ParamField>

<ParamField path="token" type="str" default="ZERORUNTIME_AUTH_TOKEN">
  Access token. Falls back to the `ZERORUNTIME_AUTH_TOKEN` environment variable, and
  raises `ValueError` if neither is present.
</ParamField>

<ParamField path="safety" type="SafetyConfig" default="SafetyConfig()">
  Watchdog, staleness, slew and failsafe policy, from
  [`SafetyConfig`](/robotics/api-reference/python/safety-config). Position
  limits come from `limits()`, not from here.
</ParamField>

<ParamField path="cameras" type="list[CameraConfig] | None">
  Which cameras to stream. `None`, the default, streams every USB camera (every
  camera on macOS, with `pip install opencv-python`), or the ones
  `ZERORUNTIME_CAMERAS` names. `[]` streams none. A list of
  [`CameraConfig`](/robotics/api-reference/python/camera-config) streams
  exactly those, on Linux only: on macOS and Windows each one is a test pattern.
</ParamField>

<ParamField path="control_hz" type="int" default="50">
  The rate [`run()`](#running-it) paces at, and
  the rate the descriptor announces. The slew clamp is per tick, so this changes
  the effective speed limit.
</ParamField>

<ParamField path="observation_hz" type="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.
</ParamField>

<ParamField path="telemetry_hz" type="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`.
</ParamField>

<ParamField path="announce_hz" type="float" default="0.5">
  Descriptor re-announce, for late joiners.
</ParamField>

<ParamField path="stats_hz" type="float" default="1.0">
  How often stats are sent. `0` stops sending them, and
  `stats()` still works.
</ParamField>

<ParamField path="record" type="str">
  A path to attach an [`EpisodeRecorder`](#recording)
  immediately. It does not open an episode.
</ParamField>

<ParamField path="video_bitrate" type="int">
  Target bitrate for the published camera tracks, in bits per second. Leave it
  out for the transport's own default.
</ParamField>

<ParamField path="name" type="str" default="robot">
  Display name in the meeting.
</ParamField>

<ParamField path="cloud_recording, cloud_recording_timeout_s" type="see Cloud recording">
  Also upload the episodes. Both are documented in
  [Cloud recording](/robotics/guides/cloud-recording).
</ParamField>

<ParamField path="transport" type="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.
</ParamField>

<Warning>
  **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.
</Warning>

## Running it

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

```python theme={null}
follower.run(hz=50)
```

That one call is the same as writing this out:

```python theme={null}
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.

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

## Methods

### You implement

```python theme={null}
@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: ...
```

<ParamField path="descriptor()" type="returns RobotDescriptor">
  Your machine's joint schema: what the joints are called, in what order, in
  what units. Called **once**, at construction. See
  [`RobotDescriptor`](/robotics/api-reference/python/robot-descriptor).
</ParamField>

<ParamField path="read_joints()" type="returns Mapping[str, float]">
  The current position of every joint, in the descriptor's units. Called every
  observation.
</ParamField>

<ParamField path="write_joints(joints)" type="returns None">
  Send those values to the motors. They are **already clamped** by the time you
  get them, so write them as they are.
</ParamField>

<ParamField path="limits()" type="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.
</ParamField>

<ParamField path="telemetry()" type="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.
</ParamField>

<ParamField path="connect()" type="returns None">
  Optional. Open your bus here rather than in `__init__`.
</ParamField>

<ParamField path="disconnect()" type="returns None">
  Optional. Close it again. Always runs, even if the session never came up.
</ParamField>

<ParamField path="read_images()" type="returns Mapping[str, Any]">
  Optional. Camera frames from the same call as `read_joints()`, for hardware
  that returns both together. The default returns `{}`, which leaves the
  cameras to the SDK (see `cameras` above).
</ParamField>

<ParamField path="push_camera_configs()" type="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 leave the cameras to the
  SDK. On Windows this pair is the only way to send video.
</ParamField>

<ParamField path="torque_enable(on)" type="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.
</ParamField>

<Warning>
  **`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.
</Warning>

### You call

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

#### Safety

<ParamField path="estop()" type="returns None">
  Stop the arm now. It stays stopped until you call `clear_estop()`.
</ParamField>

<ParamField path="clear_estop()" type="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.
</ParamField>

<ParamField path="arm()" type="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.
</ParamField>

<Note>
  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.
</Note>

#### Recording

<ParamField path="start_recording(root, **kw)" type="returns EpisodeRecorder">
  Attach a recorder. Does not open an episode.
</ParamField>

<ParamField path="start_episode(task)" type="returns str | None">
  Begin one recorded demonstration, ending any open episode first. `None` if
  there is no recorder.
</ParamField>

<ParamField path="end_episode(success=None)" type="returns None">
  Close the episode. `success` is `True`, `False`, or `None` for "not judged".
</ParamField>

<ParamField path="stop_recording()" type="returns None">
  Close the recorder and detach it.
</ParamField>

<Note>
  With `auto_episode=True`, the default, episodes open and close on the deadman.
  Squeeze to start, release to end.
</Note>

#### Monitoring

<ParamField path="stats()" type="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.
</ParamField>

<ParamField path="metrics()" type="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.
</ParamField>

```python theme={null}
control = next(r for r in follower.stats() if r["type"] == "teleop-control")
print(control["tickOverruns"], control["tickJitterMsCurrent"])

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

## Properties

```python theme={null}
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.

```python theme={null}
@follower.on_safe_state
def _(state, reason):
    print(state.value, reason.value if reason else "")
```

<ParamField path="on_ready(descriptor)" type="once, at the end of start()">
  The session is up and the descriptor has been announced.
</ParamField>

<ParamField path="on_observation(obs, state)" type="every observation">
  The follower read the joints and cameras and sent them.
</ParamField>

<ParamField path="on_state(state)" type="every observation">
  The joint values that went with that observation, on the same tick.
</ParamField>

<ParamField path="on_applied(action)" type="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.
</ParamField>

<ParamField path="on_limited(event)" type="once per clamped joint">
  A joint was held back by a limit or the speed cap, carrying a
  [`LimitEvent`](/robotics/api-reference/python/action#limitevent).
</ParamField>

<ParamField path="on_rejected(seq, reason)" type="per dropped command">
  A command was thrown away. `reason` is `stale` or `schema`.
</ParamField>

<ParamField path="on_safe_state(state, reason)" type="every transition">
  The follower changed state, for example `ACTIVE` back to `HOLDING`.
</ParamField>

<ParamField path="on_control_granted(lease, holder)" type="on grant">
  An operator took control.
</ParamField>

<ParamField path="on_control_released(lease, why)" type="on release">
  Control was dropped. `why` is `released`, `holder_left`, `estop`, or a string
  beginning `transport:`.
</ParamField>

<ParamField path="on_episode(phase, episode_id, extra)" type="per episode">
  A recorded episode opened or closed. `phase` is `start` or `end`.
</ParamField>

<ParamField path="on_stats(remote_stats)" type="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.
</ParamField>

<ParamField path="on_recording_start(recording_id)" type="once, on cloud ready">
  The cloud recorder is live. See
  [Cloud recording](/robotics/guides/cloud-recording).
</ParamField>

<ParamField path="on_recording_state(state, detail)" type="on every change">
  The cloud recorder changed state, with a reason when there is one.
</ParamField>

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

## Example

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

```python highlight={13,20-22,30-31} 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,
                                       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.

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

## Related

<CardGroup cols={2}>
  <Card title="SafetyConfig" icon="sliders" href="/robotics/api-reference/python/safety-config">
    Every safety field, its default, and what changing it does.
  </Card>

  <Card title="RobotDescriptor" icon="table-list" href="/robotics/api-reference/python/robot-descriptor">
    Declaring the joints, units and space your machine reports.
  </Card>

  <Card title="Adapters" icon="box" href="/robotics/api-reference/python/adapters/lerobot">
    The LeRobot and ROS 2 subclasses that ship with the SDK.
  </Card>

  <Card title="Leader" icon="gamepad" href="/robotics/api-reference/python/leader">
    The other half of a session.
  </Card>
</CardGroup>


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.