feat(slicing): slice_video — cut clips by time range with optional square resize

README.md

@@ -54,6 +54,18 @@ with open("video.mp4", "rb") as f:
         pass
 ```
 
+### Slice into Clips
+
+```python
+from simple_video_utils.slicing import slice_video
+
+# One MP4 (bytes) per (start, end) second range
+clips = slice_video("video.mp4", [(0.0, 1.5), (2.0, 3.2)])
+
+# Center-crop to a square and resize to 256x256 (e.g. for model input)
+clips = slice_video("video.mp4", [(0.0, 1.5)], size=256)
+```
+
 ### Remote Videos
 
 ```python

simple_video_utils/slicing.py

@@ -0,0 +1,73 @@
+"""Cut a video into clips by time range, optionally transforming to a square size."""
+
+import io
+from collections.abc import Sequence
+from fractions import Fraction
+
+import av
+import numpy as np
+
+from simple_video_utils.frames import read_frames_exact
+from simple_video_utils.metadata import video_metadata
+
+
+def _center_crop_square(frame: np.ndarray) -> np.ndarray:
+    height, width = frame.shape[:2]
+    side = min(height, width)
+    top = (height - side) // 2
+    left = (width - side) // 2
+    return frame[top : top + side, left : left + side]
+
+
+def _encode_clip(frames: Sequence[np.ndarray], fps: float, size: int | None, codec: str) -> bytes:
+    if size is not None:
+        out_width = out_height = size
+    elif len(frames):
+        out_height, out_width = frames[0].shape[:2]
+    else:
+        return b""
+
+    buffer = io.BytesIO()
+    with av.open(buffer, mode="w", format="mp4") as container:
+        stream = container.add_stream(codec, rate=Fraction(fps).limit_denominator(1000))
+        stream.width = out_width
+        stream.height = out_height
+        stream.pix_fmt = "yuv420p"
+        for frame in frames:
+            if size is not None:
+                frame = _center_crop_square(frame)
+            video_frame = av.VideoFrame.from_ndarray(frame, format="rgb24")
+            if size is not None:
+                video_frame = video_frame.reformat(width=size, height=size)
+            for packet in stream.encode(video_frame):
+                container.mux(packet)
+        for packet in stream.encode():
+            container.mux(packet)
+    return buffer.getvalue()
+
+
+def slice_video(
+    src: str,
+    slices: Sequence[tuple[float, float]],
+    size: int | None = None,
+    codec: str = "h264",
+) -> list[bytes]:
+    """Cut ``src`` into clips, one per (start, end) second range, as encoded MP4 bytes.
+
+    Args:
+        src: Path or URL to the source video.
+        slices: (start, end) second ranges; end is exclusive of the next frame.
+        size: If set, each frame is center-cropped to a square and resized to
+            ``size`` x ``size`` (e.g. 256 for models that expect square input).
+            If None, clips keep the source resolution.
+        codec: Output video codec.
+
+    Returns:
+        One ``bytes`` MP4 per slice, in the same order (empty ``bytes`` for a
+        slice that contains no frames).
+    """
+    fps = video_metadata(src).fps
+    return [
+        _encode_clip(list(read_frames_exact(src, start_time=start, end_time=end)), fps, size, codec)
+        for start, end in slices
+    ]

tests/test_slicing.py

@@ -0,0 +1,44 @@
+import io
+
+import av
+import numpy as np
+import pytest
+
+from simple_video_utils.slicing import slice_video
+
+
+@pytest.fixture
+def video(tmp_path):
+    path = tmp_path / "src.mp4"
+    with av.open(str(path), mode="w") as container:
+        stream = container.add_stream("h264", rate=30)
+        stream.width, stream.height, stream.pix_fmt = 320, 240, "yuv420p"
+        for i in range(30):
+            arr = np.full((240, 320, 3), i * 8 % 256, dtype=np.uint8)
+            for packet in stream.encode(av.VideoFrame.from_ndarray(arr, format="rgb24")):
+                container.mux(packet)
+        for packet in stream.encode():
+            container.mux(packet)
+    return str(path)
+
+
+def _dims(clip: bytes) -> tuple[int, int]:
+    with av.open(io.BytesIO(clip), mode="r") as container:
+        frame = next(container.decode(video=0))
+        return frame.width, frame.height
+
+
+def test_slice_returns_one_clip_per_range(video):
+    clips = slice_video(video, [(0.0, 0.3), (0.5, 0.8)])
+    assert len(clips) == 2
+    assert all(clip for clip in clips)
+
+
+def test_slice_keeps_source_size_by_default(video):
+    [clip] = slice_video(video, [(0.0, 0.3)])
+    assert _dims(clip) == (320, 240)
+
+
+def test_slice_center_crops_and_resizes(video):
+    [clip] = slice_video(video, [(0.0, 0.3)], size=256)
+    assert _dims(clip) == (256, 256)