Transport

Overview

What every backend provides

EVA drives every backend the same way: pull one observation, send a joint command, report link health.

• Commands are always joint angles. EEF-pose outputs are converted upstream via IK before reaching transport — see Action spaces.
• Link health. Live backends report time since last inbound message; the console flags stalled links. Dataset replay and the offline fallback do not.
• Sim vs. real target. Commands target a real robot or a simulator/3D preview. Only ros1 ships simulator publishers.
• Data collection. COLLECT records through the same transport — see Data collection.

Prerequisites

dataset — replay a recording

dataset replays one episode from a LeRobot v2.1 dataset directory (joint states in .parquet, videos in .mp4). Fully offline: no robot, no policy server. transport.dataset_keys maps the columns holding joint state, action, and each camera's video; defaults fit most recordings.

Copy from configs/00_openloop/dual_agilex_AGILEX_openloop.py:

transport = dict(
    type='dataset',
    dataset_dir='<path to your dataset folder>',   # ← set this to your recording's folder
    episode_id=0,                              # which recorded run to replay (0 = first)
    dataset_keys=dict(
        state_key='observation.qpos',         # ← change only if your data uses other column names
        action_key='action',
        video_keys=dict(
            # map each camera name to its video column:
            cam_high='observation.images.cam_high',
        ),
    ),
)

dataset_dir points at the folder containing meta/ and episode files; use an absolute path. A missing or empty folder aborts startup with an error naming dataset_dir.

eva --config configs/00_openloop/dual_agilex_AGILEX_openloop.py --web-port 8080

Blank camera panels at http://localhost:8080 usually mean video_keys names do not match the dataset's columns.

ros1 / ros2 — live ROS robot

ros1 and ros2 subscribe to camera and joint-state topics, align by timestamp, and publish joint commands. ros2 additionally decodes compressed image topics (suffix /compressed).

transport.topics is required: it maps EVA's camera and arm names to your robot's topic names. Empty raises a startup error. Set this per robot in the deploy _base.py.

Copy from configs/01_deploy/r1lite/_base.py:

transport = dict(
    type='ros2',                  # or 'ros1'
    topics=dict(
        camera_topics=dict(
            # EVA camera name : your robot's image topic
            head='<your head camera topic>',        # ← fill in
            left_wrist='<your left wrist camera topic>',  # ← fill in
        ),
        group_topics=dict(
            left_arm=dict(
                state_topic='<left arm joint-state topic>',    # ← required: where you read joint angles
                command_topic='<left arm command topic>',      # ← required: where you send joint commands
                # optional, leave out unless you need them:
                # gripper_state_topic / gripper_command_topic  (ros2 split gripper)
                # eef_state_topic     (only when the obs space is EEF)
                # sim_command_topic   (ros1 simulator only)
            ),
            # add right_arm=dict(...) the same way if your robot has one
        ),
    ),
)

Each arm requires state_topic and command_topic; other keys are optional per the comments. Add a right_arm block for a second arm.

The ROS node name defaults to eva_client; override via transport.node_name on collision.

zmq — robot behind a messaging node

zmq exchanges already-aligned observation snapshots and action messages with a separate execution-layer node over ZeroMQ.

Defaults assume the node runs on the same host on 5555/5556. Override the endpoints for a remote node. Copy from configs/01_deploy/ur5e/_base.py:

transport = dict(
    type='zmq',
    # Uncomment and edit these only if the node is not on this machine / these ports:
    # sub_endpoint='tcp://<node IP>:<obs port>',   # ← where observations arrive
    # pub_endpoint='tcp://<node IP>:<action port>', # ← where actions go
    resize_pad=False,
    image_layout='hwc',
)

Confirming you are not offline

The offline fallback is silent: no frames, all commands dropped. To detect it:

• Startup log. The fallback prints transport ros2 unavailable (...); starting offline transport. Fix the missing ROS install or topic mapping.
• Console link indicator. Offline or stalled links show as offline.
• No camera images and no joint-chart motion means no frames arriving.

Shared image settings

Image preprocessing applied by every backend before frames reach the policy. Defaults shown.

Preset inheritance: see Configuration.

How to add a new transport

A transport is a subclass of TransportBridge (src/transport/base.py) registered into TRANSPORT_REGISTRY. The package __init__ auto-imports every sibling module under src/transport/, so dropping a file there is enough to make the backend selectable.

1. Subclass TransportBridge and implement get_frame, publish_action, and close. Override the optional hooks (get_latest_qpos, seconds_since_last_recv, is_offline, the *_collection* family, create_observation_reader) only when your source supports them.
2. Register a builder with @TRANSPORT_REGISTRY.register("<your_name>"). The decorated callable receives (config, robot) and returns the transport instance — see build_transport in src/transport/base.py.
3. Save the file as src/transport/<your_name>.py. The package walker in src/transport/__init__.py imports it on launch, which triggers the registration as a side effect — no extra wiring.
4. Select it from any config: transport = dict(type='<your_name>', ...). Validate by listing the registry at startup (TRANSPORT_REGISTRY.available()) — your key must appear there.

# src/transport/mybridge.py  ← new file; auto-imported by the package
from __future__ import annotations

import numpy as np

from core.config import ConfigDict
from core.registry import TRANSPORT_REGISTRY
from core.types import Observation
from robots.base import Robot
from transport.base import TransportBridge


class MyBridge(TransportBridge):
    def __init__(self, config: ConfigDict, robot: Robot) -> None:
        self._config = config
        self._robot = robot
        # ← open sockets / file handles / etc.

    def get_frame(self) -> Observation | None:
        # ← build images dict (observation_key -> [H,W,3] uint8)
        # ← and state_qpos [state_dim] float32 in state_composition order
        return None

    def publish_action(self, action: np.ndarray, target: str = "real") -> None:
        # ← ship the flat float32 action vector to your robot/sim
        ...

    def close(self) -> None:
        # ← release whatever __init__ acquired
        ...


@TRANSPORT_REGISTRY.register("mybridge")   # ← the string used in transport.type
def _build(config: ConfigDict, robot: Robot) -> TransportBridge:
    return MyBridge(config, robot)

The contract is defined in src/transport/base.py: get_frame must key images by observation_key and concatenate state in robot.observation_schema.state_composition order, and publish_action receives a flat joint-space vector of robot.total_action_dim. For a minimal real backend mirror ZmqTransport in src/transport/zmq.py; for the ROS-style deque-sync collection path, see _RosTransportBase in src/transport/base.py.