LeRobotDataset 是每个 LeRobot 脚本读写所用的格式。它感知 episode,能够即时解码视频 observation,并可与 Hugging Face Hub 往返同步。
关于该格式及常用操作,请参阅 使用 LeRobotDataset,迁移请参阅 移植大型 dataset,CLI 请参阅 工具。
( repo_id: strroot: str | pathlib.Path | None = Noneepisodes: list[int] | None = Noneepisode_filter: collections.abc.Callable[[dict], bool] | None = Noneimage_transforms: collections.abc.Callable | None = Nonedelta_timestamps: dict[str, list[float]] | None = Nonetolerance_s: float = 0.0001revision: str | None = Noneforce_cache_sync: bool = Falsedownload_videos: bool = Truevideo_backend: str | None = Nonereturn_uint8: bool = Falsedepth_output_unit: str = 'mm'batch_encoding_size: int = 1rgb_encoder: lerobot.configs.video.RGBEncoderConfig | None = Nonedepth_encoder: lerobot.configs.video.DepthEncoderConfig | None = Noneencoder_threads: int | None = Nonestreaming_encoding: bool = Falseencoder_queue_maxsize: int = 30repo_type: str = 'dataset'token: str | bool | None = None )
( frame: dict )
向当前 episode 缓冲区添加单个帧。
委托给 DatasetWriter.add_frame。dataset 必须处于写入模式(通过 create 或 resume 创建)。
( delete_images: bool = True )
丢弃当前 episode 缓冲区而不保存。
委托给 DatasetWriter.clear_episode_buffer。用于丢弃失败或中断的采集 episode。
移除应用于视觉 observation 的变换。
( repo_id: strfps: intfeatures: dictroot: str | pathlib.Path | None = Nonerobot_type: str | None = Noneuse_videos: bool = Truetolerance_s: float = 0.0001image_writer_processes: int = 0image_writer_threads: int = 0video_backend: str | None = Nonebatch_encoding_size: int = 1rgb_encoder: lerobot.configs.video.RGBEncoderConfig | None = Nonedepth_encoder: lerobot.configs.video.DepthEncoderConfig | None = Nonemetadata_buffer_size: int = 10streaming_encoding: bool = Falseencoder_queue_maxsize: int = 30encoder_threads: int | None = Nonevideo_files_size_in_mb: int | None = Nonedata_files_size_in_mb: int | None = None )
参数
'{hf_user}/{dataset_name}'。$HF_LEROBOT_HOME/{repo_id}。True,视觉模态存储为 MP4 视频。
若为 False,则存储为图像。0 表示仅使用线程。1 表示立即编码。None 时,使用 rgb_encoder_defaults()。None 时,使用 depth_encoder_defaults()。None
让编解码器自行决定。True,在采集期间实时编码视频帧,而不是先写入图像。从头创建一个新的 LeRobotDataset 以采集数据。
返回一个带有活动 DatasetWriter 的写入模式 dataset。使用 add_frame / save_episode 填充它,完成后调用 finalize。
刷新所有待处理工作并关闭写入器。
必须在数据采集/转换之后调用,否则页脚元数据不会被写入 parquet 文件,dataset 将无效。
幂等——可安全多次调用。如果从未显式调用,DatasetWriter.del 将作为安全网。
获取未经图像变换的原始帧。
与 __getitem__ 不同,此方法返回给定索引处的原始 HF dataset 行,不进行增量时间戳扩展、视频解码或图像变换。
检查 episode 缓冲区中是否有未保存的帧。
( branch: str | None = Nonetags: list | None = Nonelicense: str | None = 'apache-2.0'tag_version: bool = Truepush_videos: bool = Trueprivate: bool | None = Noneallow_patterns: list[str] | str | None = Noneupload_large_folder: bool = False**card_kwargs )
参数
True,为当前代码库版本创建 Git 标签。False,跳过上传 videos/ 目录。True,创建私有仓库。若为 None(默认),则遵循 Hub 上的组织默认设置(仅影响组织)。True,对非常大的 dataset 使用 upload_large_folder 而非 upload_folder。将 dataset 上传到 Hugging Face Hub。
若仓库不存在则创建,上传所有 dataset 文件(可选排除视频),生成 dataset 卡片,并用当前代码库版本为修订打标签。
( repo_id: strroot: str | pathlib.Path | None = Nonetolerance_s: float = 0.0001revision: str | None = Noneforce_cache_sync: bool = Falsevideo_backend: str | None = Nonebatch_encoding_size: int = 1rgb_encoder: lerobot.configs.video.RGBEncoderConfig | None = Nonedepth_encoder: lerobot.configs.video.DepthEncoderConfig | None = Noneencoder_threads: int | None = Noneimage_writer_processes: int = 0image_writer_threads: int = 0streaming_encoding: bool = Falseencoder_queue_maxsize: int = 30token: str | bool | None = None )
参数
$HF_LEROBOT_HOME/hub 下的修订安全快照缓存。True,即使存在本地缓存,也重新从 Hub 下载元数据。None 时,使用 rgb_encoder_defaults()。None 时,使用 depth_encoder_defaults()。None 让编解码器自行决定。True,在采集期间实时编码视频。在现有 dataset 上继续采集。
从现有 dataset(本地或 Hub)加载元数据,并创建一个 DatasetWriter 用于追加新 episode。底层 HF dataset 在调用 finalize 并随后读取数据之前不会被加载。
( episode_data: dict | None = Noneparallel_encoding: bool = True )
将当前 episode 缓冲区保存到磁盘。
委托给 DatasetWriter.save_episode。编码视频、写入 parquet 数据并更新元数据。之后 episode 缓冲区会被重置。
从底层 dataset 选择特定列。
用于在重放期间提取 action sequence 而无需加载所有特征。返回仅包含所请求列的 datasets.Dataset。
替换应用于视觉 observation 的变换。
( repo_id: strroot: str | pathlib.Path | None = Nonerevision: str | None = Noneforce_cache_sync: bool = Falsemetadata_buffer_size: int = 10repo_type: typing.Literal['dataset', 'bucket'] = 'dataset'token: str | bool | None = None )
LeRobot dataset 的元数据容器。
管理描述 dataset 结构、内容和统计信息的 info.json、stats.json、tasks.parquet 和 episodes/ parquet 文件。
( repo_id: strfps: intfeatures: dictrobot_type: str | None = Noneroot: str | pathlib.Path | None = Noneuse_videos: bool = Truemetadata_buffer_size: int = 10chunks_size: int | None = Nonedata_files_size_in_mb: int | None = Nonevideo_files_size_in_mb: int | None = None )
参数
'user/my_dataset')。$HF_LEROBOT_HOME/{repo_id}。必须尚不存在。True,视觉模态编码为 MP4 视频。None 使用默认值。None 使用默认值。None 使用默认值。从头创建一个新的 LeRobot dataset 的元数据。
使用提供的特征模式和 dataset 设置在磁盘上初始化 info.json 文件。此时不写入任何 episode 数据。
确保为读取操作完全加载元数据。
幂等——当元数据已在内存中时,这只是一次 is None 检查。在同一实例上从写入模式切换到读取模式之前调用此方法。
( predicate: Callablecandidates: list[int] | None = None )
筛选元数据满足给定谓词的 episode。
获取当前的块和文件大小设置。
( ep_index: int )
返回给定 episode index 的相对 parquet 文件路径。
给定自然语言任务,如果该任务已存在于 dataset 中,则返回其 task_index,否则返回 None。
( ep_index: intvid_key: str )
返回给定 episode 和视频键的相对视频文件路径。
将深度特征统计从记录时的单位原地重缩放为 output_unit。
深度统计以帧记录时的单位(features[key]["info"]["depth_unit"])存储,而读取时帧以 output_unit 返回。此操作转换带单位的统计条目,使统计信息与消费者看到的帧一致。
( episode_index: intepisode_length: intepisode_tasks: listepisode_stats: dictepisode_metadata: dict )
持久化 episode 元数据,更新 dataset 信息并聚合统计信息。
将 episode 的元数据写入缓冲的 parquet 写入器,递增 info.json 中的总 episode/帧计数器,并将 episode 的统计信息合并到运行中的 dataset 统计信息中。
( tasks: list )
为当前 episode 注册任务并持久化到磁盘。
dataset 中尚不存在的新任务会被分配连续的任务索引,并追加到 tasks parquet 文件。
( chunks_size: int | None = Nonedata_files_size_in_mb: int | None = Nonevideo_files_size_in_mb: int | None = None )
在 dataset 创建后更新块和文件大小设置。
这允许用户自定义存储组织方式而无需修改构造函数。这些设置控制 episode 如何分块以及文件在创建新文件前可增长到多大。
( video_key: str | None = Nonevideo_encoder: lerobot.configs.video.VideoEncoderConfig | None = Nonepreserve_keys: collections.abc.Iterable[str] | None = None )
在 info.json 中填充或刷新每特征的视频信息。
警告:此函数从第一个 episode 的视频写入信息,隐含假设所有视频都以相同方式编码。此外,这意味着它假设第一个 episode 存在。
始终重新探测视频并覆盖每个重新计算的键的现有信息。preserve_keys 列出必须保留其现有值的键(例如 is_depth_map 和深度量化参数等数据固有条目),而不是重新计算。
( repo_ids: listroot: str | pathlib.Path | None = Noneepisodes: dict | None = Noneimage_transforms: collections.abc.Callable | None = Nonedelta_timestamps: dict[str, list[float]] | None = Nonetolerances_s: dict | None = Nonedownload_videos: bool = Truevideo_backend: str | None = Nonetoken: str | bool | None = None )
由多个底层 LeRobotDataset 组成的 dataset。
底层 LeRobotDataset 被有效地拼接在一起,此类采用了 LeRobotDataset 的大部分 API 结构。
移除此 dataset 及其子 dataset 的变换。
替换此 dataset 及其子 dataset 的变换。
( repo_id: strroot: str | pathlib.Path | None = Noneepisodes: list[int] | None = Noneimage_transforms: collections.abc.Callable | None = Nonedelta_timestamps: dict[list[float]] | None = Nonetolerance_s: float = 0.0001revision: str | None = Noneforce_cache_sync: bool = Falsestreaming: bool = Truebuffer_size: int = 1000max_num_shards: int = 16seed: int = 42rng: numpy.random._generator.Generator | None = Noneshuffle: bool = Truereturn_uint8: bool = Falsedepth_output_unit: str = 'mm'repo_type: typing.Literal['dataset', 'bucket'] = 'dataset'token: str | bool | None = None )
具有流式传输能力的 LeRobotDataset。
此类扩展了 LeRobotDataset 以添加流式传输功能,允许数据以流式方式处理,而非全部加载到内存中。这对于可能无法装入内存的大型 dataset,或希望在不完全下载的情况下快速浏览 dataset 时特别有用。
关键创新在于使用了可回溯迭代器(Backtrackable iterator),它维护一个有界的近期项缓冲区,使我们能够访问先前的帧以处理增量时间戳,而无需将整个 dataset 加载到内存中。
示例:
基本用法:
from lerobot.common.datasets.streaming_dataset import StreamingLeRobotDataset
# Create a streaming dataset with delta timestamps
delta_timestamps = {
"observation.image": [-1.0, -0.5, 0.0], # 1 sec ago, 0.5 sec ago, current
"action": [0.0, 0.1, 0.2], # current, 0.1 sec future, 0.2 sec future
}
dataset = StreamingLeRobotDataset(
repo_id="your-dataset-repo-id",
delta_timestamps=delta_timestamps,
streaming=True,
buffer_size=1000,
)
# Iterate over the dataset
for i, item in enumerate(dataset):
print(f"Sample {i}: Episode {item['episode_index']} Frame {item['frame_index']}")
# item will contain stacked frames according to delta_timestamps
if i >= 10:
break从 dataset 迭代器创建帧