LeRobot_v3.0数据集对齐
更新时间：2026-07-29
简介
把一个已完成的 LeRobot v3.0 数据集对齐到共享的跨本体规格（统一 fps / robot_type / features、动作与状态维度投影、可选逐维归一化），依赖 lerobot（内部走 align_lerobot_v30.align_dataset + ffmpeg 视频重编码），用于多数据集合并前的预处理，使各输入通过 merge_datasets 的兼容性校验。
功能描述
• 统一规格：按同一个 spec（plan_unification 产出）重采样 fps、投影 action / observation.state 到规范宽度、统一 robot_type 与 features schema
• 归一化：支持 none / meanstd / minmax 三种逐维动作与状态缩放
• 视频重编码：vcodec 默认 h264（比 lerobot 的 libsvtav1 快 5-10 倍），auto 可选硬件编码器
• 逐行任务：每行 src（源 v3.0 数据集根目录）对齐输出到 out，返回成功/失败字符串
算子参数
输入
| 输入 | 含义 |
|---|---|
| src | 源 v3.0 数据集根目录（string，本地或 BOS 路径）。 |
| out | 对齐后数据集的输出根目录（string，本地或 BOS 路径）。 |
输出
| 输出 | 含义 |
|---|---|
| result | 字符串（large_string），成功为 "SUCCESS: <out_path>"，失败为 "Failed: <src>: <ErrorType>: <msg>"。 |
参数
| 参数名称 | 类型 | 默认值 | 描述 | |
|---|---|---|---|---|
| spec | dict | （必填） | 统一目标规格，align_lerobot_v30.plan_unification 的输出；所有行共用同一 spec 以产出 byte-identical 的 fps/robot_type/features |
|
| normalize | str | none | 逐维动作/状态缩放，可选 none/meanstd/minmax | |
| vcodec | str | h264 | 视频编码器，可选 h264/auto/libsvtav1 等；auto 在可用时选硬件编码器 | |
| encoder_threads | int |None | 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.align_lerobot_v30_udf import AlignLeRobotV30
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 # 生产中 spec 应由 align_lerobot_v30.plan_unification([...]) 生成，此处给一个最小示例
16 spec = {
17 "fps": 20,
18 "robot_type": "generic",
19 "state_dim": 7,
20 "action_dim": 7,
21 "image_keys": ["cam_high"],
22 "image_hw": (480, 640),
23 "features": {
24 "observation.state": {"dtype": "float32", "shape": (7,)},
25 "action": {"dtype": "float32", "shape": (7,)},
26 },
27 }
28
29 samples = {
30 "src": ["bos://your-bucket/v30/dataset_a"], # MOCK
31 "out": ["bos://your-bucket/aligned/dataset_a"], # MOCK
32 }
33 ds = daft.from_pydict(samples)
34 ds = ds.with_column(
35 "result",
36 aihc_udf(
37 AlignLeRobotV30,
38 construct_args={"spec": spec, "normalize": "none", "vcodec": "h264"},
39 num_cpus=1, concurrency=1, batch_size=1,
40 )(col("src"), col("out")),
41 )
42 ds.show()评价此篇文章