Mcap格式转换处理器
更新时间：2026-09-14
简介
MCAP 格式转换算子，把 protobuf 编码的 MCAP episode（ABC-130k / YAM 双臂工作站布局，一个 episode 一个 episode.mcap）转换成 LeRobot v3.0 数据集。不需要安装 ROS，也不涉及 CDR 编码。
源数据话题：
| 话题 | schema | 说明 |
|---|---|---|
| /instruction | Instructions | 任务文本，单条 |
| /{side}-arm-{state,action} | RobotState | 每侧 6 个关节角（弧度），-state 另带 velocity/torque/pose |
| /{side}-ee-{state,action} | GripperState | 夹爪归一化开合度（0 关 1 开） |
| /top-camera | foxglove.CompressedVideo | 单目站（RealSense）顶部相机 |
| /top-{left,right}-camera | foxglove.CompressedVideo | 立体站（ZED-X）顶部双目 |
| /{side}-wrist-camera | foxglove.CompressedVideo | 腕部相机 |
| /<视频话题>-info | foxglove.CameraCalibration | 分辨率与内参 |
功能描述
- 自动探测站型与编码：单目 / 立体、每路相机逐个读
CompressedVideo.format选 H.264 或 H.265，分辨率取自CameraCalibration（探测不到才解码首帧） observation.state/action按left, right顺序拼接每侧[关节角, 夹爪]，只取position字段（与上游训练口径一致）- 严格校验每侧 arm 为 6 维、ee 为 1 维，并在写出前校验所有 episode 的 topic/schema/维度/编码/分辨率一致；不补零掩盖缺失流
- 固定时钟因果重采样：源流采样率各不相同（关节/夹爪 200~290Hz，相机约 30Hz，条数不等），行网格取
target_fps固定周期，每条流取「tick 时刻或之前最近的一条采样」（causal ZOH） CompressedVideo消息数与流里实际帧数（demux 出的 access unit 数）不一致时，按首尾消息时间戳线性重铺真实帧时间，避免把一条消息误当成一帧- 任务名优先取
/instruction，缺失时回落到任务目录名，再回落到default_task - 视频逐帧流式解码，任意时刻内存里只有一帧：立体 episode 单路 3300 帧 1920×1200 全解开是 23GB RGB
- episode 级多进程分片，分片结果用 lerobot
aggregate_datasets合并 - 写出后逐路核对输出 mp4 的帧数与 parquet 行数，不一致直接失败并删掉半成品目录：lerobot 的流式编码器在自己队列打满 100ms 时会静默丢帧（实测 1920×1200 立体 episode 三路里两路各丢 5~6 帧），行还照样写进 parquet，之后每一行都会对上错的图。本算子已把入队改成阻塞等待，核对是第二道闸
- 只认
episode.mcap；已标注 episode 目录里的annotation.mcap（子任务标签）会被跳过，不会被当成 episode - 输出已是 v3.0 且
force_conversion=False时跳过，返回SKIPPED
算子参数
输入
| 输入 | 含义 |
|---|---|
| input_path | 源数据集目录，在这个目录下递归查找 episode.mcap |
| output_path | 输出数据集目录，v3.0 数据集直接写到这里 |
输出
| 输出 | 含义 |
|---|---|
| result | SUCCESS / SKIPPED: has already been v3.0 / Failed: [repo] <异常类型>: <信息> |
参数
| 参数名称 | 类型 | 默认值 | 描述 |
|---|---|---|---|
| num_workers | int | 4 | episode 级并发进程数，<=1 时串行 |
| vcodec | str | "h264" | 输出视频编码。auto 会被映射成 h264，避开 lerobot resolve_vcodec 误选 nvenc |
| force_conversion | bool | False | True 表示输出已是 v3.0 也重新转换 |
| default_task | str | "default_task" | /instruction 和任务目录名都拿不到时的兜底任务名 |
| robot_type | str | "unknown" | 写入 v3 元数据的机器人标识 |
| target_fps | int | 30 | 输出数据集行频率，所有源流按此固定时钟重采样 |
| top_camera_policy | str | "left" | 立体站两路顶部相机的取法：left/right 取一只眼作为 observation.images.top；both 输出 top_left + top_right 两个特征；official 按 sha1(episode 目录名) 每个 episode 挑一只眼，复现上游训练集。单目站忽略此参数 |
| resize | tuple[int,int] 或 None | None | 所有相机统一缩放到 (height, width)。None 保持源分辨率——立体站是 1920×1200，全量转换建议开缩放 |
| decode_threads | int | 8 | 每路相机解码器的 FFmpeg 线程数。PyAV 默认单线程，在 1920×1200 H.265 上慢约 5 倍；不要设 0（会按核数开线程，与 episode 级并发叠加后线程过订阅） |
| repo_id | str 或 None | None | 写入 v3 元数据的逻辑数据集名。None 取 input_path 的目录名 |
注意事项
- 立体站（ZED-X）四路相机都是 H.265、1920×1200；单目站（RealSense）是 H.264。编码按路判断，不要按站型假设。
- 立体 episode 如果同时存在
/top-camera，那是把左右眼缩略图烧进同一画面的低分辨率预览图，本算子在两只眼都存在时直接丢弃它。 - 源流不是等长的，不能按消息下标配对：实测某立体 episode 的
/left-arm-state有 31954 条、/top-left-camera只有 3332 条。
调用示例
Python
1from __future__ import annotations
2
3import os
4
5import daft
6from daft import col
7
8from daft.aihc.common.udf import aihc_udf
9from daft.aihc.functions.embodied.convert_mcap_to_lerobot_v30_udf import (
10 ConvertMcapToLeRobotV30,
11)
12
13if __name__ == "__main__":
14 if os.getenv("DAFT_RUNNER", "native") == "ray":
15 import ray
16 ray.init(dashboard_host="0.0.0.0", ignore_reinit_error=True)
17 daft.set_runner_ray()
18 daft.set_execution_config(actor_udf_ready_timeout=6000, min_cpu_per_task=0)
19
20samples = {
21 "input_paths": ["/path/to/mcap_dataset/<task_name>"],
22 "output_paths": ["/tmp/mcap_lerobot_out/<task_name>"],
23 }
24 ds = daft.from_pydict(samples)
25 ds = ds.with_column(
26 "result",
27 aihc_udf(
28 ConvertMcapToLeRobotV30,
29 construct_args={
30 "num_workers": 4,
31 "vcodec": "h264",
32 "force_conversion": True,
33 "robot_type": "yam_bimanual",
34 "target_fps": 30,
35 "top_camera_policy": "left",
36 "resize": None,
37 "decode_threads": 8,
38 },
39 num_cpus=1,
40 concurrency=1,
41 batch_size=1,
42 )(col("input_paths"), col("output_paths")),
43 )
44 ds.show()评价此篇文章