Added new response type

Author: Vaghinak Basentsyan

src/superannotate/lib/app/interface/cli_interface.py

@@ -10,7 +10,6 @@
 import lib.core as constances
 import pandas as pd
 from lib import __file__ as lib_path
-from lib.app.helpers import get_annotation_paths
 from lib.app.helpers import split_project_path
 from lib.app.input_converters.conversion import import_annotation
 from lib.app.interface.base_interface import BaseInterfaceFacade
@@ -19,14 +18,14 @@
 from lib.app.interface.sdk_interface import attach_video_urls_to_project
 from lib.app.interface.sdk_interface import create_folder
 from lib.app.interface.sdk_interface import create_project
+from lib.app.interface.sdk_interface import upload_annotations_from_folder_to_project
 from lib.app.interface.sdk_interface import upload_images_from_folder_to_project
+from lib.app.interface.sdk_interface import upload_preannotations_from_folder_to_project
 from lib.app.interface.sdk_interface import upload_videos_from_folder_to_project
 from lib.app.serializers import ImageSerializer
 from lib.core.entities import ConfigEntity
 from lib.infrastructure.controller import Controller
 from lib.infrastructure.repositories import ConfigRepository
-from tqdm import tqdm
-
 
 logger = logging.getLogger()
 controller = Controller(logger)
