简介

从 LeRobot v3.0 数据集里删掉指定 episode，产出一份不含这些 episode 的新数据集，源数据集不动。包 lerobot datasets/dataset_tools.py:81 delete_episodes，外面加 crash-safe 与 skip-if-done 两层运维行为。

功能描述

• 非破坏：源数据集只读，结果写到 output_path；output_path 为空时缺省写到与源同级的 <源目录名>_modified
• 保留下来的 episode 按原顺序重索引成 0..N-1：data/**/*.parquet 的 episode_index / index / task_index 全部重编，meta/episodes/** 与 meta/info.json（total_episodes / total_frames / total_tasks / splits）同步重写
• 任务表按存活 episode 重建：删完之后没有 episode 引用的任务会从 meta/tasks.parquet 消失，task_index 相应重映射
• 视频按文件粒度处理：一个 mp4 里的 episode 全部保留时整文件拷贝（不重编码）；既有保留又有删除时用 PyAV 按帧区间重编码，并重排该文件内 episode 的 from_timestamp / to_timestamp
• meta/stats.json 由存活 episode 的逐 episode 统计重新聚合（lerobot aggregate_stats），不是照抄源 stats
• crash-safe：开跑前把已存在的输出目录整体删掉重建，中途失败再删一次半成品，避免下一次运行把半成品误判成完成态
• 完成后校验输出确实是合法 v3.0（meta/info.json + codebase_version），不是就报错
• skip-if-done：输出已是 v3.0 且 force=False 时不动手，返回 SKIPPED
• 参数校验沿用上游：episode_indices 为空、含越界索引、或把所有 episode 删光都直接失败（返回 Failed 状态行，不抛异常打断整批）

算子参数

输入

输出

参数

注意事项

• output_path 指向的目录会被整体删除后重建（force=True 时连已完成的 v3.0 输出一起删），不要指向源数据集或还有其他内容的目录。
• 输出的 episode 索引是重排过的：删掉 ep1 之后原 ep2 变成新 ep1。下游若要引用原 episode 号，需要自己记映射。
• 视频文件沿用源的 chunk / file 索引命名，删除后输出里可能出现不连续的 file_index，属正常。
• 混合文件的重编码固定用 lerobot _copy_and_reindex_videos 的默认 libsvtav1 / yuv420p（本算子未暴露），这部分视频会多一次代际压缩损失；整文件保留的视频是字节拷贝，无损失。
• 单行只返回状态串（__return_column_type__ 是 large_string），需要输出路径请直接用输入列。

调用示例

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.delete_episodes_udf import DeleteEpisodes
10
11if __name__ == "__main__":
12    if os.getenv("DAFT_RUNNER", "native") == "ray":
13        import ray
14        ray.init(dashboard_host="0.0.0.0", ignore_reinit_error=True)
15        daft.set_runner_ray()
16    daft.set_execution_config(actor_udf_ready_timeout=6000, min_cpu_per_task=0)
17
18# 夹具可用 test/_helpers/lerobot_edit_fixture.py:build_tabular_dataset 生成
19    samples = {
20        "dataset_path": ["/path/to/lerobot_dataset"],
21        "episode_indices": [[1]],
22        "output_path": ["/tmp/lerobot_deleted_out"],
23    }
24    ds = daft.from_pydict(samples)
25    ds = ds.with_column(
26        "result",
27        aihc_udf(
28            DeleteEpisodes,
29            construct_args={"repo_id": None, "force": False},
30            num_cpus=1,
31            concurrency=1,
32            batch_size=1,
33        )(col("dataset_path"), col("episode_indices"), col("output_path")),
34    )
35    ds.show()