导出为LeRobot_v3.0
更新时间：2026-07-29
简介
把视频 + 7-DoF 动作 + 状态组装导出为标准 LeRobot v3.0 数据集，AIHC 自研，是 7-DoF 动作链路的终点，依赖 lerobot（复用 LeRobotDataset.create/add_frame/save_episode writer）+ 可选 OpenCV 读帧，离线 CPU，用于把处理好的 episode 落盘为可训练的 v3.0 数据集。
功能描述
• shard 写出：每个 transform batch 在独立的 <output_path>_shard_<...> 目录里完整走 create → add_frame → save_episode → finalize，返回 shard 路径
• 主进程合并：主脚本 ds.collect() 后对返回列去重，再调 merge_shards_to_final（或 lerobot 的 merge_datasets）合并为最终 output_path
• schema 可配：action_dim/state_dim/feature_schema 兼容不同本体（如 14-DoF 双臂），不绑定固定 7-DoF
• 容错开关：fail_on_error=True（默认）遇写入/finalize 失败即抛出，避免悄悄产坏 parquet
算子参数
输入
| 输入 | 含义 |
|---|---|
| video | 该 episode 的视频本地路径或多相机 dict；use_videos=False 时可为 None。 |
| action | 动作序列 [N, action_dim]，类型 list<list |
| state | 状态序列 [N, state_dim]，类型 list<list |
| task | 该 episode 的语言指令（string）。 |
输出
| 输出 | 含义 |
|---|---|
| result | 字符串（large_string）：该 episode 所在的 shard 目录路径；写入失败的行为 None。主脚本对该列去重后 merge。 |
参数
| 参数名称 | 类型 | 默认值 | 描述 | |
|---|---|---|---|---|
| output_path | str | （必填） | 最终 v3.0 数据集输出目录；actor 在 <output_path>_shard_<xxx> 写分片，主脚本负责 merge |
|
| repo_id | str | （必填） | 数据集标识，如 aihc/hand_demo |
|
| fps | int | （必填） | 帧率（须为正） | |
| robot_type | str | generic_hand | 本体类型标识，写入 v3 元数据 | |
| action_dim | int | 7 | 动作维度（须为正；双臂等可配） | |
| state_dim | int |None | None（等于 action_dim） | 状态维度，None 时与 action_dim 相同 | |
| image_key | str | observation.images.cam | 视频 feature 的 key | |
| image_shape | tuple[int,int,int] | (480, 640, 3) | 视频帧 [H, W, C] | |
| feature_schema | dict |None | None | 完全自定义的 lerobot features dict；提供时覆盖 action/state/image 默认 | |
| use_videos | bool | True | 是否作为 video feature 写入 | |
| vcodec | str | h264 | 视频编码器 | |
| encoder_threads | int |None | None | 每个编码器实例的线程数 | |
| fail_on_error | bool | True | 遇 episode 写入/finalize 失败是否抛出（False 保留吞异常成 None 的旧行为） |
调用示例
Python
1from __future__ import annotations
2import os
3import daft
4from daft import col
5from daft.aihc.common.udf import aihc_udf
6from daft.aihc.functions.embodied.export_to_lerobot_v30 import ExportToLeRobotV30
7
8if __name__ == "__main__":
9 if os.getenv("DAFT_RUNNER", "native") == "ray":
10 import ray
11 ray.init(dashboard_host="0.0.0.0", ignore_reinit_error=True)
12 daft.set_runner_ray()
13 daft.set_execution_config(actor_udf_ready_timeout=6000, min_cpu_per_task=0)
14
15 OUTPUT_DIR = "/mnt/pfs/lerobot_v30_out/ds" # MOCK
16
17 # 一个 5 帧无视频 episode：dx 递增、gripper 常开
18 action = [[0.0 + 0.1 * i, 0.0, 0.0, 0.0, 0.0, 0.0, 1.0] for i in range(5)]
19 state = [row[:] for row in action]
20 samples = {
21 "video": [None],
22 "action": [action],
23 "state": [state],
24 "task": ["pick up the block"],
25 }
26 ds = daft.from_pydict(samples)
27 ds = ds.with_column(
28 "shard",
29 aihc_udf(
30 ExportToLeRobotV30,
31 construct_args={
32 "output_path": OUTPUT_DIR,
33 "repo_id": "aihc/hand_demo",
34 "fps": 20,
35 "robot_type": "generic_hand",
36 "action_dim": 7,
37 "state_dim": 7,
38 "use_videos": False,
39 },
40 num_cpus=1, concurrency=1, batch_size=8,
41 )(col("video"), col("action"), col("state"), col("task")),
42 )
43 ds.show()
44 # 生产中：shards = set(ds.to_pydict()["shard"]); 再调 merge_shards_to_final(shards, "aihc/hand_demo", OUTPUT_DIR)评价此篇文章