From 705c7f56bf4cab89ea4650ee089783a83144125c Mon Sep 17 00:00:00 2001 From: Rohit Garg Date: Fri, 10 Apr 2026 13:55:51 +0530 Subject: [PATCH 01/18] indexing: add understand, indexing and face store support --- videodb/__init__.py | 6 + videodb/_constants.py | 7 ++ videodb/collection.py | 57 +++++++++ videodb/face.py | 174 ++++++++++++++++++++++++++ videodb/store.py | 258 +++++++++++++++++++++++++++++++++++++++ videodb/understanding.py | 117 ++++++++++++++++++ videodb/video.py | 178 +++++++++++++++++++++++++++ 7 files changed, 797 insertions(+) create mode 100644 videodb/face.py create mode 100644 videodb/store.py create mode 100644 videodb/understanding.py diff --git a/videodb/__init__.py b/videodb/__init__.py index a5adc9c..06b0de3 100644 --- a/videodb/__init__.py +++ b/videodb/__init__.py @@ -29,6 +29,8 @@ from videodb.websocket_client import WebSocketConnection from videodb.capture import CaptureClient, Channel, AudioChannel, VideoChannel, Channels, ChannelList +from videodb.face import IndexResult, Identity, Face +from videodb.store import FaceStore from videodb.exceptions import ( VideodbError, AuthenticationError, @@ -53,6 +55,10 @@ "AuthenticationError", "InvalidRequestError", "IndexType", + "IndexResult", + "Identity", + "Face", + "FaceStore", "SearchError", "play_stream", "MediaType", diff --git a/videodb/_constants.py b/videodb/_constants.py index b546fab..462bca4 100644 --- a/videodb/_constants.py +++ b/videodb/_constants.py @@ -111,6 +111,13 @@ class ApiPath: token = "token" websocket = "websocket" export = "export" + understand = "understand" + indexes = "indexes" + store = "store" + faces = "faces" + identities = "identities" + merge = "merge" + split = "split" class Status: diff --git a/videodb/collection.py b/videodb/collection.py index 04c8d05..cf68b30 100644 --- a/videodb/collection.py +++ b/videodb/collection.py @@ -17,6 +17,7 @@ from videodb.capture_session import CaptureSession from videodb.rtstream import RTStream, RTStreamSearchResult, RTStreamShot from videodb.search import SearchFactory, SearchResult +from videodb.store import FaceStore logger = logging.getLogger(__name__) @@ -774,3 +775,59 @@ def list_capture_sessions(self, status: str = None) -> list["CaptureSession"]: ) ) return sessions + + # ── Stores ──────────────────────────────────────────────────────── + + def list_stores(self) -> list: + """List all stores in the collection. + + :return: List of store records + :rtype: list + """ + response = self._connection.get( + path=f"{ApiPath.collection}/{self.id}/{ApiPath.store}", + ) + if not response: + return [] + return response.get("stores", []) + + def get_store(self, store_type: str) -> Optional[dict]: + """Get a specific store by type. + + :param str store_type: Store type (e.g. "faces") + :return: Store record + :rtype: dict + """ + return self._connection.get( + path=f"{ApiPath.collection}/{self.id}/{ApiPath.store}/{store_type}", + ) + + @property + def face_store(self) -> FaceStore: + """Access the face store for this collection. + + Provides identity and face management via nested managers:: + + store = collection.face_store + + # Identity operations + identities = store.identities.list() + identity = store.identities.get("abc123") + identity.update(name="Ashish") + store.identities.merge(source_ids=["id1", "id2"], target_name="Final") + store.identities.split("id1", face_ids=["f1"], new_identity_name="New") + + # Face operations + faces = store.faces.list(video_id="m-xxx") + face = store.faces.get("face_abc") + store.faces.delete("face_abc") + + :return: :class:`FaceStore ` object + :rtype: :class:`videodb.store.FaceStore` + """ + if not hasattr(self, "_face_store"): + self._face_store = FaceStore( + _connection=self._connection, + _collection_id=self.id, + ) + return self._face_store diff --git a/videodb/face.py b/videodb/face.py new file mode 100644 index 0000000..7ff178f --- /dev/null +++ b/videodb/face.py @@ -0,0 +1,174 @@ +from videodb._constants import ApiPath + + +class IndexResult: + """Result of a video.index() call. + + For async index types (e.g. face), the SDK auto-polls until completion. + """ + + 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})" + ) + + +class Identity: + """A face identity within a collection's face store. + + Typically accessed via ``collection.face_store.identities.get(id)``. + """ + + def __init__( + self, + _connection=None, + _collection_id=None, + identity_id=None, + collection_id=None, + name=None, + face_count=None, + example_count=None, + preview_faces=None, + source_index_ids=None, + created_at=None, + updated_at=None, + **kwargs, + ): + self._connection = _connection + self._collection_id = _collection_id or collection_id + self.id = identity_id + self.collection_id = collection_id + self.name = name + self.face_count = face_count or 0 + self.example_count = example_count or 0 + self.representative_faces = kwargs.get("representative_faces", []) + self.preview_faces = preview_faces or [] + self.source_index_ids = source_index_ids or [] + self.created_at = created_at + self.updated_at = updated_at + + def _identity_path(self): + return ( + f"{ApiPath.collection}/{self._collection_id}" + f"/{ApiPath.store}/{ApiPath.faces}" + f"/{ApiPath.identities}/{self.id}" + ) + + def update(self, name=None, set_representative=None): + """Update this identity. + + :param str name: New name for the identity + :param dict set_representative: {"face_id": "..."} to change the representative face + :return: Updated response data + :rtype: dict + """ + data = {} + if name is not None: + data["name"] = name + if set_representative is not None: + data["set_representative"] = set_representative + if not data: + return + response = self._connection.patch( + path=self._identity_path(), + data=data, + ) + if name is not None: + self.name = name + return response + + def delete(self): + """Delete this identity and unassign all its faces.""" + self._connection.delete(path=self._identity_path()) + + def __repr__(self): + return ( + f"Identity(id={self.id}, name={self.name}, " + f"face_count={self.face_count})" + ) + + +class Face: + """A single detected and indexed face record. + + Typically accessed via ``collection.face_store.faces.get(id)``. + """ + + def __init__( + self, + _connection=None, + _collection_id=None, + face_id=None, + video_id=None, + collection_id=None, + identity_id=None, + source_index_id=None, + source=None, + timestamp_ms=None, + frame_url=None, + image_url=None, + bbox=None, + confidence=None, + created_at=None, + **kwargs, + ): + self._connection = _connection + self._collection_id = _collection_id or collection_id + self.id = face_id + self.video_id = video_id + self.collection_id = collection_id + self.identity_id = identity_id + self.source_index_id = source_index_id + self.source = source + self.timestamp_ms = timestamp_ms + self.frame_url = frame_url + self.image_url = image_url + self.bbox = bbox + self.confidence = confidence + self.created_at = created_at + + def delete(self): + """Delete this face record.""" + self._connection.delete( + path=( + f"{ApiPath.collection}/{self._collection_id}" + f"/{ApiPath.store}/{ApiPath.faces}" + f"/{ApiPath.faces}/{self.id}" + ), + ) + + def __repr__(self): + return ( + f"Face(id={self.id}, identity_id={self.identity_id}, " + f"confidence={self.confidence})" + ) diff --git a/videodb/store.py b/videodb/store.py new file mode 100644 index 0000000..de1fb64 --- /dev/null +++ b/videodb/store.py @@ -0,0 +1,258 @@ +from typing import Optional, List + +from videodb._constants import ApiPath +from videodb.face import Identity, Face + + +class IdentityManager: + """Manager for face identity operations within a FaceStore. + + Access via ``collection.face_store.identities``. + """ + + def __init__(self, _connection, _collection_id): + self._connection = _connection + self._collection_id = _collection_id + + def _path(self, *parts): + segments = [ + ApiPath.collection, self._collection_id, + ApiPath.store, ApiPath.faces, ApiPath.identities, + ] + segments.extend(parts) + return "/".join(segments) + + def list(self, named: Optional[bool] = None) -> List[Identity]: + """List all identities in the face store. + + :param bool named: Filter by named (True) or unnamed (False) identities + :return: List of :class:`Identity ` objects + :rtype: List[:class:`videodb.face.Identity`] + """ + params = {} + if named is not None: + params["named"] = str(named).lower() + + response = self._connection.get(path=self._path(), params=params) + if not response: + return [] + return [ + Identity( + _connection=self._connection, + _collection_id=self._collection_id, + **ident, + ) + for ident in response.get("identities", []) + ] + + def get(self, identity_id: str) -> Optional[Identity]: + """Get an identity by ID. + + :param str identity_id: The identity ID + :return: :class:`Identity ` object + :rtype: :class:`videodb.face.Identity` + """ + response = self._connection.get(path=self._path(identity_id)) + if not response: + return None + return Identity( + _connection=self._connection, + _collection_id=self._collection_id, + **response, + ) + + def update( + self, + identity_id: str, + name: Optional[str] = None, + set_representative: Optional[dict] = None, + ) -> Optional[dict]: + """Update an identity. + + :param str identity_id: The identity ID + :param str name: New name for the identity + :param dict set_representative: {"face_id": "..."} to change representative + :return: Response data + :rtype: dict + """ + data = {} + if name is not None: + data["name"] = name + if set_representative is not None: + data["set_representative"] = set_representative + if not data: + return None + return self._connection.patch(path=self._path(identity_id), data=data) + + def delete(self, identity_id: str) -> None: + """Delete an identity and unassign all its faces. + + :param str identity_id: The identity ID to delete + """ + self._connection.delete(path=self._path(identity_id)) + + def merge( + self, + source_ids: List[str], + target_name: Optional[str] = None, + ) -> Optional[Identity]: + """Merge multiple identities into one. + + The identity with the highest face_count is kept as the target. + + :param list source_ids: List of identity IDs to merge (min 2) + :param str target_name: Name for the merged identity + :return: :class:`Identity ` of the merged result + :rtype: :class:`videodb.face.Identity` + """ + data = {"source_ids": source_ids} + if target_name is not None: + data["target_name"] = target_name + + response = self._connection.post( + path=self._path(ApiPath.merge), data=data + ) + if not response: + return None + return Identity( + _connection=self._connection, + _collection_id=self._collection_id, + **response, + ) + + def split( + self, + identity_id: str, + face_ids: List[str], + new_identity_name: Optional[str] = None, + ) -> Optional[dict]: + """Split faces off an identity into a new one. + + :param str identity_id: Source identity ID + :param list face_ids: Face IDs to split off + :param str new_identity_name: Name for the new identity (optional) + :return: New identity data if new_identity_name was provided + :rtype: dict + """ + data = {"face_ids": face_ids} + if new_identity_name is not None: + data["new_identity_name"] = new_identity_name + return self._connection.post( + path=self._path(identity_id, ApiPath.split), data=data + ) + + +class FaceManager: + """Manager for face record operations within a FaceStore. + + Access via ``collection.face_store.faces``. + """ + + def __init__(self, _connection, _collection_id): + self._connection = _connection + self._collection_id = _collection_id + + def _path(self, *parts): + segments = [ + ApiPath.collection, self._collection_id, + ApiPath.store, ApiPath.faces, ApiPath.faces, + ] + segments.extend(parts) + return "/".join(segments) + + def list( + self, + video_id: Optional[str] = None, + identity_id: Optional[str] = None, + source_index_id: Optional[str] = None, + ) -> List[Face]: + """List face records. + + :param str video_id: Filter by video ID + :param str identity_id: Filter by identity ID + :param str source_index_id: Filter by source index ID + :return: List of :class:`Face ` objects + :rtype: List[:class:`videodb.face.Face`] + """ + params = {} + if video_id is not None: + params["video_id"] = video_id + if identity_id is not None: + params["identity_id"] = identity_id + if source_index_id is not None: + params["source_index_id"] = source_index_id + + response = self._connection.get(path=self._path(), params=params) + if not response: + return [] + return [ + Face( + _connection=self._connection, + _collection_id=self._collection_id, + **f, + ) + for f in response.get("faces", []) + ] + + def get(self, face_id: str) -> Optional[Face]: + """Get a face record by ID. + + :param str face_id: The face ID + :return: :class:`Face ` object + :rtype: :class:`videodb.face.Face` + """ + response = self._connection.get(path=self._path(face_id)) + if not response: + return None + return Face( + _connection=self._connection, + _collection_id=self._collection_id, + **response, + ) + + def delete(self, face_id: str) -> None: + """Delete a face record. + + If the face's identity has no remaining faces, the identity is also deleted. + + :param str face_id: The face ID to delete + """ + self._connection.delete(path=self._path(face_id)) + + +class FaceStore: + """Face store for a collection — manages identities and face records. + + Access via ``collection.face_store``. + + Usage:: + + store = collection.face_store + + # Identity operations + identities = store.identities.list() + identity = store.identities.get("abc123") + identity.update(name="Ashish") + store.identities.merge(source_ids=["id1", "id2"], target_name="Final") + store.identities.split("id1", face_ids=["f1", "f2"], new_identity_name="New") + + # Face operations + faces = store.faces.list(video_id="m-xxx") + face = store.faces.get("face_abc") + store.faces.delete("face_abc") + """ + + def __init__(self, _connection, _collection_id, **kwargs): + self._connection = _connection + self._collection_id = _collection_id + self.type = kwargs.get("type", "faces") + self.collection_id = _collection_id + self.status = kwargs.get("status") + self.created_at = kwargs.get("created_at") + self.updated_at = kwargs.get("updated_at") + + self.identities = IdentityManager(_connection, _collection_id) + self.faces = FaceManager(_connection, _collection_id) + + def __repr__(self): + return f"FaceStore(collection_id={self._collection_id}, status={self.status})" diff --git a/videodb/understanding.py b/videodb/understanding.py new file mode 100644 index 0000000..527f190 --- /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 = confidence + self.timestamp_ms = timestamp_ms + + 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 = timestamp_ms + 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 20996cd..64d50a8 100644 --- a/videodb/video.py +++ b/videodb/video.py @@ -15,6 +15,8 @@ from videodb.scene import Scene, SceneCollection from videodb.search import SearchFactory, SearchResult from videodb.shot import Shot +from videodb.face import IndexResult +from videodb.understanding import UnderstandingResult class Video: @@ -885,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.face.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.face.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.face.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}", + ) From 94674ce5dc5d46bb0b07dbc5882ff1b48368fe1d Mon Sep 17 00:00:00 2001 From: Rohit Garg Date: Fri, 10 Apr 2026 16:36:52 +0530 Subject: [PATCH 02/18] configurable polling args for long running jobs --- videodb/_constants.py | 2 ++ videodb/_utils/_http_client.py | 45 +++++++++++++++++++++++++--------- videodb/video.py | 30 +++++++++++++++++++++-- 3 files changed, 63 insertions(+), 14 deletions(-) diff --git a/videodb/_constants.py b/videodb/_constants.py index 462bca4..045c932 100644 --- a/videodb/_constants.py +++ b/videodb/_constants.py @@ -137,6 +137,8 @@ class HttpClientDefaultValues: timeout = 30 backoff_factor = 0.1 status_forcelist = [502, 503, 504] + max_poll_time = 500 + poll_interval = 5 class MaxSupported: diff --git a/videodb/_utils/_http_client.py b/videodb/_utils/_http_client.py index ab571f6..7d4d86e 100644 --- a/videodb/_utils/_http_client.py +++ b/videodb/_utils/_http_client.py @@ -1,8 +1,8 @@ """Http client Module.""" import logging +import time import requests -import backoff from tqdm import tqdm from typing import ( @@ -65,6 +65,8 @@ def __init__( self.base_url = base_url self.show_progress = False self.progress_bar = None + self.max_poll_time = HttpClientDefaultValues.max_poll_time + self.poll_interval = HttpClientDefaultValues.poll_interval logger.debug(f"Initialized http client with base url: {self.base_url}") def _make_request( @@ -133,23 +135,31 @@ def _handle_request_error(self, e: requests.exceptions.RequestException) -> None f"Invalid request: {str(e)}", e.response ) from None - @backoff.on_exception( - backoff.constant, Exception, max_time=500, interval=5, logger=None, jitter=None - ) def _get_output(self, url: str): - """Get the output from an async request""" - response_json = self.session.get(url).json() - if ( - response_json.get("status") == Status.in_progress - or response_json.get("status") == Status.processing - ): + """Poll an output URL until the job completes or times out.""" + start = time.monotonic() + while True: + response_json = self.session.get(url).json() + status = response_json.get("status") + if status not in (Status.in_progress, Status.processing): + break + percentage = response_json.get("data", {}).get("percentage") if percentage and self.show_progress and self.progress_bar: self.progress_bar.n = int(percentage) self.progress_bar.update(0) - logger.debug("Waiting for processing to complete") - raise Exception("Stuck on processing status") from None + elapsed = time.monotonic() - start + if elapsed >= self.max_poll_time: + raise RequestTimeoutError( + f"Polling timed out after {int(elapsed)}s " + f"(max_poll_time={self.max_poll_time})", + None, + ) + + logger.debug("Waiting for processing to complete (%.0fs elapsed)", elapsed) + time.sleep(self.poll_interval) + if self.show_progress and self.progress_bar: self.progress_bar.n = 100 self.progress_bar.update(0) @@ -209,11 +219,21 @@ def _format_headers(self, headers: dict): formatted_headers[f"x-{key}"] = value return formatted_headers + def _apply_poll_overrides(self, kwargs): + """Extract and apply per-call poll overrides from kwargs.""" + self.max_poll_time = kwargs.pop( + "max_poll_time", HttpClientDefaultValues.max_poll_time + ) + self.poll_interval = kwargs.pop( + "poll_interval", HttpClientDefaultValues.poll_interval + ) + def get( self, path: str, show_progress: Optional[bool] = False, **kwargs ) -> requests.Response: """Make a get request""" self.show_progress = show_progress + self._apply_poll_overrides(kwargs) return self._make_request(method=self.session.get, path=path, **kwargs) def post( @@ -221,6 +241,7 @@ def post( ) -> requests.Response: """Make a post request""" self.show_progress = show_progress + self._apply_poll_overrides(kwargs) return self._make_request(self.session.post, path, json=data, **kwargs) def put(self, path: str, data=None, **kwargs) -> requests.Response: diff --git a/videodb/video.py b/videodb/video.py index 64d50a8..12d4826 100644 --- a/videodb/video.py +++ b/videodb/video.py @@ -936,15 +936,28 @@ def index( **response, ) - def get_index(self, index_id: str) -> Optional[IndexResult]: + def get_index( + self, + index_id: str, + max_poll_time: Optional[int] = None, + poll_interval: Optional[int] = None, + ) -> Optional[IndexResult]: """Get an index by its ID. :param str index_id: The index ID + :param int max_poll_time: Max seconds to poll if still processing (default: 500) + :param int poll_interval: Seconds between polls (default: 5) :return: :class:`IndexResult ` object :rtype: :class:`videodb.face.IndexResult` """ + poll_kwargs = {} + if max_poll_time is not None: + poll_kwargs["max_poll_time"] = max_poll_time + if poll_interval is not None: + poll_kwargs["poll_interval"] = poll_interval response = self._connection.get( path=f"{ApiPath.video}/{self.id}/{ApiPath.indexes}/{index_id}", + **poll_kwargs, ) if not response: return None @@ -1020,15 +1033,28 @@ def understand( return None return UnderstandingResult(_connection=self._connection, **response) - def get_understanding(self, understanding_id: str) -> Optional[UnderstandingResult]: + def get_understanding( + self, + understanding_id: str, + max_poll_time: Optional[int] = None, + poll_interval: Optional[int] = None, + ) -> Optional[UnderstandingResult]: """Fetch a stored understanding result. :param str understanding_id: The understanding result ID + :param int max_poll_time: Max seconds to poll if still processing (default: 500) + :param int poll_interval: Seconds between polls (default: 5) :return: :class:`UnderstandingResult ` object :rtype: :class:`videodb.understanding.UnderstandingResult` """ + poll_kwargs = {} + if max_poll_time is not None: + poll_kwargs["max_poll_time"] = max_poll_time + if poll_interval is not None: + poll_kwargs["poll_interval"] = poll_interval response = self._connection.get( path=f"{ApiPath.video}/{self.id}/{ApiPath.understand}/{understanding_id}", + **poll_kwargs, ) if not response: return None From fa63876551358b7da2fcd885464a166785e8f4b3 Mon Sep 17 00:00:00 2001 From: Rohit Garg Date: Fri, 10 Apr 2026 17:12:15 +0530 Subject: [PATCH 03/18] make understand and index functions async --- videodb/_utils/_http_client.py | 9 +++++++-- videodb/video.py | 30 +++++++++++++++++------------- 2 files changed, 24 insertions(+), 15 deletions(-) diff --git a/videodb/_utils/_http_client.py b/videodb/_utils/_http_client.py index 7d4d86e..ecab5de 100644 --- a/videodb/_utils/_http_client.py +++ b/videodb/_utils/_http_client.py @@ -75,6 +75,7 @@ def _make_request( path: str, base_url: Optional[str] = None, headers: Optional[dict] = None, + wait: bool = True, **kwargs, ): """Make a request to the api @@ -83,6 +84,7 @@ def _make_request( :param str path: The path to make the request to :param str base_url: (optional) The base url to use for the request :param dict headers: (optional) The headers to use for the request + :param bool wait: If False, return raw data without polling (default True) :param kwargs: The keyword arguments to pass to the request method :return: json response from the request """ @@ -92,6 +94,8 @@ def _make_request( request_headers = {**self.session.headers, **(headers or {})} response = method(url, headers=request_headers, timeout=timeout, **kwargs) response.raise_for_status() + if not wait: + return response.json().get("data") return self._parse_response(response) except requests.exceptions.RequestException as e: @@ -237,12 +241,13 @@ def get( return self._make_request(method=self.session.get, path=path, **kwargs) def post( - self, path: str, data=None, show_progress: Optional[bool] = False, **kwargs + self, path: str, data=None, show_progress: Optional[bool] = False, + wait: bool = True, **kwargs, ) -> requests.Response: """Make a post request""" self.show_progress = show_progress self._apply_poll_overrides(kwargs) - return self._make_request(self.session.post, path, json=data, **kwargs) + return self._make_request(self.session.post, path, json=data, wait=wait, **kwargs) def put(self, path: str, data=None, **kwargs) -> requests.Response: """Make a put request""" diff --git a/videodb/video.py b/videodb/video.py index 12d4826..5f23bb0 100644 --- a/videodb/video.py +++ b/videodb/video.py @@ -897,17 +897,20 @@ def index( use_for: Optional[List[str]] = None, name: Optional[str] = None, callback_url: Optional[str] = None, - ) -> Optional[IndexResult]: + ) -> Optional[str]: """Create an index on this video. + Returns the index_id immediately. Use ``get_index(id)`` to poll + for the result. + :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.face.IndexResult` + :return: index_id string + :rtype: Optional[str] """ data = {} if source is not None: @@ -927,14 +930,11 @@ def index( response = self._connection.post( path=f"{ApiPath.video}/{self.id}/{ApiPath.indexes}", data=data, + wait=False, ) if not response: return None - return IndexResult( - _connection=self._connection, - video_id=self.id, - **response, - ) + return response.get("index_id") def get_index( self, @@ -1002,8 +1002,11 @@ def understand( transform: Optional[dict] = None, store: bool = False, callback_url: Optional[str] = None, - ) -> Optional[UnderstandingResult]: - """Run face understanding (detection) on the video. + ) -> Optional[str]: + """Launch understanding (detection) on the video. + + Returns the understanding_id immediately. Use + ``get_understanding(id)`` to poll for the result. :param list extract: What to extract, e.g. ["faces"] :param dict segmentation: Segmentation config, e.g. {"type": "time", "window": "1s"} @@ -1011,8 +1014,8 @@ def understand( :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` + :return: understanding_id string + :rtype: Optional[str] """ data = { "extract": extract, @@ -1028,10 +1031,11 @@ def understand( response = self._connection.post( path=f"{ApiPath.video}/{self.id}/{ApiPath.understand}", data=data, + wait=False, ) if not response: return None - return UnderstandingResult(_connection=self._connection, **response) + return response.get("understanding_id") def get_understanding( self, From 71aee120ed2cbd0cdfc35c1934c6dffb5e54abc5 Mon Sep 17 00:00:00 2001 From: Rohit Garg Date: Fri, 10 Apr 2026 19:40:09 +0530 Subject: [PATCH 04/18] indexing: add config support in understanding module --- videodb/video.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/videodb/video.py b/videodb/video.py index 5f23bb0..c9b58c1 100644 --- a/videodb/video.py +++ b/videodb/video.py @@ -1000,6 +1000,7 @@ def understand( segmentation: Optional[dict] = None, sampling: Optional[dict] = None, transform: Optional[dict] = None, + config: Optional[dict] = None, store: bool = False, callback_url: Optional[str] = None, ) -> Optional[str]: @@ -1012,6 +1013,8 @@ def understand( :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 dict config: Per-extract-type config, e.g. + ``{"faces": {"confidence_threshold": 0.6, "min_face_size": 30}}`` :param bool store: Whether to persist the understanding result :param str callback_url: URL to receive callback when done (optional) :return: understanding_id string @@ -1025,6 +1028,8 @@ def understand( } if transform: data["transform"] = transform + if config: + data["config"] = config if callback_url: data["callback_url"] = callback_url From 14da21437afb4f298bf414691468bd979efcc6b3 Mon Sep 17 00:00:00 2001 From: Rohit Garg Date: Fri, 10 Apr 2026 21:17:25 +0530 Subject: [PATCH 05/18] indexing: proper cast for float and int types in face data --- videodb/face.py | 4 ++-- videodb/understanding.py | 6 +++--- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/videodb/face.py b/videodb/face.py index 7ff178f..6ce14f0 100644 --- a/videodb/face.py +++ b/videodb/face.py @@ -150,11 +150,11 @@ def __init__( self.identity_id = identity_id self.source_index_id = source_index_id self.source = source - self.timestamp_ms = timestamp_ms + self.timestamp_ms = int(timestamp_ms) if timestamp_ms is not None else None self.frame_url = frame_url self.image_url = image_url self.bbox = bbox - self.confidence = confidence + self.confidence = float(confidence) if confidence is not None else None self.created_at = created_at def delete(self): diff --git a/videodb/understanding.py b/videodb/understanding.py index 527f190..2950e42 100644 --- a/videodb/understanding.py +++ b/videodb/understanding.py @@ -4,8 +4,8 @@ class FaceDetection: 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 = confidence - self.timestamp_ms = timestamp_ms + 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})" @@ -15,7 +15,7 @@ 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 = timestamp_ms + 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 From 810a2298f5a933fc15a2b332bfb6467cccb93b85 Mon Sep 17 00:00:00 2001 From: ashish ashish-spext Date: Thu, 16 Apr 2026 21:11:06 +0530 Subject: [PATCH 06/18] Add prompt_url fallback for large generate_text payloads --- README.md | 3 +++ openapi.yaml | 13 ++++++++++-- videodb/_upload.py | 48 +++++++++++++++++++++++++++++++------------ videodb/collection.py | 35 ++++++++++++++++++++++++++----- videodb/editor.py | 24 ++++++++-------------- 5 files changed, 88 insertions(+), 35 deletions(-) diff --git a/README.md b/README.md index be59222..00b7f95 100644 --- a/README.md +++ b/README.md @@ -610,6 +610,9 @@ response = coll.generate_text( model_name="pro", # basic, pro, or ultra response_type="text" # text or json ) + +# Large prompts are uploaded automatically with a unique filename and +# sent as prompt_url instead of inline JSON to avoid request payload limits. ``` ### Video Dubbing and Translation diff --git a/openapi.yaml b/openapi.yaml index 417b87b..e1c79a6 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -2022,12 +2022,16 @@ paths: application/json: schema: type: object - required: - - prompt + description: Provide either `prompt` for inline text or `prompt_url` for a previously uploaded prompt file. properties: prompt: type: string + description: Inline prompt text. Use for smaller prompts. example: "Summarize the content of this video" + prompt_url: + type: string + description: URL to a previously uploaded plain-text prompt file. Use for larger prompts. + example: "https://storage.googleapis.com/.../generate_text_prompt_123abc.txt" video_id: type: string example: "m-12345" @@ -2043,6 +2047,11 @@ paths: callback_url: type: string example: "https://webhook.example.com/callback" + oneOf: + - required: + - prompt + - required: + - prompt_url responses: '200': description: Text generation started or completed diff --git a/videodb/_upload.py b/videodb/_upload.py index 0291abe..0a25e5d 100644 --- a/videodb/_upload.py +++ b/videodb/_upload.py @@ -1,9 +1,9 @@ -import requests - -from typing import Optional +import os +from typing import Optional, Union from urllib.parse import urlparse + +import requests from requests import HTTPError -import os from videodb._constants import ( @@ -20,6 +20,31 @@ def _is_url(path: str) -> bool: return all([parsed.scheme in ("http", "https"), parsed.netloc]) +def upload_bytes( + _connection, + content: Union[str, bytes], + name: str, + content_type: str = "application/octet-stream", + collection_id: Optional[str] = None, +) -> str: + """Upload in-memory content using a presigned upload URL and return the object URL.""" + collection_id = collection_id or _connection.collection_id + upload_url_data = _connection.get( + path=f"{ApiPath.collection}/{collection_id}/{ApiPath.upload_url}", + params={"name": name}, + ) + upload_url = upload_url_data.get("upload_url") + + try: + files = {"file": (name, content, content_type)} + response = requests.post(upload_url, files=files) + response.raise_for_status() + except requests.exceptions.RequestException as e: + raise VideodbError("Error while uploading content", cause=e) + + return upload_url + + def upload( _connection, source: Optional[str] = None, @@ -71,16 +96,13 @@ def upload( if file_path: try: name = os.path.splitext(os.path.basename(file_path))[0] if not name else name - upload_url_data = _connection.get( - path=f"{ApiPath.collection}/{collection_id}/{ApiPath.upload_url}", - params={"name": name}, - ) - upload_url = upload_url_data.get("upload_url") with open(file_path, "rb") as file: - files = {"file": (name, file)} - response = requests.post(upload_url, files=files) - response.raise_for_status() - url = upload_url + url = upload_bytes( + _connection=_connection, + content=file, + name=name, + collection_id=collection_id, + ) except FileNotFoundError as e: raise VideodbError("File not found", cause=e) diff --git a/videodb/collection.py b/videodb/collection.py index 7f57fc3..5c7a0a3 100644 --- a/videodb/collection.py +++ b/videodb/collection.py @@ -1,8 +1,11 @@ +import json import logging +import uuid from typing import Optional, Union, List, Dict, Any, Literal from videodb._upload import ( upload, + upload_bytes, ) from videodb._constants import ( ApiPath, @@ -21,6 +24,8 @@ logger = logging.getLogger(__name__) +MAX_GENERATE_TEXT_PAYLOAD_SIZE = 250 * 1024 + class Collection: """Collection class to interact with the Collection. @@ -422,20 +427,40 @@ def generate_text( ) -> Union[str, dict]: """Generate text from a prompt using genai offering. + Small prompts are sent inline as JSON. When the serialized request body + approaches the observed gateway limit (~256 KB), the prompt is uploaded + via the collection presigned upload URL and referenced as ``prompt_url`` + to avoid API Gateway/Lambda payload size limits. + :param str prompt: Prompt for the text generation :param str model_name: Model name to use ("basic", "pro" or "ultra") :param str response_type: Desired response type ("text" or "json") :return: Generated text response :rtype: Union[str, dict] """ + payload = { + "prompt": prompt, + "model_name": model_name, + "response_type": response_type, + } - return self._connection.post( - path=f"{ApiPath.collection}/{self.id}/{ApiPath.generate}/{ApiPath.text}", - data={ - "prompt": prompt, + payload_size = len(json.dumps(payload).encode("utf-8")) + if payload_size > MAX_GENERATE_TEXT_PAYLOAD_SIZE: + payload = { + "prompt_url": upload_bytes( + _connection=self._connection, + content=prompt, + name=f"generate_text_prompt_{uuid.uuid4().hex}.txt", + content_type="text/plain; charset=utf-8", + collection_id=self.id, + ), "model_name": model_name, "response_type": response_type, - }, + } + + return self._connection.post( + path=f"{ApiPath.collection}/{self.id}/{ApiPath.generate}/{ApiPath.text}", + data=payload, ) def dub_video( diff --git a/videodb/editor.py b/videodb/editor.py index 1dec1e8..85f0981 100644 --- a/videodb/editor.py +++ b/videodb/editor.py @@ -1,12 +1,12 @@ import json import logging -import requests import warnings from typing import List, Optional, Union from enum import Enum from videodb._constants import ApiPath +from videodb._upload import upload_bytes from videodb._utils._video import build_iframe_embed_code from videodb.exceptions import InvalidRequestError @@ -1148,26 +1148,20 @@ def _upload_timeline_data(self, json_str: str) -> str: :rtype: str :raises InvalidRequestError: If upload fails """ - # Get a presigned upload URL - upload_url_data = self.connection.get( - path=f"{ApiPath.collection}/{self.connection.collection_id}/{ApiPath.upload_url}", - params={"name": "timeline_data.json"}, - ) - upload_url = upload_url_data.get("upload_url") - - # Upload the JSON data as a file try: - files = {"file": ("timeline_data.json", json_str, "application/json")} - response = requests.post(upload_url, files=files) - response.raise_for_status() - except requests.exceptions.RequestException as e: + return upload_bytes( + _connection=self.connection, + content=json_str, + name="timeline_data.json", + content_type="application/json", + collection_id=self.connection.collection_id, + ) + except Exception as e: raise InvalidRequestError( f"Failed to upload timeline data: {str(e)}", getattr(e, "response", None), ) from None - return upload_url - def download_stream(self, stream_url: str) -> dict: """Download a stream from the timeline. From e20831ed07eb96281f554a1f92fa08ff0e75c864 Mon Sep 17 00:00:00 2001 From: Ankit raj <113342181+ankit-v2-3@users.noreply.github.com> Date: Fri, 17 Apr 2026 17:05:31 +0530 Subject: [PATCH 07/18] feat: make text gen. async --- videodb/_constants.py | 2 ++ videodb/_utils/_http_client.py | 10 +++++++--- videodb/client.py | 9 +++++++++ videodb/collection.py | 9 ++++++++- 4 files changed, 26 insertions(+), 4 deletions(-) diff --git a/videodb/_constants.py b/videodb/_constants.py index 99db326..9e35a39 100644 --- a/videodb/_constants.py +++ b/videodb/_constants.py @@ -123,11 +123,13 @@ class ApiPath: identities = "identities" merge = "merge" split = "split" + async_response = "async-response" class Status: processing = "processing" in_progress = "in progress" + complete = "complete" class MeetingStatus: diff --git a/videodb/_utils/_http_client.py b/videodb/_utils/_http_client.py index ecab5de..32fbba8 100644 --- a/videodb/_utils/_http_client.py +++ b/videodb/_utils/_http_client.py @@ -95,7 +95,10 @@ def _make_request( response = method(url, headers=request_headers, timeout=timeout, **kwargs) response.raise_for_status() if not wait: - return response.json().get("data") + data = response.json().get("data") + if data is not None: + return data + return response.json() return self._parse_response(response) except requests.exceptions.RequestException as e: @@ -185,6 +188,7 @@ def _parse_response(self, response: requests.Response): response_json.get("status") == Status.processing and response_json.get("request_type", "sync") == "sync" ): + print("inside loop") if self.show_progress: self.progress_bar = tqdm( total=100, @@ -233,12 +237,12 @@ def _apply_poll_overrides(self, kwargs): ) def get( - self, path: str, show_progress: Optional[bool] = False, **kwargs + self, path: str, show_progress: Optional[bool] = False, wait: bool = True, **kwargs ) -> requests.Response: """Make a get request""" self.show_progress = show_progress self._apply_poll_overrides(kwargs) - return self._make_request(method=self.session.get, path=path, **kwargs) + return self._make_request(method=self.session.get, path=path, wait=wait, **kwargs) def post( self, path: str, data=None, show_progress: Optional[bool] = False, diff --git a/videodb/client.py b/videodb/client.py index 7eac7bb..81bfa96 100644 --- a/videodb/client.py +++ b/videodb/client.py @@ -494,3 +494,12 @@ def generate_client_token(self, expires_in: int = 86400) -> str: data={"expires_in": expires_in}, ) return response.get("token") + + def get_async_response(self, id: str) -> dict: + """Get the details of an async response. + + :param str id: ID of the async response + :return: Details of the async response + :rtype: dict + """ + return self.get(path=f"{ApiPath.async_response}/{id}", wait=False) diff --git a/videodb/collection.py b/videodb/collection.py index 73e06e9..07f94ba 100644 --- a/videodb/collection.py +++ b/videodb/collection.py @@ -425,6 +425,8 @@ def generate_text( prompt: str, model_name: Literal["basic", "pro", "ultra"] = "basic", response_type: Literal["text", "json"] = "text", + wait: bool = True, + callback_url: Optional[str] = None, ) -> Union[str, dict]: """Generate text from a prompt using genai offering. @@ -436,13 +438,16 @@ def generate_text( :param str prompt: Prompt for the text generation :param str model_name: Model name to use ("basic", "pro" or "ultra") :param str response_type: Desired response type ("text" or "json") - :return: Generated text response + :param bool wait: Wait for the text generation to complete (default: True) + :param str callback_url: URL to receive the callback (optional) + :return: Generated text response if wait is False, otherwise job id of the text generation :rtype: Union[str, dict] """ payload = { "prompt": prompt, "model_name": model_name, "response_type": response_type, + "callback_url": callback_url, } payload_size = len(json.dumps(payload).encode("utf-8")) @@ -457,11 +462,13 @@ def generate_text( ), "model_name": model_name, "response_type": response_type, + "callback_url": callback_url, } return self._connection.post( path=f"{ApiPath.collection}/{self.id}/{ApiPath.generate}/{ApiPath.text}", data=payload, + wait=wait, ) def dub_video( From a1967b97cc60013fd324ab145a884e3a8972cf7c Mon Sep 17 00:00:00 2001 From: Ankit raj <113342181+ankit-v2-3@users.noreply.github.com> Date: Fri, 17 Apr 2026 18:10:28 +0530 Subject: [PATCH 08/18] fix: remove print --- videodb/_utils/_http_client.py | 1 - 1 file changed, 1 deletion(-) diff --git a/videodb/_utils/_http_client.py b/videodb/_utils/_http_client.py index 32fbba8..6507e38 100644 --- a/videodb/_utils/_http_client.py +++ b/videodb/_utils/_http_client.py @@ -188,7 +188,6 @@ def _parse_response(self, response: requests.Response): response_json.get("status") == Status.processing and response_json.get("request_type", "sync") == "sync" ): - print("inside loop") if self.show_progress: self.progress_bar = tqdm( total=100, From 40f61305119cf95625acf2f85963597b9acbfe88 Mon Sep 17 00:00:00 2001 From: Rohit Garg Date: Tue, 5 May 2026 19:28:25 +0530 Subject: [PATCH 09/18] feat: add async generation job support --- openapi.yaml | 63 ++++++++++++++++++++++-- videodb/__init__.py | 2 + videodb/_constants.py | 1 + videodb/client.py | 60 +++++++++++++++++++++++ videodb/collection.py | 70 ++++++++++++++++++++------ videodb/job.py | 111 ++++++++++++++++++++++++++++++++++++++++++ 6 files changed, 288 insertions(+), 19 deletions(-) create mode 100644 videodb/job.py diff --git a/openapi.yaml b/openapi.yaml index 417b87b..7d50656 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -1915,6 +1915,14 @@ paths: aspect_ratio: type: string example: "16:9" + model_name: + type: string + description: Use `flux` for FLUX self-inference image generation. + example: "flux" + config: + type: object + description: Model configuration. For FLUX supports size, num_inference_steps, guidance_scale, negative_prompt, and seed. + additionalProperties: true callback_url: type: string example: "https://webhook.example.com/callback" @@ -1983,16 +1991,32 @@ paths: schema: type: object required: - - prompt - audio_type properties: prompt: type: string + description: Prompt for music or sound effect generation. example: "Generate upbeat background music" + text: + type: string + description: Text to convert to speech when `audio_type` is `voice`. + example: "Hello, welcome to VideoDB." audio_type: type: string - enum: [speech, sound_effect, music] - example: "music" + enum: [voice, sound_effect, music] + example: "voice" + voice_name: + type: string + description: Voice name for hosted text-to-speech. + example: "Default" + model_name: + type: string + description: Use `omnivoice` for OmniVoice self-inference text-to-speech. + example: "omnivoice" + config: + type: object + description: Voice configuration. For OmniVoice supports instructions, ref_audio, ref_text, response_format, speed, language, and max_new_tokens. + additionalProperties: true callback_url: type: string example: "https://webhook.example.com/callback" @@ -2004,6 +2028,39 @@ paths: schema: $ref: '#/components/schemas/AsyncResponse' + /job/{job_id}: + get: + summary: Get self-inference generation job status + security: + - ApiKeyAuth: [] + parameters: + - name: job_id + in: path + required: true + schema: + type: string + example: "550e8400-e29b-41d4-a716-446655440000" + responses: + '200': + description: Generation job status or final generated asset + content: + application/json: + schema: + type: object + properties: + success: + type: boolean + status: + type: string + enum: [processing, done, failed] + data: + type: object + additionalProperties: true + message: + type: string + '404': + description: Job not found + /collection/{collection_id}/generate/text/: post: summary: Generate text using AI diff --git a/videodb/__init__.py b/videodb/__init__.py index 59d2a1b..039c6e9 100644 --- a/videodb/__init__.py +++ b/videodb/__init__.py @@ -25,6 +25,7 @@ RTStreamChannelType, ) from videodb.client import Connection +from videodb.job import GenerationJob from videodb.capture_session import CaptureSession from videodb.websocket_client import WebSocketConnection from videodb.capture import CaptureClient, Channel, AudioChannel, VideoChannel, Channels, ChannelList @@ -42,6 +43,7 @@ __all__ = [ "connect", "CaptureSession", + "GenerationJob", "WebSocketConnection", "CaptureClient", "Channel", diff --git a/videodb/_constants.py b/videodb/_constants.py index f7de578..ba36148 100644 --- a/videodb/_constants.py +++ b/videodb/_constants.py @@ -116,6 +116,7 @@ class ApiPath: token = "token" websocket = "websocket" export = "export" + job = "job" class Status: diff --git a/videodb/client.py b/videodb/client.py index 7eac7bb..8011f75 100644 --- a/videodb/client.py +++ b/videodb/client.py @@ -1,4 +1,5 @@ import logging +import requests from typing import ( Optional, @@ -11,6 +12,7 @@ TranscodeMode, VideoConfig, AudioConfig, + HttpClientDefaultValues, ) from videodb.collection import Collection @@ -264,6 +266,64 @@ def get_transcode_details(self, job_id: str) -> dict: """ return self.get(path=f"{ApiPath.transcode}/{job_id}") + def get_job_status(self, job_id: str) -> dict: + """Get status/details for a self-inference generation job. + + This preserves and normalizes the job response wrapper instead of using + the standard SDK response parser, because callers need the top-level + job status while polling. + + :param str job_id: ID of the generation job + :return: Normalized job response with success, status, data, message + :rtype: dict + """ + try: + url = f"{self.base_url}/{ApiPath.job}/{job_id}" + response = self.session.get(url, timeout=HttpClientDefaultValues.timeout) + response.raise_for_status() + response_json = response.json() + except requests.exceptions.RequestException as e: + self._handle_request_error(e) + except ValueError: + from videodb.exceptions import InvalidRequestError + + raise InvalidRequestError("Invalid request: Unable to parse job response") from None + + data = response_json.get("data") or {} + status = response_json.get("status") or data.get("status") + success = response_json.get("success", False) + + if not status: + status = "done" if success else "failed" + + return { + "success": success, + "status": status, + "data": data, + "message": response_json.get("message"), + } + + def wait_for_job( + self, + job_id: str, + timeout: int = 600, + interval: int = 5, + result_type: str = None, + ): + """Poll a self-inference generation job until completion. + + :param str job_id: ID of the generation job + :param int timeout: Maximum seconds to wait + :param int interval: Seconds between polls + :param str result_type: Optional expected result type, "audio" or "image" + :return: Generated SDK asset or final job data + """ + from videodb.job import GenerationJob + + return GenerationJob( + self, job_id=job_id, result_type=result_type + ).wait(timeout=timeout, interval=interval) + def upload( self, source: Optional[str] = None, diff --git a/videodb/collection.py b/videodb/collection.py index 7f57fc3..86cfaf0 100644 --- a/videodb/collection.py +++ b/videodb/collection.py @@ -14,6 +14,7 @@ from videodb.video import Video from videodb.audio import Audio from videodb.image import Image +from videodb.job import GenerationJob from videodb.meeting import Meeting from videodb.capture_session import CaptureSession from videodb.rtstream import RTStream, RTStreamSearchResult, RTStreamShot @@ -274,25 +275,47 @@ def generate_image( prompt: str, aspect_ratio: Optional[Literal["1:1", "9:16", "16:9", "4:3", "3:4"]] = "1:1", callback_url: Optional[str] = None, - ) -> Image: + model_name: Optional[str] = None, + config: Optional[dict] = None, + wait: bool = False, + poll_interval: int = 5, + timeout: int = 600, + ) -> Union[Image, GenerationJob]: """Generate an image from a prompt. :param str prompt: Prompt for the image generation - :param str aspect_ratio: Aspect ratio of the image (optional) + :param str aspect_ratio: Aspect ratio of the image (optional, hosted models) :param str callback_url: URL to receive the callback (optional) - :return: :class:`Image ` object - :rtype: :class:`videodb.image.Image` + :param str model_name: Model name. Use ``"flux"`` for FLUX self-inference. + :param dict config: Model configuration. Used by FLUX. + :param bool wait: If True, wait for self-inference jobs and return Image. + :param int poll_interval: Seconds between job polls when wait=True. + :param int timeout: Maximum seconds to wait when wait=True. + :return: :class:`Image ` or :class:`GenerationJob ` + :rtype: Union[:class:`videodb.image.Image`, :class:`videodb.job.GenerationJob`] """ + payload = { + "prompt": prompt, + "aspect_ratio": aspect_ratio, + "callback_url": callback_url, + } + if model_name: + payload["model_name"] = model_name + if config is not None: + payload["config"] = config + image_data = self._connection.post( path=f"{ApiPath.collection}/{self.id}/{ApiPath.generate}/{ApiPath.image}", - data={ - "prompt": prompt, - "aspect_ratio": aspect_ratio, - "callback_url": callback_url, - }, + data=payload, ) - if image_data: - return Image(self._connection, **image_data) + if not image_data: + return None + if image_data.get("job_id"): + job = GenerationJob.from_data( + self._connection, image_data, result_type="image" + ) + return job.wait(timeout=timeout, interval=poll_interval) if wait else job + return Image(self._connection, **image_data) def generate_music( self, prompt: str, duration: int = 5, callback_url: Optional[str] = None @@ -352,15 +375,23 @@ def generate_voice( voice_name: str = "Default", config: dict = {}, callback_url: Optional[str] = None, - ) -> Audio: + model_name: str = "elevenlabs", + wait: bool = False, + poll_interval: int = 5, + timeout: int = 600, + ) -> Union[Audio, GenerationJob]: """Generate voice from text. :param str text: Text to convert to voice :param str voice_name: Name of the voice to use :param dict config: Configuration for the voice generation :param str callback_url: URL to receive the callback (optional) - :return: :class:`Audio