@@ -107,13 +106,11 @@ def upload_images(
         """
         if not isinstance(extensions, list):
             extensions = extensions.split(",")
-
         upload_images_from_folder_to_project(
             project,
             folder_path=folder,
             extensions=extensions,
             annotation_status=set_annotation_status,
-            from_s3_bucket=None,
             exclude_file_patterns=exclude_file_patterns,
             recursive_subfolders=recursive_subfolders,
             image_quality_in_editor=image_quality_in_editor,
@@ -202,7 +199,6 @@ def _upload_annotations(
             data_set_name = ""
         if not task:
             task = "object_detection"
-
         annotations_path = folder
         with tempfile.TemporaryDirectory() as temp_dir:
             if format != "SuperAnnotate":
@@ -222,22 +218,14 @@ def _upload_annotations(
                 project_name=project_name,
                 annotation_classes=json.load(open(classes_path)),
             )
-            annotation_paths = get_annotation_paths(annotations_path)
-            chunk_size = 10
-            with tqdm(total=len(annotation_paths)) as progress_bar:
-                for i in range(0, len(annotation_paths), chunk_size):
-                    response = self.controller.upload_annotations_from_folder(
-                        project_name=project["project"].name,
-                        folder_name=folder_name,
-                        folder_path=annotations_path,
-                        annotation_paths=annotation_paths[
-                            i : i + chunk_size  # noqa: E203
-                        ],
-                        is_pre_annotations=pre,
-                    )
-                    if response.errors:
-                        logger.warning(response.errors)
-                    progress_bar.update(chunk_size)
+            if pre:
+                upload_preannotations_from_folder_to_project(
+                    project_name, annotations_path
+                )
+            else:
+                upload_annotations_from_folder_to_project(
+                    project_name, annotations_path
+                )
         sys.exit(0)
 
     def attach_image_urls(

src/superannotate/lib/app/interface/sdk_interface.py

@@ -13,6 +13,7 @@
 from typing import Iterable
 from typing import List
 from typing import Optional
+from typing import Tuple
 from typing import Union
 
 import boto3
@@ -1338,7 +1339,9 @@ def get_image_annotations(project: Union[str, dict], image_name: str):
 def upload_images_from_folder_to_project(
     project: Union[str, dict],
     folder_path: Union[str, Path],
-    extensions: Optional[Iterable[str]] = constances.DEFAULT_IMAGE_EXTENSIONS,
+    extensions: Optional[
+        Union[List[str], Tuple[str]]
+    ] = constances.DEFAULT_IMAGE_EXTENSIONS,
     annotation_status="NotStarted",
     from_s3_bucket=None,
     exclude_file_patterns: Optional[
@@ -1382,8 +1385,8 @@ def upload_images_from_folder_to_project(
         logger.info(
             "When using recursive subfolder parsing same name images in different subfolders will overwrite each other."
         )
-
     if not isinstance(extensions, (list, tuple)):
+        print(extensions)
         raise AppException(
             "extensions should be a list or a tuple in upload_images_from_folder_to_project"
         )
@@ -1660,7 +1663,9 @@ def prepare_export(
 def upload_videos_from_folder_to_project(
     project: Union[str, dict],
     folder_path: Union[str, Path],
-    extensions: Optional[Iterable[str]] = constances.DEFAULT_VIDEO_EXTENSIONS,
+    extensions: Optional[
+        Union[Tuple[str], List[str]]
+    ] = constances.DEFAULT_VIDEO_EXTENSIONS,
     exclude_file_patterns: Optional[Iterable[str]] = (),
     recursive_subfolders: Optional[StrictBool] = False,
     target_fps: Optional[int] = None,

src/superannotate/lib/core/__init__.py

@@ -55,6 +55,8 @@
 TOKEN_UUID = "token"
 
 
+DEPRECATED_VIDEO_PROJECTS_MESSAGE = "The function does not support projects containing videos attached with URLs"
+
 __version__ = "?"
 
 __alL__ = (

src/superannotate/lib/core/service_types.py

@@ -0,0 +1,74 @@
+from typing import Dict
+from typing import List
+from typing import Optional
+from typing import Union
+
+from lib.core.exceptions import AppException
+from pydantic import BaseModel
+from pydantic import Extra
+
+
+class UserLimits(BaseModel):
+    super_user_limit: int
+    project_limit: int
+    folder_limit: int
+
+    def has_enough_slots(self, count: int):
+        if count > self.super_user_limit:
+            raise AppException(
+                "The number of items you want to upload exceeds the limit of your subscription plan."
+            )
+        if count > self.project_limit:
+            raise AppException(
+                "You have exceeded the limit of 500 000 items per project."
+            )
+        if count > self.folder_limit:
+            raise AppException(
+                "“You have exceeded the limit of 50 000 items per folder."
+            )
+        return True
+
+
+class UploadAnnotationAuthData(BaseModel):
+    access_key: str
+    secret_key: str
+    session_token: str
+    region: str
+    bucket: str
+    images: Dict[int, dict]
+
+    class Config:
+        extra = Extra.allow
+        fields = {
+            "access_key": "accessKeyId",
+            "secret_key": "secretAccessKey",
+            "session_token": "sessionToken",
+            "region": "region",
+        }
+
+    def __init__(self, **data):
+        credentials = data["creds"]
+        data.update(credentials)
+        del data["creds"]
+        super().__init__(**data)
+
+
+class ServiceResponse(BaseModel):
+    status: int
+    reason: str
+    content: Union[bytes, str]
+    data: Optional[Union[UserLimits, UploadAnnotationAuthData]]
+
+    def __init__(self, response, content_type):
+        data = {
+            "status": response.status_code,
+            "reason": response.reason,
+            "content": response.content,
+        }
+        if response.ok:
+            data["data"] = content_type(**response.json())
+        super().__init__(**data)
+
+    @property
+    def ok(self):
+        return 199 < self.status < 300

src/superannotate/lib/core/serviceproviders.py

@@ -5,6 +5,8 @@
 from typing import List
 from typing import Tuple
 
+from lib.core.service_types import ServiceResponse
+
 
 class SingleInstanceMetaClass(type):
     _instances = {}
@@ -255,7 +257,7 @@ def get_pre_annotation_upload_data(
 
     def get_annotation_upload_data(
         self, project_id: int, team_id: int, image_ids: List[int], folder_id: int
-    ):
+    ) -> ServiceResponse:
         raise NotImplementedError
 
     def get_templates(self, team_id: int):
@@ -310,3 +312,8 @@ def get_annotations_delete_progress(
         self, team_id: int, project_id: int, poll_id: int
     ):
         raise NotImplementedError
+
+    def get_limits(
+        self, team_id: int, project_id: int, folder_id: int = None
+    ) -> ServiceResponse:
+        raise NotImplementedError