diff --git a/tests/test_search_legacy_routing.py b/tests/test_search_legacy_routing.py new file mode 100644 index 0000000..8ccfe70 --- /dev/null +++ b/tests/test_search_legacy_routing.py @@ -0,0 +1,70 @@ +import warnings + +from videodb.collection import Collection +from videodb.video import Video +import videodb.search as search_module + + +class FakeConnection: + def __init__(self, response=None): + self.posts = [] + self.response = response or {"results": []} + + def post(self, path, data=None, **kwargs): + self.posts.append({"path": path, "data": data or {}, "kwargs": kwargs}) + return self.response + + +def setup_function(): + search_module._RESPONSE_WARNING_EMITTED.clear() + + +def test_video_search_routes_stitch_rerank_options_to_legacy_endpoint_without_local_warning(): + conn = FakeConnection() + video = Video(conn, id="video-1", collection_id="collection-1") + + with warnings.catch_warnings(record=True) as emitted: + warnings.simplefilter("always") + result = video.search( + "find moments", + stitch=False, + rerank=True, + rerank_params={"top_k": 3}, + ) + + assert len(emitted) == 0 + assert result.warnings == [] + assert len(conn.posts) == 1 + assert conn.posts[0]["path"] == "video/video-1/search" + assert conn.posts[0]["data"]["stitch"] is False + assert conn.posts[0]["data"]["rerank"] is True + assert conn.posts[0]["data"]["rerank_params"] == {"top_k": 3} + + +def test_collection_search_routes_stitch_rerank_options_to_legacy_endpoint(): + conn = FakeConnection() + collection = Collection(conn, id="collection-1") + + collection.search( + "find moments", + stitch=False, + rerank=True, + rerank_params={"rank_fields": ["text"]}, + ) + + assert len(conn.posts) == 1 + assert conn.posts[0]["path"] == "collection/collection-1/search" + assert conn.posts[0]["data"]["stitch"] is False + assert conn.posts[0]["data"]["rerank"] is True + assert conn.posts[0]["data"]["rerank_params"] == {"rank_fields": ["text"]} + + +def test_video_search_still_routes_to_v2_when_no_legacy_params_are_present(): + conn = FakeConnection({"response_type": "shots", "results": []}) + video = Video(conn, id="video-1", collection_id="collection-1") + + video.search("find moments") + + assert len(conn.posts) == 1 + assert conn.posts[0]["path"] == "video/video-1/search/v2" + assert conn.posts[0]["data"] == {"query": "find moments"} diff --git a/videodb/__about__.py b/videodb/__about__.py index b4566c7..7f103dd 100644 --- a/videodb/__about__.py +++ b/videodb/__about__.py @@ -2,7 +2,7 @@ -__version__ = "0.4.5" +__version__ = "0.5.0" __title__ = "videodb" __author__ = "videodb" __email__ = "contact@videodb.io" diff --git a/videodb/__init__.py b/videodb/__init__.py index 59d2a1b..b736c88 100644 --- a/videodb/__init__.py +++ b/videodb/__init__.py @@ -8,6 +8,8 @@ from videodb._constants import ( VIDEO_DB_API, IndexType, + IndexCapability, + FieldGroup, SceneExtractionType, MediaType, SearchType, @@ -25,6 +27,8 @@ RTStreamChannelType, ) from videodb.client import Connection +from videodb.search import AskResponse, SearchResponse, SearchResult +from videodb.understanding import Understanding, UnderstandingAnalyzer from videodb.capture_session import CaptureSession from videodb.websocket_client import WebSocketConnection from videodb.capture import CaptureClient, Channel, AudioChannel, VideoChannel, Channels, ChannelList @@ -53,7 +57,14 @@ "AuthenticationError", "InvalidRequestError", "IndexType", + "IndexCapability", + "FieldGroup", "SearchError", + "SearchResult", + "SearchResponse", + "AskResponse", + "Understanding", + "UnderstandingAnalyzer", "play_stream", "build_iframe_embed_code", "MediaType", diff --git a/videodb/_constants.py b/videodb/_constants.py index f7de578..cdfcb5c 100644 --- a/videodb/_constants.py +++ b/videodb/_constants.py @@ -36,6 +36,23 @@ class IndexType: scene = "scene" +class IndexCapability: + """Retrieval capabilities an index can be built for (``use_for``).""" + + semantic = "semantic" + query = "query" + aggregate = "aggregate" + + +class FieldGroup: + """Field groups that map artifact fields to retrieval capabilities.""" + + semantic = "semantic" + filter = "filter" + aggregate = "aggregate" + sort = "sort" + + class SceneExtractionType: shot_based = "shot" time_based = "time" @@ -80,7 +97,14 @@ class ApiPath: upload_url = "upload_url" transcription = "transcription" index = "index" + indexes = "indexes" + records = "records" + understand = "understand" search = "search" + ask = "ask" + semantic_search = "semantic-search" + query = "query" + aggregate = "aggregate" compile = "compile" workflow = "workflow" timeline = "timeline" @@ -123,6 +147,11 @@ class Status: in_progress = "in progress" +INDEX_TERMINAL_STATUSES = {"ready", "failed"} +UNDERSTANDING_TERMINAL_STATUSES = {"done", "failed"} +ANALYZER_TERMINAL_STATUSES = {"done", "failed", "skipped", "cancelled"} + + class MeetingStatus: initializing = "initializing" processing = "processing" diff --git a/videodb/collection.py b/videodb/collection.py index 7f57fc3..6414080 100644 --- a/videodb/collection.py +++ b/videodb/collection.py @@ -1,6 +1,6 @@ import logging -from typing import Optional, Union, List, Dict, Any, Literal +from typing import Optional, Union, List, Dict, Any, Literal, Tuple from videodb._upload import ( upload, ) @@ -17,7 +17,13 @@ from videodb.meeting import Meeting from videodb.capture_session import CaptureSession from videodb.rtstream import RTStream, RTStreamSearchResult, RTStreamShot -from videodb.search import SearchFactory, SearchResult +from videodb.search import ( + AskResponse, + SearchFactory, + SearchResponse, + SearchResult, + warn_response_warnings_once, +) logger = logging.getLogger(__name__) @@ -463,6 +469,243 @@ def dub_video( return Video(self._connection, **dub_data) def search( + self, + query: str, + *args, + config: Optional[Dict[str, Any]] = None, + **kwargs, + ) -> Union[SearchResponse, SearchResult, RTStreamSearchResult]: + """Search this collection using Search V2 by default. + + Pass Search V2 options such as ``top_k``, ``mode``, ``return_fields``, + ``include_clip``, ``session_id``, or ``config`` for the new search API. + Calls with legacy options such as ``search_type``, ``index_type``, + ``result_threshold``, ``scene_index_id``, ``index_id``, ``stitch``, + ``rerank``, or ``rerank_params`` are routed to :meth:`legacy_search`. + Do not mix Search V2 and legacy options in one call. + + Server-provided warnings are exposed on ``response.warnings`` and are + emitted as Python ``UserWarning`` messages. + + :param str query: Natural-language search query. + :param dict config: Optional Search V2 request configuration. + :return: ``SearchResponse`` for Search V2, ``SearchResult`` for legacy, or ``RTStreamSearchResult`` for RTStream legacy search. + """ + old_params = { + "search_type", + "index_type", + "result_threshold", + "dynamic_score_percentage", + "scene_index_id", + "index_id", + "algorithm", + "sort_docs_on", + "namespace", + "stitch", + "rerank", + "rerank_params", + } + new_params = { + "top_k", + "mode", + "return_fields", + "include_clip", + "session_id", + "config", + } + unsupported_params = {"index_name", "index_names", "index_ids"} + + if config is not None: + kwargs["config"] = config + + if args: + legacy_arg_names = [ + "search_type", + "index_type", + "result_threshold", + "score_threshold", + "dynamic_score_percentage", + "filter", + "sort_docs_on", + "namespace", + "scene_index_id", + ] + for name, value in zip(legacy_arg_names, args): + kwargs.setdefault(name, value) + + has_old = bool(args) or any(k in kwargs and kwargs[k] is not None for k in old_params) + has_new = any(k in kwargs and kwargs[k] is not None for k in new_params) + has_unsupported = any(k in kwargs and kwargs[k] is not None for k in unsupported_params) + + if has_old and (has_new or has_unsupported): + raise ValueError( + "Cannot mix legacy search parameters with Search V2 parameters. " + "Use legacy_search(...) for older indexes, or remove legacy parameters and use Search V2 search(...)." + ) + if has_unsupported: + raise ValueError( + "search() chooses indexes automatically and does not accept index selectors. " + "Use semantic_search() for semantic index selection, query() for structured filtering, or aggregate() for counts and facets." + ) + + if has_old: + return self.legacy_search(query=query, _skip_warning=True, **kwargs) + + return self._new_search(query=query, **kwargs) + + def _new_search(self, query: str, **kwargs) -> SearchResponse: + payload = {"query": query, **{k: v for k, v in kwargs.items() if v is not None}} + search_data = self._connection.post( + path=f"{ApiPath.collection}/{self.id}/{ApiPath.search}/v2", + data=payload, + ) + return SearchResponse(self._connection, **search_data) + + def ask( + self, + question: str, + top_k: int = 15, + mode: str = "default", + include_sources: bool = False, + ) -> AskResponse: + """Ask a question over this collection's Search V2 indexes. + + ``ask()`` is Search V2 only; it does not search legacy indexes. If the + server cannot answer from indexed content, warnings are available on + ``response.warnings``. + + :param str question: Question to answer from the collection. + :param int top_k: Maximum number of source shots to retrieve. + :param str mode: Search mode to use. + :param bool include_sources: Include source shots in the response. + :return: ``AskResponse`` with ``answer``, ``sources``, and ``warnings``. + """ + ask_data = self._connection.post( + path=f"{ApiPath.collection}/{self.id}/{ApiPath.ask}", + data={ + "question": question, + "top_k": top_k, + "mode": mode, + "include_sources": include_sources, + }, + ) + return AskResponse(self._connection, **ask_data) + + def semantic_search( + self, + query: str, + index_names: Optional[Union[List[str], str]] = None, + top_k: int = 10, + score_threshold: Optional[float] = None, + filter: Optional[Union[List, Dict]] = None, + return_fields: Optional[Union[List, Dict, str]] = None, + index_ids: Optional[Union[List[str], str]] = None, + ) -> SearchResult: + """Run direct Search V2 semantic retrieval for this collection. + + Use ``index_names`` or ``index_ids`` to target semantic indexes. Singular + legacy selectors such as ``index_name``/``index_id`` are not accepted by + this method. + + :param str query: Natural-language query. + :param index_names: Optional Search V2 semantic index name or names. + :param int top_k: Maximum number of shots to return. + :param float score_threshold: Optional minimum similarity score. + :param filter: Optional Search V2 filter. + :param return_fields: Optional metadata fields to include. + :param index_ids: Optional Search V2 index ID or IDs. + :return: ``SearchResult`` with shots and server-provided ``warnings``. + """ + search_data = self._connection.post( + path=f"{ApiPath.collection}/{self.id}/{ApiPath.semantic_search}", + data={ + "query": query, + "index_names": index_names, + "index_ids": index_ids, + "top_k": top_k, + "score_threshold": score_threshold, + "filter": filter, + "return_fields": return_fields, + }, + ) + return SearchResult(self._connection, **search_data) + + def query( + self, + index_name: Optional[str] = None, + filter: Optional[Union[List, Dict]] = None, + limit: int = 100, + return_fields: Optional[Union[List, Dict, str]] = None, + sort: Optional[Union[str, List[Tuple[str, str]]]] = None, + index_id: Optional[str] = None, + ) -> SearchResult: + """Run a structured Search V2 query for this collection. + + ``query()`` is V2-only and is intended for filtering, sorting, and + retrieving indexed records without natural-language planning. + + :param str index_name: Optional Search V2 index name. + :param filter: Optional Search V2 filter. + :param int limit: Maximum number of records to return. + :param return_fields: Optional fields to include in each result. + :param sort: Optional sort field or ``[(field, direction)]`` list. + :param str index_id: Optional Search V2 index ID. + :return: ``SearchResult`` with shots and server-provided ``warnings``. + """ + query_data = self._connection.post( + path=f"{ApiPath.collection}/{self.id}/{ApiPath.query}", + data={ + "index_name": index_name, + "index_id": index_id, + "filter": filter, + "limit": limit, + "return_fields": return_fields, + "sort": sort, + }, + ) + return SearchResult(self._connection, **query_data) + + def aggregate( + self, + index_name: Optional[str] = None, + filter: Optional[Union[List, Dict]] = None, + group_by: Optional[str] = None, + metric: str = "count", + limit: int = 100, + sort: Optional[Union[str, List[Tuple[str, str]]]] = None, + index_id: Optional[str] = None, + ) -> Union[Dict, List[Dict]]: + """Run a Search V2 aggregate over this collection's indexed records. + + Use this for counts, facets, and grouped metrics. ``aggregate()`` is + V2-only and returns the server aggregate payload directly. + + :param str index_name: Optional Search V2 index name. + :param filter: Optional Search V2 filter. + :param str group_by: Optional field to group by. + :param str metric: Aggregate metric, default ``"count"``. + :param int limit: Maximum number of aggregate rows. + :param sort: Optional sort field or ``[(field, direction)]`` list. + :param str index_id: Optional Search V2 index ID. + :return: Aggregate dict/list; dict responses may include ``warnings``. + """ + aggregate_data = self._connection.post( + path=f"{ApiPath.collection}/{self.id}/{ApiPath.aggregate}", + data={ + "index_name": index_name, + "index_id": index_id, + "filter": filter, + "group_by": group_by, + "metric": metric, + "limit": limit, + "sort": sort, + }, + ) + if isinstance(aggregate_data, dict): + warn_response_warnings_once(aggregate_data.get("warnings") or []) + return aggregate_data + + def legacy_search( self, query: str, search_type: Optional[str] = SearchType.semantic, @@ -474,25 +717,43 @@ def search( sort_docs_on: Optional[str] = None, namespace: Optional[str] = None, scene_index_id: Optional[str] = None, + index_id: Optional[str] = None, + algorithm: Optional[str] = None, + stitch: Optional[bool] = None, + rerank: Optional[bool] = None, + rerank_params: Optional[Dict[str, Any]] = None, + _skip_warning: bool = False, ) -> Union[SearchResult, RTStreamSearchResult]: - """Search for a query in the collection. - - :param str query: Query to search for - :param SearchType search_type: Type of search to perform (optional) - :param IndexType index_type: Type of index to search (optional) - :param int result_threshold: Number of results to return (optional) - :param float score_threshold: Threshold score for the search (optional) - :param float dynamic_score_percentage: Percentage of dynamic score to consider (optional) - :param list filter: Additional metadata filters (optional) - :param str sort_docs_on: Sort docs within each video by "score" or "start" (optional) - :param str namespace: Search namespace (optional, "rtstream" to search RTStreams) - :param str scene_index_id: Filter by specific scene index (optional) - :raise SearchError: If the search fails - :return: :class:`SearchResult ` or - :class:`RTStreamSearchResult ` object - :rtype: Union[:class:`videodb.search.SearchResult`, - :class:`videodb.rtstream.RTStreamSearchResult`] + """Search this collection using legacy spoken-word, scene, or RTStream indexes. + + Use this when you intentionally want older indexes. New applications + should prefer :meth:`search`, :meth:`semantic_search`, :meth:`query`, + :meth:`aggregate`, or :meth:`ask` for Search V2 indexes. Set + ``namespace="rtstream"`` to search legacy RTStream results. The + ``index_id`` keyword is accepted as an alias for ``scene_index_id``. + + Server-provided migration warnings are available on ``result.warnings``. + + :param str query: Query to search for. + :param SearchType search_type: Legacy search type. + :param IndexType index_type: Legacy index type. + :param int result_threshold: Number of results to return. + :param float score_threshold: Minimum score threshold. + :param float dynamic_score_percentage: Dynamic score percentage. + :param list filter: Legacy metadata filters. + :param str sort_docs_on: Sort docs within each video by ``"score"`` or ``"start"``. + :param str namespace: Search namespace; use ``"rtstream"`` for RTStream legacy search. + :param str scene_index_id: Filter by a specific scene index. + :param str index_id: Alias for ``scene_index_id``. + :param str algorithm: Legacy keyword search algorithm when applicable. + :param bool stitch: Whether legacy search should stitch adjacent results. + :param bool rerank: Whether legacy search should rerank results. + :param dict rerank_params: Legacy reranking parameters. + :return: ``SearchResult`` with warnings, or ``RTStreamSearchResult`` for RTStream legacy search. """ + if scene_index_id is None and index_id is not None: + scene_index_id = index_id + if namespace == "rtstream": data = {"query": query} if scene_index_id is not None: @@ -528,6 +789,14 @@ def search( ] return RTStreamSearchResult(collection_id=self.id, shots=shots) + legacy_options = {} + if stitch is not None: + legacy_options["stitch"] = stitch + if rerank is not None: + legacy_options["rerank"] = rerank + if rerank_params is not None: + legacy_options["rerank_params"] = rerank_params + search = SearchFactory(self._connection).get_search(search_type) return search.search_inside_collection( collection_id=self.id, @@ -539,6 +808,9 @@ def search( dynamic_score_percentage=dynamic_score_percentage, sort_docs_on=sort_docs_on, filter=filter, + scene_index_id=scene_index_id, + algorithm=algorithm, + **legacy_options, ) def search_title(self, query) -> List[Video]: diff --git a/videodb/index.py b/videodb/index.py new file mode 100644 index 0000000..9c2d93b --- /dev/null +++ b/videodb/index.py @@ -0,0 +1,273 @@ +import time + +from typing import List, Optional + +from videodb._constants import ApiPath, INDEX_TERMINAL_STATUSES + + +class FieldSchema: + """Schema details for a single indexed field. + + :ivar str type: Data type of the field (e.g. ``"string"``, ``"string_array"``, ``"number"``, ``"text"``, ``"boolean"``) + :ivar list groups: Field groups this field belongs to (e.g. ``["semantic", "filter"]``) + :ivar list operators: Filter operators supported by the field (for filterable fields) + """ + + def __init__( + self, + type: Optional[str] = None, + groups: Optional[List[str]] = None, + operators: Optional[List[str]] = None, + ) -> None: + self.type = type + self.groups = groups or [] + self.operators = operators or [] + + def __repr__(self) -> str: + return ( + f"FieldSchema(" + f"type={self.type}, " + f"groups={self.groups}, " + f"operators={self.operators})" + ) + + def __getitem__(self, key): + return self.__dict__[key] + + +class IndexRecord: + """A single indexed record (one temporal scene of an index). + + :ivar str video_id: ID of the video the record belongs to + :ivar str understanding_id: ID of the understanding run that produced the record + :ivar str scene_id: ID of the scene within the artifact + :ivar float start: Start time of the scene in seconds + :ivar float end: End time of the scene in seconds + :ivar dict data: Indexed field values for the scene + :ivar str segment_id: Deprecated alias of ``scene_id`` + :ivar float start_sec: Deprecated alias of ``start`` + :ivar float end_sec: Deprecated alias of ``end`` + """ + + def __init__( + self, + video_id: Optional[str] = None, + understanding_id: Optional[str] = None, + scene_id: Optional[str] = None, + start: Optional[float] = None, + end: Optional[float] = None, + data: Optional[dict] = None, + segment_id: Optional[str] = None, + start_sec: Optional[float] = None, + end_sec: Optional[float] = None, + ) -> None: + self.video_id = video_id + self.understanding_id = understanding_id + self.scene_id = scene_id if scene_id is not None else segment_id + self.start = start if start is not None else start_sec + self.end = end if end is not None else end_sec + self.data = data or {} + self.segment_id = self.scene_id + self.start_sec = self.start + self.end_sec = self.end + + def __repr__(self) -> str: + return ( + f"IndexRecord(" + f"video_id={self.video_id}, " + f"scene_id={self.scene_id}, " + f"start={self.start}, " + f"end={self.end}, " + f"data={self.data})" + ) + + def __getitem__(self, key): + return self.__dict__[key] + + +class RecordPage: + """A page of indexed records returned by :meth:`Index.records`. + + :ivar list[IndexRecord] records: Records in this page + :ivar str next_cursor: Cursor for the next page, ``None`` if there are no more records + """ + + def __init__( + self, + records: Optional[List[IndexRecord]] = None, + next_cursor: Optional[str] = None, + ) -> None: + self.records = records or [] + self.next_cursor = next_cursor + + def __repr__(self) -> str: + return ( + f"RecordPage(records={len(self.records)}, next_cursor={self.next_cursor})" + ) + + def __iter__(self): + return iter(self.records) + + def __getitem__(self, key): + return self.records[key] + + +class Index: + """Index manifest for a retrieval-ready index built from an understanding artifact. + + Note: Users should not initialize this class directly. + Instead use :meth:`Video.index() `, + :meth:`Video.get_index() `, or + :meth:`Video.list_indexes() `. + + :ivar str index_id: Unique identifier for the index + :ivar str video_id: ID of the video this index belongs to + :ivar str collection_id: ID of the collection this index belongs to + :ivar str name: User-facing name of the index + :ivar str status: Build status of the index (e.g. ``"building"``, ``"ready"``, ``"failed"``) + :ivar list use_for: Retrieval capabilities the index supports + (subset of ``"semantic"``, ``"query"``, ``"aggregate"``) + :ivar source: Source artifact reference or records the index was built from + :ivar int record_count: Number of records in the index + :ivar dict fields: Field groups mapping (``semantic``, ``filter``, + ``aggregate``, ``sort``) to lists of field names + :ivar dict field_schema: Mapping of field name to :class:`FieldSchema ` + """ + + def __init__( + self, _connection, video_id: str, collection_id: str = None, **kwargs + ) -> None: + self._connection = _connection + self.video_id = video_id + self.collection_id = collection_id + self.update_from_response(kwargs) + + def update_from_response(self, data: dict) -> "Index": + data = data or {} + self.index_id = data.get("index_id") or getattr(self, "index_id", None) + self.name = data.get("name") + self.status = data.get("status") + self.error = data.get("error") + self.use_for = data.get("use_for", []) + self.source = data.get("source") + self.record_count = data.get("record_count") + self.fields = data.get("fields", {}) + self.field_schema = { + field: FieldSchema( + type=schema.get("type"), + groups=schema.get("groups"), + operators=schema.get("operators"), + ) + for field, schema in (data.get("field_schema") or {}).items() + } + return self + + def __repr__(self) -> str: + return ( + f"Index(" + f"index_id={self.index_id}, " + f"video_id={self.video_id}, " + f"name={self.name}, " + f"status={self.status}, " + + (f"error={self.error}, " if self.error else "") + + f"use_for={self.use_for}, " + f"record_count={self.record_count})" + ) + + def __getitem__(self, key): + return self.__dict__[key] + + @property + def is_complete(self) -> bool: + """Return True when the index build is in a terminal status.""" + return self.status in INDEX_TERMINAL_STATUSES + + @property + def is_successful(self) -> bool: + """Return True when the index build completed successfully.""" + return self.status == "ready" + + def refresh(self) -> "Index": + """Refresh the index manifest and build status from the API.""" + data = self._connection.get( + path=f"{ApiPath.video}/{self.video_id}/{ApiPath.indexes}/{self.index_id}", + params={"collection_id": self.collection_id} + if self.collection_id + else None, + ) + return self.update_from_response(data) + + def wait_until_complete( + self, + timeout: int = 1800, + poll_interval: int = 10, + ) -> "Index": + """Poll this index until it reaches a terminal status. + + Terminal statuses are ``ready`` and ``failed``. + + :param int timeout: Maximum time to wait, in seconds + :param int poll_interval: Seconds between status checks + :raises TimeoutError: If the build does not complete before timeout + :return: This index with refreshed status + """ + deadline = time.time() + timeout + while True: + self.refresh() + if self.is_complete: + return self + if time.time() >= deadline: + raise TimeoutError(f"Index {self.index_id} did not complete within {timeout}s") + time.sleep(poll_interval) + + def records(self, limit: int = 20, cursor: Optional[str] = None) -> RecordPage: + """Preview the records stored in the index. + + Intended for inspection and debugging. Records are paginated via a cursor. + + :param int limit: (optional) Maximum number of records to return (default: 20) + :param str cursor: (optional) Cursor returned by a previous page to fetch the next page + :return: A page of indexed records, :class:`RecordPage ` object + :rtype: :class:`videodb.index.RecordPage` + """ + params = {"limit": limit, "collection_id": self.collection_id} + if cursor is not None: + params["cursor"] = cursor + records_data = self._connection.get( + path=f"{ApiPath.video}/{self.video_id}/{ApiPath.indexes}/{self.index_id}/{ApiPath.records}", + params={key: value for key, value in params.items() if value is not None}, + ) + if not records_data: + return RecordPage() + records = [ + IndexRecord( + video_id=record.get("video_id"), + understanding_id=record.get("understanding_id"), + scene_id=record.get("scene_id"), + start=record.get("start"), + end=record.get("end"), + data=record.get("data"), + segment_id=record.get("segment_id"), + start_sec=record.get("start_sec"), + end_sec=record.get("end_sec"), + ) + for record in records_data.get("records", []) + ] + return RecordPage(records=records, next_cursor=records_data.get("next_cursor")) + + def delete(self) -> None: + """Delete the index. + + Removes the index's retrieval structures. It does not delete the original + video or stored understanding artifacts. + + :raises InvalidRequestError: If the delete fails + :return: None if the delete is successful + :rtype: None + """ + self._connection.delete( + path=f"{ApiPath.video}/{self.video_id}/{ApiPath.indexes}/{self.index_id}", + params={"collection_id": self.collection_id} + if self.collection_id + else None, + ) diff --git a/videodb/rtstream.py b/videodb/rtstream.py index 0cf35c6..4c2a97a 100644 --- a/videodb/rtstream.py +++ b/videodb/rtstream.py @@ -368,6 +368,246 @@ def disable_alert(self, alert_id): ) +class RTStreamUnderstanding: + """RTStreamUnderstanding class to interact with a continuous understanding job. + + Produced by :meth:`RTStream.understand`. It runs VLM analysis over stream + windows and, when ``store=True``, persists the output so it can be indexed + later. Understanding is independent of scene indexing. + + :ivar str id: Understanding id (``und-...``) + :ivar str rtstream_id: ID of the parent RTStream + :ivar str status: Job status (``running`` | ``stopped`` | ``failed``) + :ivar bool store: Whether output is persisted for later indexing + :ivar dict segmentation: Time segmentation, e.g. ``{"type": "time", "window": "10s"}`` + :ivar list analyzers: Analyzer specs for this understanding + :ivar dict outputs: Named output source descriptors, e.g. ``outputs["scene"]`` + """ + + def __init__( + self, + _connection, + understanding_id: str = None, + rtstream_id: str = None, + id: str = None, + **kwargs, + ) -> None: + self._connection = _connection + self.id = understanding_id or id + self.rtstream_id = rtstream_id + self.status = kwargs.get("status", None) + self.store = kwargs.get("store", True) + self.segmentation = kwargs.get("segmentation", {}) + self.analyzers = kwargs.get("analyzers", []) + self.outputs = kwargs.get("outputs", {}) + + def __repr__(self) -> str: + return ( + f"RTStreamUnderstanding(" + f"id={self.id}, " + f"rtstream_id={self.rtstream_id}, " + f"status={self.status}, " + f"store={self.store}, " + f"analyzers={len(self.analyzers)})" + ) + + def refresh(self) -> "RTStreamUnderstanding": + """Reload this understanding from the server. + + :return: This understanding, updated + :rtype: :class:`RTStreamUnderstanding ` + """ + data = self._connection.get( + f"{ApiPath.rtstream}/{self.rtstream_id}/{ApiPath.understand}/{self.id}" + ) + if data: + data.setdefault("understanding_id", self.id) + data.setdefault("rtstream_id", self.rtstream_id) + self.__init__(self._connection, **data) + return self + + def start(self): + """Resume processing new stream windows. + + :return: None + :rtype: None + """ + self._connection.patch( + f"{ApiPath.rtstream}/{self.rtstream_id}/{ApiPath.understand}/{self.id}/{ApiPath.status}", + data={"action": "start"}, + ) + self.status = "running" + + def stop(self): + """Pause processing new stream windows. Existing records remain available. + + :return: None + :rtype: None + """ + self._connection.patch( + f"{ApiPath.rtstream}/{self.rtstream_id}/{ApiPath.understand}/{self.id}/{ApiPath.status}", + data={"action": "stop"}, + ) + self.status = "stopped" + + def get_records( + self, + start: float, + end: float, + output: str = "scene", + page: int = 1, + page_size: int = 100, + ): + """Get understanding output records for a time range. + + :param float start: Start Unix timestamp + :param float end: End Unix timestamp + :param str output: Analyzer output name (default: ``"scene"``) + :param int page: Page number (default: 1) + :param int page_size: Records per page (default: 100) + :return: Records payload with ``records`` and ``next_page`` + :rtype: dict + """ + params = { + "start": start, + "end": end, + "output": output, + "page": page, + "page_size": page_size, + } + return self._connection.get( + f"{ApiPath.rtstream}/{self.rtstream_id}/{ApiPath.understand}/{self.id}/{ApiPath.records}", + params={k: v for k, v in params.items() if v is not None}, + ) + + +class RTStreamIndex: + """RTStreamIndex — a continuous index over an understanding output. + + Produced by :meth:`RTStream.index`. Materializes an understanding's stored + output into a searchable index; has its own lifecycle, separate from the + understanding. + + :ivar str id: Index id (``idx-...``) + :ivar str rtstream_id: ID of the parent RTStream + :ivar str name: Index name + :ivar str status: ``running`` | ``stopped`` | ``failed`` + :ivar list use_for: Index capabilities, e.g. ``["semantic"]`` + :ivar str source_understanding_id: Understanding this index consumes + :ivar str output: Analyzer output name being indexed (e.g. ``"scene"``) + """ + + def __init__(self, _connection, index_id=None, rtstream_id=None, id=None, **kwargs): + self._connection = _connection + self.id = index_id or id + self.rtstream_id = rtstream_id + self.name = kwargs.get("name") + self.status = kwargs.get("status") + self.use_for = kwargs.get("use_for") or ["semantic"] + self.source_understanding_id = kwargs.get("source_understanding_id") + self.output = kwargs.get("output", "scene") + + def __repr__(self) -> str: + return ( + f"RTStreamIndex(" + f"id={self.id}, " + f"rtstream_id={self.rtstream_id}, " + f"status={self.status}, " + f"use_for={self.use_for}, " + f"source_understanding_id={self.source_understanding_id})" + ) + + def refresh(self) -> "RTStreamIndex": + """Reload this index from the server.""" + data = self._connection.get( + f"{ApiPath.rtstream}/{self.rtstream_id}/{ApiPath.indexes}/{self.id}" + ) + if data: + data.setdefault("index_id", self.id) + data.setdefault("rtstream_id", self.rtstream_id) + self.__init__(self._connection, **data) + return self + + def start(self): + """Resume materializing new understanding output into the index.""" + self._connection.patch( + f"{ApiPath.rtstream}/{self.rtstream_id}/{ApiPath.indexes}/{self.id}/{ApiPath.status}", + data={"action": "start"}, + ) + self.status = "running" + + def stop(self): + """Pause materializing. Existing indexed records remain searchable.""" + self._connection.patch( + f"{ApiPath.rtstream}/{self.rtstream_id}/{ApiPath.indexes}/{self.id}/{ApiPath.status}", + data={"action": "stop"}, + ) + self.status = "stopped" + + def get_records(self, start=None, end=None, page: int = 1, page_size: int = 100): + """Get materialized index records. + + :param int start: Start Unix timestamp (optional) + :param int end: End Unix timestamp (optional) + :param int page: Page number + :param int page_size: Records per page + :return: Records payload + :rtype: dict + """ + params = {"page": page, "page_size": page_size} + if start is not None: + params["start"] = start + if end is not None: + params["end"] = end + return self._connection.get( + f"{ApiPath.rtstream}/{self.rtstream_id}/{ApiPath.indexes}/{self.id}/{ApiPath.records}", + params=params, + ) + + def create_alert(self, event_id, callback_url, ws_connection_id=None) -> str: + """Attach an event alert to this index. + + :param str event_id: ID of the event + :param str callback_url: URL to receive the alert callback + :param str ws_connection_id: WebSocket connection ID for real-time alerts (optional) + :return: Alert ID + :rtype: str + """ + data = {"event_id": event_id, "callback_url": callback_url} + if ws_connection_id: + data["ws_connection_id"] = ws_connection_id + alert_data = self._connection.post( + f"{ApiPath.rtstream}/{self.rtstream_id}/{ApiPath.indexes}/{self.id}/{ApiPath.alert}", + data=data, + ) + return (alert_data or {}).get("alert_id") + + def list_alerts(self): + """List all alerts on this index. + + :return: List of alerts + :rtype: List[dict] + """ + alert_data = self._connection.get( + f"{ApiPath.rtstream}/{self.rtstream_id}/{ApiPath.indexes}/{self.id}/{ApiPath.alert}" + ) + return (alert_data or {}).get("alerts", []) + + def enable_alert(self, alert_id): + """Enable an alert on this index.""" + self._connection.patch( + f"{ApiPath.rtstream}/{self.rtstream_id}/{ApiPath.indexes}/{self.id}/{ApiPath.alert}/{alert_id}/{ApiPath.status}", + data={"action": "enable"}, + ) + + def disable_alert(self, alert_id): + """Disable an alert on this index.""" + self._connection.patch( + f"{ApiPath.rtstream}/{self.rtstream_id}/{ApiPath.indexes}/{self.id}/{ApiPath.alert}/{alert_id}/{ApiPath.status}", + data={"action": "disable"}, + ) + + class RTStream: """RTStream class to interact with the RTStream @@ -829,6 +1069,133 @@ def get_scene_index(self, index_id: str) -> RTStreamSceneIndex: status=index_data.get("status"), ) + def understand( + self, + segmentation: Dict = None, + analyzers: List[Dict] = None, + store: bool = True, + ws_connection_id: str = None, + ) -> "RTStreamUnderstanding": + """Start a continuous VLM understanding job on the stream. + + Understanding is independent of indexing: it produces VLM output per + stream window and (when ``store=True``) persists it so it can be indexed + later. Initial support is one ``vlm`` analyzer with time segmentation. + + :param dict segmentation: Time segmentation, e.g. ``{"type": "time", "window": "10s"}`` + :param list analyzers: Exactly one VLM analyzer spec, e.g. + ``[{"type": "vlm", "name": "scene", "sampling": {"frame_count": 5}, "config": {"prompt": "...", "model": "basic"}}]`` + :param bool store: Persist output for later indexing (default: True) + :param str ws_connection_id: WebSocket connection ID for real-time updates (optional) + :return: The understanding job, :class:`RTStreamUnderstanding ` object + :rtype: :class:`videodb.rtstream.RTStreamUnderstanding` + """ + data = { + "segmentation": segmentation or {}, + "analyzers": analyzers or [], + "store": store, + } + if ws_connection_id: + data["ws_connection_id"] = ws_connection_id + understanding_data = self._connection.post( + f"{ApiPath.rtstream}/{self.id}/{ApiPath.understand}", + data=data, + ) + if not understanding_data: + return None + understanding_data.setdefault("rtstream_id", self.id) + return RTStreamUnderstanding( + _connection=self._connection, **understanding_data + ) + + def get_understanding(self, understanding_id: str) -> "RTStreamUnderstanding": + """Get an understanding job by id. + + :param str understanding_id: ID of the understanding job + :return: The understanding job, :class:`RTStreamUnderstanding ` object + :rtype: :class:`videodb.rtstream.RTStreamUnderstanding` + """ + if not understanding_id: + raise ValueError("understanding_id is required") + understanding_data = self._connection.get( + f"{ApiPath.rtstream}/{self.id}/{ApiPath.understand}/{understanding_id}" + ) + if not understanding_data: + return None + understanding_data.setdefault("rtstream_id", self.id) + return RTStreamUnderstanding( + _connection=self._connection, **understanding_data + ) + + def list_understanding(self) -> List["RTStreamUnderstanding"]: + """List all understanding jobs on the stream. + + :return: List of understanding jobs + :rtype: List[:class:`RTStreamUnderstanding `] + """ + data = self._connection.get( + f"{ApiPath.rtstream}/{self.id}/{ApiPath.understand}" + ) + results = (data or {}).get("understandings") or [] + for item in results: + item.setdefault("rtstream_id", self.id) + return [ + RTStreamUnderstanding(_connection=self._connection, **item) + for item in results + ] + + def index(self, source, name=None, use_for=None) -> "RTStreamIndex": + """Materialize an understanding output into a searchable index. + + :param dict source: understanding output descriptor, e.g. + ``understanding.outputs["scene"]`` + :param str name: index name (optional) + :param list use_for: capabilities; defaults to ``["semantic"]`` + :return: The index, :class:`RTStreamIndex ` object + :rtype: :class:`videodb.rtstream.RTStreamIndex` + """ + data = {"source": source} + if name is not None: + data["name"] = name + if use_for is not None: + data["use_for"] = use_for + index_data = self._connection.post( + f"{ApiPath.rtstream}/{self.id}/{ApiPath.indexes}", data=data + ) + if not index_data: + return None + index_data.setdefault("rtstream_id", self.id) + return RTStreamIndex(_connection=self._connection, **index_data) + + def get_index(self, index_id: str) -> "RTStreamIndex": + """Get an index by id. + + :param str index_id: ID of the index + :return: :class:`RTStreamIndex ` object + :rtype: :class:`videodb.rtstream.RTStreamIndex` + """ + if not index_id: + raise ValueError("index_id is required") + index_data = self._connection.get( + f"{ApiPath.rtstream}/{self.id}/{ApiPath.indexes}/{index_id}" + ) + if not index_data: + return None + index_data.setdefault("rtstream_id", self.id) + return RTStreamIndex(_connection=self._connection, **index_data) + + def list_indexes(self) -> List["RTStreamIndex"]: + """List all indexes on the stream. + + :return: List of indexes + :rtype: List[:class:`RTStreamIndex `] + """ + data = self._connection.get(f"{ApiPath.rtstream}/{self.id}/{ApiPath.indexes}") + results = (data or {}).get("indexes") or [] + for item in results: + item.setdefault("rtstream_id", self.id) + return [RTStreamIndex(_connection=self._connection, **item) for item in results] + def get_transcript( self, page=1, diff --git a/videodb/search.py b/videodb/search.py index 7cb87f9..fac8682 100644 --- a/videodb/search.py +++ b/videodb/search.py @@ -1,4 +1,6 @@ from abc import ABC, abstractmethod +import warnings + from videodb._utils._video import play_stream, build_iframe_embed_code from videodb._constants import ( IndexType, @@ -14,6 +16,42 @@ from videodb.shot import Shot +_RESPONSE_WARNING_EMITTED = set() + + +def _warn_once(key: str, message: str, stacklevel: int = 4): + if key in _RESPONSE_WARNING_EMITTED: + return + _RESPONSE_WARNING_EMITTED.add(key) + warnings.warn(message, UserWarning, stacklevel=stacklevel) + + +def _warning_docs_text(warning: dict) -> str: + docs = warning.get("docs") or [] + parts = [] + for item in docs: + if not isinstance(item, dict): + continue + label = item.get("label") + url = item.get("url") + if label and url: + parts.append(f"{label}: {url}") + elif url: + parts.append(str(url)) + return f" Docs: {'; '.join(parts)}" if parts else "" + + +def warn_response_warnings_once(response_warnings, stacklevel: int = 5): + if isinstance(response_warnings, dict): + response_warnings = [response_warnings] + for warning in response_warnings or []: + if not isinstance(warning, dict): + continue + code = str(warning.get("code") or warning.get("message") or "response_warning") + message = str(warning.get("message") or f"VideoDB warning: {code}").strip() + _warn_once(f"response:{code}", f"{message}{_warning_docs_text(warning)}", stacklevel=stacklevel) + + class SearchResult: """SearchResult class to interact with search results @@ -29,13 +67,16 @@ def __init__(self, _connection, **kwargs): self.stream_url = None self.player_url = None self.collection_id = "default" + self.warnings = kwargs.get("warnings") or [] + if kwargs.get("_emit_warnings", True): + warn_response_warnings_once(self.warnings, stacklevel=5) self._results = kwargs.get("results", []) self._format_results() def _format_results(self): for result in self._results: self.collection_id = result.get("collection_id") - for doc in result.get("docs"): + for doc in result.get("docs") or []: self.shots.append( Shot( self._connection, @@ -49,7 +90,7 @@ def _format_results(self): scene_index_id=doc.get("scene_index_id"), scene_index_name=doc.get("scene_index_name"), metadata=doc.get("metadata"), - stream_url=doc.get("stream_link"), + stream_url=doc.get("stream_link") or doc.get("stream_url"), player_url=doc.get("player_url"), ) ) @@ -63,6 +104,15 @@ def __repr__(self) -> str: f"shots={self.shots})" ) + def __iter__(self): + return iter(self.shots) + + def __len__(self): + return len(self.shots) + + def __getitem__(self, index): + return self.shots[index] + def get_shots(self) -> List[Shot]: return self.shots @@ -139,6 +189,141 @@ def get_embed_code( ) +class AskResponse: + """Response returned by ``ask()``. + + :ivar str answer: Text answer generated from retrieved video context. + :ivar list[Shot] sources: Source shots selected by the LLM when requested. + """ + + def __init__(self, _connection, **kwargs): + self._connection = _connection + self.answer = kwargs.get("answer") or "" + self.warnings = kwargs.get("warnings") or [] + warn_response_warnings_once(self.warnings, stacklevel=5) + self.sources = SearchResult( + _connection, + results=kwargs.get("sources") or [], + warnings=self.warnings, + _emit_warnings=False, + ).shots + + def __repr__(self) -> str: + return f"AskResponse(answer={self.answer!r}, sources={self.sources})" + + +class SearchResponse: + """Envelope returned by high-level Search v2. + + For ``response_type='shots'`` or ``response_type='deepsearch'``, ``results`` is a :class:`SearchResult`. + For ``response_type='aggregate'``, ``results`` is the aggregate dict/list returned by the server. + """ + + def __init__(self, _connection, **kwargs): + self._connection = _connection + self.response_type = kwargs.get("response_type") + self.session_id = kwargs.get("session_id") + self.waiting_for = kwargs.get("waiting_for") or "none" + self.clarification = kwargs.get("clarification") + self.trace = kwargs.get("trace") + self.warnings = kwargs.get("warnings") or [] + warn_response_warnings_once(self.warnings, stacklevel=6) + raw_results = kwargs.get("results", []) + if self.response_type in {"shots", "deepsearch"}: + self.results = SearchResult( + _connection, + results=raw_results, + warnings=self.warnings, + _emit_warnings=False, + ) + self.shots = self.results.shots + else: + self.results = raw_results + self.shots = [] + + def __repr__(self) -> str: + if self.response_type == "deepsearch": + return ( + "SearchResponse(" + f"response_type={self.response_type}, " + f"session_id={self.session_id!r}, " + f"waiting_for={self.waiting_for!r}, " + f"clarification={self.clarification!r}, " + f"results={self.results})" + ) + return f"SearchResponse(response_type={self.response_type}, results={self.results})" + + def __iter__(self): + if self.response_type in {"shots", "deepsearch"}: + return iter(self.results) + if isinstance(self.results, list): + return iter(self.results) + return iter([self.results]) + + def __len__(self): + if self.response_type in {"shots", "deepsearch"}: + return len(self.results) + if isinstance(self.results, list): + return len(self.results) + return 1 if self.results is not None else 0 + + def __getitem__(self, index): + if self.response_type in {"shots", "deepsearch"}: + return self.results[index] + return self.results[index] + + def get_shot_results(self) -> SearchResult: + """Return the underlying shot results for media helper methods. + + ``search()`` now returns a SearchResponse envelope, but older SDK code often + chained ``.compile()``, ``.play()``, or ``.get_embed_code()`` directly from + ``search()``. Keep that flow working for response types backed by shots. + """ + if self.response_type in {"shots", "deepsearch"} and isinstance(self.results, SearchResult): + return self.results + raise SearchError( + "This SearchResponse does not contain shot results. " + "compile(), play(), and get_embed_code() are only available when response_type is 'shots' or 'deepsearch'." + ) + + def compile(self) -> str: + """Compile shot results into a stream URL. + + Available for ``response_type='shots'`` and ``response_type='deepsearch'``. + """ + return self.get_shot_results().compile() + + def play(self) -> str: + """Compile if needed, then play shot results. + + Available for ``response_type='shots'`` and ``response_type='deepsearch'``. + """ + return self.get_shot_results().play() + + def get_embed_code( + self, + width: str = "100%", + height: int = 405, + title: str = "VideoDB Player", + allow_fullscreen: bool = True, + auto_generate: bool = True, + ) -> str: + """Generate an iframe embed code for shot results. + + Available for ``response_type='shots'`` and ``response_type='deepsearch'``. + """ + return self.get_shot_results().get_embed_code( + width=width, + height=height, + title=title, + allow_fullscreen=allow_fullscreen, + auto_generate=auto_generate, + ) + + def get_shots(self) -> List[Shot]: + return self.shots + + class Search(ABC): """Search interface inside video or collection""" diff --git a/videodb/understanding.py b/videodb/understanding.py new file mode 100644 index 0000000..aaba338 --- /dev/null +++ b/videodb/understanding.py @@ -0,0 +1,292 @@ +import time +from typing import Any, Dict, List, Optional + +from videodb._constants import ( + ApiPath, + ANALYZER_TERMINAL_STATUSES, + UNDERSTANDING_TERMINAL_STATUSES, +) + +class UnderstandingAnalyzer: + """Analyzer status and output handle for one analyzer in an understanding run.""" + + def __init__( + self, + understanding: "Understanding", + id: Optional[str] = None, + name: Optional[str] = None, + type: Optional[str] = None, + status: Optional[str] = None, + **kwargs, + ) -> None: + self.understanding = understanding + self.id = id + self.name = name + self.type = type + self.status = status + self.extra = kwargs + + def __repr__(self) -> str: + return ( + f"UnderstandingAnalyzer(" + f"id={self.id}, " + f"name={self.name}, " + f"type={self.type}, " + f"status={self.status})" + ) + + def __getitem__(self, key): + return self.__dict__[key] + + @property + def is_complete(self) -> bool: + """Return True when the analyzer is in a terminal status.""" + return self.status in ANALYZER_TERMINAL_STATUSES + + @property + def is_successful(self) -> bool: + """Return True when the analyzer completed successfully.""" + return self.status == "done" + + def refresh(self) -> "UnderstandingAnalyzer": + """Refresh this analyzer's status from the API.""" + analyzer = self.understanding.get_analyzer(self.name or self.id, refresh=True) + self.id = analyzer.id + self.name = analyzer.name + self.type = analyzer.type + self.status = analyzer.status + self.extra = analyzer.extra + return self + + def wait_until_complete( + self, + timeout: int = 1800, + poll_interval: int = 10, + ) -> "UnderstandingAnalyzer": + """Poll this analyzer until it reaches a terminal status. + + :param int timeout: Maximum time to wait, in seconds + :param int poll_interval: Seconds between status checks + :raises TimeoutError: If the analyzer does not complete before timeout + :return: This analyzer with refreshed status + """ + deadline = time.time() + timeout + while True: + self.refresh() + if self.is_complete: + return self + if time.time() >= deadline: + raise TimeoutError( + f"Analyzer {self.name or self.id} did not complete within {timeout}s" + ) + time.sleep(poll_interval) + + def get_output(self) -> Any: + """Return this analyzer's output. + + The current API returns the analyzer's segments output. + """ + identifier = self.name or self.id + if not identifier: + raise ValueError("Analyzer id or name is required") + return self.understanding.get_analyzer_output(identifier) + + def to_index_source(self) -> Dict: + """Serialize this analyzer as an index ``source`` reference. + + Sends only identifiers — the server re-fetches the analyzer's output from its + own store, so scenes never round-trip through the client: + + for analyzer in understanding.list_analyzers(): + if analyzer.is_successful: + video.index(name=analyzer.name, source=analyzer) + + :raises ValueError: If the analyzer has no id or its understanding id is unknown + """ + understanding_id = getattr(self.understanding, "id", None) + if not understanding_id or not self.id: + raise ValueError("analyzer source requires understanding id and analyzer id") + # ids + type — the server's analyzer record stays the source of truth for the + return { + "understanding_id": understanding_id, + "analyzer_id": self.id, + "analyzer_type": self.type, + } + + +class Understanding: + """A video understanding run. + + Use :meth:`list_analyzers` or :meth:`get_analyzer` to inspect analyzers and + fetch analyzer outputs. + """ + + def __init__( + self, + _connection, + video_id: str, + collection_id: Optional[str] = None, + understanding_id: Optional[str] = None, + id: Optional[str] = None, + status: Optional[str] = None, + analyzers: Optional[List[Dict[str, Any]]] = None, + output_url: Optional[str] = None, + **kwargs, + ) -> None: + self._connection = _connection + self.video_id = video_id + self.collection_id = collection_id + self.id = understanding_id or id + self.status = status + self.output_url = output_url + self.extra = kwargs + self.analyzers = [self.create_analyzer(item) for item in (analyzers or [])] + + def __repr__(self) -> str: + return ( + f"Understanding(" + f"id={self.id}, " + f"video_id={self.video_id}, " + f"status={self.status}, " + f"analyzers={len(self.analyzers)})" + ) + + def __getitem__(self, key): + return self.__dict__[key] + + @property + def is_complete(self) -> bool: + """Return True when the understanding run is in a terminal status.""" + return self.status in UNDERSTANDING_TERMINAL_STATUSES + + @property + def is_successful(self) -> bool: + """Return True when the understanding run completed successfully.""" + return self.status == "done" + + def create_analyzer(self, data: Dict[str, Any]) -> UnderstandingAnalyzer: + return UnderstandingAnalyzer(self, **(data or {})) + + def update_from_response(self, data: Dict[str, Any]) -> "Understanding": + data = data or {} + self.status = data.get("status", self.status) + if data.get("understanding_id") or data.get("id"): + self.id = data.get("understanding_id") or data.get("id") + if data.get("video_id"): + self.video_id = data.get("video_id") + if data.get("collection_id"): + self.collection_id = data.get("collection_id") + if data.get("output_url"): + self.output_url = data.get("output_url") + if "analyzers" in data: + self.analyzers = [self.create_analyzer(item) for item in data.get("analyzers") or []] + return self + + def refresh(self) -> "Understanding": + """Refresh understanding and analyzer statuses from the API.""" + data = self._connection.get( + path=f"{ApiPath.video}/{self.video_id}/{ApiPath.understand}/{self.id}" + ) + return self.update_from_response(data) + + def wait_until_complete( + self, + timeout: int = 1800, + poll_interval: int = 10, + ) -> "Understanding": + """Poll this understanding until it reaches a terminal status. + + Terminal statuses are ``done`` and ``failed``. + + :param int timeout: Maximum time to wait, in seconds + :param int poll_interval: Seconds between status checks + :raises TimeoutError: If the run does not complete before timeout + :return: This understanding with refreshed status + """ + deadline = time.time() + timeout + while True: + self.refresh() + if self.is_complete: + return self + if time.time() >= deadline: + raise TimeoutError(f"Understanding {self.id} did not complete within {timeout}s") + time.sleep(poll_interval) + + def list_analyzers(self) -> List[UnderstandingAnalyzer]: + """Return analyzers in this understanding run.""" + return list(self.analyzers) + + def get_analyzer(self, name_or_id: str, refresh: bool = False) -> UnderstandingAnalyzer: + """Return an analyzer by user-facing name or internal analyzer id. + + :param str name_or_id: Analyzer name returned by the API, or an internal id such as ``"an_..."`` + :param bool refresh: When True, fetch the latest analyzer status first + :raises ValueError: If no analyzer matches + :return: :class:`UnderstandingAnalyzer ` object + """ + if refresh: + data = self._connection.get( + path=f"{ApiPath.video}/{self.video_id}/{ApiPath.understand}/{self.id}", + params={"analyzer": name_or_id}, + ) or {} + self.status = data.get("status", self.status) + analyzer_data = (data.get("analyzers") or [None])[0] + if analyzer_data: + refreshed = self.create_analyzer(analyzer_data) + for index, analyzer in enumerate(self.analyzers): + if analyzer.name == refreshed.name or analyzer.id == refreshed.id: + self.analyzers[index] = refreshed + return refreshed + self.analyzers.append(refreshed) + return refreshed + + for analyzer in self.analyzers: + if analyzer.name == name_or_id or analyzer.id == name_or_id: + return analyzer + raise ValueError(f"Analyzer not found: {name_or_id}") + + def get_analyzer_output(self, name_or_id: str) -> Any: + """Return output for an analyzer by name or id.""" + return self._connection.get( + path=( + f"{ApiPath.video}/{self.video_id}/{ApiPath.understand}/{self.id}" + f"/analyzers/{name_or_id}/output" + ) + ) + + def delete(self) -> None: + """Delete this understanding run.""" + self._connection.delete( + path=f"{ApiPath.video}/{self.video_id}/{ApiPath.understand}/{self.id}" + ) + + +def normalize_understanding_analyzers(analyzers: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + """Normalize analyzer payloads to the server contract. + + The public SDK accepts friendly analyzer types like ``spoken_words``. Names + remain optional; the server assigns a unique name and id when omitted. Use + explicit names when another analyzer references one through ``inputs``. + """ + if not isinstance(analyzers, list) or not analyzers: + raise ValueError("analyzers must be a non-empty list") + + normalized = [] + names = set() + + for index, analyzer in enumerate(analyzers): + if not isinstance(analyzer, dict): + raise ValueError(f"analyzers[{index}] must be a dict") + if not analyzer.get("type"): + raise ValueError(f"analyzers[{index}].type is required") + + item = dict(analyzer) + name = item.get("name") + if name: + if name in names: + raise ValueError(f"Duplicate analyzer name: {name}") + names.add(name) + + normalized.append(item) + + return normalized diff --git a/videodb/video.py b/videodb/video.py index 37d0526..75cc8ae 100644 --- a/videodb/video.py +++ b/videodb/video.py @@ -12,8 +12,16 @@ Workflows, ) from videodb.image import Image, Frame +from videodb.index import Index +from videodb.understanding import Understanding, normalize_understanding_analyzers from videodb.scene import Scene, SceneCollection -from videodb.search import SearchFactory, SearchResult +from videodb.search import ( + AskResponse, + SearchFactory, + SearchResponse, + SearchResult, + warn_response_warnings_once, +) from videodb.shot import Shot _VALID_SEGMENTERS = {Segmenter.word, Segmenter.sentence, Segmenter.time} @@ -83,6 +91,240 @@ def update(self, name: Optional[str] = None) -> None: self.name = response_data.get("name", name) def search( + self, + query: str, + *args, + config: Optional[Dict[str, Any]] = None, + **kwargs, + ) -> Union[SearchResponse, SearchResult]: + """Search this video using Search V2 by default. + + Pass Search V2 options such as ``top_k``, ``mode``, ``return_fields``, + ``include_clip``, ``session_id``, or ``config`` for the new search API. + Calls with legacy options such as ``search_type``, ``index_type``, + ``result_threshold``, ``scene_index_id``, ``index_id``, ``stitch``, + ``rerank``, or ``rerank_params`` are routed to :meth:`legacy_search`. + Do not mix Search V2 and legacy options in one call. + + Server-provided warnings are exposed on ``response.warnings`` and are + emitted as Python ``UserWarning`` messages. + + :param str query: Natural-language search query. + :param dict config: Optional Search V2 request configuration. + :return: ``SearchResponse`` for Search V2, or ``SearchResult`` for legacy-routed calls. + """ + old_params = { + "search_type", + "index_type", + "result_threshold", + "dynamic_score_percentage", + "scene_index_id", + "index_id", + "algorithm", + "sort_docs_on", + "namespace", + "stitch", + "rerank", + "rerank_params", + } + new_params = { + "top_k", + "mode", + "return_fields", + "include_clip", + "session_id", + "config", + } + unsupported_params = {"index_name", "index_names", "index_ids"} + + if config is not None: + kwargs["config"] = config + + if args: + legacy_arg_names = [ + "search_type", + "index_type", + "result_threshold", + "score_threshold", + "dynamic_score_percentage", + "filter", + ] + for name, value in zip(legacy_arg_names, args): + kwargs.setdefault(name, value) + + has_old = bool(args) or any(k in kwargs and kwargs[k] is not None for k in old_params) + has_new = any(k in kwargs and kwargs[k] is not None for k in new_params) + has_unsupported = any(k in kwargs and kwargs[k] is not None for k in unsupported_params) + + if has_old and (has_new or has_unsupported): + raise ValueError( + "Cannot mix legacy search parameters with Search V2 parameters. " + "Use legacy_search(...) for older indexes, or remove legacy parameters and use Search V2 search(...)." + ) + if has_unsupported: + raise ValueError( + "search() chooses indexes automatically and does not accept index selectors. " + "Use semantic_search() for semantic index selection, query() for structured filtering, or aggregate() for counts and facets." + ) + + if has_old: + return self.legacy_search(query=query, _skip_warning=True, **kwargs) + + return self._new_search(query=query, **kwargs) + + def _new_search(self, query: str, **kwargs) -> SearchResponse: + payload = {"query": query, **{k: v for k, v in kwargs.items() if v is not None}} + search_data = self._connection.post( + path=f"{ApiPath.video}/{self.id}/{ApiPath.search}/v2", + data=payload, + ) + return SearchResponse(self._connection, **search_data) + + def ask( + self, + question: str, + top_k: int = 15, + mode: str = "default", + include_sources: bool = False, + ) -> AskResponse: + """Ask a question over this video's Search V2 indexes. + + ``ask()`` is Search V2 only; it does not search legacy indexes. If the + server cannot answer from indexed content, warnings are available on + ``response.warnings``. + + :param str question: Question to answer from the video. + :param int top_k: Maximum number of source shots to retrieve. + :param str mode: Search mode to use. + :param bool include_sources: Include source shots in the response. + :return: ``AskResponse`` with ``answer``, ``sources``, and ``warnings``. + """ + ask_data = self._connection.post( + path=f"{ApiPath.video}/{self.id}/{ApiPath.ask}", + data={ + "question": question, + "top_k": top_k, + "mode": mode, + "include_sources": include_sources, + }, + ) + return AskResponse(self._connection, **ask_data) + + def semantic_search( + self, + query: str, + index_names: Optional[Union[List[str], str]] = None, + top_k: int = 10, + score_threshold: Optional[float] = None, + filter: Optional[Union[List, Dict]] = None, + return_fields: Optional[Union[List, Dict, str]] = None, + index_ids: Optional[Union[List[str], str]] = None, + ) -> SearchResult: + """Run direct Search V2 semantic retrieval for this video. + + Use ``index_names`` or ``index_ids`` to target semantic indexes. Singular + legacy selectors such as ``index_name``/``index_id`` are not accepted by + this method. + + :param str query: Natural-language query. + :param index_names: Optional Search V2 semantic index name or names. + :param int top_k: Maximum number of shots to return. + :param float score_threshold: Optional minimum similarity score. + :param filter: Optional Search V2 filter. + :param return_fields: Optional metadata fields to include. + :param index_ids: Optional Search V2 index ID or IDs. + :return: ``SearchResult`` with shots and server-provided ``warnings``. + """ + search_data = self._connection.post( + path=f"{ApiPath.video}/{self.id}/{ApiPath.semantic_search}", + data={ + "query": query, + "index_names": index_names, + "index_ids": index_ids, + "top_k": top_k, + "score_threshold": score_threshold, + "filter": filter, + "return_fields": return_fields, + }, + ) + return SearchResult(self._connection, **search_data) + + def query( + self, + index_name: Optional[str] = None, + filter: Optional[Union[List, Dict]] = None, + limit: int = 100, + return_fields: Optional[Union[List, Dict, str]] = None, + sort: Optional[Union[str, List[Tuple[str, str]]]] = None, + index_id: Optional[str] = None, + ) -> SearchResult: + """Run a structured Search V2 query for this video. + + ``query()`` is V2-only and is intended for filtering, sorting, and + retrieving indexed records without natural-language planning. + + :param str index_name: Optional Search V2 index name. + :param filter: Optional Search V2 filter. + :param int limit: Maximum number of records to return. + :param return_fields: Optional fields to include in each result. + :param sort: Optional sort field or ``[(field, direction)]`` list. + :param str index_id: Optional Search V2 index ID. + :return: ``SearchResult`` with shots and server-provided ``warnings``. + """ + query_data = self._connection.post( + path=f"{ApiPath.video}/{self.id}/{ApiPath.query}", + data={ + "index_name": index_name, + "index_id": index_id, + "filter": filter, + "limit": limit, + "return_fields": return_fields, + "sort": sort, + }, + ) + return SearchResult(self._connection, **query_data) + + def aggregate( + self, + index_name: Optional[str] = None, + filter: Optional[Union[List, Dict]] = None, + group_by: Optional[str] = None, + metric: str = "count", + limit: int = 100, + sort: Optional[Union[str, List[Tuple[str, str]]]] = None, + index_id: Optional[str] = None, + ) -> Union[Dict, List[Dict]]: + """Run a Search V2 aggregate over this video's indexed records. + + Use this for counts, facets, and grouped metrics. ``aggregate()`` is + V2-only and returns the server aggregate payload directly. + + :param str index_name: Optional Search V2 index name. + :param filter: Optional Search V2 filter. + :param str group_by: Optional field to group by. + :param str metric: Aggregate metric, default ``"count"``. + :param int limit: Maximum number of aggregate rows. + :param sort: Optional sort field or ``[(field, direction)]`` list. + :param str index_id: Optional Search V2 index ID. + :return: Aggregate dict/list; dict responses may include ``warnings``. + """ + aggregate_data = self._connection.post( + path=f"{ApiPath.video}/{self.id}/{ApiPath.aggregate}", + data={ + "index_name": index_name, + "index_id": index_id, + "filter": filter, + "group_by": group_by, + "metric": metric, + "limit": limit, + "sort": sort, + }, + ) + if isinstance(aggregate_data, dict): + warn_response_warnings_once(aggregate_data.get("warnings") or []) + return aggregate_data + + def legacy_search( self, query: str, search_type: Optional[str] = SearchType.semantic, @@ -93,18 +335,28 @@ def search( filter: List[Dict[str, Any]] = [], **kwargs, ) -> SearchResult: - """Search for a query in the video. + """Search this video using legacy spoken-word or scene indexes. + + Use this when you intentionally want older indexes. New applications + should prefer :meth:`search`, :meth:`semantic_search`, :meth:`query`, + :meth:`aggregate`, or :meth:`ask` for Search V2 indexes. The ``index_id`` + keyword is accepted as an alias for ``scene_index_id``. + + Server-provided migration warnings are available on ``result.warnings``. :param str query: Query to search for. - :param SearchType search_type: (optional) Type of search to perform :class:`SearchType ` object - :param IndexType index_type: (optional) Type of index to search :class:`IndexType ` object - :param int result_threshold: (optional) Number of results to return - :param float score_threshold: (optional) Threshold score for the search - :param float dynamic_score_percentage: (optional) Percentage of dynamic score to consider - :raise SearchError: If the search fails - :return: :class:`SearchResult ` object - :rtype: :class:`videodb.search.SearchResult` + :param SearchType search_type: Legacy search type. + :param IndexType index_type: Legacy index type. + :param int result_threshold: Number of results to return. + :param float score_threshold: Minimum score threshold. + :param float dynamic_score_percentage: Dynamic score percentage. + :param list filter: Legacy metadata filters. + :return: ``SearchResult`` with shots and server-provided ``warnings``. """ + kwargs.pop("_skip_warning", False) + if kwargs.get("scene_index_id") is None and kwargs.get("index_id") is not None: + kwargs["scene_index_id"] = kwargs.get("index_id") + kwargs.pop("index_id", None) search = SearchFactory(self._connection).get_search(search_type) return search.search_inside_video( video_id=self.id, @@ -733,6 +985,257 @@ def delete_scene_index(self, scene_index_id: str) -> None: path=f"{ApiPath.video}/{self.id}/{ApiPath.index}/{ApiPath.scene}/{scene_index_id}" ) + def understand( + self, + analyzers: List[Dict[str, Any]], + segmentation: Optional[Dict[str, Any]] = None, + sampling: Optional[Dict[str, Any]] = None, + transform: Optional[Dict[str, Any]] = None, + audio_chunking: Optional[Dict[str, Any]] = None, + callback_url: Optional[str] = None, + **kwargs, + ) -> Understanding: + """Create an understanding run for this video. + + :param list analyzers: Analyzer definitions, such as analyzer type + ``spoken_words``. + :param dict segmentation: Optional run-level segmentation config + :param dict sampling: Optional run-level sampling config + :param dict transform: Optional run-level transform config + :param dict audio_chunking: Optional run-level audio chunking config + :param str callback_url: Optional URL called when the run completes + :return: :class:`Understanding ` object + """ + normalized_analyzers = normalize_understanding_analyzers(analyzers) + payload = {"analyzers": normalized_analyzers} + optional_fields = { + "segmentation": segmentation, + "sampling": sampling, + "transform": transform, + "audio_chunking": audio_chunking, + "callback_url": callback_url, + **kwargs, + } + payload.update({key: value for key, value in optional_fields.items() if value is not None}) + + data = self._connection.post( + path=f"{ApiPath.video}/{self.id}/{ApiPath.understand}", + data=payload, + ) or {} + data.setdefault( + "analyzers", + [ + { + "name": analyzer.get("name"), + "type": analyzer.get("type"), + "status": "pending", + } + for analyzer in normalized_analyzers + ], + ) + data.setdefault("video_id", self.id) + data.setdefault("collection_id", self.collection_id) + return Understanding(self._connection, **data) + + def get_understanding(self, understanding_id: str) -> Understanding: + """Get an understanding run by id. + + :param str understanding_id: Understanding run id + :return: :class:`Understanding ` object + """ + if not understanding_id: + raise ValueError("understanding_id is required") + data = self._connection.get( + path=f"{ApiPath.video}/{self.id}/{ApiPath.understand}/{understanding_id}" + ) or {} + data.setdefault("video_id", self.id) + data.setdefault("collection_id", self.collection_id) + data.setdefault("understanding_id", understanding_id) + return Understanding(self._connection, **data) + + def list_understandings(self) -> List[Understanding]: + """List understanding runs for this video.""" + data = self._connection.get(path=f"{ApiPath.video}/{self.id}/{ApiPath.understand}") + results = (data or {}).get("understanding_results") or [] + understandings = [] + for item in results: + data = dict(item) + data.setdefault("video_id", self.id) + data.setdefault("collection_id", self.collection_id) + understandings.append(Understanding(self._connection, **data)) + return understandings + + def delete_understanding(self, understanding_id: str) -> None: + """Delete an understanding run.""" + if not understanding_id: + raise ValueError("understanding_id is required") + self._connection.delete( + path=f"{ApiPath.video}/{self.id}/{ApiPath.understand}/{understanding_id}" + ) + + @staticmethod + def _format_index_source(source: Union[object, Dict]) -> Dict: + """Format an index *source* into the request payload. + + Exactly two source kinds are supported: + + - an :class:`UnderstandingAnalyzer ` + (anything exposing ``to_index_source()``) — serialized as a light reference + ``{understanding_id, analyzer_id, ...}``; the server re-fetches the analyzer + output from its own store, so scenes never round-trip through the client + - a dict carrying either ``scenes`` (user-provided temporal records) or an + ``understanding_id`` reference (optionally with ``analyzer_id`` / + ``analyzer_type``), passed through as-is + - a bare list of temporal record dicts — sugar for ``{"scenes": [...]}`` + + :param source: The analyzer object, source dict, or list of temporal records + :raises ValueError: If the source is missing or of an unsupported type + :return: The serialized ``source`` payload + :rtype: dict + """ + if source is None: + raise ValueError("source is required") + + if hasattr(source, "to_index_source"): + return source.to_index_source() + + if isinstance(source, list): + return {"scenes": source} + + if isinstance(source, dict): + if isinstance(source.get("scenes"), list) or source.get("understanding_id"): + return source + raise ValueError( + "source dict must carry 'scenes' (temporal records) or an " + "'understanding_id' reference" + ) + + raise ValueError( + "source must be an analyzer object, a dict with 'scenes' or " + "'understanding_id', or a list of temporal records — got " + + type(source).__name__ + ) + + def _format_index(self, index_data: dict) -> Index: + index_data = dict(index_data) + video_id = index_data.pop("video_id", None) or self.id + collection_id = index_data.pop("collection_id", None) or self.collection_id + return Index( + self._connection, + video_id=video_id, + collection_id=collection_id, + **index_data, + ) + + def index( + self, + source: Union[object, Dict, List], + name: Optional[str] = None, + use_for: Optional[List[str]] = None, + fields: Optional[Dict[str, List[str]]] = None, + callback_url: Optional[str] = None, + ) -> Optional[Index]: + """Create a retrieval-ready index from an understanding artifact. + + Turns an understanding artifact (or user-provided temporal records) into an + index that declares retrieval capabilities (``use_for``) and field-level + indexing configuration (``fields``). + + :param source: An :class:`UnderstandingAnalyzer` object (indexed by reference — + scenes never leave the server), or a dict carrying ``scenes`` (temporal + records) or an ``understanding_id`` reference + :param str name: (optional) User-facing index name. Defaults to the + artifact/source name on the server. + :param list use_for: (optional) Retrieval capabilities to enable, any of + :attr:`IndexCapability.semantic `, + :attr:`IndexCapability.query `, + :attr:`IndexCapability.aggregate `. + Defaults to the artifact's defaults on the server. + :param dict fields: (optional) Field-level indexing configuration mapping + field groups (``semantic``, ``filter``, ``aggregate``, + ``sort``) to lists of field names + :param str callback_url: (optional) URL called when indexing completes + :raises ValueError: If ``source`` is missing or of an unsupported type + :raises InvalidRequestError: If the index creation fails + :return: The created index, :class:`Index ` object + :rtype: :class:`videodb.index.Index` + """ + index_data = self._connection.post( + path=f"{ApiPath.video}/{self.id}/{ApiPath.indexes}", + data={ + "source": self._format_index_source(source), + "name": name, + "use_for": use_for, + "fields": fields, + "callback_url": callback_url, + }, + ) + if not index_data: + return None + return self._format_index(index_data) + + def get_index( + self, index_id: Optional[str] = None, name: Optional[str] = None + ) -> Optional[Index]: + """Get an index manifest by its ID or name. + + :param str index_id: (optional) The id of the index + :param str name: (optional) The name of the index + :raises ValueError: If neither ``index_id`` nor ``name`` is provided + :return: The index, :class:`Index ` object + :rtype: :class:`videodb.index.Index` + """ + if not index_id and not name: + raise ValueError("Either index_id or name is required") + params = {"collection_id": self.collection_id} + if index_id: + path = f"{ApiPath.video}/{self.id}/{ApiPath.indexes}/{index_id}" + else: + path = f"{ApiPath.video}/{self.id}/{ApiPath.indexes}" + params["name"] = name + index_data = self._connection.get(path=path, params=params) + if not index_data: + return None + return self._format_index(index_data) + + def list_indexes(self, use_for: Optional[str] = None) -> List[Index]: + """List all the indexes of the video. + + :param str use_for: (optional) Filter by retrieval capability, any of + :attr:`IndexCapability.semantic `, + :attr:`IndexCapability.query `, + :attr:`IndexCapability.aggregate ` + :return: List of :class:`Index ` objects + :rtype: list[:class:`videodb.index.Index`] + """ + params = {"collection_id": self.collection_id} + if use_for is not None: + params["use_for"] = use_for + index_data = self._connection.get( + path=f"{ApiPath.video}/{self.id}/{ApiPath.indexes}", + params=params, + ) + return [self._format_index(index) for index in index_data.get("indexes", [])] + + def delete_index(self, index_id: str) -> None: + """Delete an index. + + Removes the index's retrieval structures. It does not delete the original + video or stored understanding artifacts. + + :param str index_id: The id of the index to be deleted + :raises ValueError: If ``index_id`` is not provided + :raises InvalidRequestError: If the delete fails + :return: None if the delete is successful + :rtype: None + """ + if not index_id: + raise ValueError("index_id is required") + self._connection.delete( + path=f"{ApiPath.video}/{self.id}/{ApiPath.indexes}/{index_id}", + params={"collection_id": self.collection_id}, + ) + def add_subtitle(self, style: SubtitleStyle = SubtitleStyle()) -> str: """Add subtitles to the video.