feat: add image and ImageAsset

Author: ankit-v2-3

videodb/_constants.py

@@ -7,6 +7,7 @@
 class MediaType:
     video = "video"
     audio = "audio"
+    image = "image"
 
 
 class SearchType:
@@ -31,6 +32,7 @@ class ApiPath:
     upload = "upload"
     video = "video"
     audio = "audio"
+    image = "image"
     stream = "stream"
     thumbnail = "thumbnail"
     upload_url = "upload_url"

videodb/asset.py

@@ -85,3 +85,35 @@ def __repr__(self) -> str:
             f"fade_in_duration={self.fade_in_duration}, "
             f"fade_out_duration={self.fade_out_duration})"
         )
+
+
+class ImageAsset(MediaAsset):
+    def __init__(
+        self,
+        asset_id: str,
+        width: Optional[int] = 100,
+        height: Optional[int] = 100,
+        position_x: Optional[int] = 80,
+        position_y: Optional[int] = 20,
+        end: Optional[Union[int, None]] = None,
+    ) -> None:
+        super().__init__(asset_id)
+        self.width = width
+        self.height = height
+        self.position_x = position_x
+        self.position_y = position_y
+        self.end = end
+
+    def to_json(self) -> dict:
+        return copy.deepcopy(self.__dict__)
+
+    def __repr__(self) -> str:
+        return (
+            f"ImageAsset("
+            f"asset_id={self.asset_id}, "
+            f"width={self.width}, "
+            f"height={self.height}, "
+            f"position_x={self.position_x}, "
+            f"position_y={self.position_y}, "
+            f"end={self.end})"
+        )

videodb/client.py

@@ -13,6 +13,7 @@
 from videodb._utils._http_client import HttpClient
 from videodb.video import Video
 from videodb.audio import Audio
+from videodb.image import Image
 
 from videodb._upload import (
     upload,
@@ -46,7 +47,7 @@ def upload(
         name: Optional[str] = None,
         description: Optional[str] = None,
         callback_url: Optional[str] = None,
-    ) -> Union[Video, Audio, None]:
+    ) -> Union[Video, Audio, Image, None]:
         upload_data = upload(
             self,
             file_path,
@@ -60,3 +61,5 @@ def upload(
             return Video(self, **upload_data) if upload_data else None
         elif upload_data.get("id").startswith("a-"):
             return Audio(self, **upload_data) if upload_data else None
+        elif upload_data.get("id").startswith("i-"):
+            return Image(self, **upload_data) if upload_data else None

videodb/collection.py

@@ -13,6 +13,7 @@
 )
 from videodb.video import Video
 from videodb.audio import Audio
+from videodb.image import Image
 from videodb.search import SearchFactory, SearchResult
 
 logger = logging.getLogger(__name__)
@@ -54,6 +55,17 @@ def get_audio(self, audio_id: str) -> Audio:
     def delete_audio(self, audio_id: str) -> None:
         return self._connection.delete(path=f"{ApiPath.audio}/{audio_id}")
 
+    def get_images(self) -> list[Image]:
+        images_data = self._connection.get(path=f"{ApiPath.image}")
+        return [Image(self._connection, **image) for image in images_data.get("images")]
+
+    def get_image(self, image_id: str) -> Image:
+        image_data = self._connection.get(path=f"{ApiPath.image}/{image_id}")
+        return Image(self._connection, **image_data)
+
+    def delete_image(self, image_id: str) -> None:
+        return self._connection.delete(path=f"{ApiPath.image}/{image_id}")
+
     def search(
         self,
         query: str,
@@ -79,7 +91,7 @@ def upload(
         name: Optional[str] = None,
         description: Optional[str] = None,
         callback_url: Optional[str] = None,
-    ) -> Union[Video, Audio, None]:
+    ) -> Union[Video, Audio, Image, None]:
         upload_data = upload(
             self._connection,
             file_path,
@@ -93,3 +105,5 @@ def upload(
             return Video(self._connection, **upload_data) if upload_data else None
         elif upload_data.get("id").startswith("a-"):
             return Audio(self._connection, **upload_data) if upload_data else None
+        elif upload_data.get("id").startswith("i-"):
+            return Image(self._connection, **upload_data) if upload_data else None

videodb/image.py

@@ -0,0 +1,22 @@
+from videodb._constants import (
+    ApiPath,
+)
+
+
+class Image:
+    def __init__(self, _connection, id: str, collection_id: str, **kwargs) -> None:
+        self._connection = _connection
+        self.id = id
+        self.collection_id = collection_id
+        self.name = kwargs.get("name", None)
+
+    def __repr__(self) -> str:
+        return (
+            f"Image("
+            f"id={self.id}, "
+            f"collection_id={self.collection_id}, "
+            f"name={self.name})"
+        )
+
+    def delete(self) -> None:
+        self._connection.delete(f"{ApiPath.image}/{self.id}")

videodb/timeline.py

@@ -1,7 +1,7 @@
 from typing import Union
 
 from videodb._constants import ApiPath
-from videodb.asset import VideoAsset, AudioAsset
+from videodb.asset import VideoAsset, AudioAsset, ImageAsset
 
 
 class Timeline(object):
@@ -28,9 +28,9 @@ def add_inline(self, asset: Union[VideoAsset]) -> None:
             raise ValueError("asset must be of type VideoAsset")
         self._timeline.append(asset)
 
-    def add_overlay(self, start: int, asset: Union[AudioAsset]) -> None:
-        if not isinstance(asset, AudioAsset):
-            raise ValueError("asset must be of type AudioAsset")
+    def add_overlay(self, start: int, asset: Union[AudioAsset, ImageAsset]) -> None:
+        if not isinstance(asset, AudioAsset) and not isinstance(asset, ImageAsset):
+            raise ValueError("asset must be of type AudioAsset or ImageAsset")
         self._timeline.append((start, asset))
 
     def generate_stream(self) -> str: