feat(examples): add Daft-native LeRobot and DROID reading scripts

examples/README.md

@@ -4,6 +4,25 @@ Three complete walkthroughs, generated by the `daft-physical-ai` CLI. Hand
 tracking lives in [hands/](hands/); reward scoring in [rewards/](rewards/);
 motion trimming in [trim/](trim/).
 
+Daft already reads robot data. These scripts are the copyable pattern per
+format - no extra package surface:
+
+- [`lerobot_episode_index.py`](lerobot_episode_index.py) - episode / task /
+  frame views of a LeRobot v3 dataset, filtered without decoding video
+- [`merge_lerobot_datasets.py`](merge_lerobot_datasets.py) - re-index
+  `episode_index` and the global frame `index` before concatenating two
+  recording sessions
+- [`droid_episode_index.py`](droid_episode_index.py) - lazy DROID episode
+  index: successful episodes, projected columns, plan only
+- [`egodex_raw_hdf5_video.py`](egodex_raw_hdf5_video.py) - raw EgoDex HDF5 +
+  video through `daft_physical_ai.datasets.egodex`
+
+```bash
+uv run python examples/lerobot_episode_index.py
+uv run python examples/merge_lerobot_datasets.py
+uv run python examples/droid_episode_index.py
+```
+
 ## Hand tracking
 
 A complete hand-tracking walkthrough, generated by the `daft-physical-ai` CLI:

examples/droid_episode_index.py

@@ -0,0 +1,34 @@
+"""Build a lazy DROID episode index with Daft's native reader.
+
+`daft.datasets.droid.raw()` catalogs episodes without decoding video. Filter
+and project first; materialize later. This script only prints the plan.
+"""
+
+from __future__ import annotations
+
+import daft
+from daft.datasets import droid
+
+
+def build_episode_index() -> daft.DataFrame:
+    episodes = droid.raw()
+    successful = episodes.where(daft.col("success") == daft.lit(True))
+    return successful.select(
+        "uuid",
+        "scene_id",
+        "building",
+        "current_task",
+        "success",
+        "trajectory_length",
+        "wrist_cam_video",
+        "ext1_cam_video",
+        "ext2_cam_video",
+    )
+
+
+def main() -> None:
+    build_episode_index().explain(show_all=True)
+
+
+if __name__ == "__main__":
+    main()

examples/lerobot_episode_index.py

@@ -0,0 +1,45 @@
+"""Index a LeRobot v3 dataset without decoding a single video frame.
+
+`daft.datasets.lerobot` reads a LeRobot dataset lazily: `read_episodes` gives
+one row per episode from the metadata, `read_tasks` the task table, and `read`
+one row per frame with episode metadata broadcast on. Video stays undecoded
+until you ask for it with `load_video_frames`.
+
+The default dataset is a tiny EgoDex sample (3 episodes / 632 frames) in
+LeRobot v3 format.
+"""
+
+from __future__ import annotations
+
+import argparse
+
+from daft import col
+from daft.datasets import lerobot
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Episode/task/frame views of a LeRobot dataset.")
+    parser.add_argument("--dataset", default="pepijn223/egodex-test", help="HF repo id or path")
+    parser.add_argument("--min-length", type=int, default=100, help="episode-length filter to demo")
+    args = parser.parse_args()
+
+    episodes = lerobot.read_episodes(args.dataset)
+    index = episodes.select("episode_index", "tasks", "length").sort("episode_index").to_pydict()
+    print(f"{args.dataset}: {len(index['episode_index'])} episodes")
+    for episode_index, tasks, length in zip(index["episode_index"], index["tasks"], index["length"]):
+        print(f"  episode {episode_index}: {length:4d} frames  {tasks[0][:70]}")
+
+    tasks = lerobot.read_tasks(args.dataset).to_pydict()
+    print(f"\n{len(next(iter(tasks.values()), []))} distinct tasks in the task table")
+
+    long_episodes = episodes.where(col("length") >= args.min_length)
+    frames = lerobot.read(args.dataset).join(long_episodes.select("episode_index"), on="episode_index", how="semi")
+    print(
+        f"frames in episodes with >= {args.min_length} steps: {frames.count_rows()} "
+        f"(selected without touching any video)"
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

examples/merge_lerobot_datasets.py

@@ -0,0 +1,61 @@
+"""Merge two LeRobot recording sessions into one table.
+
+Yesterday's teleop and today's land as two LeRobot datasets, both numbering
+episodes from zero. `episode_index` and the global frame `index` collide, so
+the second session is re-indexed before concat.
+
+This demo reads the same tiny v3 dataset twice as session A and session B
+(a second public v3 dataset is not available yet). The mechanics are those of
+merging distinct recordings from one rig. Task strings ride on every frame, so
+task identity survives without a `task_index` remap.
+"""
+
+from __future__ import annotations
+
+import argparse
+
+from daft import col, lit
+from daft.datasets import lerobot
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Merge two LeRobot sessions with re-indexed episodes.")
+    parser.add_argument("--session-a", default="pepijn223/egodex-test")
+    parser.add_argument("--session-b", default="pepijn223/egodex-test")
+    args = parser.parse_args()
+
+    frames_a = lerobot.read(args.session_a)
+    frames_b = lerobot.read(args.session_b)
+
+    extent = frames_a.agg(
+        (col("episode_index").max() + lit(1)).alias("n_episodes"),
+        (col("index").max() + lit(1)).alias("n_frames"),
+    ).to_pydict()
+    episode_offset, frame_offset = extent["n_episodes"][0], extent["n_frames"][0]
+
+    merged = frames_a.concat(
+        frames_b.with_column("episode_index", col("episode_index") + lit(episode_offset)).with_column(
+            "index", col("index") + lit(frame_offset)
+        )
+    )
+
+    lengths = (
+        merged.groupby("episode_index")
+        .agg(col("frame_index").count().alias("frames"))
+        .sort("episode_index")
+        .to_pydict()
+    )
+    print(f"merged: {len(lengths['episode_index'])} episodes / {sum(lengths['frames'])} frames")
+    for episode_index, frames in zip(lengths["episode_index"], lengths["frames"]):
+        source = "A" if episode_index < episode_offset else "B"
+        print(f"  episode {episode_index} (session {source}): {frames} frames")
+
+    n_frames = merged.count_rows()
+    n_distinct = merged.select("index").distinct().count_rows()
+    assert n_frames == n_distinct, "global frame index must stay unique after the merge"
+    print(f"\nglobal frame index unique after re-indexing: {n_distinct}/{n_frames}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())