简介

对已落盘的 LeRobot v3.0 数据集重算 meta/stats.json。lerobot 0.4.4 只在写数据集的过程中算统计（LeRobotDataset.save_episode → compute_episode_stats → aggregate_stats → write_stats），没有给已成型数据集重算的入口，本算子补这个缺口：表列特征直接从 parquet 归约，视频特征抽帧统计，聚合与落盘复用 lerobot 自己的 aggregate_stats / write_stats。

源码锚点（lerobot 0.4.4）：

功能描述

• 表列特征（state / action 等）直接读 data/**/*.parquet：pyarrow 列投影 + 按 episode_index 行过滤，成本随请求的特征数走，不随数据集整体规模走
• 视频 / 图像特征按 video_frames_per_episode 个等距时间戳抽帧（decode_video_frames），逐通道在 (0, 2, 3) 上归约后除以 255，形状 (3, 1, 1)，与写入端 compute_episode_stats 的图像分支口径一致；从不全量解码
• 统计集合与写入端一致：跳过 index / episode_index / task_index 三个元数据列和所有 string 特征；features 白名单只能在这个集合内再收窄，传了未知键或元数据键直接报错并列出可用键
• 逐 episode 统计用 lerobot aggregate_stats 聚合、write_stats 落盘，产出布局与新写数据集的 meta/stats.json 一致
• v3.0 里多个 episode 顺序共享一个 mp4，采样区间按该 episode 元数据的 videos/<key>/from_timestamp 偏移，不是从文件 0 秒起算
• 抽样型数据集的 mp4 常出现「帧率元数据与真实 pts 栅格不一致 / 物理视频比元数据声称的短」，为此有三层防御：按容器 duration 与帧数钳制采样区间、采样点向下（floor）吸附到帧栅格且容差抬到 1/4 帧、按时间戳解码抛异常时回落到 pyav 按帧索引抽样（最多解 1024 帧后均匀取样）
• num_workers > 1 时 episode 级线程池并发（视频解码释放 GIL）
• skip-if-done：meta/stats.json 已存在且 force=False 时不重算，返回 SKIPPED，并把已有 stats.json 的内容原样回填到 features 里
• 每个特征的统计字典以 JSON 字符串返回，输出 schema 不随数据集特征集变化（__return_column_type__ 是静态的）
• write=False 时只算不落盘，stats_path 返回空串

算子参数

输入

输出

参数

注意事项

• 分位数 q01..q99 的聚合语义是「各 episode 分位数按帧数加权平均」（lerobot aggregate_feature_stats），不是全量数据的真分位数；单 episode 内部也是 5000 bin 直方图近似（RunningQuantileStats），样本不足 2 行时 q* 直接取 mean。
• 图像统计前会走 lerobot auto_downsample_height_width（长边 ≥300 时按整数因子下采样到约 150 像素），这是写入端的既有行为，统计值与全分辨率精确统计会有细微差别。
• 视频 / 图像特征统计是逐通道 3 个值、值域 [0,1]；表列特征逐维。
• 输入不是完成态 v3.0 数据集时直接失败，不会去猜 v2.x 布局。

调用示例

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.lerobot.compute_dataset_stats_udf import (
10    ComputeDatasetStats,
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        "dataset_path": ["/path/to/lerobot_dataset"],
22    }
23    ds = daft.from_pydict(samples)
24    ds = ds.with_column(
25        "result",
26        aihc_udf(
27            ComputeDatasetStats,
28            construct_args={
29                "features": None,
30                "num_workers": 1,
31                "video_frames_per_episode": 32,
32                "write": True,
33                "force": False,
34            },
35            num_cpus=1,
36            concurrency=1,
37            batch_size=1,
38        )(col("dataset_path")),
39    )
40    ds.show()