dataset

LeRobotDataset 是每个 LeRobot 脚本读写所用的格式。它感知 episode,能够即时解码视频 observation,并可与 Hugging Face Hub 往返同步。

关于该格式及常用操作,请参阅 使用 LeRobotDataset,迁移请参阅 移植大型 dataset,CLI 请参阅 工具

LeRobotDataset

class lerobot.datasets.LeRobotDataset

< >

( 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 )

add_frame

< >

( frame: dict )

参数

  • frame — 将该帧的特征名称映射到其值的字典。 必须包含 'task' 键。Torch 张量会转换为 numpy。

抛出异常

RuntimeError

  • RuntimeError — 如果 dataset 为只读(无写入器)。

向当前 episode 缓冲区添加单个帧。

委托给 DatasetWriter.add_frame。dataset 必须处于写入模式(通过 createresume 创建)。

clear_episode_buffer

< >

( delete_images: bool = True )

参数

  • delete_images — 若为 True,同时删除为当前 episode 写入磁盘的临时图像文件。

抛出异常

RuntimeError

  • RuntimeError — 如果 dataset 为只读(无写入器)。

丢弃当前 episode 缓冲区而不保存。

委托给 DatasetWriter.clear_episode_buffer。用于丢弃失败或中断的采集 episode。

clear_image_transforms

< >

( )

移除应用于视觉 observation 的变换。

create

< >

( 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 )

参数

  • repo_id — 仓库标识符,通常为 '{hf_user}/{dataset_name}'
  • fps — 数据采集时使用的每秒帧数。
  • features — 将特征名称映射到其类型/形状元数据的特征规格字典。
  • root — dataset 存储的本地目录。默认为 $HF_LEROBOT_HOME/{repo_id}
  • robot_type — 存储在元数据中的可选机器人类型字符串。
  • use_videos — 若为 True,视觉模态存储为 MP4 视频。 若为 False,则存储为图像。
  • tolerance_s — 时间戳同步容差,单位为秒。
  • image_writer_processes — 异步图像写入的子进程数。0 表示仅使用线程。
  • image_writer_threads — 异步图像写入的线程数。
  • video_backend — 视频解码后端(在回读时使用)。
  • batch_encoding_size — 在批量编码视频之前累积的 episode 数。1 表示立即编码。
  • rgb_encoder — 相机的视频编码器设置(编解码器、质量等)。 当为 None 时,使用 rgb_encoder_defaults()
  • depth_encoder — 深度相机的视频编码器设置(编解码器、质量等)。 当为 None 时,使用 depth_encoder_defaults()
  • encoder_threads — 编码器线程数(全局)。None 让编解码器自行决定。
  • metadata_buffer_size — 在刷新到 parquet 之前缓冲的 episode 元数据记录数。
  • streaming_encoding — 若为 True,在采集期间实时编码视频帧,而不是先写入图像。
  • encoder_queue_maxsize — 使用流式编码时每台相机的最大缓冲帧数。

从头创建一个新的 LeRobotDataset 以采集数据。

返回一个带有活动 DatasetWriter 的写入模式 dataset。使用 add_frame / save_episode 填充它,完成后调用 finalize

finalize

< >

( )

刷新所有待处理工作并关闭写入器。

必须在数据采集/转换之后调用,否则页脚元数据不会被写入 parquet 文件,dataset 将无效。

幂等——可安全多次调用。如果从未显式调用,DatasetWriter.del 将作为安全网。

get_raw_item

< >

( idx )

获取未经图像变换的原始帧。

__getitem__ 不同,此方法返回给定索引处的原始 HF dataset 行,不进行增量时间戳扩展、视频解码或图像变换。

has_pending_frames

< >

( )

检查 episode 缓冲区中是否有未保存的帧。

push_to_hub

< >

( 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 )

参数

  • branch — 可选的推送目标分支。若不存在,则从当前修订创建。
  • tags — dataset 卡片使用的可选标签列表。
  • license — dataset 卡片的许可证标识符。
  • tag_version — 若为 True,为当前代码库版本创建 Git 标签。
  • push_videos — 若为 False,跳过上传 videos/ 目录。
  • private — 若为 True,创建私有仓库。若为 None(默认),则遵循 Hub 上的组织默认设置(仅影响组织)。
  • allow_patterns — 限制上传哪些文件的 glob 模式。
  • upload_large_folder — 若为 True,对非常大的 dataset 使用 upload_large_folder 而非 upload_folder
  • **card_kwargs — 转发给 dataset 卡片创建的额外关键字参数。

将 dataset 上传到 Hugging Face Hub。

若仓库不存在则创建,上传所有 dataset 文件(可选排除视频),生成 dataset 卡片,并用当前代码库版本为修订打标签。

resume

< >

( 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 )

参数

  • repo_id — 现有 dataset 的仓库标识符。
  • root — dataset 的本地目录。提供时,Hub 下载会直接落到该目录。省略时,Hub 下载使用 $HF_LEROBOT_HOME/hub 下的修订安全快照缓存。
  • tolerance_s — 时间戳同步容差,单位为秒。
  • revision — Git 修订(分支、标签或提交哈希)。默认为当前代码库版本标签。
  • force_cache_sync — 若为 True,即使存在本地缓存,也重新从 Hub 下载元数据。
  • video_backend — 用于回读数据的视频解码后端。
  • batch_encoding_size — 在批量编码视频之前累积的 episode 数。
  • rgb_encoder — 相机的视频编码器设置(编解码器、质量等)。当为 None 时,使用 rgb_encoder_defaults()
  • depth_encoder — 深度相机的视频编码器设置(编解码器、质量等)。当为 None 时,使用 depth_encoder_defaults()
  • encoder_threads — 编码器线程数(全局)。None 让编解码器自行决定。
  • image_writer_processes — 异步图像写入的子进程。
  • image_writer_threads — 异步图像写入的线程。
  • streaming_encoding — 若为 True,在采集期间实时编码视频。
  • encoder_queue_maxsize — 流式处理时每台相机的最大缓冲帧数。
  • token — 当必须从 Hub 下载元数据时使用的认证令牌。令牌不会保留在 dataset 实例上。

在现有 dataset 上继续采集。

从现有 dataset(本地或 Hub)加载元数据,并创建一个 DatasetWriter 用于追加新 episode。底层 HF dataset 在调用 finalize 并随后读取数据之前不会被加载。

save_episode

< >

( episode_data: dict | None = Noneparallel_encoding: bool = True )

参数

  • episode_data — 可选的预构建 episode 字典。若为 None,则使用由 add_frame 填充的内部 episode 缓冲区。
  • parallel_encoding — 若为 True 且存在多个相机,则使用进程池并行编码视频。

抛出异常

RuntimeError

  • RuntimeError — 如果 dataset 为只读(无写入器)。

将当前 episode 缓冲区保存到磁盘。

委托给 DatasetWriter.save_episode。编码视频、写入 parquet 数据并更新元数据。之后 episode 缓冲区会被重置。

select_columns

< >

( column_names: str | list[str] )

从底层 dataset 选择特定列。

用于在重放期间提取 action sequence 而无需加载所有特征。返回仅包含所请求列的 datasets.Dataset

set_image_transforms

< >

( image_transforms: collections.abc.Callable | None )

替换应用于视觉 observation 的变换。

LeRobotDatasetMetadata

class lerobot.datasets.LeRobotDatasetMetadata

< >

