From a16ef496e8b949f0b287166329faee5230388c51 Mon Sep 17 00:00:00 2001 From: sankalp nagaonkar Date: Tue, 15 Sep 2026 19:20:25 +0530 Subject: [PATCH] Expose native semantic-search windows and result metadata --- README.md | 39 +++++++++ tests/test_windowed_semantic_search.py | 109 +++++++++++++++++++++++++ videodb/__init__.py | 2 + videodb/collection.py | 6 ++ videodb/search.py | 3 + videodb/shot.py | 7 +- videodb/video.py | 6 ++ videodb/window.py | 26 ++++++ 8 files changed, 197 insertions(+), 1 deletion(-) create mode 100644 tests/test_windowed_semantic_search.py create mode 100644 videodb/window.py diff --git a/README.md b/README.md index d10f4a2..7de05d7 100644 --- a/README.md +++ b/README.md @@ -275,6 +275,45 @@ answer = video.ask( The same `search()`, `semantic_search()`, `query()`, `aggregate()`, and `ask()` methods are available on collections for retrieval across multiple videos. Existing applications can continue using `index_spoken_words()`, `index_scenes()`, and `legacy_search()` for legacy indexes. +### Windowed Semantic Search + +On a server with native window support, direct `semantic_search()` can merge nearby +hits into longer intervals. This works on both collections and videos with Search +V2 indexes; `ask()` remains a separate method. + +```python +result = collection.semantic_search( + "turn off tap", index_names=["episodes"], top_k=10, window="moment" +) +print(result.window_plan) # None if the server did not resolve windows +for shot in result.shots: + print(shot.start, shot.end, shot.window.merged_from, shot.window.coverage) + +# Explicit durations and context. All values are seconds. +result = video.semantic_search( + "opening the fridge", + window={"level": "moment", "min_seconds": 10, "max_seconds": 30}, +) +``` + +- `None` (default): unchanged retrieval of indexed intervals. +- `moment` / `span`: merge nearby hits using the server's configured gap (currently + 5 / 30 seconds by default). These are not fixed output lengths. +- `session`: return a whole-video interval per matching video; requires video length metadata. +- `auto`: keyword-based selection of moment/span/session, defaulting to moment. +- A dict accepts `level`, `level_seconds`, `merge_gap_seconds`, `min_seconds`, + `max_seconds`, `pad_before_seconds`, `pad_after_seconds`, and `whole_video`. + +`shot.window` is a `WindowInfo` with level, duration, supporting hit count/IDs, +coverage, and `to_dict()`. Coverage is the fraction of the returned interval covered +by retrieved hits, not ground-truth recall. Text/metadata represent the best +supporting hit, not a generated summary. Requests without windows have +`window_plan=None` and `shot.window=None`. + +There is no SDK fallback or extra retrieval call. Verify `result.window_plan` before +claiming server window support. Evidence, answers, time-lapse, and multi-scale +indexing are not part of this addition. + ### Working with Transcripts ```python diff --git a/tests/test_windowed_semantic_search.py b/tests/test_windowed_semantic_search.py new file mode 100644 index 0000000..3fa6e08 --- /dev/null +++ b/tests/test_windowed_semantic_search.py @@ -0,0 +1,109 @@ +"""Thin native window contract. No runtime patch or network dependencies.""" + +from copy import deepcopy + +import pytest +from videodb import WindowInfo +from videodb.collection import Collection +from videodb.search import SearchResult +from videodb.video import Video + + +class FakeConnection: + def __init__(self, response=None): + self.calls = [] + self.response = response or {"results": []} + + def post(self, path, data): + self.calls.append((path, deepcopy(data))) + return deepcopy(self.response) + + +@pytest.mark.parametrize("cls,scope", [(Collection, "collection"), (Video, "video")]) +def test_no_window_snapshot(cls, scope): + conn = FakeConnection() + obj = cls(conn, id="test", **({"collection_id": "c"} if cls is Video else {})) + result = obj.semantic_search("tap", ["episodes"], 10, 0.2, [], ["source"], ["idx"]) + assert conn.calls == [ + ( + f"{scope}/test/semantic-search", + { + "query": "tap", + "index_names": ["episodes"], + "index_ids": ["idx"], + "top_k": 10, + "score_threshold": 0.2, + "filter": [], + "return_fields": ["source"], + }, + ) + ] + assert result.window_plan is None + + +@pytest.mark.parametrize( + "window", ["moment", "span", "session", "auto", {"min_seconds": 10, "max_seconds": 20}] +) +@pytest.mark.parametrize("cls,scope", [(Collection, "collection"), (Video, "video")]) +def test_window_sent_once_without_fallback(window, cls, scope): + conn = FakeConnection() + result = cls(conn, id="test", **({"collection_id": "c"} if cls is Video else {})).semantic_search( + "tap", window=window + ) + assert len(conn.calls) == 1 + assert conn.calls[0][0] == f"{scope}/test/semantic-search" + assert conn.calls[0][1]["window"] == window + assert result.window_plan is None # An older server cannot be mistaken for native support. + + +def test_native_grouped_response_and_rank(): + info = { + "level": "moment", + "level_seconds": 5, + "merged_from": 2, + "seconds": 8, + "coverage": 0.75, + "hit_scene_ids": ["s1", "s2"], + "future_field": "ignored", + } + doc = { + "start": 0, + "end": 8, + "text": "tap", + "score": 0.9, + "window": info, + "metadata": {"source": "epic"}, + "scene_index_id": "idx", + "scene_index_name": "episodes", + } + groups = [ + {"collection_id": "c", "video_id": vid, "length": 300, "title": "Kitchen", "docs": [doc]} + for vid in ["a", "b", "a"] + ] + conn = FakeConnection({"results": groups, "window_plan": {"level": "moment", "how": "explicit"}}) + result = Collection(conn, "c").semantic_search("tap", window="moment") + assert [s.video_id for s in result.shots] == ["a", "b", "a"] + assert result.window_plan == {"level": "moment", "how": "explicit"} + shot = result.shots[0] + assert isinstance(shot.window, WindowInfo) + assert shot.window.hit_scene_ids == ["s1", "s2"] + assert shot.window.to_dict()["merged_from"] == 2 + assert shot.metadata == {"source": "epic"} + assert shot.scene_index_id == "idx" and shot.video_title == "Kitchen" + assert shot.search_score == 0.9 and len(conn.calls) == 1 + + +def test_old_shots_and_warnings(): + with pytest.warns(UserWarning, match="window metadata unavailable"): + result = SearchResult( + FakeConnection(), + results=[{"video_id": "a", "docs": [{"start": 0, "end": 3}]}], + warnings=[{"code": "window_test_warning", "message": "window metadata unavailable"}], + ) + assert result.shots[0].window is None + assert result.window_plan is None + + +def test_keyword_only_parameter(): + with pytest.raises(TypeError): + Collection(FakeConnection(), "c").semantic_search("q", None, 10, None, None, None, None, "moment") diff --git a/videodb/__init__.py b/videodb/__init__.py index 79293de..a8d7db4 100644 --- a/videodb/__init__.py +++ b/videodb/__init__.py @@ -30,6 +30,7 @@ ) from videodb.client import Connection from videodb.search import AskResponse, SearchResponse, SearchResult +from videodb.window import WindowInfo from videodb.understanding import Understanding, UnderstandingAnalyzer from videodb.job import GenerationJob from videodb.sandbox import Sandbox @@ -50,6 +51,7 @@ __all__ = [ "connect", + "WindowInfo", "CaptureSession", "GenerationJob", "Sandbox", diff --git a/videodb/collection.py b/videodb/collection.py index a36966f..742ffe5 100644 --- a/videodb/collection.py +++ b/videodb/collection.py @@ -751,6 +751,8 @@ def semantic_search( filter: Optional[Union[List, Dict]] = None, return_fields: Optional[Union[List, Dict, str]] = None, index_ids: Optional[Union[List[str], str]] = None, + *, + window: Optional[Union[str, Dict[str, Any]]] = None, ) -> SearchResult: """Run direct Search V2 semantic retrieval for this collection. @@ -765,6 +767,9 @@ def semantic_search( :param filter: Optional Search V2 filter. :param return_fields: Optional metadata fields to include. :param index_ids: Optional Search V2 index ID or IDs. + :param window: None: as indexed; 'auto', 'moment', 'span', 'session', or a bounds dict; + the server returns ``window_plan`` and ``shot.window``. Named levels guide + merging, not fixed clip lengths. Requires server support; no client fallback. :return: ``SearchResult`` with shots and server-provided ``warnings``. """ search_data = self._connection.post( @@ -777,6 +782,7 @@ def semantic_search( "score_threshold": score_threshold, "filter": filter, "return_fields": return_fields, + **({"window": window} if window is not None else {}), }, ) return SearchResult(self._connection, **search_data) diff --git a/videodb/search.py b/videodb/search.py index fac8682..b827db6 100644 --- a/videodb/search.py +++ b/videodb/search.py @@ -59,6 +59,7 @@ class SearchResult: :ivar str stream_url: URL to stream the search result :ivar str player_url: URL to play the search result in a player :ivar list[Shot] shots: List of shots in the search result + :ivar Optional[dict] window_plan: Native server window plan, otherwise None """ def __init__(self, _connection, **kwargs): @@ -67,6 +68,7 @@ def __init__(self, _connection, **kwargs): self.stream_url = None self.player_url = None self.collection_id = "default" + self.window_plan = kwargs.get("window_plan") self.warnings = kwargs.get("warnings") or [] if kwargs.get("_emit_warnings", True): warn_response_warnings_once(self.warnings, stacklevel=5) @@ -92,6 +94,7 @@ def _format_results(self): metadata=doc.get("metadata"), stream_url=doc.get("stream_link") or doc.get("stream_url"), player_url=doc.get("player_url"), + window=doc.get("window"), ) ) diff --git a/videodb/shot.py b/videodb/shot.py index ce4357c..473d24d 100644 --- a/videodb/shot.py +++ b/videodb/shot.py @@ -1,4 +1,6 @@ -from typing import Optional +from typing import Any, Dict, Optional, Union + +from videodb.window import WindowInfo from videodb._utils._video import play_stream, build_iframe_embed_code from videodb._constants import ( ApiPath, @@ -20,6 +22,7 @@ class Shot: :ivar Optional[str] scene_index_id: ID of the scene index for scene search results :ivar Optional[str] scene_index_name: Name of the scene index for scene search results :ivar Optional[dict] metadata: Additional metadata for the shot + :ivar Optional[WindowInfo] window: Server-resolved interval metadata, otherwise None """ def __init__( @@ -37,6 +40,7 @@ def __init__( metadata: Optional[dict] = None, stream_url: Optional[str] = None, player_url: Optional[str] = None, + window: Optional[Union[WindowInfo, Dict[str, Any]]] = None, ) -> None: self._connection = _connection self.video_id = video_id @@ -51,6 +55,7 @@ def __init__( self.metadata = metadata self.stream_url = stream_url self.player_url = player_url + self.window = window if isinstance(window, WindowInfo) else WindowInfo.from_dict(window) def __repr__(self) -> str: repr_str = ( diff --git a/videodb/video.py b/videodb/video.py index c3e014f..4b815a9 100644 --- a/videodb/video.py +++ b/videodb/video.py @@ -219,6 +219,8 @@ def semantic_search( filter: Optional[Union[List, Dict]] = None, return_fields: Optional[Union[List, Dict, str]] = None, index_ids: Optional[Union[List[str], str]] = None, + *, + window: Optional[Union[str, Dict[str, Any]]] = None, ) -> SearchResult: """Run direct Search V2 semantic retrieval for this video. @@ -233,6 +235,9 @@ def semantic_search( :param filter: Optional Search V2 filter. :param return_fields: Optional metadata fields to include. :param index_ids: Optional Search V2 index ID or IDs. + :param window: None: as indexed; 'auto', 'moment', 'span', 'session', or a bounds dict; + the server returns ``window_plan`` and ``shot.window``. Named levels guide + merging, not fixed clip lengths. Requires server support; no client fallback. :return: ``SearchResult`` with shots and server-provided ``warnings``. """ search_data = self._connection.post( @@ -245,6 +250,7 @@ def semantic_search( "score_threshold": score_threshold, "filter": filter, "return_fields": return_fields, + **({"window": window} if window is not None else {}), }, ) return SearchResult(self._connection, **search_data) diff --git a/videodb/window.py b/videodb/window.py new file mode 100644 index 0000000..a5cc2b5 --- /dev/null +++ b/videodb/window.py @@ -0,0 +1,26 @@ +"""Metadata for server-resolved semantic-search windows (no client fallback).""" + +from dataclasses import asdict, dataclass, field, fields +from typing import Any, Dict, List, Optional + + +@dataclass +class WindowInfo: + """How a returned interval was formed; coverage is hit coverage, not recall.""" + + level: Optional[str] = None + level_seconds: Optional[float] = None + merged_from: int = 0 + coverage: float = 0.0 + seconds: float = 0.0 + hit_scene_ids: List[str] = field(default_factory=list) + + @classmethod + def from_dict(cls, value: Optional[Dict[str, Any]]) -> Optional["WindowInfo"]: + if value is None: + return None + names = {item.name for item in fields(cls)} + return cls(**{key: item for key, item in value.items() if key in names}) + + def to_dict(self) -> Dict[str, Any]: + return asdict(self)