diff --git a/scripts/benchmark_clip_modes.py b/scripts/benchmark_clip_modes.py new file mode 100644 index 00000000..f31c04c3 --- /dev/null +++ b/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 ") + 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() diff --git a/src/vidxp/capabilities/action/config.py b/src/vidxp/capabilities/action/config.py index 97431dfa..d7598add 100644 --- a/src/vidxp/capabilities/action/config.py +++ b/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: diff --git a/src/vidxp/capabilities/action/indexing.py b/src/vidxp/capabilities/action/indexing.py index cf397418..d6d58c38 100644 --- a/src/vidxp/capabilities/action/indexing.py +++ b/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,7 +63,7 @@ def encode_video_clips( def videoprism_records( - clips: Sequence[Sequence[FrameSample]], + clips: Sequence[VideoClip], vectors: Sequence[Sequence[float]], info, config: IndexConfig, @@ -67,7 +71,7 @@ def videoprism_records( 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,7 +95,7 @@ def videoprism_records( "end": end, "fps": info.fps, "duration": info.duration, - "sample_count": len(clip), + "sample_count": clip.sample_count, }, ) ) @@ -99,7 +103,7 @@ def videoprism_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 diff --git a/src/vidxp/capabilities/actor/indexing.py b/src/vidxp/capabilities/actor/indexing.py index 35da06c2..58ab1314 100644 --- a/src/vidxp/capabilities/actor/indexing.py +++ b/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)) diff --git a/src/vidxp/capabilities/scene/indexing.py b/src/vidxp/capabilities/scene/indexing.py index 5f721e61..0e8e56df 100644 --- a/src/vidxp/capabilities/scene/indexing.py +++ b/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, diff --git a/src/vidxp/capabilities/visual.py b/src/vidxp/capabilities/visual.py index a49e2dd6..cd6ff74c 100644 --- a/src/vidxp/capabilities/visual.py +++ b/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( diff --git a/src/vidxp/core/clip.py b/src/vidxp/core/clip.py new file mode 100644 index 00000000..3a6f7c36 --- /dev/null +++ b/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, + ) diff --git a/src/vidxp/core/scene_boundaries.py b/src/vidxp/core/scene_boundaries.py new file mode 100644 index 00000000..88d4561b --- /dev/null +++ b/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] diff --git a/tests/test_videoprism.py b/tests/test_videoprism.py index a7140eb9..b4af3e5f 100644 --- a/tests/test_videoprism.py +++ b/tests/test_videoprism.py @@ -12,7 +12,7 @@ ) from vidxp.capabilities.action.models import normalize_pooled_output from vidxp.capabilities.action.specs import VIDEOPRISM_MODEL -from vidxp.core.contracts import CancellationToken, IndexConfig +from vidxp.core.contracts import CancellationToken, IndexConfig, VideoSource from vidxp.core.video import FrameSample, VideoInfo @@ -67,7 +67,7 @@ def test_streaming_index_groups_clips_and_pads_only_the_tail(self): self.assertEqual(summary, {"videoprism_clips": 2}) self.assertEqual(operations, 2) self.assertEqual( - [len(call.args[0][0]) for call in encode.call_args_list], + [call.args[0][0].sample_count for call in encode.call_args_list], [CLIP_FRAMES, CLIP_FRAMES], ) tail = storage.upsert.call_args_list[1].args[1][0] @@ -87,5 +87,73 @@ def test_model_contract_pins_the_pytorch_checkpoint(self): self.assertEqual(VIDEOPRISM_MODEL.download_size_bytes, 993_993_146) + def test_scene_clip_mode_flushes_clips_at_shot_boundaries(self): + config = IndexConfig( + video_id="video-1", + enabled_modalities=("action",), + capability_options={"action": {"clip_mode": "scene"}}, + ) + source = VideoSource(path="fake.mp4") + + with ( + patch( + "vidxp.capabilities.action.indexing.detect_shot_boundaries", + return_value=[2.0, 5.0], + ) as boundaries_mock, + patch( + "vidxp.capabilities.action.indexing.get_videoprism_model", + return_value=Mock(), + ), + ): + state = VISUAL_PROCESSOR.prepare(config, Mock(), None, source=source) + + boundaries_mock.assert_called_once_with("fake.mp4") + self.assertEqual(state.accumulator.boundaries, [2.0, 5.0]) + + info = VideoInfo(30.0, 270, 9.0, 2, 2) + samples = [ + FrameSample(index * 15, index / 2, object()) + for index in range(18) + ] + storage = Mock() + storage.upsert.side_effect = lambda _name, records, **_kwargs: len( + records + ) + + with patch( + "vidxp.capabilities.action.indexing.encode_video_clips", + side_effect=lambda clips, _provider: [[0.1] for _ in clips], + ): + process_videoprism_samples( + samples, + state=state, + info=info, + config=config, + storage=storage, + cancellation=CancellationToken(), + ) + summary, operations = VISUAL_PROCESSOR.finalize( + state, + config=config, + storage=storage, + ) + + self.assertEqual(summary, {"videoprism_clips": 3}) + self.assertEqual(operations, 3) + + records = [call.args[1][0] for call in storage.upsert.call_args_list] + self.assertEqual( + [record.metadata["sample_count"] for record in records], + [4, 6, 8], + ) + self.assertEqual( + [ + (record.metadata["start"], record.metadata["end"]) + for record in records + ], + [(0.0, 2.0), (2.0, 5.0), (5.0, 9.0)], + ) + + if __name__ == "__main__": unittest.main()