( 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.jsonstats.jsontasks.parquetepisodes/ parquet 文件。

create

< >

( 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 )

参数

  • repo_id — 仓库标识符(例如 'user/my_dataset')。
  • fps — 数据采集时使用的每秒帧数。
  • features — 将特征名称映射到其类型/形状元数据的特征规格字典。
  • robot_type — 存储在元数据中的可选机器人类型字符串。
  • root — dataset 的本地目录。默认为 $HF_LEROBOT_HOME/{repo_id}。必须尚不存在。
  • use_videos — 若为 True,视觉模态编码为 MP4 视频。
  • metadata_buffer_size — 在刷新到 parquet 之前缓冲的 episode 元数据记录数。
  • chunks_size — 每个块目录的最大文件数。None 使用默认值。
  • data_files_size_in_mb — parquet 文件最大大小,单位为 MB。None 使用默认值。
  • video_files_size_in_mb — 视频文件最大大小,单位为 MB。None 使用默认值。

从头创建一个新的 LeRobot dataset 的元数据。

使用提供的特征模式和 dataset 设置在磁盘上初始化 info.json 文件。此时不写入任何 episode 数据。

ensure_readable

< >

( )

确保为读取操作完全加载元数据。

幂等——当元数据已在内存中时,这只是一次 is None 检查。在同一实例上从写入模式切换到读取模式之前调用此方法。

filter_episodes

< >

( predicate: Callablecandidates: list[int] | None = None )

参数

  • predicate — 用于选择 episode 的每 episode 元数据行谓词。
  • candidates — 可选的 episode index 列表,用于限制评估范围。

筛选元数据满足给定谓词的 episode。

finalize

< >

( )

刷新元数据缓冲区并关闭 parquet 写入器。

幂等——可安全多次调用。

get_chunk_settings

< >

( )

获取当前的块和文件大小设置。

get_data_file_path

< >

( ep_index: int )

参数

  • ep_index — 从零开始的 episode index。

抛出异常

IndexError

  • IndexError — 如果 ep_index 超出范围。

返回给定 episode index 的相对 parquet 文件路径。

get_task_index

< >

( task: str )

给定自然语言任务,如果该任务已存在于 dataset 中,则返回其 task_index,否则返回 None。

get_video_file_path

< >

( ep_index: intvid_key: str )

参数

  • ep_index — 从零开始的 episode index。
  • vid_key — 标识视频流的特征键(例如 'observation.images.laptop')。

抛出异常

IndexError

  • IndexError — 如果 ep_index 超出范围。

返回给定 episode 和视频键的相对视频文件路径。

rescale_depth_stats

< >

( output_unit: str )

将深度特征统计从记录时的单位原地重缩放为 output_unit

深度统计以帧记录时的单位(features[key]["info"]["depth_unit"])存储,而读取时帧以 output_unit 返回。此操作转换带单位的统计条目,使统计信息与消费者看到的帧一致。

save_episode

< >

( episode_index: intepisode_length: intepisode_tasks: listepisode_stats: dictepisode_metadata: dict )

参数

  • episode_index — 正在保存的 episode 的从零开始索引。
  • episode_length — 此 episode 中的帧数。
  • episode_tasks — 此 episode 的任务描述列表。
  • episode_stats — 此 episode 的每特征统计信息。
  • episode_metadata — 额外元数据(块/文件索引、帧范围、视频时间戳等)。

持久化 episode 元数据,更新 dataset 信息并聚合统计信息。

将 episode 的元数据写入缓冲的 parquet 写入器,递增 info.json 中的总 episode/帧计数器,并将 episode 的统计信息合并到运行中的 dataset 统计信息中。

save_episode_tasks

< >

( tasks: list )

参数

  • tasks — 自然语言的唯一任务描述列表。

抛出异常

ValueError

  • ValueError — 如果 tasks 包含重复项。

为当前 episode 注册任务并持久化到磁盘。

dataset 中尚不存在的新任务会被分配连续的任务索引,并追加到 tasks parquet 文件。

update_chunk_settings

< >

( chunks_size: int | None = Nonedata_files_size_in_mb: int | None = Nonevideo_files_size_in_mb: int | None = None )

参数

  • chunks_size — 每个块目录的最大文件数。若为 None,则保留当前值。
  • data_files_size_in_mb — 数据 parquet 文件的最大大小,单位为 MB。若为 None,则保留当前值。
  • video_files_size_in_mb — 视频文件的最大大小,单位为 MB。若为 None,则保留当前值。

在 dataset 创建后更新块和文件大小设置。

这允许用户自定义存储组织方式而无需修改构造函数。这些设置控制 episode 如何分块以及文件在创建新文件前可增长到多大。

update_video_info

< >

( video_key: str | None = Nonevideo_encoder: lerobot.configs.video.VideoEncoderConfig | None = Nonepreserve_keys: collections.abc.Iterable[str] | None = None )

参数

  • video_key — 若提供,则仅更新此视频键。否则更新 dataset 中的所有视频键。
  • video_encoder — 用于生成视频的编码器配置。提供时,其字段会与从流派生的 video.* 条目一起记录为 video.<field> 条目(参见 get_video_info)。
  • preserve_keys — 保留其现有值而非重新计算的键。None(默认)重新计算每个键。

info.json 中填充或刷新每特征的视频信息。

警告:此函数从第一个 episode 的视频写入信息,隐含假设所有视频都以相同方式编码。此外,这意味着它假设第一个 episode 存在。

始终重新探测视频并覆盖每个重新计算的键的现有信息。preserve_keys 列出必须保留其现有值的键(例如 is_depth_map 和深度量化参数等数据固有条目),而不是重新计算。

MultiLeRobotDataset

class lerobot.datasets.MultiLeRobotDataset

< >

( 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 结构。

clear_image_transforms

< >

( )

移除此 dataset 及其子 dataset 的变换。

set_image_transforms

< >

( image_transforms: collections.abc.Callable | None )

替换此 dataset 及其子 dataset 的变换。

StreamingLeRobotDataset

class lerobot.datasets.StreamingLeRobotDataset

< >

( 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

make_frame

< >

( dataset_iterator: Backtrackable )

从 dataset 迭代器创建帧

在 GitHub 上更新