lerobot

lerobot

Dataset schema for LeRobot dataset V3.0.

LeRobot is a HuggingFace robotics dataset format using: - Parquet files for tabular data (observation.state, action, timestamps) - MP4 videos or embedded PNG bytes for camera observations - JSON metadata files (info.json, stats.json, tasks.jsonl/tasks.parquet)

Reference: https://github.com/huggingface/lerobot

LeRobotDatasetMetadataV30

LeRobotDatasetMetadataV30(dataset_path)

Metadata handler for LeRobot dataset format V3.0.

This class parses and provides access to LeRobot dataset metadata, including: - info.json: Dataset configuration (version, paths, features, etc.) - tasks.parquet: Task descriptions indexed by task_index - episodes/*.parquet: Per-episode metadata (chunk/file indices, video timestamps)

The LeRobot V3.0 format organizes data in chunks for efficient storage: - Data files: data/chunk-{chunk_index}/episode_{file_index}.parquet - Videos: videos/chunk-{chunk_index}/{video_key}episode.mp4

Attributes:

Name	Type	Description
dataset_path		Root path to the LeRobot dataset directory.
info		Parsed info.json containing dataset configuration.
tasks		DataFrame of task descriptions from tasks.parquet.
episodes		Concatenated DataFrame of all episode metadata.

Source code in src/versatil/data/raw/schemas/lerobot.py

def __init__(self, dataset_path: str | Path):
    self.dataset_path = Path(dataset_path)
    self.info = self._load_info()
    self.tasks = pq.read_table(
        self.dataset_path / LeRobotPathsV30.DEFAULT_TASKS_PATH
    )
    self.episodes = self._load_episodes()

get_version

get_version()

Return the LeRobot codebase version recorded in the dataset info.

Source code in src/versatil/data/raw/schemas/lerobot.py

def get_version(self) -> str:
    """Return the LeRobot codebase version recorded in the dataset info."""
    return self.info["codebase_version"]

get_data_file_path

get_data_file_path(episode_index)

Get the path to the parquet data file for a specific episode.

LeRobot stores episode data in chunked parquet files. This method resolves the chunk and file indices from episode metadata.

Parameters:

Name	Type	Description	Default
episode_index	int	Index of the episode to locate.	required

Returns:

Type	Description
Path	Absolute path to the episode's parquet data file.

Source code in src/versatil/data/raw/schemas/lerobot.py

def get_data_file_path(self, episode_index: int) -> Path:
    """Get the path to the parquet data file for a specific episode.

    LeRobot stores episode data in chunked parquet files. This method
    resolves the chunk and file indices from episode metadata.

    Args:
        episode_index: Index of the episode to locate.

    Returns:
        Absolute path to the episode's parquet data file.
    """
    episode = self.get_episode_meta(episode_index)
    chunk_idx = episode["data/chunk_index"][0].as_py()
    file_idx = episode["data/file_index"][0].as_py()
    file_path = self.info["data_path"].format(
        chunk_index=chunk_idx, file_index=file_idx
    )
    return self.dataset_path / Path(file_path)

get_image_file_path

get_image_file_path(episode_index, image_key, frame_index)

Get the path to a specific image file.

Used when images are stored as individual files on disk rather than embedded in parquet or encoded as video.

Parameters:

Name	Type	Description	Default
episode_index	int	Index of the episode.	required
image_key	str	Camera/image stream identifier (e.g., "observation.images.top").	required
frame_index	int	Frame number within the episode.	required

Returns:

Type	Description
Path	Absolute path to the image file.

Source code in src/versatil/data/raw/schemas/lerobot.py

def get_image_file_path(
    self, episode_index: int, image_key: str, frame_index: int
) -> Path:
    """Get the path to a specific image file.

    Used when images are stored as individual files on disk rather than
    embedded in parquet or encoded as video.

    Args:
        episode_index: Index of the episode.
        image_key: Camera/image stream identifier (e.g., "observation.images.top").
        frame_index: Frame number within the episode.

    Returns:
        Absolute path to the image file.
    """
    file_path = LeRobotPathsV30.DEFAULT_IMAGE_PATH.format(
        image_key=image_key, episode_index=episode_index, frame_index=frame_index
    )
    return self.dataset_path / Path(file_path)

get_video_file_path

get_video_file_path(episode_index, video_key)

Get the path to a video file for a specific episode and camera.

Videos are stored in chunks similar to data files. This method resolves the chunk and file indices from episode metadata.

Parameters:

Name	Type	Description	Default
episode_index	int	Index of the episode.	required
video_key	str	Camera/video stream identifier (e.g., "observation.images.top").	required

Returns:

Type	Description
Path	Absolute path to the video file (typically MP4).

Source code in src/versatil/data/raw/schemas/lerobot.py

def get_video_file_path(self, episode_index: int, video_key: str) -> Path:
    """Get the path to a video file for a specific episode and camera.

    Videos are stored in chunks similar to data files. This method
    resolves the chunk and file indices from episode metadata.

    Args:
        episode_index: Index of the episode.
        video_key: Camera/video stream identifier (e.g., "observation.images.top").

    Returns:
        Absolute path to the video file (typically MP4).
    """
    episode = self.get_episode_meta(episode_index)
    chunk_idx = episode[f"videos/{video_key}/chunk_index"][0].as_py()
    file_idx = episode[f"videos/{video_key}/file_index"][0].as_py()
    file_path = self.info["video_path"].format(
        video_key=video_key, chunk_index=chunk_idx, file_index=file_idx
    )
    return self.dataset_path / Path(file_path)

get_episode_meta

get_episode_meta(episode_index)

Get metadata for a specific episode.

Parameters:

Name	Type	Description	Default
episode_index	int	Index of the episode to retrieve.	required

Returns:

Type	Description
Table	PyArrow Table row(s) containing the episode's metadata.

Source code in src/versatil/data/raw/schemas/lerobot.py

def get_episode_meta(self, episode_index: int) -> pa.Table:
    """Get metadata for a specific episode.

    Args:
        episode_index: Index of the episode to retrieve.

    Returns:
        PyArrow Table row(s) containing the episode's metadata.
    """
    mask = pa.compute.equal(self.episodes["episode_index"], episode_index)
    return self.episodes.filter(mask)

get_features

get_features()

Get the feature schema definition from info.json.

Features describe the data columns and their types (e.g., state, action, image, video).

Returns:

Type	Description
dict[str, Any]	Dictionary mapping feature names to their specifications.

Source code in src/versatil/data/raw/schemas/lerobot.py

def get_features(self) -> dict[str, Any]:
    """Get the feature schema definition from info.json.

    Features describe the data columns and their types (e.g., state, action,
    image, video).

    Returns:
        Dictionary mapping feature names to their specifications.
    """
    return self.info["features"]

get_image_keys

get_image_keys()

Get feature keys for image-type observations.

Returns:

Type	Description
list[str]	List of image key names (e.g., ["observation.images.top"]).

Source code in src/versatil/data/raw/schemas/lerobot.py

def get_image_keys(self) -> list[str]:
    """Get feature keys for image-type observations.

    Returns:
        List of image key names (e.g., ["observation.images.top"]).
    """
    return [
        key for key, ft in self.get_features().items() if ft["dtype"] == "image"
    ]

get_video_keys

get_video_keys()

Get feature keys for video-type observations.

Returns:

Type	Description
list[str]	List of video key names (e.g., ["observation.images.top"]).

Source code in src/versatil/data/raw/schemas/lerobot.py

def get_video_keys(self) -> list[str]:
    """Get feature keys for video-type observations.

    Returns:
        List of video key names (e.g., ["observation.images.top"]).
    """
    return [
        key for key, ft in self.get_features().items() if ft["dtype"] == "video"
    ]

get_total_episodes

get_total_episodes()

Return the number of episodes recorded in the dataset info.

Source code in src/versatil/data/raw/schemas/lerobot.py

def get_total_episodes(self) -> int:
    """Return the number of episodes recorded in the dataset info."""
    return self.info["total_episodes"]

LeRobotDatasetSchemaV30

LeRobotDatasetSchemaV30(dataset_path, zarr_path, metadata, dataset_type)

Bases: DatasetSchema

Dataset schema for converting LeRobot V3.0 datasets to Zarr format.

This schema handles the conversion of LeRobot datasets (stored as parquet files with MP4 videos or PNG images) into the Zarr format. It supports: - Video-based observations (MP4 files decoded at specific timestamps) - Image-based observations (PNG files or embedded bytes in parquet) - State/action vectors from parquet columns - Language instructions from task descriptions

The conversion flow: 1. Load episode data from parquet files 2. Extract frames from videos or load images 3. Map LeRobot features to observation/action keys 4. Apply resizing transforms to images 5. Return data dictionary ready for Zarr storage

Attributes:

Name	Type	Description
dataset_path		Root path to the LeRobot dataset directory.
lerobot_metadata	LeRobotDatasetMetadataV30	Metadata handler for accessing dataset structure, loaded lazily on first conversion-time access.
dataset_type		String with the dataset type used by the schema (e.g. libero, metaworld, etc.)

Source code in src/versatil/data/raw/schemas/lerobot.py

def __init__(
    self,
    dataset_path: str,
    zarr_path: str,
    metadata: DatasetMetadata,
    dataset_type: str,
):
    self.dataset_path = Path(dataset_path)
    super().__init__(
        zarr_path=zarr_path, metadata=metadata, dataset_type=dataset_type
    )
    self._lerobot_metadata: LeRobotDatasetMetadataV30 | None = None

lerobot_metadata property writable

lerobot_metadata

Raw LeRobot metadata, loaded on first access.

Lazy loading keeps zarr-only workflows independent of the raw dataset directory: when the zarr store already exists, conversion never runs and the raw data does not need to be present.

get_episode_parquet

get_episode_parquet(episode_id)

Load and filter the parquet table rows belonging to one episode.

Source code in src/versatil/data/raw/schemas/lerobot.py

def get_episode_parquet(self, episode_id: int) -> pa.Table:
    """Load and filter the parquet table rows belonging to one episode."""
    table = pq.read_table(self.lerobot_metadata.get_data_file_path(episode_id))
    mask = pa.compute.equal(table["episode_index"], episode_id)
    return table.filter(mask)

get_episode_videos_frames

get_episode_videos_frames(episode_id, preloaded_episode_table=None)

Extract all video frames for an episode.

Parameters:

Name	Type	Description	Default
episode_id	int	Index of the episode.	required
preloaded_episode_table	Table	Optional pre-loaded episode Table to avoid redundant parquet reads.	None

Returns:

Type	Description
dict[str, list[ndarray]]	Dictionary mapping video keys to lists of cv2 images (numpy arrays), one per timestep.
dict[str, list[ndarray]]	Returns empty dict if dataset has no video features.

Source code in src/versatil/data/raw/schemas/lerobot.py

def get_episode_videos_frames(
    self, episode_id: int, preloaded_episode_table: pa.Table = None
) -> dict[str, list[np.ndarray]]:
    """Extract all video frames for an episode.

    Args:
        episode_id: Index of the episode.
        preloaded_episode_table: Optional pre-loaded episode Table to avoid
            redundant parquet reads.

    Returns:
        Dictionary mapping video keys to lists of cv2 images (numpy arrays), one per timestep.
        Returns empty dict if dataset has no video features.
    """
    episode_table = (
        preloaded_episode_table
        if preloaded_episode_table is not None
        else self.get_episode_parquet(episode_id)
    )

    video_keys = self.lerobot_metadata.get_video_keys()

    frames = {}
    if video_keys:
        # Use episode timestamps to query video frames
        timestamps = episode_table["timestamp"].to_pylist()
        query_timestamps = dict.fromkeys(video_keys, timestamps)
        frames = self._get_frames_from_videos(query_timestamps, episode_id)

    return frames

get_episode_images

get_episode_images(episode_id, preloaded_episode_table=None)

Extract all images for an episode.

Images in LeRobot can be stored in three ways: 1. Embedded as bytes in the parquet file (encoded_image['bytes']) 2. As paths in the parquet file pointing to external files (encoded_image['path']) 3. As separate files on disk (loaded via _get_images_from_filesystem)

Parameters:

Name	Type	Description	Default
episode_id	int	Index of the episode.	required
preloaded_episode_table	Table	Optional pre-loaded episode Table to avoid redundant parquet reads.	None

Returns:

Type	Description
dict[str, list[ndarray]]	Dictionary mapping image keys to lists of cv2 images (numpy arrays), one per timestep.
dict[str, list[ndarray]]	Returns empty dict if dataset has no image features.

Source code in src/versatil/data/raw/schemas/lerobot.py

def get_episode_images(
    self, episode_id: int, preloaded_episode_table: pa.Table = None
) -> dict[str, list[np.ndarray]]:
    """Extract all images for an episode.

    Images in LeRobot can be stored in three ways:
    1. Embedded as bytes in the parquet file (encoded_image['bytes'])
    2. As paths in the parquet file pointing to external files (encoded_image['path'])
    3. As separate files on disk (loaded via _get_images_from_filesystem)

    Args:
        episode_id: Index of the episode.
        preloaded_episode_table: Optional pre-loaded episode Table to avoid
            redundant parquet reads.

    Returns:
        Dictionary mapping image keys to lists of cv2 images (numpy arrays), one per timestep.
        Returns empty dict if dataset has no image features.
    """
    episode_table = (
        preloaded_episode_table
        if preloaded_episode_table is not None
        else self.get_episode_parquet(episode_id)
    )

    image_keys = self.lerobot_metadata.get_image_keys()
    images = {}

    if image_keys:
        frame_indexes = episode_table["frame_index"].to_pylist()
        episode_cols = episode_table.column_names

        for image_key in image_keys:
            frames = []

            if image_key in episode_cols:
                # Images are embedded in parquet (as bytes or paths)
                encoded_images = episode_table[image_key].to_pylist()
                for encoded_image in encoded_images:
                    if "bytes" in encoded_image:
                        img = cv2.imdecode(
                            np.frombuffer(encoded_image["bytes"], np.uint8),
                            cv2.IMREAD_COLOR,
                        )
                        img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
                        frames.append(img)

                    else:
                        img_path = self.dataset_path / Path(encoded_image["path"])
                        img = cv2.imread(str(img_path))
                        img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
                        frames.append(img)
            else:
                # Images stored as separate files on disk
                frames = self._get_images_from_filesystem(
                    episode_id, image_key, frame_indexes
                )

            images[image_key] = frames

    return images

get_episode_language_instructions

get_episode_language_instructions(episode_id, preloaded_episode_table=None)

Return per-frame language instructions for one episode.

Source code in src/versatil/data/raw/schemas/lerobot.py

def get_episode_language_instructions(
    self, episode_id: int, preloaded_episode_table: pa.Table = None
) -> list[list[str]]:
    """Return per-frame language instructions for one episode."""
    episode_table = (
        preloaded_episode_table
        if preloaded_episode_table is not None
        else self.get_episode_parquet(episode_id)
    )

    # Map task indices to task names (language instructions)
    task_names = self.lerobot_metadata.tasks[1].to_pylist()
    language_instructions = [
        [task_names[i]] for i in episode_table["task_index"].to_pylist()
    ]

    return language_instructions

extract_episode

extract_episode(episode_id)

Extract and convert a complete episode to zarr format.

Parameters:

Name	Type	Description	Default
episode_id	int	Index of the episode to extract. (currently unused but kept for API consistency with base class).	required

Returns:

Type	Description
dict[str, ndarray]	Dictionary mapping Zarr keys to numpy arrays ready for storage.
dict[str, ndarray]	Keys correspond to observations and precomputed_actions defined
dict[str, ndarray]	in the DatasetMetadata.

Source code in src/versatil/data/raw/schemas/lerobot.py

def extract_episode(
    self,
    episode_id: int,
) -> dict[str, np.ndarray]:
    """Extract and convert a complete episode to zarr format.

    Args:
        episode_id: Index of the episode to extract.                (currently unused but kept for API consistency with base class).

    Returns:
        Dictionary mapping Zarr keys to numpy arrays ready for storage.
        Keys correspond to observations and precomputed_actions defined
        in the DatasetMetadata.
    """
    # Load all raw data for this episode
    episode_table = self.get_episode_parquet(episode_id)
    language_instructions = self.get_episode_language_instructions(
        episode_id, episode_table
    )
    videos = self.get_episode_videos_frames(episode_id, episode_table)
    images = self.get_episode_images(episode_id, episode_table)
    # Merge video and image frames (video keys take precedence if overlap)
    frames = videos | images

    data = {}

    # Process observations based on their type
    for zarr_key, obs in self.metadata.observations.items():
        if zarr_key == ObsKey.LANGUAGE.value:
            # Language instructions are stored as string arrays
            data[zarr_key] = np.array(language_instructions)

        elif isinstance(obs, CameraMetadata):
            # Camera observations: extract frames and resize
            camera_key = obs.raw_camera_key

            if camera_key not in frames:
                raise ValueError(
                    f"The camera key: {camera_key} does not exist in the dataset."
                )

            camera_resizer = build_camera_resizer(camera_metadata=obs)
            resized_frames = [
                camera_resizer(image=np.array(f))["image"]
                for f in frames[camera_key]
            ]
            data[zarr_key] = np.stack(resized_frames).astype(obs.dtype)

        else:
            # Vector observations (proprioceptive state, etc.)
            col_key = obs.raw_data_column_keys[0]

            if col_key not in episode_table.column_names:
                raise ValueError(
                    f"The column {col_key} does not exist in the dataset."
                )

            obs_array = np.stack(episode_table[col_key].to_pylist())
            # Apply optional slicing to extract specific dimensions
            if obs.slice_start is not None and obs.slice_end is not None:
                obs_array = obs_array[:, obs.slice_start : obs.slice_end]
            data[zarr_key] = obs_array.astype(obs.dtype)

    # Process precomputed actions (pre-computed in LeRobot format)
    for zarr_key, action in self.metadata.precomputed_actions.items():
        col_key = action.raw_data_column_keys[0]

        if col_key not in episode_table.column_names:
            raise ValueError(f"The column {col_key} does not exist in the dataset.")

        action_array = np.stack(episode_table[col_key].to_pylist())
        # Apply optional slicing to extract specific action dimensions
        if action.slice_start is not None and action.slice_end is not None:
            action_array = action_array[:, action.slice_start : action.slice_end]
        data[zarr_key] = action_array.astype(action.dtype)

    return data

decode_video_frames

decode_video_frames(video_path, timestamps, tolerance_s=0.01)

Load frames from a video file at specified timestamps using PyAV.

Returns:

Type	Description
list[ndarray]	List of RGB numpy arrays (uint8) corresponding to each requested timestamp.

Source code in src/versatil/data/raw/schemas/lerobot.py

def decode_video_frames(
    video_path: Path,
    timestamps: list[float],
    tolerance_s: float = 0.01,
) -> list[np.ndarray]:
    """Load frames from a video file at specified timestamps using PyAV.

    Returns:
        List of RGB numpy arrays (uint8) corresponding to each requested timestamp.
    """
    container = av.open(str(video_path))
    stream = container.streams.video[0]

    # Ensure timestamps are sorted for efficient seeking
    sorted_indices = np.argsort(timestamps)
    sorted_timestamps = [timestamps[i] for i in sorted_indices]

    loaded_frames = [None] * len(timestamps)

    for sorted_pos, timestamp in enumerate(sorted_timestamps):
        # Seek close to requested timestamp (in stream time_base units)
        seek_ts = int(timestamp / stream.time_base)
        container.seek(seek_ts, any_frame=False, backward=True, stream=stream)

        frame_found = False

        for frame in container.decode(stream):
            actual_timestamp_s = float(frame.pts * stream.time_base)

            if actual_timestamp_s + tolerance_s < timestamp:
                continue

            if abs(actual_timestamp_s - timestamp) > tolerance_s:
                raise RuntimeError(
                    f"Timestamp tolerance exceeded: requested={timestamp:.4f}, "
                    f"got={actual_timestamp_s:.4f}, video={video_path}"
                )

            img_rgb = frame.to_ndarray(format="rgb24")
            original_index = sorted_indices[sorted_pos]
            loaded_frames[original_index] = img_rgb
            frame_found = True
            break

        if not frame_found:
            raise RuntimeError(
                f"Failed to read frame at timestamp={timestamp:.4f}s "
                f"from video: {video_path}"
            )

    container.close()

    return loaded_frames