diff --git a/videodb/__init__.py b/videodb/__init__.py index 59d2a1b..6f2dfb0 100644 --- a/videodb/__init__.py +++ b/videodb/__init__.py @@ -29,6 +29,7 @@ from videodb.websocket_client import WebSocketConnection from videodb.capture import CaptureClient, Channel, AudioChannel, VideoChannel, Channels, ChannelList +from videodb.index import IndexResult from videodb.exceptions import ( VideodbError, AuthenticationError, @@ -53,6 +54,7 @@ "AuthenticationError", "InvalidRequestError", "IndexType", + "IndexResult", "SearchError", "play_stream", "build_iframe_embed_code", diff --git a/videodb/_constants.py b/videodb/_constants.py index f7de578..aba9195 100644 --- a/videodb/_constants.py +++ b/videodb/_constants.py @@ -116,6 +116,9 @@ class ApiPath: token = "token" websocket = "websocket" export = "export" + understand = "understand" + indexes = "indexes" + async_response = "async-response" class Status: diff --git a/videodb/index.py b/videodb/index.py new file mode 100644 index 0000000..4dc8cfe --- /dev/null +++ b/videodb/index.py @@ -0,0 +1,37 @@ +class IndexResult: + """Result of a ``video.index()`` call.""" + + def __init__( + self, + _connection=None, + index_id=None, + video_id=None, + collection_id=None, + name=None, + type=None, + status=None, + source=None, + config=None, + use_for=None, + output_url=None, + created_at=None, + **kwargs, + ): + self._connection = _connection + self.id = index_id + self.video_id = video_id + self.collection_id = collection_id + self.name = name + self.type = type + self.status = status + self.source = source or {} + self.config = config or {} + self.use_for = use_for or [] + self.output_url = output_url + self.created_at = created_at + + def __repr__(self): + return ( + f"IndexResult(id={self.id}, type={self.type}, " + f"status={self.status}, name={self.name})" + ) diff --git a/videodb/understanding.py b/videodb/understanding.py new file mode 100644 index 0000000..2950e42 --- /dev/null +++ b/videodb/understanding.py @@ -0,0 +1,117 @@ +class FaceDetection: + """A single detected face occurrence in a frame.""" + + def __init__(self, bbox, confidence, face_detection_id=None, timestamp_ms=None, **kwargs): + self.face_detection_id = face_detection_id + self.bbox = bbox + self.confidence = float(confidence) if confidence is not None else None + self.timestamp_ms = int(timestamp_ms) if timestamp_ms is not None else None + + def __repr__(self): + return f"FaceDetection(bbox={self.bbox}, confidence={self.confidence})" + + +class SegmentResult: + """Detection results for a single time segment / frame.""" + + def __init__(self, timestamp_ms=None, frame_url=None, detections=None, **kwargs): + self.timestamp_ms = int(timestamp_ms) if timestamp_ms is not None else None + self.frame_url = frame_url + self.detections = [ + FaceDetection(**d) if isinstance(d, dict) else d + for d in (detections or []) + ] + + def __repr__(self): + return f"SegmentResult(timestamp_ms={self.timestamp_ms}, faces={len(self.detections)})" + + +class UnderstandingResult: + """Result of a video.understand() call.""" + + def __init__( + self, + _connection=None, + understanding_id=None, + video_id=None, + collection_id=None, + extract=None, + status=None, + store=None, + config=None, + results=None, + **kwargs, + ): + self._connection = _connection + self.id = understanding_id + self.video_id = video_id + self.collection_id = collection_id + self.extract = extract or [] + self.status = status + self.store = store + self.config = config or {} + self.results_raw = results or {} + self.results = self._parse_results(results) + + @staticmethod + def _parse_results(results): + """Parse results from either dict format (get) or list format (create). + + Server get_understanding returns: + {"faces": {"status": "done", "data": [segment dicts]}} + Server create/understand returns: + [segment dicts] + """ + if results is None: + return [] + + # Dict format from get_understanding: {extract_type: {status, data}} + if isinstance(results, dict): + segments = [] + for extract_type, extract_data in results.items(): + if isinstance(extract_data, dict): + data = extract_data.get("data") + if isinstance(data, list): + segments.extend(data) + elif isinstance(extract_data, list): + segments.extend(extract_data) + return [ + SegmentResult(**s) if isinstance(s, dict) else s + for s in segments + ] + + # List format from create understanding + if isinstance(results, list): + return [ + SegmentResult(**r) if isinstance(r, dict) else r + for r in results + ] + + return [] + + def to_source_dict(self) -> dict: + """Serialize detection data for passing to video.index(source=...).""" + return { + "type": "faces", + "detections": [ + { + "timestamp_ms": seg.timestamp_ms, + "frame_url": seg.frame_url, + "detections": [ + { + "face_detection_id": d.face_detection_id, + "bbox": d.bbox, + "confidence": d.confidence, + } + for d in seg.detections + ], + } + for seg in self.results + ], + } + + def __repr__(self): + return ( + f"UnderstandingResult(id={self.id}, status={self.status}, " + f"segments={len(self.results)})" + ) diff --git a/videodb/video.py b/videodb/video.py index 37d0526..7d9b312 100644 --- a/videodb/video.py +++ b/videodb/video.py @@ -1,5 +1,5 @@ from typing import Literal, Optional, Union, List, Dict, Tuple, Any -from videodb._utils._video import play_stream, build_iframe_embed_code +from videodb._utils._video import play_stream from videodb._constants import ( ApiPath, IndexType, @@ -15,8 +15,8 @@ from videodb.scene import Scene, SceneCollection from videodb.search import SearchFactory, SearchResult from videodb.shot import Shot - -_VALID_SEGMENTERS = {Segmenter.word, Segmenter.sentence, Segmenter.time} +from videodb.index import IndexResult +from videodb.understanding import UnderstandingResult class Video: @@ -156,9 +156,7 @@ def generate_stream( "length": self.length, }, ) - self.stream_url = stream_data.get("stream_url") - self.player_url = stream_data.get("player_url") - return self.stream_url + return stream_data.get("stream_url", None) def generate_thumbnail(self, time: Optional[float] = None) -> Union[str, Image]: """Generate the thumbnail of the video. @@ -204,19 +202,6 @@ def _fetch_transcript( length: int = 1, force: bool = None, ) -> None: - if segmenter not in _VALID_SEGMENTERS: - raise ValueError( - f"Invalid segmenter '{segmenter}'. " - f"Must be one of: {', '.join(sorted(_VALID_SEGMENTERS))}" - ) - if start is not None and start < 0: - raise ValueError(f"start must be non-negative, got {start}") - if end is not None and end < 0: - raise ValueError(f"end must be non-negative, got {end}") - if start is not None and end is not None and start > end: - raise ValueError( - f"start ({start}) must be less than or equal to end ({end})" - ) if ( self.transcript and not start @@ -250,18 +235,12 @@ def get_transcript( ) -> List[Dict[str, Union[float, str]]]: """Get timestamped transcript segments for the video. - :param int start: Start time in seconds (must be >= 0 and <= end) - :param int end: End time in seconds (must be >= 0 and >= start) - :param Segmenter segmenter: How to split the transcript into segments. - Must be one of :attr:`Segmenter.word` (default, one segment per word), - :attr:`Segmenter.sentence` (one segment per sentence), or - :attr:`Segmenter.time` (fixed-duration segments controlled by *length*) - :param int length: Duration in seconds for each segment when - *segmenter* is :attr:`Segmenter.time` (default 1) - :param bool force: Force re-fetch transcript from the server, - bypassing the local cache - :raises ValueError: If *segmenter* is not a valid value, or if - *start*/*end* are negative or *start* > *end* + :param int start: Start time in seconds + :param int end: End time in seconds + :param Segmenter segmenter: Segmentation type (:class:`Segmenter.word`, + :class:`Segmenter.sentence`, :class:`Segmenter.time`) + :param int length: Length of segments when using time segmenter + :param bool force: Force fetch new transcript :return: List of dicts with keys: start (float), end (float), text (str) :rtype: List[Dict[str, Union[float, str]]] """ @@ -294,10 +273,7 @@ def generate_transcript( """Generate transcript for the video. :param bool force: Force generate new transcript - :param str language_code: (optional) Language code for transcription. - Use ISO 639-1 codes (e.g., "en", "hi", "fr") or regional - variants with underscores (e.g., "en_us", "en_uk", "en_au"). - Defaults to "en_us" if not specified. + :param str language_code: (optional) Language code of the video :return: Full transcript text as string :rtype: str """ @@ -350,10 +326,7 @@ def index_spoken_words( ) -> None: """Semantic indexing of spoken words in the video. - :param str language_code: (optional) Language code for transcription. - Use ISO 639-1 codes (e.g., "en", "hi", "fr") or regional - variants with underscores (e.g., "en_us", "en_uk", "en_au"). - Defaults to "en_us" if not specified. + :param str language_code: (optional) Language code of the video :param SegmentationType segmentation_type: (optional) Segmentation type used for indexing, :class:`SegmentationType ` object :param bool force: (optional) Force to index the video :param str callback_url: (optional) URL to receive the callback @@ -654,10 +627,7 @@ def index_audio( :param str prompt: (optional) Prompt for processing transcript segments :param str model_name: (optional) LLM tier to use (e.g. "basic", "pro", "ultra") :param dict model_config: (optional) Model configuration - :param str language_code: (optional) Language code for transcription. - Use ISO 639-1 codes (e.g., "en", "hi", "fr") or regional - variants with underscores (e.g., "en_us", "en_uk", "en_au"). - Defaults to "en_us" if not specified. + :param str language_code: (optional) Language code for transcription :param dict batch_config: (optional) Segmentation config with keys: - "type": Segmentation type ("word", "sentence", or "time") - "value": Segment length (words, sentences, or seconds) @@ -822,41 +792,6 @@ def play(self) -> str: """ return play_stream(self.stream_url) - 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 HTML iframe embed code for the video. - - :param str width: Width of the iframe (default: "100%") - :param int height: Height of the iframe in pixels (default: 405) - :param str title: Title attribute for the iframe (default: "VideoDB Player") - :param bool allow_fullscreen: Whether to allow fullscreen (default: True) - :param bool auto_generate: If True and player_url is missing, auto-generate it (default: True) - :return: HTML iframe string - :rtype: str - :raises ValueError: If player_url is not available - """ - if not self.player_url and auto_generate: - self.generate_stream() - - if not self.player_url: - raise ValueError( - "player_url not available. Call generate_stream() first or set auto_generate=True." - ) - - return build_iframe_embed_code( - player_url=self.player_url, - width=width, - height=height, - title=title, - allow_fullscreen=allow_fullscreen, - ) - def get_meeting(self): """Get meeting information associated with the video. @@ -952,3 +887,179 @@ def download(self, name: Optional[str] = None) -> dict: download_name = name or self.name or f"video_{self.id}" return self._connection.download(self.stream_url, download_name) + + # ── Generic Index ────────────────────────────────────────────────── + + def index( + self, + source=None, + config: Optional[Dict] = None, + use_for: Optional[List[str]] = None, + name: Optional[str] = None, + callback_url: Optional[str] = None, + ) -> Optional[IndexResult]: + """Create an index on this video. + + :param source: Source data — an :class:`UnderstandingResult` object + (calls ``to_source_dict()`` automatically), or a raw dict + :param dict config: Configuration for the index + :param list use_for: What this index is used for (e.g. ["search", "query"]) + :param str name: Name for the index + :param str callback_url: URL to receive callback when done (optional) + :return: :class:`IndexResult ` object + :rtype: :class:`videodb.index.IndexResult` + """ + data = {} + if source is not None: + if hasattr(source, "to_source_dict"): + data["source"] = source.to_source_dict() + elif isinstance(source, dict): + data["source"] = source + if config is not None: + data["config"] = config + if use_for is not None: + data["use_for"] = use_for + if name is not None: + data["name"] = name + if callback_url is not None: + data["callback_url"] = callback_url + + response = self._connection.post( + path=f"{ApiPath.video}/{self.id}/{ApiPath.indexes}", + data=data, + ) + if not response: + return None + return IndexResult( + _connection=self._connection, + video_id=self.id, + **response, + ) + + def get_index(self, index_id: str) -> Optional[IndexResult]: + """Get an index by its ID. + + :param str index_id: The index ID + :return: :class:`IndexResult ` object + :rtype: :class:`videodb.index.IndexResult` + """ + response = self._connection.get( + path=f"{ApiPath.video}/{self.id}/{ApiPath.indexes}/{index_id}", + ) + if not response: + return None + return IndexResult( + _connection=self._connection, + video_id=self.id, + **response, + ) + + def list_indexes(self) -> List[IndexResult]: + """List all indexes for this video. + + :return: List of :class:`IndexResult ` objects + :rtype: List[:class:`videodb.index.IndexResult`] + """ + response = self._connection.get( + path=f"{ApiPath.video}/{self.id}/{ApiPath.indexes}", + ) + if not response: + return [] + return [ + IndexResult(_connection=self._connection, video_id=self.id, **idx) + for idx in response.get("indexes", []) + ] + + def delete_index(self, index_id: str) -> None: + """Delete an index. + + :param str index_id: The index ID to delete + """ + self._connection.delete( + path=f"{ApiPath.video}/{self.id}/{ApiPath.indexes}/{index_id}", + ) + + # ── Understanding ────────────────────────────────────────────────── + + def understand( + self, + extract: list, + segmentation: Optional[dict] = None, + sampling: Optional[dict] = None, + transform: Optional[dict] = None, + store: bool = False, + callback_url: Optional[str] = None, + ) -> Optional[UnderstandingResult]: + """Run face understanding (detection) on the video. + + :param list extract: What to extract, e.g. ["faces"] + :param dict segmentation: Segmentation config, e.g. {"type": "time", "window": "1s"} + :param dict sampling: Sampling config, e.g. {"frame_count": 2} + :param dict transform: Transform config, e.g. {"frame_size": "480p"} + :param bool store: Whether to persist the understanding result + :param str callback_url: URL to receive callback when done (optional) + :return: :class:`UnderstandingResult ` object + :rtype: :class:`videodb.understanding.UnderstandingResult` + """ + data = { + "extract": extract, + "segmentation": segmentation or {"type": "time", "window": "1s"}, + "sampling": sampling or {"frame_count": 2}, + "store": store, + } + if transform: + data["transform"] = transform + if callback_url: + data["callback_url"] = callback_url + + response = self._connection.post( + path=f"{ApiPath.video}/{self.id}/{ApiPath.understand}", + data=data, + ) + if not response: + return None + return UnderstandingResult(_connection=self._connection, **response) + + def get_understanding(self, understanding_id: str) -> Optional[UnderstandingResult]: + """Fetch a stored understanding result. + + :param str understanding_id: The understanding result ID + :return: :class:`UnderstandingResult ` object + :rtype: :class:`videodb.understanding.UnderstandingResult` + """ + response = self._connection.get( + path=f"{ApiPath.video}/{self.id}/{ApiPath.understand}/{understanding_id}", + ) + if not response: + return None + return UnderstandingResult(_connection=self._connection, **response) + + def list_understanding(self, extract: Optional[list] = None) -> list: + """List understanding results for this video. + + :param list extract: Filter by extract type, e.g. ["faces"] + :return: List of :class:`UnderstandingResult ` summaries + :rtype: list + """ + params = {} + if extract: + params["extract"] = ",".join(extract) if isinstance(extract, list) else extract + response = self._connection.get( + path=f"{ApiPath.video}/{self.id}/{ApiPath.understand}", + params=params, + ) + if not response: + return [] + return [ + UnderstandingResult(_connection=self._connection, **r) + for r in response.get("understanding_results", []) + ] + + def delete_understanding(self, understanding_id: str) -> None: + """Delete a stored understanding result. + + :param str understanding_id: The understanding result ID + """ + self._connection.delete( + path=f"{ApiPath.video}/{self.id}/{ApiPath.understand}/{understanding_id}", + )