传输层

传输层连接 EVA 与数据源——读取观测、发送关节命令。通过transport.type选择。

概览

`transport.type`	适用场景
`ros1`	基于 ROS1 栈的真实机器人。
`ros2`	基于 ROS2 栈的真实机器人；还能解码压缩相机画面。
`zmq`	不经 ROS、通过独立的 ZeroMQ 执行节点接入的真实机器人。
`dataset`	回放一份录制运行——无需机器人、无需策略服务。

离线降级。若在未装 ROS 的主机上选了 ROS 后端，EVA 会打印警告，转而运行一个空的离线链路——不收帧、丢弃所有命令——以便 Web 控制台仍能打开。参见确认你没有处于离线状态。

每个后端都提供什么

EVA 驱动各后端的方式相同：取一帧观测、发一条关节命令、报告链路健康。

命令永远是关节角度。EEF 位姿输出会在到达传输层之前通过 IK 转换为关节角度——参见动作空间。
链路健康。在线后端报告距上一条入站消息的时间，控制台据此标记卡住的链路；dataset 回放与离线降级不报告。
仿真目标与真实目标。命令可发往真机或仿真/3D 预览。仅ros1自带仿真发布器。
数据采集。COLLECT 通过同一传输层录制——参见数据采集。

前置条件

后端	必须具备的条件
`dataset`	仅需 EVA。
`zmq`	EVA，外加一个在配置端点上发布观测的执行层节点。
`ros1` / `ros2`	可用的 ROS 安装，以及在线发布映射话题的机器人。未装 ROS 时，EVA 会静默降级到离线链路。

dataset —— 回放录制

dataset从一份 LeRobot v2.1 数据集目录里回放一个 episode（关节状态存于.parquet，视频为.mp4）。完全离线：无机器人、无策略服务。transport.dataset_keys映射存放关节状态、动作及各相机视频的列名；默认值适配多数录制。

从configs/00_openloop/dual_agilex_AGILEX_openloop.py复制：

python

transport = dict(
    type='dataset',
    dataset_dir='<你的数据集文件夹路径>',   # ← 改成你录制数据所在的文件夹
    episode_id=0,                        # 回放第几次录制（0 = 第一次）
    dataset_keys=dict(
        state_key='observation.qpos',   # ← 仅当你的数据用了别的列名时才改
        action_key='action',
        video_keys=dict(
            # 把每个相机名映射到它的视频列：
            cam_high='observation.images.cam_high',
        ),
    ),
)

dataset_dir指向包含meta/子目录和各回合文件的文件夹，建议用绝对路径。指向不存在或为空的文件夹会在启动时报错，错误信息中点名dataset_dir。

bash

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

访问http://localhost:8080后若相机面板空白，多半是video_keys里的列名与数据集列名对不上。

ros1 / ros2 —— 在线 ROS 机器人

ros1与ros2订阅相机和关节状态话题，按时间戳对齐入站消息，并发布关节命令。ros2还能解码压缩图像话题（名以/compressed结尾）。

transport.topics必填：把 EVA 的相机名和机械臂名映射到你机器人的真实话题名。留空会在启动时报错。在部署用的_base.py里按机器人设置。

话题名因机器人而异。把下面的占位符换成你机器人的话题名（可用ros2 topic list列出）。

从configs/01_deploy/r1lite/_base.py复制：

python

transport = dict(
    type='ros2',                  # 或 'ros1'
    topics=dict(
        camera_topics=dict(
            # EVA 相机名 ： 你机器人的图像话题
            head='<你的头部相机话题>',        # ← 填入
            left_wrist='<你的左腕相机话题>',    # ← 填入
        ),
        group_topics=dict(
            left_arm=dict(
                state_topic='<左臂关节状态话题>',    # ← 必填：从哪里读关节角度
                command_topic='<左臂命令话题>',      # ← 必填：往哪里发关节命令
                # 可选，除非需要否则不必写：
                # gripper_state_topic / gripper_command_topic （ros2 拆分式夹爪）
                # eef_state_topic     （仅当观测空间为 EEF 时）
                # sim_command_topic   （仅 ros1 仿真）
            ),
            # 若机器人有右臂，照同样方式加 right_arm=dict(...)
        ),
    ),
)

每条机械臂至少需要state_topic和command_topic；其余键为可选，仅在注释所述情况下使用。第二条臂照样追加right_arm块。

ROS 节点名默认eva_client；如需避免重名，用transport.node_name覆盖。

zmq —— 经消息节点接入的机器人

zmq通过 ZeroMQ 与独立的执行层节点交换已对齐的观测快照与动作消息。

字段	默认值	含义
`sub_endpoint`	`tcp://127.0.0.1:5555`	从哪个地址接收观测。
`pub_endpoint`	`tcp://127.0.0.1:5556`	往哪个地址发送动作。
`disabled_cameras`	`[]`	从入站帧中忽略的相机名。
`disabled_groups`	`[]`	需要忽略的机械臂/执行器组名。

默认值假设节点在本机 5555/5556 端口。远程节点请覆盖端点。从configs/01_deploy/ur5e/_base.py复制：

python

transport = dict(
    type='zmq',
    # 仅当节点不在本机/不是这些端口时，才取消注释并修改：
    # sub_endpoint='tcp://<节点IP>:<观测端口>',   # ← 观测从哪里来
    # pub_endpoint='tcp://<节点IP>:<动作端口>', # ← 动作往哪里去
    resize_pad=False,
    image_layout='hwc',
)

确认你没有处于离线状态

离线降级是静默的：收不到任何帧、丢弃所有命令。识别方式：

启动日志。降级时打印transport ros2 unavailable (...); starting offline transport。先修好缺失的 ROS 安装或话题映射。
控制台链路指示。离线或卡住的链路显示为 offline。
无相机图像、关节图表不动，意味着没有帧到达。

共享图像设置

每个后端在帧送达策略前应用同一套图像预处理，默认值如下。

字段	默认值	含义
`image_height` / `image_width`	`224 / 224`	送入策略的图像尺寸。
`resize_pad`	`True`	`True`用填充保持纵横比；`False`直接拉伸。与策略训练时一致。
`image_layout`	`"chw"`	像素排列：`chw`或`hwc`。
`convert_bgr_to_rgb`	`True`	交换蓝/红通道。

预设继承：参见配置。

如何添加新的传输层

一个传输层就是TransportBridge（src/transport/base.py）的子类，并注册到TRANSPORT_REGISTRY。src/transport/__init__.py会自动导入该目录下的所有同级模块，因此把文件放进去即可让后端可被选用。

继承TransportBridge，实现get_frame、publish_action和close。可选钩子（get_latest_qpos、seconds_since_last_recv、is_offline、*_collection*一族、create_observation_reader）仅在数据源支持时再覆盖。
用@TRANSPORT_REGISTRY.register("<你的名字>")注册一个构造函数。被装饰的可调用对象接收(config, robot)，返回传输层实例——参见src/transport/base.py中的build_transport。
把文件保存为src/transport/<你的名字>.py。src/transport/__init__.py里的包遍历会在启动时导入它，注册以副作用方式完成——无需其它接线。
在任意配置中选用：transport = dict(type='<你的名字>', ...)。启动时列出注册表（TRANSPORT_REGISTRY.available()）即可验证你的键名是否在列。

python

# src/transport/mybridge.py  ← 新建文件；由包自动导入
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
        # ← 在这里打开 socket / 文件句柄等

    def get_frame(self) -> Observation | None:
        # ← 构建 images 字典（observation_key -> [H,W,3] uint8）
        # ← 与按 state_composition 顺序拼接的 state_qpos [state_dim] float32
        return None

    def publish_action(self, action: np.ndarray, target: str = "real") -> None:
        # ← 把扁平 float32 动作向量送达机器人/仿真
        ...

    def close(self) -> None:
        # ← 释放 __init__ 中申请的资源
        ...


@TRANSPORT_REGISTRY.register("mybridge")   # ← 这个字符串就是 transport.type 用的值
def _build(config: ConfigDict, robot: Robot) -> TransportBridge:
    return MyBridge(config, robot)

契约见src/transport/base.py：get_frame需按observation_key组织图像，并按robot.observation_schema.state_composition顺序拼接关节状态；publish_action接收长度为robot.total_action_dim的扁平关节空间向量。最小可用的真机后端可参考src/transport/zmq.py中的ZmqTransport；若需 ROS 风格的双端队列同步采集路径，请参考src/transport/base.py中的_RosTransportBase。