内置导出写的是 LeRobot v3。如果你的训练代码需要别的格式,比如 CSV、RLDS、HDF5 或者团队内部格式,导出算法就负责把数据集版本写成那种格式。平台会保存这些文件,并像其他导出一样提供下载。
导出算法每个数据集版本运行一次。它读取 version,并通过 out(一个 ArtifactWriter)写出 artifact:
| 方法 | 作用 |
|---|---|
out.format(name) | 说明文件的格式名称。 |
out.path(rel) | 获得导出内的一个写入路径,上级文件夹会自动创建。 |
out.add(src, rel) | 把一个文件或文件夹复制进导出。 |
out.summary(dict) | 描述这次导出,例如数量和大小。 |
示例:把任务片段导出为 CSV
这个算法为版本里的每个任务片段写一行 CSV。文件是 packages/sdk-python/examples/v1/export-csv/main.py。
"""Export: one CSV row per episode of the dataset version.
In: the dataset version: its episodes, with their recording, window, task and split.
Out: an artifact: episodes.csv plus a short summary. The platform checks and hashes every file.
Why: the smallest complete export; copy it to write the files another tool expects (a table, a manifest).
"""
import csv
from embodiflow import ArtifactWriter, DatasetVersion, algorithm
@algorithm.export(name="episodes-csv", title="Episodes as CSV")
def run(version: DatasetVersion, out: ArtifactWriter) -> None:
out.format("csv")
with open(out.path("episodes.csv"), "w", newline="", encoding="utf-8") as handle:
writer = csv.writer(handle)
writer.writerow(["episode_id", "recording_id", "start_s", "end_s", "duration_s", "task", "split"])
for member in version.members:
writer.writerow([member.episode_id, member.recording_id, member.start_s, member.end_s, f"{member.duration_s:.3f}", member.task or "", member.split or ""])
out.summary({"episodes": len(version.members)})它的测试是 packages/sdk-python/examples/v1/export-csv/test_main.py:用一份很小的测试数据,通过真实的运行器运行这段代码,并检查写出的值。想从它开始,运行 embodiflow algorithm init my-algorithm --category export,它会写出 main.py 和 test_main.py;再运行 embodiflow algorithm test my-algorithm(需要装有 pytest 和 SDK 的 Python)。
值得知道
需要数据本身、而不只是任务片段的表格时,version.members[i].recording() 会下载该任务片段的录制窗口。
导出工作流只接收一个 artifact。无论什么格式,导出时都会检查脱敏,所以自定义格式不会成为绕过它的途径。
内置的 LeRobot v3 导出有参数 video_codec、max_width、unsplit、splits 和 include_labels,版本的导出选项会映射到它们。默认之外的任务语言会被 EXPORT_OPTION_UNSUPPORTED 拒绝,不会被悄悄丢掉。
和整理算法一样,你可以在编辑器的「试跑」里用数据集版本的前几个任务片段试跑导出算法。要生成真正的文件,请从数据集版本发起导出;启动导出工作流运行,需要有运行工作流和写入导出的权限。