feat(action): add reusable clip/window abstraction for video understanding

scripts/benchmark_clip_modes.py

@@ -0,0 +1,107 @@
+"""Standalone benchmark for #83 clip_mode comparison.
+
+Not part of the automated test suite. Run manually:
+
+    python scripts/benchmark_clip_modes.py path/to/video.mp4
+
+This is a stand-in for the harness in #76, which does not exist yet.
+It only measures the action/VideoPrism capability in isolation.
+"""
+
+from __future__ import annotations
+
+import sys
+import time
+import tracemalloc
+from pathlib import Path
+
+from unittest.mock import patch
+
+from vidxp.capabilities.registry import create_capability_registry
+from vidxp.capabilities.visual import index_visuals
+from vidxp.core.contracts import CancellationToken, IndexConfig, VideoSource
+from vidxp.runtime import ModelRuntime
+from vidxp.settings import VidXPSettings
+
+# This benchmark measures clip windowing / shot-detection overhead only,
+# not VideoPrism model inference (which requires a ~1GB local download).
+# The model layer is mocked, same as the unit tests do.
+
+
+class _NullStorage:
+    def upsert(self, _modality, records, **_kwargs):
+        return len(records)
+
+    def delete_records(self, *_args, **_kwargs):
+        pass
+
+    def delete_video(self, *_args, **_kwargs):
+        pass
+
+
+def run_once(video_path: str, clip_mode: str) -> dict:
+    config = IndexConfig(
+        video_id="benchmark-video",
+        enabled_modalities=("action",),
+        capability_options={"action": {"clip_mode": clip_mode}},
+    )
+    registry = create_capability_registry()
+    runtime = ModelRuntime(
+        VidXPSettings(repository_root="unused", runtime_backend="cpu")
+    )
+    source = VideoSource(path=video_path, video_id="benchmark-video")
+
+    tracemalloc.start()
+    started = time.perf_counter()
+    with (
+        patch(
+            "vidxp.capabilities.action.indexing.get_videoprism_model",
+            return_value=object(),
+        ),
+        patch(
+            "vidxp.capabilities.action.indexing.encode_video_clips",
+            side_effect=lambda clips, _provider: [[0.1] for _ in clips],
+        ),
+    ):
+        result = index_visuals(
+            source,
+            config=config,
+            storage=_NullStorage(),
+            cancellation=CancellationToken(),
+            registry=registry,
+            runtime=runtime,
+        )
+    elapsed = time.perf_counter() - started
+    _current, peak = tracemalloc.get_traced_memory()
+    tracemalloc.stop()
+
+    return {
+        "clip_mode": clip_mode,
+        "seconds": round(elapsed, 3),
+        "peak_memory_mb": round(peak / (1024 * 1024), 2),
+        "clips": result.summary.get("videoprism_clips"),
+    }
+
+
+def main() -> None:
+    if len(sys.argv) != 2:
+        print("Usage: python scripts/benchmark_clip_modes.py <video_path>")
+        raise SystemExit(1)
+    video_path = sys.argv[1]
+    if not Path(video_path).is_file():
+        print(f"Not a file: {video_path}")
+        raise SystemExit(1)
+
+    print(f"Benchmarking {video_path}\n")
+    for clip_mode in ("fixed", "scene"):
+        stats = run_once(video_path, clip_mode)
+        print(
+            f"  clip_mode={stats['clip_mode']:<7} "
+            f"time={stats['seconds']:>7}s  "
+            f"peak_mem={stats['peak_memory_mb']:>7}MB  "
+            f"clips={stats['clips']}"
+        )
+
+
+if __name__ == "__main__":
+    main()

src/vidxp/capabilities/action/config.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+from typing import Literal
+
 from pydantic import Field
 
 from vidxp.capabilities.contracts import CapabilityConfig
@@ -9,6 +11,7 @@
 class VideoPrismConfig(CapabilityConfig):
     batch_size: int = Field(default=1, gt=0)
     sample_fps: float = Field(default=2.0, gt=0)
+    clip_mode: Literal["fixed", "scene"] = Field(default="fixed")
 
 
 def videoprism_config(config: IndexConfig) -> VideoPrismConfig:

src/vidxp/capabilities/action/indexing.py

@@ -18,6 +18,8 @@
     stable_source_id,
 )
 from vidxp.core.indexing_common import ProgressCallback, report_progress
