Record a LeRobot Dataset

SUMMARY

Define a LeRobot dataset, create its logger, record one or more episodes, and finalize the dataset when collection is complete.

Recording Workflow

Recording a LeRobot dataset starts by defining what each frame contains and how frequently frames are collected. LeRobotDatasetLogger then manages the episode lifecycle while your application supplies observations, actions, and task information.

The steps below show how to define the dataset, create the logger, record one or more episodes, and finalize the recording.

1. Define Dataset

A. Repository ID and Local Path

Choose the dataset identity and local storage location before defining its contents.

import shutil
from pathlib import Path

repo_id = "user/my_record_example"
local_path = (
    Path(__file__).resolve().parent.parent.parent.parent.parent
    / "results"
    / repo_id
)
if local_path.exists():
    shutil.rmtree(local_path)

• repo_id identifies the dataset, usually as organization/dataset-name.
• local_path selects its local storage directory. When omitted, the default ~/.cache/telekinesis/lerobot/<repo_id> cache directory is used.

B. Feature Schema

The keys, shapes, types, and dimension names define the values accepted by log(frame).

features = {
    "observation.camera_rgb": {
        "dtype": "video",
        "shape": [3, 480, 640],
        "names": ["channel", "height", "width"],
    },
    "observation.state": {
        "dtype": "float32",
        "shape": [7],
        "names": [
            "shoulder_pan", "shoulder_lift", "elbow",
            "wrist_1", "wrist_2", "wrist_3", "gripper",
        ],
    },
    "action": {
        "dtype": "float32",
        "shape": [7],
        "names": [
            "shoulder_pan", "shoulder_lift", "elbow",
            "wrist_1", "wrist_2", "wrist_3", "gripper",
        ],
    },
}

See Feature Schema for the supported dtype values and their storage behavior.

C. Writer Configuration

Configure writer timing and choose the frame rate based on the acquisition speed:

from telekinesis.dataengine import datasets

config = datasets.LeRobotDatasetWriterConfig(
    tolerance_s=1e-4,
)
dataset_fps = 30

LeRobotDatasetWriterConfig controls writer and encoding behavior. Only values that differ from the defaults need to be specified. See the configuration reference for all available options.

2. Create the Logger

Use mode="create" for a new dataset. To overwrite an existing dataset, use mode="overwrite" to replace it.

from telekinesis.dataengine import data_loggers

lerobot_logger = data_loggers.LeRobotDatasetLogger(
    repo_id=repo_id,
    local_path=local_path,
    mode="create",
    fps=dataset_fps,
    features=features,
    robot_type="my_dummy_ur",
    config=config,
)

3. Record Episodes

Set the number of task attempts or demonstrations required by the collection run:

num_episodes = 5

Loop over the requested episodes, log frames at dataset_fps, and persist only successful episodes:

try:
    for episode_index in range(num_episodes):
        # Start episode
        lerobot_logger.start_episode()
        frame_index = 0

        try:
            while not task_complete(frame_index):
                frame_start = time.perf_counter()
                
                task = "Pick up the blue cube"
                frame = make_frame(task=task)
                # Log frame
                lerobot_logger.log(frame)
                
                frame_index += 1
                wait_for_next_frame(frame_start, dataset_fps)
        
        except (Exception, KeyboardInterrupt):
            lerobot_logger.discard_episode()
            logger.exception(
                "Episode recording failed. Current episode discarded."
            )
            raise
        # Stop episode
        else:   
            lerobot_logger.stop_episode()
            logger.info(f"Episode {episode_index} saved.")

except KeyboardInterrupt:
    logger.info("Stopping data collection.")

The example code below defines the frame-construction, task-completion, and FPS pacing helpers used by this loop.

4. Finalize Recording

Clean up an active episode first, then close the logger as the last operation:

finally:
    logger.info("Cleaning up active episode if any.")
    if lerobot_logger.episode_active:
        try:
            lerobot_logger.discard_episode()
        except Exception:
            logger.exception(
                "Failed to discard the active episode during cleanup."
            )

    lerobot_logger.close()
    logger.info("Logging complete.")

Handling Interrupted Recording

Discard incomplete episodes before finalizing the dataset. A finally block ensures cleanup also runs after an exception or keyboard interrupt:

try:
    lerobot_logger.start_episode()
    try:
        for frame in demonstration:
            lerobot_logger.log(frame)
    except (Exception, KeyboardInterrupt):
        lerobot_logger.discard_episode()
        raise
    else:
        lerobot_logger.stop_episode()
finally:
    if lerobot_logger.episode_active:
        lerobot_logger.discard_episode()
    lerobot_logger.close()

Class Reference

Parameter	Type	Default	Description
repo_id	str	required	Dataset repository identifier, typically "{hf_user}/{dataset_name}".
mode	Literal["create", "overwrite", "resume"]	"create"	Creates a dataset, replaces an existing local dataset, or resumes one.
local_path	str | Path | None	None	Local dataset directory; otherwise uses the default LeRobot cache path.
fps	int | None	None	Collection frame rate. Required in "create" and "overwrite" modes.
features	dict | None	None	Dataset feature schema. Required in "create" and "overwrite" modes.
robot_type	str | None	None	Optional robot type stored in the dataset metadata.
use_videos	bool	True	Whether to encode visual observations as videos.
config	LeRobotDatasetWriterConfig | None	None	Recording and encoding configuration; uses the default configuration when omitted.
force_cache_sync	bool	False	Refreshes existing metadata before resuming; unused when creating or overwriting.

