From b05161cd14f2a0f993f1c31ea2a0c86e2ebac6fb Mon Sep 17 00:00:00 2001 From: ik020 Date: Sat, 5 Sep 2026 09:33:39 +0500 Subject: [PATCH 1/4] refactor(action): introduce VideoClip abstraction for #83 Extracts clip grouping into core/clip.py (VideoClip, ClipStreamAccumulator), decoupling clip boundaries from the VideoPrism encoder. Fixed-window mode preserves existing behavior exactly (verified by existing test suite). --- src/vidxp/capabilities/action/config.py | 1 + src/vidxp/capabilities/action/indexing.py | 41 ++++++++++--------- src/vidxp/core/clip.py | 50 +++++++++++++++++++++++ tests/test_videoprism.py | 2 +- 4 files changed, 73 insertions(+), 21 deletions(-) create mode 100644 src/vidxp/core/clip.py diff --git a/src/vidxp/capabilities/action/config.py b/src/vidxp/capabilities/action/config.py index 97431dfa..003025ce 100644 --- a/src/vidxp/capabilities/action/config.py +++ b/src/vidxp/capabilities/action/config.py @@ -9,6 +9,7 @@ class VideoPrismConfig(CapabilityConfig): batch_size: int = Field(default=1, gt=0) sample_fps: float = Field(default=2.0, gt=0) + clip_mode: str = 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..c5da76a2 100644 --- a/src/vidxp/capabilities/action/indexing.py +++ b/src/vidxp/capabilities/action/indexing.py @@ -20,6 +20,7 @@ from vidxp.core.indexing_common import ProgressCallback, report_progress from vidxp.core.video import FrameSample, FrameSampling from vidxp.ports import IndexStore, ModelRuntimePort +from vidxp.core.clip import ClipStreamAccumulator, VideoClip CLIP_FRAMES = 16 @@ -28,7 +29,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 +44,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 +62,7 @@ def encode_video_clips( def videoprism_records( - clips: Sequence[Sequence[FrameSample]], + clips: Sequence[VideoClip], vectors: Sequence[Sequence[float]], info, config: IndexConfig, @@ -67,7 +70,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 +94,7 @@ def videoprism_records( "end": end, "fps": info.fps, "duration": info.duration, - "sample_count": len(clip), + "sample_count": clip.sample_count, }, ) ) @@ -99,7 +102,7 @@ def videoprism_records( def _store_clips( - clips: Sequence[Sequence[FrameSample]], + clips: Sequence[VideoClip], *, state: VideoPrismIndexState, info, @@ -110,8 +113,13 @@ def _store_clips( settings = videoprism_config(config) 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, @@ -199,18 +200,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/core/clip.py b/src/vidxp/core/clip.py new file mode 100644 index 00000000..203a976e --- /dev/null +++ b/src/vidxp/core/clip.py @@ -0,0 +1,50 @@ +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: + self.pending.append(sample) + crossed = ( + self.boundaries is not None + and self._next_boundary < len(self.boundaries) + and sample.timestamp >= self.boundaries[self._next_boundary] + ) + if crossed: + self._next_boundary += 1 + if crossed or 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/tests/test_videoprism.py b/tests/test_videoprism.py index a7140eb9..86d9f716 100644 --- a/tests/test_videoprism.py +++ b/tests/test_videoprism.py @@ -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] From 9861bfbfce48513aa4df91326dcae8d0db8fdcfd Mon Sep 17 00:00:00 2001 From: ik020 Date: Sat, 5 Sep 2026 09:46:07 +0500 Subject: [PATCH 2/4] feat(action): wire PySceneDetect shot boundaries for #83 Adds core/scene_boundaries.py (PySceneDetect wrapper) and threads source through the shared VisualProcessor.prepare() so action's indexer can detect real shot boundaries when clip_mode='scene'. Default clip_mode='fixed' behavior is unchanged. --- src/vidxp/capabilities/action/indexing.py | 18 +++++++++++++++++- src/vidxp/capabilities/actor/indexing.py | 1 + src/vidxp/capabilities/scene/indexing.py | 1 + src/vidxp/capabilities/visual.py | 5 ++++- src/vidxp/core/scene_boundaries.py | 15 +++++++++++++++ 5 files changed, 38 insertions(+), 2 deletions(-) create mode 100644 src/vidxp/core/scene_boundaries.py diff --git a/src/vidxp/capabilities/action/indexing.py b/src/vidxp/capabilities/action/indexing.py index c5da76a2..d457ab3e 100644 --- a/src/vidxp/capabilities/action/indexing.py +++ b/src/vidxp/capabilities/action/indexing.py @@ -21,6 +21,7 @@ from vidxp.core.video import FrameSample, FrameSampling from vidxp.ports import IndexStore, ModelRuntimePort from vidxp.core.clip import ClipStreamAccumulator, VideoClip +from vidxp.core.scene_boundaries import detect_shot_boundaries CLIP_FRAMES = 16 @@ -166,13 +167,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, 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/scene_boundaries.py b/src/vidxp/core/scene_boundaries.py new file mode 100644 index 00000000..7ba12609 --- /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 (or detection failed) - + 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] From 845ae862193e180ccefffc06290000f5213e7b6e Mon Sep 17 00:00:00 2001 From: ik020 Date: Sat, 5 Sep 2026 09:48:51 +0500 Subject: [PATCH 3/4] test(action): cover clip_mode=scene shot-boundary flushing Verifies prepare() calls detect_shot_boundaries and passes them into ClipStreamAccumulator, and that process/finalize produce clips whose start/end/sample_count respect the shot boundaries rather than fixed 16-frame windows. --- tests/test_videoprism.py | 70 +++++++++++++++++++++++++++++++++++++++- 1 file changed, 69 insertions(+), 1 deletion(-) diff --git a/tests/test_videoprism.py b/tests/test_videoprism.py index 86d9f716..c69fb3af 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 @@ -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], + [5, 6, 7], + ) + self.assertEqual( + [ + (record.metadata["start"], record.metadata["end"]) + for record in records + ], + [(0.0, 2.5), (2.5, 5.5), (5.5, 9.0)], + ) + + if __name__ == "__main__": unittest.main() From 9983e019e15f8ee8fc1e99c5e2478668235ee85f Mon Sep 17 00:00:00 2001 From: ik020 Date: Sat, 5 Sep 2026 10:49:50 +0500 Subject: [PATCH 4/4] feat(action): finalize clip mode abstraction for #83 --- scripts/benchmark_clip_modes.py | 107 ++++++++++++++++++++++ src/vidxp/capabilities/action/config.py | 4 +- src/vidxp/capabilities/action/indexing.py | 5 +- src/vidxp/core/clip.py | 13 ++- src/vidxp/core/scene_boundaries.py | 4 +- tests/test_videoprism.py | 4 +- 6 files changed, 124 insertions(+), 13 deletions(-) create mode 100644 scripts/benchmark_clip_modes.py 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 003025ce..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,7 +11,7 @@ class VideoPrismConfig(CapabilityConfig): batch_size: int = Field(default=1, gt=0) sample_fps: float = Field(default=2.0, gt=0) - clip_mode: str = Field(default="fixed") + 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 d457ab3e..d6d58c38 100644 --- a/src/vidxp/capabilities/action/indexing.py +++ b/src/vidxp/capabilities/action/indexing.py @@ -18,10 +18,10 @@ stable_source_id, ) from vidxp.core.indexing_common import ProgressCallback, report_progress -from vidxp.core.video import FrameSample, FrameSampling -from vidxp.ports import IndexStore, ModelRuntimePort 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 CLIP_FRAMES = 16 @@ -114,7 +114,6 @@ def _store_clips( settings = videoprism_config(config) for group in batched(clips, settings.batch_size): cancellation.raise_if_cancelled() - model_clips = [ VideoClip( samples=clip.samples + (clip.samples[-1],) * (CLIP_FRAMES - clip.sample_count), diff --git a/src/vidxp/core/clip.py b/src/vidxp/core/clip.py index 203a976e..3a6f7c36 100644 --- a/src/vidxp/core/clip.py +++ b/src/vidxp/core/clip.py @@ -26,15 +26,18 @@ class ClipStreamAccumulator: def add(self, samples: Iterable[FrameSample]) -> Iterator[VideoClip]: for sample in samples: - self.pending.append(sample) - crossed = ( + while ( self.boundaries is not None and self._next_boundary < len(self.boundaries) and sample.timestamp >= self.boundaries[self._next_boundary] - ) - if crossed: + ): + if self.pending: + yield self._flush() self._next_boundary += 1 - if crossed or len(self.pending) >= self.clip_frames: + + self.pending.append(sample) + + if len(self.pending) >= self.clip_frames: yield self._flush() def finalize(self) -> VideoClip | None: diff --git a/src/vidxp/core/scene_boundaries.py b/src/vidxp/core/scene_boundaries.py index 7ba12609..88d4561b 100644 --- a/src/vidxp/core/scene_boundaries.py +++ b/src/vidxp/core/scene_boundaries.py @@ -6,8 +6,8 @@ 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 (or detection failed) - - callers should treat that as "one shot covering the whole video". + 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 diff --git a/tests/test_videoprism.py b/tests/test_videoprism.py index c69fb3af..b4af3e5f 100644 --- a/tests/test_videoprism.py +++ b/tests/test_videoprism.py @@ -144,14 +144,14 @@ def test_scene_clip_mode_flushes_clips_at_shot_boundaries(self): records = [call.args[1][0] for call in storage.upsert.call_args_list] self.assertEqual( [record.metadata["sample_count"] for record in records], - [5, 6, 7], + [4, 6, 8], ) self.assertEqual( [ (record.metadata["start"], record.metadata["end"]) for record in records ], - [(0.0, 2.5), (2.5, 5.5), (5.5, 9.0)], + [(0.0, 2.0), (2.0, 5.0), (5.0, 9.0)], )