+from vidxp.core.clip import ClipStreamAccumulator, VideoClip
+from vidxp.core.scene_boundaries import detect_shot_boundaries
 from vidxp.core.video import FrameSample, FrameSampling
 from vidxp.ports import IndexStore, ModelRuntimePort
 
@@ -28,7 +30,9 @@
 @dataclass
 class VideoPrismIndexState:
     provider: VideoPrismModel
-    pending: list[FrameSample] = field(default_factory=list)
+    accumulator: ClipStreamAccumulator = field(
+        default_factory=lambda: ClipStreamAccumulator(clip_frames=CLIP_FRAMES)
+    )
     stored_clips: int = 0
     video_info: Any | None = None
 
@@ -41,13 +45,13 @@ def videoprism_sampling(config: IndexConfig, info) -> FrameSampling:
 
 
 def encode_video_clips(
-    clips: Sequence[Sequence[FrameSample]],
+    clips: Sequence[VideoClip],
     provider: VideoPrismModel,
 ) -> list[list[float]]:
     import torch
 
     inputs = provider.processor(
-        videos=[[sample.frame for sample in clip] for clip in clips],
+        videos=[[sample.frame for sample in clip.samples] for clip in clips],
         do_sample_frames=False,
         return_tensors="pt",
     )
@@ -59,15 +63,15 @@ def encode_video_clips(
 
 
 def videoprism_records(
-    clips: Sequence[Sequence[FrameSample]],
+    clips: Sequence[VideoClip],
     vectors: Sequence[Sequence[float]],
     info,
     config: IndexConfig,
 ) -> list[StorageRecord]:
     records = []
     cadence = 1 / min(info.fps, videoprism_config(config).sample_fps)
     for clip, vector in zip(clips, vectors):
-        first, last = clip[0], clip[-1]
+        first, last = clip.samples[0], clip.samples[-1]
         end = min(info.duration, last.timestamp + cadence)
         if end <= first.timestamp:
             end = first.timestamp + 1 / info.fps
@@ -91,15 +95,15 @@ def videoprism_records(
                     "end": end,
                     "fps": info.fps,
                     "duration": info.duration,
-                    "sample_count": len(clip),
+                    "sample_count": clip.sample_count,
                 },
             )
         )
     return records
 
 
 def _store_clips(
-    clips: Sequence[Sequence[FrameSample]],
+    clips: Sequence[VideoClip],
     *,
     state: VideoPrismIndexState,
     info,
@@ -111,7 +115,11 @@ def _store_clips(
     for group in batched(clips, settings.batch_size):
         cancellation.raise_if_cancelled()
         model_clips = [
-            list(clip) + [clip[-1]] * (CLIP_FRAMES - len(clip))
+            VideoClip(
+                samples=clip.samples + (clip.samples[-1],) * (CLIP_FRAMES - clip.sample_count),
+                start=clip.start,
+                end=clip.end,
+            )
             for clip in group
         ]
         vectors = encode_video_clips(model_clips, state.provider)
@@ -133,16 +141,9 @@ def process_videoprism_samples(
     cancellation: CancellationToken,
 ) -> None:
     state.video_info = info
-    state.pending.extend(samples)
-    complete = len(state.pending) // CLIP_FRAMES
-    if not complete:
+    clips = list(state.accumulator.add(samples))
+    if not clips:
         return
-    consumed = complete * CLIP_FRAMES
-    clips = [
-        state.pending[start : start + CLIP_FRAMES]
-        for start in range(0, consumed, CLIP_FRAMES)
-    ]
-    del state.pending[:consumed]
     _store_clips(
         clips,
         state=state,
@@ -165,13 +166,28 @@ def prepare(
         config: IndexConfig,
         runtime: ModelRuntimePort,
         progress: ProgressCallback | None,
+        source=None,
     ) -> VideoPrismIndexState:
         report_progress(
             progress,
             "preparing_videoprism_model",
             f"Preparing VideoPrism {VIDEOPRISM_MODEL.model_id}.",
         )
-        return VideoPrismIndexState(get_videoprism_model(runtime))
+        boundaries = None
+        if videoprism_config(config).clip_mode == "scene" and source is not None:
+            report_progress(
+                progress,
+                "detecting_shot_boundaries",
+                "Detecting shot boundaries with PySceneDetect.",
+            )
+            boundaries = detect_shot_boundaries(source.path)
+        return VideoPrismIndexState(
+            provider=get_videoprism_model(runtime),
+            accumulator=ClipStreamAccumulator(
+                clip_frames=CLIP_FRAMES,
+                boundaries=boundaries,
+            ),
+        )
 
     def process(
         self,
@@ -199,18 +215,18 @@ def finalize(
         config: IndexConfig,
         storage: IndexStore,
     ) -> tuple[dict[str, Any], int]:
-        if state.pending:
+        tail = state.accumulator.finalize()
+        if tail is not None:
             if state.video_info is None:
                 raise RuntimeError("VideoPrism indexing is missing video metadata.")
             _store_clips(
-                [state.pending],
+                [tail],
                 state=state,
                 info=state.video_info,
                 config=config,
                 storage=storage,
                 cancellation=CancellationToken(),
             )
-            state.pending.clear()
         return {"videoprism_clips": state.stored_clips}, state.stored_clips
 
 

src/vidxp/capabilities/actor/indexing.py

@@ -271,6 +271,7 @@ def prepare(
         config: IndexConfig,
         runtime: ModelRuntimePort,
         progress: ProgressCallback | None,
+        source=None,
     ) -> ActorIndexState:
         return ActorIndexState(models=get_actor_models(runtime))
 

src/vidxp/capabilities/scene/indexing.py

@@ -123,6 +123,7 @@ def prepare(
         config: IndexConfig,
         runtime: ModelRuntimePort,
         progress: ProgressCallback | None,
+        source=None,
     ) -> SceneIndexState:
         report_progress(
             progress,

src/vidxp/capabilities/visual.py

@@ -32,6 +32,7 @@ def prepare(
         config: IndexConfig,
         runtime: ModelRuntimePort,
         progress: ProgressCallback | None,
+        source: VideoSource | None = None,
     ) -> Any: ...
 
     def process(
@@ -84,6 +85,7 @@ def _participants(
     runtime: ModelRuntimePort,
     progress: ProgressCallback | None,
     timings: dict[str, float],
+    source: VideoSource | None = None,
 ) -> list[_Participant]:
     participants = []
     for name in names:
@@ -93,7 +95,7 @@ def _participants(
                 f"Capability {name!r} does not provide a visual processor."
             )
         started = perf_counter()
-        state = processor.prepare(config, runtime, progress)
+        state = processor.prepare(config, runtime, progress, source=source)
         timings[name] = perf_counter() - started
         sampling_factory = getattr(type(processor), "sampling", None)
         sampling = (
@@ -266,6 +268,7 @@ def index_visuals(
         runtime=runtime,
         progress=progress,
         timings=timings,
+        source=source,
     )
     expected = _expected_sample_count(info, participants)
     report_progress(

src/vidxp/core/clip.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Iterable, Iterator, Sequence
+
+from vidxp.core.video import FrameSample
+
+
+@dataclass(frozen=True)
+class VideoClip:
+    samples: tuple[FrameSample, ...]
+    start: float
+    end: float
+
+    @property
+    def sample_count(self) -> int:
+        return len(self.samples)
+
+
+@dataclass
+class ClipStreamAccumulator:
+    clip_frames: int
+    boundaries: Sequence[float] | None = None
+    pending: list = field(default_factory=list)
+    _next_boundary: int = 0
+
+    def add(self, samples: Iterable[FrameSample]) -> Iterator[VideoClip]:
+        for sample in samples:
+            while (
+                self.boundaries is not None
+                and self._next_boundary < len(self.boundaries)
+                and sample.timestamp >= self.boundaries[self._next_boundary]
+            ):
+                if self.pending:
+                    yield self._flush()
+                self._next_boundary += 1
+
+            self.pending.append(sample)
+
+            if len(self.pending) >= self.clip_frames:
+                yield self._flush()
+
+    def finalize(self) -> VideoClip | None:
+        return self._flush() if self.pending else None
+
+    def _flush(self) -> VideoClip:
+        samples = tuple(self.pending)
+        self.pending.clear()
+        return VideoClip(
+            samples=samples,
+            start=samples[0].timestamp,
+            end=samples[-1].timestamp,
+        )

src/vidxp/core/scene_boundaries.py

@@ -0,0 +1,15 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+
+def detect_shot_boundaries(path: str | Path) -> list[float]:
+    """Return sorted shot-end timestamps (seconds) using PySceneDetect.
+
+    An empty list means no cuts were found, so callers should treat
+    that as "one shot covering the whole video".
+    """
+    from scenedetect import ContentDetector, detect
+
+    scene_list = detect(str(path), ContentDetector())
+    return [end.get_seconds() for _start, end in scene_list]