简介

Rerun .rrd 格式转换算子，把 VLA 数采导出的 rrd recording（一个 episode 一个 .rrd）转换成 LeRobot v3.0 数据集。通过 rerun.dataframe 直接读文件，不起 viewer、不走 gRPC。

源数据实体：

功能描述

• 特征集自动探测：相机来自 VideoStream 实体，编码读静态 VideoStream:codec（fourcc avc1→H.264、hev1→H.265）逐路判断，分辨率解首帧拿到
• 相机特征名取 metadata.json 的 videos[].key（按实体叶子名匹配），即 observation.images.head_color / left_arm_color / right_arm_color；metadata.json 缺失或对不上时回落到 observation.images.<实体叶子名>
• observation.state / action 按 metadata.json 的 state_space / action_space 声明顺序拼接（实测 29 维：torso 3 + 双臂各 6 + 双手各 6 + head 2）。实体扫描是按字母序的，与采集端的关节顺序不同，所以只有 metadata 缺失时才回落到字母序
• 写出前逐 episode 校验 state/action 维度顺序、相机实体、编码、分辨率和 fps；schema 漂移直接失败，不把错误值写进同一数据集
• 行与帧按下标配对：采集端的同步器每个 tick 产出一行标量 + 每路相机一条 sample，健康 episode 各流条数相等，这个配对就是权威口径。两条时间线自己会相对漂移（实测 ±16.6ms，接近半个帧周期），按时间戳重新配对会把约 60% 的行挪到相邻帧上。只有条数不一致的残缺 episode 才回落到因果 ZOH（取「tick 时刻或之前最近的一条」），并打 warning
• VideoStream sample 条数与流里实际帧数（demux 出的 access unit 数）不一致时，按首尾 sample 时间戳线性重铺真实帧时间；残缺流在首个 tick 前没有历史样本时直接失败，禁止使用未来首帧
• 任务名优先取 metadata.json 的 task_prompt，再回落 task_description、上级目录名、default_task
• 视频样本先落盘成 Annex-B 流，再逐帧拉取，任意时刻内存里只有一帧（1240 帧 1280×720 全解开是 3.4GB RGB）
• 标量列中途有空洞时按前值保持；开头就缺样本的行（还没有任何因果值）整行裁掉，不补 0 伪造观测。裁行只改输出起点，帧配对仍按原始时间线的下标算，不会把相机重置回第 0 帧
• episode 级多进程分片，分片结果用 lerobot aggregate_datasets 合并
• 写出后逐路核对输出 mp4 的帧数与 parquet 行数，不一致直接失败并删掉半成品目录：lerobot 的流式编码器在自己队列打满 100ms 时会静默丢帧（实测 1920×1200 立体 episode 三路里两路各丢 5~6 帧），行还照样写进 parquet，之后每一行都会对上错的图。本算子已把入队改成阻塞等待，核对是第二道闸
• 输出已是 v3.0 且 force_conversion=False 时跳过，返回 SKIPPED

算子参数

输入

输出

参数

注意事项

• 只支持 VideoStream 里的 H.264 / H.265，其他编码直接抛 UnsupportedRrdSchemaError，不会交给 demuxer 静默产出坏图。
• 输入 rrd 由哪个 rerun 版本写的不重要，但读的版本不能太旧：rerun 0.26.2 读 0.27.2 写的文件会打前向兼容 warning（实测能读，结果与 0.27.3 一致）。
• metadata.json 的 total_frames 与实际标量行数不一致时只打 warning，按实际行数转换。
• 相机条数与标量行数不一致时会退到因果 ZOH，这时输出会重复或丢帧，日志里有明确 warning，值得回头查采集侧。

调用示例

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_rrd_to_lerobot_v30_udf import (
10    ConvertRrdToLeRobotV30,
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/rrd_dataset/<task_name>"],
22        "output_paths": ["/tmp/rrd_lerobot_out/<task_name>"],
23    }
24    ds = daft.from_pydict(samples)
25    ds = ds.with_column(
26        "result",
27        aihc_udf(
28            ConvertRrdToLeRobotV30,
29            construct_args={
30                "num_workers": 4,
31                "vcodec": "h264",
32                "force_conversion": True,
33                "robot_type": "EHR02",
34                "fps": None,
35                "resize": None,
36                "decode_threads": 8,
37            },
38            num_cpus=1,
39            concurrency=1,
40            batch_size=1,
41        )(col("input_paths"), col("output_paths")),
42    )
43    ds.show()