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

# Leader

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

`Leader` is the base class for the machine that drives. Subclass it to read
your own controller, and the SDK handles the session, the lease, video sync and
recording.

```python theme={null}
from zeroruntime.teleops import Leader

class MyController(Leader):
    def descriptor(self):  return single_group_descriptor(...)
    def read_joints(self): return self.device.pose()
```

## Constructor

```python theme={null}
Leader(meeting_id=None, *, token=None, publish_hz=50, heartbeat_hz=10,
       sync_buffer_ms=120.0, sync_video=True, max_misalignment=None,
       stats_hz=1.0, name="operator", transport=None)
```

<ParamField path="meeting_id" type="str | None">
  The room the follower is in. Both ends must join the same room.
</ParamField>

<ParamField path="token" type="str" default="ZERORUNTIME_AUTH_TOKEN">
  Access token.
</ParamField>

<ParamField path="publish_hz" type="int" default="50">
  The rate `run()` paces at. Heartbeats flow at this rate whether or not the
  deadman is held.
</ParamField>

<ParamField path="heartbeat_hz" type="int" default="10">
  Heartbeat rate. Must be comfortably faster than the follower's
  `watchdog_timeout_s`. Heartbeats flow whether or not the deadman is held, so
  the follower can tell "operator let go" from "link died".
</ParamField>

