chore(format): formatting code

Add `episodes_per_video_file` to `episodes_factory`: episode `i` goes to
`file_index = i // episodes_per_video_file` and `from_timestamp` restarts at 0 on
each rollover. `create_videos` encodes one .mp4 per file, and
`mock_snapshot_download` lists and creates every video file rather than only
file-000. Without the option the fixtures emit the same single-file layout as
before.

Add streaming tests over that layout, on the plain and the delta path.

src/lerobot/datasets/dataset_reader.py

@@ -240,7 +240,12 @@ class DatasetReader:
     def _get_query_indices(
         self, abs_idx: int, ep_idx: int
     ) -> tuple[dict[str, list[int]], dict[str, torch.Tensor]]:
-        """Compute query indices for delta timestamps."""
+        """Compute query indices for delta timestamps.
+
+        A delta is padding when ``abs_idx + delta`` falls outside the episode's
+        ``[dataset_from_index, dataset_to_index)`` range, and is clamped back into it
+        otherwise.
+        """
         ep = self._meta.episodes[ep_idx]
         ep_start = ep["dataset_from_index"]
         ep_end = ep["dataset_to_index"]

src/lerobot/datasets/streaming_dataset.py

@@ -32,8 +32,6 @@ from .feature_utils import get_delta_indices
 from .io_utils import item_to_torch
 from .utils import (
     check_version_compatibility,
-    find_float_index,
-    is_float_in_list,
     safe_shard,
 )
 from .video_utils import (
@@ -478,49 +476,28 @@ class StreamingLeRobotDataset(torch.utils.data.IterableDataset):
         lookback, lookahead = self._get_window_steps(self.delta_timestamps)
         return Backtrackable(dataset, history=lookback, lookahead=lookahead)
 
-    def _make_timestamps_from_indices(
-        self, start_ts: float, indices: dict[str, list[int]] | None = None
-    ) -> dict[str, list[float]]:
-        if indices is not None:
-            return {
-                key: (
-                    start_ts + torch.tensor(indices[key]) / self.fps
-                ).tolist()  # NOTE: why not delta_timestamps directly?
-                for key in self.delta_timestamps
-            }
-        else:
-            return dict.fromkeys(self.meta.video_keys, [start_ts])
+    def _get_query_indices(
+        self, abs_idx: int, ep_idx: int
+    ) -> tuple[dict[str, list[int]], dict[str, torch.BoolTensor]]:
+        """Video-key query indices and padding from integer episode boundaries.
 
-    def _make_padding_camera_frame(self, camera_key: str):
-        """Variable-shape padding frame for given camera keys, given in (H, W, C)"""
-        return torch.zeros(self.meta.info.features[camera_key]["shape"]).permute(-1, 0, 1)
-
-    def _get_video_frame_padding_mask(
-        self,
-        video_frames: dict[str, torch.Tensor],
-        query_timestamps: dict[str, list[float]],
-        original_timestamps: dict[str, list[float]],
-    ) -> dict[str, torch.BoolTensor]:
-        padding_mask = {}
-
-        for video_key, timestamps in original_timestamps.items():
-            if video_key not in video_frames:
-                continue  # only padding on video keys that are available
-            frames = []
-            mask = []
-            padding_frame = self._make_padding_camera_frame(video_key)
-            for ts in timestamps:
-                if is_float_in_list(ts, query_timestamps[video_key]):
-                    idx = find_float_index(ts, query_timestamps[video_key])
-                    frames.append(video_frames[video_key][idx, :])
-                    mask.append(False)
-                else:
-                    frames.append(padding_frame)
-                    mask.append(True)
-
-            padding_mask[f"{video_key}_is_pad"] = torch.BoolTensor(mask)
-
-        return padding_mask
+        Mirrors ``DatasetReader._get_query_indices`` but only for video keys.
+        """
+        ep = self.meta.episodes[ep_idx]
+        ep_start, ep_end = ep["dataset_from_index"], ep["dataset_to_index"]
+        query_indices = {
+            key: [max(ep_start, min(ep_end - 1, abs_idx + delta)) for delta in delta_idx]
+            for key, delta_idx in self.delta_indices.items()
+            if key in self.meta.video_keys
+        }
+        padding = {
+            f"{key}_is_pad": torch.BoolTensor(
+                [(abs_idx + delta < ep_start) or (abs_idx + delta >= ep_end) for delta in delta_idx]
+            )
+            for key, delta_idx in self.delta_indices.items()
+            if key in self.meta.video_keys
+        }
+        return query_indices, padding
 
     def make_frame(self, dataset_iterator: Backtrackable) -> Generator:
         """Makes a frame starting from a dataset iterator"""
@@ -533,19 +510,11 @@ class StreamingLeRobotDataset(torch.utils.data.IterableDataset):
 
         updates = []  # list of "updates" to apply to the item retrieved from hf_dataset (w/o camera features)
 
-        # Get episode index from the item
+        # Get episode index and absolute frame index from the item
         ep_idx = item["episode_index"]
-
-        # "timestamp" restarts from 0 for each episode, whereas we need a global timestep within the single .mp4 file (given by index/fps)
-        current_ts = item["index"] / self.fps
-
-        episode_boundaries_ts = {
-            key: (
-                self.meta.episodes[ep_idx][f"videos/{key}/from_timestamp"],
-                self.meta.episodes[ep_idx][f"videos/{key}/to_timestamp"],
-            )
-            for key in self.meta.video_keys
-        }
+        abs_idx = int(item["index"])
+        ep_start = self.meta.episodes[ep_idx]["dataset_from_index"]
+        current_ts = float(item["timestamp"])
 
         # Apply delta querying logic if necessary
         if self.delta_indices is not None:
@@ -555,12 +524,19 @@ class StreamingLeRobotDataset(torch.utils.data.IterableDataset):
 
         # Load video frames, when needed
         if len(self.meta.video_keys) > 0:
-            original_timestamps = self._make_timestamps_from_indices(current_ts, self.delta_indices)
+            query_indices = None
+            if self.delta_indices is not None:
+                query_indices, video_padding = self._get_query_indices(abs_idx, ep_idx)
 
-            # Some timestamps might not result available considering the episode's boundaries
-            query_timestamps = self._get_query_timestamps(
-                current_ts, self.delta_indices, episode_boundaries_ts
-            )
+            # Episode-local timestamps; `_query_videos` shifts them by the per-key `from_timestamp` at decode.
+            query_timestamps = {
+                key: (
+                    [(idx - ep_start) / self.fps for idx in query_indices[key]]
+                    if query_indices is not None and key in query_indices
+                    else [current_ts]
+                )
+                for key in self.meta.video_keys
+            }
             video_frames = self._query_videos(query_timestamps, ep_idx)
 
             if self.image_transforms is not None:
@@ -572,10 +548,7 @@ class StreamingLeRobotDataset(torch.utils.data.IterableDataset):
 
             if self.delta_indices is not None:
                 # We always return the same number of frames. Unavailable frames are padded.
-                padding_mask = self._get_video_frame_padding_mask(
-                    video_frames, query_timestamps, original_timestamps
-                )
-                updates.append(padding_mask)
+                updates.append(video_padding)
 
         result = item.copy()
         for update in updates:
@@ -594,27 +567,6 @@ class StreamingLeRobotDataset(torch.utils.data.IterableDataset):
 
         yield result
 
-    def _get_query_timestamps(
-        self,
-        current_ts: float,
-        query_indices: dict[str, list[int]] | None = None,
-        episode_boundaries_ts: dict[str, tuple[float, float]] | None = None,
-    ) -> dict[str, list[float]]:
-        query_timestamps = {}
-        keys_to_timestamps = self._make_timestamps_from_indices(current_ts, query_indices)
-        for key in self.meta.video_keys:
-            if query_indices is not None and key in query_indices:
-                timestamps = keys_to_timestamps[key]
-                # Clamp out timesteps outside of episode boundaries
-                query_timestamps[key] = torch.clamp(
-                    torch.tensor(timestamps), *episode_boundaries_ts[key]
-                ).tolist()
-
-            else:
-                query_timestamps[key] = [current_ts]
-
-        return query_timestamps
-
     def _query_videos(self, query_timestamps: dict[str, list[float]], ep_idx: int) -> dict:
         """Note: When using data workers (e.g. DataLoader with num_workers>0), do not call this function
         in the main process (e.g. by using a second Dataloader with num_workers=0). It will result in a
@@ -622,8 +574,14 @@ class StreamingLeRobotDataset(torch.utils.data.IterableDataset):
         the main process and a subprocess fails to access it.
         """
 
+        ep = self.meta.episodes[ep_idx]
         item = {}
-        for video_key, query_ts in query_timestamps.items():
+        for video_key, ep_local_ts in query_timestamps.items():
+            # Episode-local timestamps restart from 0 each episode; shift by the per-key
+            # `from_timestamp` to reach the episode's segment within its video file, matching
+            # `DatasetReader._decode_single`.
+            from_timestamp = ep[f"videos/{video_key}/from_timestamp"]
+            query_ts = [from_timestamp + ts for ts in ep_local_ts]
             root = self.meta.url_root if self.streaming and not self.streaming_from_local else self.root
             video_path = f"{root}/{self.meta.get_video_file_path(ep_idx, video_key)}"
             if video_key in self.meta.depth_keys:

src/lerobot/datasets/utils.py

@@ -523,17 +523,6 @@ def create_lerobot_dataset_card(
     )
 
 
-def is_float_in_list(target, float_list, threshold=1e-6):
-    return any(abs(target - x) <= threshold for x in float_list)
-
-
-def find_float_index(target, float_list, threshold=1e-6):
-    for i, x in enumerate(float_list):
-        if abs(target - x) <= threshold:
-            return i
-    return -1
-
-
 def safe_shard(dataset: datasets.IterableDataset, index: int, num_shards: int) -> datasets.Dataset:
     """
     Safe shards the dataset.

tests/datasets/test_streaming.py

@@ -24,11 +24,17 @@ pytest.importorskip("datasets", reason="datasets is required (install lerobot[da
 
 import lerobot.datasets.streaming_dataset as streaming_dataset_module
 from lerobot.datasets.dataset_metadata import LeRobotDatasetMetadata
+from lerobot.datasets.lerobot_dataset import LeRobotDataset
 from lerobot.datasets.streaming_dataset import StreamingLeRobotDataset
-from lerobot.datasets.utils import safe_shard
 from lerobot.utils.constants import ACTION
 from tests.fixtures.constants import DUMMY_REPO_ID
 
+# A dataset whose videos roll over into a second file, as v3.0 does past
+# DEFAULT_VIDEO_FILE_SIZE_IN_MB: episodes 4-7 live in file-001, whose timeline restarts at 0.
+MULTI_FILE_EPISODES = 8
+MULTI_FILE_FRAMES = 200
+MULTI_FILE_EPISODES_PER_VIDEO_FILE = 4
+
 
 def get_frames_expected_order(streaming_ds: StreamingLeRobotDataset) -> list[int]:
     """Replicates the shuffling logic of StreamingLeRobotDataset to get the expected order of indices."""
@@ -80,14 +86,9 @@ def get_frames_expected_order(streaming_ds: StreamingLeRobotDataset) -> list[int
 @pytest.mark.parametrize("from_local", [False, True])
 def test_streaming_dataset_forwards_hub_token_only_for_remote_data(tmp_path, monkeypatch, token, from_local):
     requested_root = tmp_path / "local" if from_local else None
-    metadata = SimpleNamespace(
+    metadata = _fake_meta(
         root=requested_root or tmp_path / "snapshot",
         revision=streaming_dataset_module.CODEBASE_VERSION,
-        _version=streaming_dataset_module.CODEBASE_VERSION,
-        features={},
-        depth_keys=[],
-        image_keys=[],
-        rescale_depth_stats=Mock(),
     )
     metadata_cls = Mock(return_value=metadata)
     load_dataset = Mock(return_value=SimpleNamespace(num_shards=1))
@@ -111,8 +112,68 @@ def test_streaming_dataset_forwards_hub_token_only_for_remote_data(tmp_path, mon
     assert not hasattr(dataset, "_token")
 
 
+def assert_videos_roll_over(ds: LeRobotDataset) -> None:
+    """Videos spanning several files must decode each frame from its own file (v3.0 rollover)."""
+    for key in ds.meta.video_keys:
+        episodes = [ds.meta.episodes[ep_idx] for ep_idx in range(ds.meta.total_episodes)]
+        file_indices = {ep[f"videos/{key}/file_index"] for ep in episodes}
+        assert len(file_indices) > 1, f"{key} is not split across video files (file_index: {file_indices})"
+
+        # The property under test: a later file's timeline starts back at 0, so an episode's
+        # `from_timestamp` is no longer its global position in the dataset.
+        assert any(
+            ep[f"videos/{key}/file_index"] > 0 and ep[f"videos/{key}/from_timestamp"] == 0.0
+            for ep in episodes
+        ), f"No episode of {key} restarts a video file's timeline at 0"
+
+
+def assert_frame_matches(streaming_frame: dict, target_frame: dict, ds: LeRobotDataset, context: str) -> None:
+    """Assert a streamed frame equals the same frame read by the non-streaming reader."""
+    assert set(streaming_frame.keys()) == set(target_frame.keys()), (
+        f"Keys differ between streaming frame and target one ({context}). "
+        f"Differ at: {set(streaming_frame.keys()) ^ set(target_frame.keys())}"
+    )
+
+    mismatched = []
+    for key in streaming_frame:
+        left, right = streaming_frame[key], target_frame[key]
+
+        if isinstance(left, str):
+            check = left == right
+
+        elif isinstance(left, float):
+            check = left == right.item()  # right is a torch.Tensor
+
+        elif isinstance(left, torch.Tensor):
+            if key not in ds.meta.camera_keys and "is_pad" not in key and f"{key}_is_pad" in streaming_frame:
+                # comparing frames only on non-padded regions. Padding is applied to last-valid broadcasting
+                left = left[~streaming_frame[f"{key}_is_pad"]]
+                right = right[~target_frame[f"{key}_is_pad"]]
+
+            check = left.shape == right.shape and torch.allclose(left, right)
+
+        else:
+            check = left == right
+
+        if not check:
+            mismatched.append(key)
+
+    assert not mismatched, f"Streaming and target frames differ on {mismatched} ({context})"
+
+
+def assert_stream_matches_reference(
+    streaming_ds: StreamingLeRobotDataset, ds: LeRobotDataset, num_frames: int
+) -> None:
+    """Stream ``num_frames`` frames and assert each equals the same frame from the non-streaming reader."""
+    stream = iter(streaming_ds)
+    for i in range(num_frames):
+        streaming_frame = next(stream)
+        frame_idx = streaming_frame["index"]
+        assert_frame_matches(streaming_frame, ds[frame_idx], ds, context=f"i: {i}, frame_idx: {frame_idx}")
+
+
 def test_single_frame_consistency(tmp_path, lerobot_dataset_factory):
-    """Test if are correctly accessed"""
+    """Streaming without deltas returns the same frames as the non-streaming reader."""
     ds_num_frames = 400
     ds_num_episodes = 10
     buffer_size = 100
@@ -127,32 +188,8 @@ def test_single_frame_consistency(tmp_path, lerobot_dataset_factory):
         total_frames=ds_num_frames,
     )
 
-    streaming_ds = iter(StreamingLeRobotDataset(repo_id=repo_id, root=local_path, buffer_size=buffer_size))
-
-    key_checks = []
-    for _ in range(ds_num_frames):
-        streaming_frame = next(streaming_ds)
-        frame_idx = streaming_frame["index"]
-        target_frame = ds[frame_idx]
-
-        for key in streaming_frame:
-            left = streaming_frame[key]
-            right = target_frame[key]
-
-            if isinstance(left, str):
-                check = left == right
-
-            elif isinstance(left, torch.Tensor):
-                check = torch.allclose(left, right) and left.shape == right.shape
-
-            elif isinstance(left, float):
-                check = left == right.item()  # right is a torch.Tensor
-
-            key_checks.append((key, check))
-
-        assert all(t[1] for t in key_checks), (
-            f"Checking {list(filter(lambda t: not t[1], key_checks))[0][0]} left and right were found different (frame_idx: {frame_idx})"
-        )
+    streaming_ds = StreamingLeRobotDataset(repo_id=repo_id, root=local_path, buffer_size=buffer_size)
+    assert_stream_matches_reference(streaming_ds, ds, ds_num_frames)
 
 
 @pytest.mark.parametrize(
@@ -302,87 +339,7 @@ def test_iter_raises_on_nested_generator_error(tmp_path, lerobot_dataset_factory
         next(iter(streaming_ds))
 
 
-@pytest.mark.parametrize(
-    "state_deltas, action_deltas",
-    [
-        ([-1, -0.5, -0.20, 0], [0, 1, 2, 3]),
-        ([-1, -0.5, -0.20, 0], [-1.5, -1, -0.5, -0.20, -0.10, 0]),
-        ([-2, -1, -0.5, 0], [0, 1, 2, 3]),
-        ([-2, -1, -0.5, 0], [-1.5, -1, -0.5, -0.20, -0.10, 0]),
-    ],
-)
-def test_frames_with_delta_consistency(tmp_path, lerobot_dataset_factory, state_deltas, action_deltas):
-    ds_num_frames = 500
-    ds_num_episodes = 10
-    buffer_size = 100
-
-    seed = 42
-
-    local_path = tmp_path / "test"
-    repo_id = f"{DUMMY_REPO_ID}-ciao"
-    camera_key = "phone"
-
-    delta_timestamps = {
-        camera_key: state_deltas,
-        "state": state_deltas,
-        ACTION: action_deltas,
-    }
-
-    ds = lerobot_dataset_factory(
-        root=local_path,
-        repo_id=repo_id,
-        total_episodes=ds_num_episodes,
-        total_frames=ds_num_frames,
-        delta_timestamps=delta_timestamps,
-    )
-
-    streaming_ds = iter(
-        StreamingLeRobotDataset(
-            repo_id=repo_id,
-            root=local_path,
-            buffer_size=buffer_size,
-            seed=seed,
-            shuffle=False,
-            delta_timestamps=delta_timestamps,
-        )
-    )
-
-    for i in range(ds_num_frames):
-        streaming_frame = next(streaming_ds)
-        frame_idx = streaming_frame["index"]
-        target_frame = ds[frame_idx]
-
-        assert set(streaming_frame.keys()) == set(target_frame.keys()), (
-            f"Keys differ between streaming frame and target one. Differ at: {set(streaming_frame.keys()) - set(target_frame.keys())}"
-        )
-
-        key_checks = []
-        for key in streaming_frame:
-            left = streaming_frame[key]
-            right = target_frame[key]
-
-            if isinstance(left, str):
-                check = left == right
-
-            elif isinstance(left, torch.Tensor):
-                if (
-                    key not in ds.meta.camera_keys
-                    and "is_pad" not in key
-                    and f"{key}_is_pad" in streaming_frame
-                ):
-                    # comparing frames only on non-padded regions. Padding is applied to last-valid broadcasting
-                    left = left[~streaming_frame[f"{key}_is_pad"]]
-                    right = right[~target_frame[f"{key}_is_pad"]]
-
-                check = torch.allclose(left, right) and left.shape == right.shape
-
-            key_checks.append((key, check))
-
-        assert all(t[1] for t in key_checks), (
-            f"Checking {list(filter(lambda t: not t[1], key_checks))[0][0]} left and right were found different (i: {i}, frame_idx: {frame_idx})"
-        )
-
-
+@pytest.mark.parametrize("sharded", [False, True])
 @pytest.mark.parametrize(
     "state_deltas, action_deltas",
     [
@@ -392,94 +349,40 @@ def test_frames_with_delta_consistency(tmp_path, lerobot_dataset_factory, state_
         ([-2, -1, -0.5, 0], [-20, -1.5, -1, -0.5, -0.20, -0.10, 0]),
     ],
 )
-def test_frames_with_delta_consistency_with_shards(
-    tmp_path, lerobot_dataset_factory, state_deltas, action_deltas
+def test_frames_with_delta_consistency(
+    tmp_path, lerobot_dataset_factory, sharded, state_deltas, action_deltas
 ):
-    ds_num_frames = 100
-    ds_num_episodes = 10
-    buffer_size = 10
-    data_file_size_mb = 0.001
-    chunks_size = 1
-
-    seed = 42
-
+    """Delta-window frames streamed match the non-streaming reader, with and without sharding."""
     local_path = tmp_path / "test"
     repo_id = f"{DUMMY_REPO_ID}-ciao"
-    camera_key = "phone"
+    delta_timestamps = {"phone": state_deltas, "state": state_deltas, ACTION: action_deltas}
 
-    delta_timestamps = {
-        camera_key: state_deltas,
-        "state": state_deltas,
-        ACTION: action_deltas,
-    }
+    if sharded:
+        num_frames, buffer_size = 100, 10
+        factory_extra = {"data_files_size_in_mb": 0.001, "chunks_size": 1}
+        stream_extra = {"max_num_shards": 4}
+    else:
+        num_frames, buffer_size = 500, 100
+        factory_extra, stream_extra = {}, {}
 
     ds = lerobot_dataset_factory(
         root=local_path,
         repo_id=repo_id,
-        total_episodes=ds_num_episodes,
-        total_frames=ds_num_frames,
+        total_episodes=10,
+        total_frames=num_frames,
         delta_timestamps=delta_timestamps,
-        data_files_size_in_mb=data_file_size_mb,
-        chunks_size=chunks_size,
+        **factory_extra,
     )
     streaming_ds = StreamingLeRobotDataset(
         repo_id=repo_id,
         root=local_path,
         buffer_size=buffer_size,
-        seed=seed,
+        seed=42,
         shuffle=False,
         delta_timestamps=delta_timestamps,
-        max_num_shards=4,
+        **stream_extra,
     )
-
-    iter(streaming_ds)
-
-    num_shards = 4
-    shards_indices = []
-    for shard_idx in range(num_shards):
-        shard = safe_shard(streaming_ds.hf_dataset, shard_idx, num_shards)
-        shard_indices = [item["index"] for item in shard]
-        shards_indices.append(shard_indices)
-
-    streaming_ds = iter(streaming_ds)
-
-    for i in range(ds_num_frames):
-        streaming_frame = next(streaming_ds)
-        frame_idx = streaming_frame["index"]
-        target_frame = ds[frame_idx]
-
-        assert set(streaming_frame.keys()) == set(target_frame.keys()), (
-            f"Keys differ between streaming frame and target one. Differ at: {set(streaming_frame.keys()) - set(target_frame.keys())}"
-        )
-
-        key_checks = []
-        for key in streaming_frame:
-            left = streaming_frame[key]
-            right = target_frame[key]
-
-            if isinstance(left, str):
-                check = left == right
-
-            elif isinstance(left, torch.Tensor):
-                if (
-                    key not in ds.meta.camera_keys
-                    and "is_pad" not in key
-                    and f"{key}_is_pad" in streaming_frame
-                ):
-                    # comparing frames only on non-padded regions. Padding is applied to last-valid broadcasting
-                    left = left[~streaming_frame[f"{key}_is_pad"]]
-                    right = right[~target_frame[f"{key}_is_pad"]]
-
-                check = torch.allclose(left, right) and left.shape == right.shape
-
-            elif isinstance(left, float):
-                check = left == right.item()  # right is a torch.Tensor
-
-            key_checks.append((key, check))
-
-        assert all(t[1] for t in key_checks), (
-            f"Checking {list(filter(lambda t: not t[1], key_checks))[0][0]} left and right were found different (i: {i}, frame_idx: {frame_idx})"
-        )
+    assert_stream_matches_reference(streaming_ds, ds, num_frames)
 
 
 class _StopConstructionError(Exception):
@@ -493,7 +396,9 @@ def _fake_meta(*args, **kwargs):
     revision = kwargs.get("revision", args[2] if len(args) > 2 else None)
     meta.root = root or "/tmp/_streaming_meta"
     meta.revision = revision or "v0"
-    meta._version = "v3.0"
+    meta._version = streaming_dataset_module.CODEBASE_VERSION
+    meta.features = {}
+    meta.video_keys = []
     meta.depth_keys = []
     meta.image_keys = []
     meta.rescale_depth_stats = lambda *_a, **_k: None
@@ -645,3 +550,42 @@ def test_bucket_root_caches_metadata_without_switching_to_local_streaming(tmp_pa
 def test_invalid_repo_type_fails_before_io():
     with pytest.raises(ValueError, match="repo_type must be 'dataset' or 'bucket'"):
         StreamingLeRobotDataset(DUMMY_REPO_ID, repo_type="space")
+
+
+@pytest.mark.parametrize(
+    "state_deltas, action_deltas",
+    [
+        (None, None),
+        ([-1, -0.5, -0.20, 0], [0, 1, 2, 3]),
+        ([-2, -1, -0.5, 0], [-1.5, -1, -0.5, -0.20, -0.10, 0]),
+    ],
+)
+def test_consistency_across_video_files(tmp_path, lerobot_dataset_factory, state_deltas, action_deltas):
+    """Videos spanning several files must decode each frame from its own file (v3.0 rollover)."""
+    local_path = tmp_path / "test"
+    repo_id = f"{DUMMY_REPO_ID}-video-rollover"
+    delta_timestamps = (
+        None
+        if state_deltas is None
+        else {"phone": state_deltas, "state": state_deltas, ACTION: action_deltas}
+    )
+
+    ds = lerobot_dataset_factory(
+        root=local_path,
+        repo_id=repo_id,
+        total_episodes=MULTI_FILE_EPISODES,
+        total_frames=MULTI_FILE_FRAMES,
+        episodes_per_video_file=MULTI_FILE_EPISODES_PER_VIDEO_FILE,
+        delta_timestamps=delta_timestamps,
+    )
+    assert_videos_roll_over(ds)
+
+    streaming_ds = StreamingLeRobotDataset(
+        repo_id=repo_id,
+        root=local_path,
+        buffer_size=100,
+        seed=42,
+        shuffle=False,
+        delta_timestamps=delta_timestamps,
+    )
+    assert_stream_matches_reference(streaming_ds, ds, MULTI_FILE_FRAMES)

tests/fixtures/dataset_factories.py

@@ -268,7 +268,17 @@ def episodes_factory(tasks_factory, stats_factory):
         video_keys: list[str] | None = None,
         tasks: pd.DataFrame | None = None,
         multi_task: bool = False,
+        episodes_per_video_file: int | None = None,
     ):
+        """Build episode metadata.
+
+        ``episodes_per_video_file`` splits the video keys across several files, as v3.0 does
+        once a video grows past ``DEFAULT_VIDEO_FILE_SIZE_IN_MB``: episode ``i`` lands in
+        ``file_index = i // episodes_per_video_file`` and each file's timeline restarts at 0,
+        so ``from_timestamp`` is relative to the file the episode lives in — not to the
+        dataset. Left ``None``, everything goes to ``file-000`` with a cumulative
+        ``from_timestamp`` (the single-file layout, where the two happen to coincide).
+        """
         if total_episodes <= 0 or total_frames <= 0:
             raise ValueError("num_episodes and total_length must be positive integers.")
         if total_frames < total_episodes:
@@ -310,8 +320,16 @@ def episodes_factory(tasks_factory, stats_factory):
             d[stats_key] = []
 
         num_frames = 0
+        # Frames written to the current video file. Resets on every file rollover, since each
+        # .mp4 carries its own timeline.
+        num_frames_in_video_file = 0
+        video_file_index = 0
         remaining_tasks = list(tasks.index)
         for ep_idx in range(total_episodes):
+            if episodes_per_video_file is not None and ep_idx // episodes_per_video_file != video_file_index:
+                video_file_index = ep_idx // episodes_per_video_file
+                num_frames_in_video_file = 0
+
             num_tasks_in_episode = random.randint(1, min(3, num_tasks_available)) if multi_task else 1
             tasks_to_sample = remaining_tasks if len(remaining_tasks) > 0 else list(tasks.index)
             episode_tasks = random.sample(tasks_to_sample, min(num_tasks_in_episode, len(tasks_to_sample)))
@@ -333,21 +351,44 @@ def episodes_factory(tasks_factory, stats_factory):
             if video_keys is not None:
                 for video_key in video_keys:
                     d[f"videos/{video_key}/chunk_index"].append(0)
-                    d[f"videos/{video_key}/file_index"].append(0)
-                    d[f"videos/{video_key}/from_timestamp"].append(num_frames / fps)
-                    d[f"videos/{video_key}/to_timestamp"].append((num_frames + lengths[ep_idx]) / fps)
+                    d[f"videos/{video_key}/file_index"].append(video_file_index)
+                    d[f"videos/{video_key}/from_timestamp"].append(num_frames_in_video_file / fps)
+                    d[f"videos/{video_key}/to_timestamp"].append(
+                        (num_frames_in_video_file + lengths[ep_idx]) / fps
+                    )
 
             # Add stats columns like "stats/action/max"
             for stats_key, stats in flatten_dict({"stats": stats_factory(features)}).items():
                 d[stats_key].append(stats)
 
             num_frames += lengths[ep_idx]
+            num_frames_in_video_file += lengths[ep_idx]
 
         return Dataset.from_dict(d)
 
     return _create_episodes
 
 
+def video_file_frames(
+    episodes: datasets.Dataset | None, video_key: str, total_frames: int
+) -> dict[int, list[int]]:
+    """Map each of ``video_key``'s files to the global frame indices it holds, in file order.
+
+    ``episodes`` is the source of truth for how a video key is split across files. Without it
+    (or without the videos columns), the whole key is one file — the pre-v3.0 layout.
+    """
+    if episodes is None or f"videos/{video_key}/file_index" not in episodes.column_names:
+        return {0: list(range(total_frames))}
+
+    frames_per_file: dict[int, list[int]] = {}
+    for ep in episodes:
+        file_index = ep[f"videos/{video_key}/file_index"]
+        frames = range(ep["dataset_from_index"], ep["dataset_to_index"])
+        frames_per_file.setdefault(file_index, []).extend(frames)
+
+    return frames_per_file
+
+
 @pytest.fixture(scope="session")
 def create_videos(info_factory, img_array_factory):
     def _create_video_directory(
@@ -356,6 +397,7 @@ def create_videos(info_factory, img_array_factory):
         total_episodes: int = 3,
         total_frames: int = 150,
         total_tasks: int = 1,
+        episodes: datasets.Dataset | None = None,
     ):
         if info is None:
             info = info_factory(
@@ -364,21 +406,27 @@ def create_videos(info_factory, img_array_factory):
 
         video_feats = {key: feats for key, feats in info.features.items() if feats["dtype"] == "video"}
         for key, ft in video_feats.items():
-            # create and save images with identifiable content
-            tmp_dir = root / "tmp_images"
-            tmp_dir.mkdir(parents=True, exist_ok=True)
-            for frame_index in range(info.total_frames):
-                content = f"{key}-{frame_index}"
-                img = img_array_factory(height=ft["shape"][0], width=ft["shape"][1], content=content)
-                pil_img = PIL.Image.fromarray(img)
-                path = tmp_dir / f"frame-{frame_index:06d}.png"
-                pil_img.save(path)
+            for file_index, frame_indices in video_file_frames(episodes, key, info.total_frames).items():
+                # create and save images with identifiable content
+                tmp_dir = root / "tmp_images"
+                tmp_dir.mkdir(parents=True, exist_ok=True)
+                for position, frame_index in enumerate(frame_indices):
+                    # Content stays keyed on the *global* frame index so a frame remains
+                    # identifiable across files, but its position in the file is what the
+                    # file's timeline addresses.
+                    content = f"{key}-{frame_index}"
+                    img = img_array_factory(height=ft["shape"][0], width=ft["shape"][1], content=content)
+                    pil_img = PIL.Image.fromarray(img)
+                    path = tmp_dir / f"frame-{position:06d}.png"
+                    pil_img.save(path)
 
-            video_path = root / DEFAULT_VIDEO_PATH.format(video_key=key, chunk_index=0, file_index=0)
-            video_path.parent.mkdir(parents=True, exist_ok=True)
-            # Use the global fps from info, not video-specific fps which might not exist
-            encode_video_frames(tmp_dir, video_path, fps=info.fps)
-            shutil.rmtree(tmp_dir)
+                video_path = root / DEFAULT_VIDEO_PATH.format(
+                    video_key=key, chunk_index=0, file_index=file_index
+                )
+                video_path.parent.mkdir(parents=True, exist_ok=True)
+                # Use the global fps from info, not video-specific fps which might not exist
+                encode_video_frames(tmp_dir, video_path, fps=info.fps)
+                shutil.rmtree(tmp_dir)
 
     return _create_video_directory
 
@@ -520,6 +568,7 @@ def lerobot_dataset_factory(
         data_files_size_in_mb: float = DEFAULT_DATA_FILE_SIZE_IN_MB,
         chunks_size: int = DEFAULT_CHUNK_SIZE,
         camera_features: dict | None = None,
+        episodes_per_video_file: int | None = None,
         **kwargs,
     ) -> LeRobotDataset:
         # Instantiate objects
@@ -557,6 +606,7 @@ def lerobot_dataset_factory(
                 video_keys=video_keys,
                 tasks=tasks,
                 multi_task=multi_task,
+                episodes_per_video_file=episodes_per_video_file,
             )
         if hf_dataset is None:
             hf_dataset = hf_dataset_factory(

tests/fixtures/hub.py

@@ -29,6 +29,7 @@ from lerobot.datasets.utils import (
     STATS_PATH,
 )
 from tests.fixtures.constants import LEROBOT_TEST_DIR
+from tests.fixtures.dataset_factories import video_file_frames
 
 
 @pytest.fixture(scope="session")
@@ -99,7 +100,12 @@ def mock_snapshot_download_factory(
 
             video_keys = [key for key, feats in info.features.items() if feats["dtype"] == "video"]
             for key in video_keys:
-                all_files.append(DEFAULT_VIDEO_PATH.format(video_key=key, chunk_index=0, file_index=0))
+                # A video key spans one file per `videos/<key>/file_index` in the episodes
+                # metadata — several of them once v3.0 rolls a video over.
+                for file_index in video_file_frames(episodes, key, info.total_frames):
+                    all_files.append(
+                        DEFAULT_VIDEO_PATH.format(video_key=key, chunk_index=0, file_index=file_index)
+                    )
 
             allowed_files = filter_repo_objects(
                 all_files, allow_patterns=allow_patterns, ignore_patterns=ignore_patterns
@@ -138,7 +144,7 @@ def mock_snapshot_download_factory(
             if request_data:
                 create_hf_dataset(local_dir, hf_dataset, data_files_size_in_mb, chunks_size)
             if request_videos:
-                create_videos(root=local_dir, info=info)
+                create_videos(root=local_dir, info=info, episodes=episodes)
 
             return str(local_dir)