flaminis/dalaran/dalaran-lerobot

Dalaran LeRobot ingestion

Dalaran has a built-in LeRobot importer: point log_file_from_path (or the viewer, or dalaran <dir> on the CLI) at the dataset _directory_ and it ingests episodes, camera videos, and state/action tables with no conversion code. There is no chunk-level LeRobotReader; the chunk-processing route is to import first, then reprocess the resulting DLR with RrdReader.

The download step needs huggingface_hub.

Step 1: dataset -> one combined DLR

from huggingface_hub import snapshot_download
import dalaran as rr
dataset_dir = snapshot_download(repo_id="dalaran/so101-pick-and-place", repo_type="dataset", local_dir=dest)
with rr.RecordingStream("dalaran_example_lerobot") as rec:
    rec.save(str(combined_dlr))
    rec.log_file_from_path(str(dataset_dir))  # the built-in importer

The importer emits one recording per episode (recording ids like episode_1), plus a metadata-only root recording, all into the single DLR.

rr.RecordingStream + log_file_from_path here is the importer bootstrap — the one place RecordingStream is correct in an ingestion pipeline (it drives the built-in importer, not per-message logging). Do not generalize it to rr.log-per-message loops; for everything after import, reprocess the DLR with RrdReader + lenses (see dalaran-chunk-processing: Chunk API vs logging API).

Step 2: split into per-episode RRDs

Catalog segments are one-recording-per-file, and recording_id becomes the segment id on registration. Split with RrdReader:

reader = rr.experimental.RrdReader(str(combined_dlr))
for entry in reader.recordings():
    store = reader.store(store=entry)
    if not store.schema().entity_paths():  # skip the metadata-only root recording
        continue
    episode_id = zero_pad(entry.recording_id)  # episode_1 -> episode_00001
    with rr.RecordingStream("dalaran_example_lerobot", recording_id=episode_id, send_properties=False) as rec:
        rec.save(str(dlr_dir / f"{episode_id}.dlr"))
        rec.send_chunks(store)

Two non-obvious moves:

• Zero-pad the episode id. episode_10 sorts before episode_2

lexicographically; segment tables and viewers sort lexicographically. Pad to a fixed width when re-assigning recording_id.

• `send_properties=False` on the new stream, so the copy doesn't inject

fresh recording properties on top of the copied chunks.

send_chunks does not preserve the source store's identity; the new stream's recording_id wins, which is exactly what makes the rename work.

If episodes need cleanup (drop topics, fix data, add derived components), run the store through lenses between read and write: reader.stream(store=entry).drop(...).lenses(...) then collect().write_dlr(..., recording_id=episode_id) (see dalaran-chunk-processing).

Computed layers and per-episode properties then follow the standard patterns in dalaran-data-model (layer recording_id must equal the episode segment id).

Gotchas

1. log_file_from_path must target the dataset root directory, not a file inside it.
2. Unpadded episode ids sort incorrectly downstream; pad before registering.
3. The combined DLR contains a metadata-only root recording; skip stores with no entity paths or you register an empty segment.

References

• https://github.com/Flaminis/Dalaran/tree/main/examples/python/dataloader prepare_dataset.py (download → import → split → register, complete and runnable) and train.py (training-side consumption via dalaran.experimental.dataloader)