<ParamField path="sync_buffer_ms" type="float" default="120.0">
  How long each video frame is held so state samples can arrive either side of
  it. `0` means align on arrival, never "skip alignment". See
  [Sync](/robotics/concepts/sync#buffer).
</ParamField>

<ParamField path="sync_video" type="bool" default="True">
  Receive the follower's video. `False` sends and receives joints only, for a
  controller that never shows a frame, and saves the cost of decoding video.
  [`observation()`](#observing) still pairs joints with frames whenever there
  are frames, and returns `images == {}` when there are none.
</ParamField>

<ParamField path="max_misalignment" type="float | None" default="None">
  Refuse to arm while any joint differs from the follower by more than this, in
  descriptor units. Set it on real hardware. See
  [`arm_when_aligned()`](#starting-to-drive).
</ParamField>

<ParamField path="stats_hz" type="float" default="1.0">
  How often this leader publishes its own stats. `0` stops publishing them, and
  `get_stats()` still works. Mirrors the follower's `stats_hz`.
</ParamField>

<ParamField path="name" type="str" default="operator">
  Display name in the meeting.
</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>
  Subclasses call `super().__init__()` **last**, for the same reason the
  follower does: the base constructor calls your `descriptor()`.
</Warning>

## Running it

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

```python theme={null}
leader.start()
for i in leader.ticks(hz=30):
    obs = leader.observation()      # your work each period
leader.stop()
```

`leader.run(hz=30)` is the same three steps in one call, for when there is
nothing to read each period. On the leader there usually is, so `ticks()` is
the one you want.

<Note>
  Already have a scheduler, such as a ROS 2 timer? Call `leader.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]: ...   # the operator's arm
```

<ParamField path="descriptor()" type="returns RobotDescriptor">
  The joint schema, matching the follower's. Called **once**, at construction.
  See [`RobotDescriptor`](/robotics/api-reference/python/robot-descriptor).
</ParamField>

<ParamField path="read_joints()" type="returns Mapping[str, float]">
  Where the operator's controller is right now. Called every tick.
</ParamField>

`read_joints()` is whatever the controller is: a backdrivable arm, a VR headset
pose, a simulator, or a policy's output.

### You call

Provided by the base class.

#### Taking control

<ParamField path="claim_control(timeout_s=10.0)" type="returns bool">
  Ask the follower for control and wait for its answer. `True` means you have
  it, `False` means you were refused or nobody answered.
</ParamField>

<ParamField path="release_control()" type="returns None">
  Give control back. The follower stops applying your commands.
</ParamField>

<ParamField path="hold_deadman(held)" type="returns None">
  `True` starts sending commands, `False` stops. Releasing is never refused.

  With `max_misalignment` set, `hold_deadman(True)` raises `NotAligned` when the
  worst joint is further out than that, or when no follower pose has arrived
  yet. The check lives here as well as in `arm_when_aligned()`.
</ParamField>

<ParamField path="estop()" type="returns None">
  Stop the follower now, on a channel of its own so it cannot queue behind
  video.
</ParamField>

```python theme={null}
if not leader.claim_control(timeout_s=10):
    raise SystemExit("no control: refused, or the follower never answered")
```

<Warning>
  Control is first come and unauthenticated, with no handover queue. Anyone in
  the room can take it.
</Warning>

#### Starting to drive

<ParamField path="arm_when_aligned(hold_s=2.0, poll=0.1, timeout=None, on_status=None, tick=None)" type="returns None">
  Wait until the two arms match, hold still for `hold_s`, then start sending.
  Holding the pose is the trigger, since a leader arm falls the moment the
  operator lets go to press a key.

  Every argument is keyword only. `on_status` runs each poll if you want to show
  progress, and `timeout` raises `NotAligned` instead of waiting forever.
</ParamField>

<ParamField path="alignment()" type="returns dict[str, float] | None">
  How far each joint is from the follower's, leader minus follower, in
  descriptor units. `None` until the follower's pose is known.
</ParamField>

<Note>
  These compare **positions**, so they fit an arm driven by another arm, a VR
  hand or a wearable controller. For a base or a drone, which report speed
  rather than position, leave `max_misalignment` unset and call
  `hold_deadman(True)` directly.
</Note>

Both raise `NotAligned`, so catch it if you want to tell the operator rather
than exit:

```python theme={null}
from zeroruntime.teleops import NotAligned

try:
    leader.arm_when_aligned(hold_s=2.0, timeout=30.0)
except NotAligned as e:
    print(e)                       # names the joint and how far out it is
```

#### Observing

<ParamField path="observation()" type="returns SyncedObservation">
  The latest frame from the follower with the joint positions from the instant
  that frame was taken. Until video is up you get joints and
  `images == {}`, which is how you know it is not ready yet. See
  [Observation](/robotics/api-reference/python/observation).
</ParamField>

#### 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>
  Episodes do not open on their own here, so call `start_episode()` and
  `end_episode()` yourself. The follower does that from the deadman.
</Note>

<Warning>
  A leader recording is indexed by video frame, not by observation, so it is a
  different sampling of the session from the follower's and must not be joined
  to it row for row. See [Recording](/robotics/concepts/recording#how-to-record).
</Warning>

#### Monitoring

<ParamField path="get_stats()" type="returns list[dict]">
  This leader's own health report: transport, latency and session. Same shape as
  the follower's, so a collector scrapes both unchanged.

  Named `get_stats()` here and `stats()` on the
  [Follower](/robotics/api-reference/python/follower#monitoring). Two spellings, one
  contract. `leader.stats()` does not exist.
</ParamField>

<Note>
  Tracking error, clamps and loop jitter are not in here. They belong to the
  machine with the motors and arrive on `remote_stats` from its own report,
  rather than being guessed at from this end.
</Note>

## Properties

```python theme={null}
leader.has_control        # True while you hold control
leader.lease              # str | None, the lease you hold
leader.commands_sent      # int
leader.observations_seen  # int
```

What the follower reports about itself:

| Property | What it is |
| - | - |
| `robot_safe_state` | the arm's own word for its state, as a `str`. Empty before its first state frame. This is the one for an operator screen |
| `robot_lease_holder` | who the arm believes holds the lease. Compare with `lease` to catch a silent handover |
| `remote_stats` | the arm's newest stats report, or `[]` if it has sent none |
| `remote_stats_age_ms` | how old that report is, or `None`. A number that keeps climbing is the arm going quiet |
| `cameras` | the camera labels the follower announced |
| `camera_sizes` | `{label: (height, width)}` for each of them |
| `schema` | the schema hash the session is running on |

Acknowledgement and sync quality:

| Property | What it is |
| - | - |
| `seq` | the sequence number of the last command you sent |
| `applied_seq` | the last one the arm reported applying, or `-1` before its first state frame |
| `sync_quality` | live counters for the current episode: how many frames were bracketed by real samples, and how far the rest were interpolated |

`seq - applied_seq` is how many commands are in flight or were dropped. Both
numbers come from this side, so the gap needs no clock sync.

<Note>
  `cameras` stays empty on a follower that has no cameras, so it is a "the far
  machine spoke" signal only when there are cameras to announce.
  `on_observation` is the unconditional one.
</Note>

## Events

<ParamField path="on_ready(descriptor)" type="once, at the end of start()">
  The transport is up. `descriptor` is this leader's own, not the follower's,
  so it says nothing about the far machine yet.

  For that, watch `on_observation` for the first state frame, or read
  [`cameras`](#properties), which stays empty until the follower's first
  announce arrives.
</ParamField>

<ParamField path="on_control_granted(lease)" type="on grant">
  You have control.
</ParamField>

<ParamField path="on_stats(reports)" type="on each report from the follower">
  The arm published its stats. The same list `remote_stats` holds.
</ParamField>

<ParamField path="on_synced_frame(frame)" type="per frame joined to state">
  A video frame was matched to the joint values from its own capture instant.

  This one runs on the media thread, so stash the frame and return. Decoding or
  writing to disk here holds up all video. Unlike `observation()`, it does not
  stamp the next command.
</ParamField>

<ParamField path="on_control_denied(reason)" type="on refusal">
  Your claim was refused, or the joints did not match. It does **not** fire when
  `claim_control()` simply times out, so a `False` return covers both refused
  and never answered.
</ParamField>

<ParamField path="on_observation(obs, staleness_ms)" type="per new observation">
  A newer observation arrived from the follower.
</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 controller on a generic device, driving one follower.

```python highlight={12-13,24,31,36} theme={null}
from typing import Mapping
import zeroruntime.teleops as teleop
from zeroruntime.teleops import Leader, Units, single_group_descriptor

JOINTS = ("shoulder", "elbow", "wrist", "gripper")


class MyController(Leader):
    def __init__(self, meeting_id, port, **kw):
        self.robot_id = "my-controller"
        self.port = port
        self._device = open_my_device(port)      # no connect() hook here
        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._device.pose()               # raises if the device fails


leader = MyController("abcd-efgh-ijkl", port="/dev/ttyACM1",
                      sync_buffer_ms=120.0, max_misalignment=20.0)

@leader.on_control_denied
def _(reason):
    print("denied:", reason)

leader.start()
if not leader.claim_control(timeout_s=10):
    leader.stop()
    raise SystemExit("no control: refused, or the follower never answered")

print("Hold the controller matching the follower to arm.")
leader.arm_when_aligned(hold_s=2.0)

try:
    for i in leader.ticks(hz=30):
        obs = leader.observation()
        if i % 60 == 0:
            print(f"sent={leader.commands_sent} "
                  f"acting_on={obs.observation_id} "
                  f"interp={obs.interpolated_ms:.1f}ms")
        render(obs.images.get("wrist"))
except KeyboardInterrupt:
    leader.hold_deadman(False)
finally:
    leader.stop()
```

Four lines in there matter more than the rest:

* **Open your device before `super().__init__()`.** The leader has no
  `connect()` hook, so this is the only place to do it.
* **`max_misalignment` stops the follower moving suddenly** to meet a
  controller that is somewhere else.
* **Check what `claim_control()` returns.** `False` means you do not have
  control, and commands go nowhere.
* **`arm_when_aligned()` is what starts the commands.** Nothing moves before
  it.

## Related

<CardGroup cols={2}>
  <Card title="Follower" icon="robot" href="/robotics/api-reference/python/follower">
    The other half of a session.
  </Card>

  <Card title="Observation" icon="eye" href="/robotics/api-reference/python/observation">
    What observation() hands back, field by field.
  </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="Sync" icon="film" href="/robotics/concepts/sync">
    How joints and frames are matched to the same instant.
  </Card>
</CardGroup>


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