Method Reference

Method	Purpose
start_episode()	Start a new episode.
log(frame)	Add one schema-complete frame to the active episode.
stop_episode(parallel_encoding=True)	Persist the active episode, optionally encoding multiple camera streams in parallel.
discard_episode()	Clear the active episode without saving it.
close()	Finalize the dataset when no episode is active.

Attribute Reference

Attribute	Type	Access	Description
episode_active	bool	Read-only	Whether an episode is currently being recorded.

Example

This script creates a LeRobot dataset, records synchronized frames across multiple episodes at the configured frame rate, and finalizes all writers. Replace the generated observations, actions, and termination condition with application-specific interfaces.

"""Example script demonstrating how to log a LeRobot dataset using the Telekinesis Data Engine."""

import shutil
import time
from pathlib import Path

import numpy as np
from loguru import logger

from telekinesis.dataengine import datasets, data_loggers


def record_lerobot_dataset_example():
    """Example function to record a LeRobot dataset."""

    # Step 1: Define Dataset
    repo_id = "user/my_record_example"
    local_path = (
        Path(__file__).resolve().parent.parent.parent.parent.parent
        / "results"
        / repo_id
    )
    if local_path.exists():
        shutil.rmtree(local_path)

    features = {
        "observation.camera_rgb": {
            "dtype": "video",
            "shape": [3, 480, 640],
            "names": ["channel", "height", "width"],
        },
        "observation.state": {
            "dtype": "float32",
            "shape": [7],
            "names": [
                "shoulder_pan", "shoulder_lift", "elbow",
                "wrist_1", "wrist_2", "wrist_3", "gripper",
            ],
        },
        "action": {
            "dtype": "float32",
            "shape": [7],
            "names": [
                "shoulder_pan", "shoulder_lift", "elbow",
                "wrist_1", "wrist_2", "wrist_3", "gripper",
            ],
        },
    }

    config = datasets.LeRobotDatasetWriterConfig(tolerance_s=1e-4)
    dataset_fps = 30

    # Step 2: Create the Logger
    lerobot_logger = data_loggers.LeRobotDatasetLogger(
        repo_id=repo_id,
        local_path=local_path,
        mode="create",
        fps=dataset_fps,
        features=features,
        robot_type="my_dummy_ur",
        config=config,
    )

    # Step 3: Record Episodes
    num_episodes = 5

    try:
        for episode_index in range(num_episodes):
            lerobot_logger.start_episode()
            frame_index = 0

            try:
                while not task_complete(frame_index):
                    frame_start = time.perf_counter()
                    task = "Pick up the blue cube"
                    frame = make_frame(task=task)
                    lerobot_logger.log(frame)
                    frame_index += 1
                    wait_for_next_frame(frame_start, dataset_fps)
            except (Exception, KeyboardInterrupt):
                lerobot_logger.discard_episode()
                logger.exception(
                    "Episode recording failed. Current episode discarded."
                )
                raise
            else:
                lerobot_logger.stop_episode()
                logger.info(f"Episode {episode_index} saved.")
    except KeyboardInterrupt:
        logger.info("Stopping data collection.")
    # Step 4: Finalize recording
    finally:
        logger.info("Cleaning up active episode if any.")
        if lerobot_logger.episode_active:
            try:
                lerobot_logger.discard_episode()
            except Exception:
                logger.exception(
                    "Failed to discard the active episode during cleanup."
                )

        lerobot_logger.close()
        logger.info("Logging complete.")


def read_observation_camera() -> np.ndarray:
    """Return the latest RGB camera observation."""
    return np.random.rand(3, 480, 640).astype("float32")


def read_observation_robot_state() -> np.ndarray:
    """Return the current robot state."""
    return np.random.rand(7).astype("float32")


def get_robot_action() -> np.ndarray:
    """Return the current action applied to the robot."""
    return np.random.rand(7).astype("float32")


def make_frame(task: str) -> dict:
    """Assemble a dataset frame from observations and an action."""
    return {
        "observation.camera_rgb": read_observation_camera(),
        "observation.state": read_observation_robot_state(),
        "action": get_robot_action(),
        "task": task,
    }


def task_complete(frame_index: int) -> bool:
    """Replace this with the application's task termination condition."""
    max_frames = 5
    return frame_index >= max_frames


def wait_for_next_frame(start_time: float, fps: int) -> None:
    """Wait until the next dataset frame should be recorded."""
    frame_period = 1.0 / fps
    elapsed = time.perf_counter() - start_time
    remaining = frame_period - elapsed

    if remaining > 0:
        time.sleep(remaining)


if __name__ == "__main__":
    record_lerobot_dataset_example()

Next Steps

Visualize the Dataset

Open a recorded LeRobot dataset in Rerun and inspect its synchronized observations and actions.

Visualize dataset →

Resume LeRobot Recording

Reopen an existing dataset and append additional demonstration episodes.

Continue recording →

Load a LeRobot Dataset

Open an existing local or remote LeRobot dataset and access its recorded frames.

Load dataset →

Inspect Dataset Metadata

Review the feature schema, frame rate, episodes, statistics, and storage metadata.

Inspect metadata →