Merge pull request #785 from superannotateai/FRIDAY_3640

Add generate_items method

Author: VaghinakDev

docs/source/api_reference/api_item.rst

@@ -8,6 +8,7 @@ Items
 .. automethod:: superannotate.SAClient.list_items
 .. automethod:: superannotate.SAClient.search_items
 .. automethod:: superannotate.SAClient.attach_items
+.. automethod:: superannotate.SAClient.generate_items
 .. automethod:: superannotate.SAClient.item_context
 .. autoclass:: superannotate.ItemContext
     :members: get_metadata, get_component_value, set_component_value

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

@@ -354,14 +354,21 @@ def get_item_by_id(self, project_id: int, item_id: int):
 
         return BaseSerializer(item).serialize(exclude={"url", "meta"})
 
-    def get_team_metadata(self):
-        """Returns team metadata
+    def get_team_metadata(self, include: List[Literal["scores"]] = None):
+        """
+        Returns team metadata, including optionally, scores
 
         :return: team metadata
         :rtype: dict
         """
         response = self.controller.get_team()
-        return TeamSerializer(response.data).serialize()
+        team = response.data
+        if include and "scores" in include:
+            team.scores = [
+                i.name
+                for i in self.controller.work_management.list_score_templates().data
+            ]
+        return TeamSerializer(team).serialize(exclude_unset=True)
 
     def get_user_metadata(
         self, pk: Union[int, str], include: List[Literal["custom_fields"]] = None
@@ -4014,6 +4021,35 @@ def attach_items(
             ]
         return uploaded, fails, duplicated
 
+    def generate_items(
+        self,
+        project: Union[NotEmptyStr, Tuple[int, int], Tuple[str, str]],
+        count: int,
+        name: str,
+    ):
+        """
+        Generate multiple items in a specific project and folder.`
+        If there are no items in the folder, it will generate a blank item otherwise, it will generate items based on the Custom Form.
+
+        :param project: Project and folder as a tuple, folder is optional.
+        :type project: Union[str, Tuple[int, int], Tuple[str, str]]
+
+        :param count: the count of items to generate
+        :type count: int
+
+        :param name: the name of the item. After generating the items,
+                     the item names will contain the provided name and a numeric suffix based on the item count.
+        :type name: str
+        """
+        project, folder = self.controller.get_project_folder(project)
+
+        response = self.controller.items.generate_items(
+            project=project, folder=folder, count=count, name=name
+        )
+        if response.errors:
+            raise AppException(response.errors)
+        logger.info(f"{response.data} items successfully generated.")
+
     def copy_items(
         self,
         source: Union[NotEmptyStr, dict],

src/superannotate/lib/core/entities/project.py

@@ -180,6 +180,7 @@ class TeamEntity(BaseModel):
     pending_invitations: Optional[List[Any]]
     creator_id: Optional[str]
     owner_id: Optional[str]
+    scores: Optional[List[str]]
 
     class Config:
         extra = Extra.ignore

src/superannotate/lib/core/serviceproviders.py

@@ -399,9 +399,9 @@ def attach(
         project: entities.ProjectEntity,
         folder: entities.FolderEntity,
         attachments: List[Attachment],
-        annotation_status_code,
         upload_state_code,
-        meta: Dict[str, AttachmentMeta],
+        annotation_status_code=None,
+        meta: Dict[str, AttachmentMeta] = None,
     ) -> ServiceResponse:
         raise NotImplementedError
 

src/superannotate/lib/core/usecases/items.py

@@ -1,9 +1,11 @@
 import logging
+import re
 import traceback
 from collections import defaultdict
 from concurrent.futures import as_completed
 from concurrent.futures import ThreadPoolExecutor
 from typing import Dict
+from typing import Generator
 from typing import List
 from typing import Union
 
@@ -386,6 +388,88 @@ def execute(self) -> Response:
         return self._response
 
 
+class GenerateItems(BaseReportableUseCase):
+    CHUNK_SIZE = 500
+    INVALID_CHARS_PATTERN = re.compile(r"[<>:\"'/\\|?*&$!+]")
+
+    def __init__(
+        self,
+        reporter: Reporter,
+        project: ProjectEntity,
+        folder: FolderEntity,
+        name_prefix: str,
+        count: int,
+        service_provider: BaseServiceProvider,
+    ):
+        super().__init__(reporter)
+        self._project = project
+        self._folder = folder
+        self._name_prefix = name_prefix
+        self._count = count
+        self._service_provider = service_provider
+
+    def validate_name(self):
+        if (
+            len(self._name_prefix) > 114
+            or self.INVALID_CHARS_PATTERN.search(self._name_prefix) is not None
+        ):
+            raise AppException("Invalid item name.")
+
+    def validate_limitations(self):
+        response = self._service_provider.get_limitations(
+            project=self._project, folder=self._folder
+        )
+        if not response.ok:
+            raise AppValidationException(response.error)
+        if self._count > response.data.folder_limit.remaining_image_count:
+            raise AppValidationException(constants.ATTACH_FOLDER_LIMIT_ERROR_MESSAGE)
+        if self._count > response.data.project_limit.remaining_image_count:
+            raise AppValidationException(constants.ATTACH_PROJECT_LIMIT_ERROR_MESSAGE)
+        if (
+            response.data.user_limit
+            and self._count > response.data.user_limit.remaining_image_count
+        ):
+            raise AppValidationException(constants.ATTACH_USER_LIMIT_ERROR_MESSAGE)
+
+    def validate_project_type(self):
+        if self._project.type != constants.ProjectType.MULTIMODAL:
+            raise AppException(
+                "This function is only supported for Multimodal projects."
+            )
+
+    @staticmethod
+    def generate_attachments(
+        name: str, start: int, end: int, chunk_size: int
+    ) -> Generator[List[Attachment], None, None]:
+        chunk = []
+        for i in range(start, end + 1):
+            chunk.append(Attachment(name=f"{name}_{i:05d}", path="custom_llm"))
+            if len(chunk) == chunk_size:
+                yield chunk
+                chunk = []
+        if chunk:
+            yield chunk
+
+    def execute(self) -> Response:
+        if self.is_valid():
+            attached_items_count = 0
+            for chunk in self.generate_attachments(
+                self._name_prefix, start=1, end=self._count, chunk_size=self.CHUNK_SIZE
+            ):
+                backend_response = self._service_provider.items.attach(
+                    project=self._project,
+                    folder=self._folder,
+                    attachments=chunk,
+                    upload_state_code=3,
+                )
+                if not backend_response.ok:
+                    self._response.errors = AppException(backend_response.error)
+                    return self._response
+                attached_items_count += len(chunk)
+            self._response.data = attached_items_count
+        return self._response
+
+
 class CopyItems(BaseReportableUseCase):
     """
     Copy items in bulk between folders in a project.

src/superannotate/lib/infrastructure/controller.py

@@ -127,6 +127,9 @@ def __init__(self, service_provider: ServiceProvider):
 
 
 class WorkManagementManager(BaseManager):
+    def list_score_templates(self):
+        return self.service_provider.work_management.list_scores()
+
     def get_user_metadata(
         self, pk: Union[str, int], include: List[Literal["custom_fields"]] = None
     ):
@@ -914,6 +917,23 @@ def attach(
         )
         return use_case.execute()
 
+    def generate_items(
+        self,
+        project: ProjectEntity,
+        folder: FolderEntity,
+        count: int,
+        name: str,
+    ):
+        use_case = usecases.GenerateItems(
+            reporter=Reporter(),
+            project=project,
+            folder=folder,
+            name_prefix=name,
+            count=count,
+            service_provider=self.service_provider,
+        )
+        return use_case.execute()
+
     def delete(
         self,
         project: ProjectEntity,

src/superannotate/lib/infrastructure/services/item.py

@@ -37,7 +37,7 @@ def attach(
         folder: entities.FolderEntity,
         attachments: List[Attachment],
         upload_state_code,
-        meta: Dict[str, AttachmentMeta],
+        meta: Dict[str, AttachmentMeta] = None,
         annotation_status_code=None,
     ):
         data = {
@@ -46,8 +46,10 @@ def attach(
             "team_id": project.team_id,
             "images": [i.dict() for i in attachments],
             "upload_state": upload_state_code,
-            "meta": meta,
+            "meta": {},
         }
+        if meta:
+            data["meta"] = meta
         if annotation_status_code:
             data["annotation_status"] = annotation_status_code
         return self.client.request(self.URL_ATTACH, "post", data=data)

tests/integration/items/test_generate_items.py

@@ -0,0 +1,57 @@
+from src.superannotate import AppException
+from src.superannotate import SAClient
+from tests.integration.base import BaseTestCase
+
+sa = SAClient()
+
+
+class TestGenerateItemsMM(BaseTestCase):
+    PROJECT_NAME = "TestGenerateItemsMM"
+    PROJECT_DESCRIPTION = "TestGenerateItemsMM"
+    PROJECT_TYPE = "Multimodal"
+    FOLDER_NAME = "test_folder"
+
+    def test_generate_items_root(self):
+        sa.generate_items(self.PROJECT_NAME, 100, name="a")
+        items = sa.list_items(self.PROJECT_NAME)
+
+        assert len(items) == 100
+
+        expected_names = {f"a_{i:05d}" for i in range(1, 101)}
+        actual_names = {item["name"] for item in items}
+
+        assert actual_names == expected_names
+
+    def test_generate_items_in_folder(self):
+        path = f"{self.PROJECT_NAME}/{self.FOLDER_NAME}"
+
+        sa.create_folder(self.PROJECT_NAME, self.FOLDER_NAME)
+        sa.generate_items(path, 100, name="a")
+        items = sa.list_items(project=self.PROJECT_NAME, folder=self.FOLDER_NAME)
+
+        assert len(items) == 100
+
+        expected_names = {f"a_{i:05d}" for i in range(1, 101)}
+        actual_names = {item["name"] for item in items}
+
+        assert actual_names == expected_names
+
+    def test_invalid_name(self):
+        with self.assertRaisesRegexp(
+            AppException,
+            "Invalid item name.",
+        ):
+            sa.generate_items(self.PROJECT_NAME, 100, name="a" * 115)
+
+        with self.assertRaisesRegexp(
+            AppException,
+            "Invalid item name.",
+        ):
+            sa.generate_items(self.PROJECT_NAME, 100, name="m<:")
+
+    def test_item_count(self):
+        with self.assertRaisesRegexp(
+            AppException,
+            "The number of items you want to attach exceeds the limit of 50 000 items per folder.",
+        ):
+            sa.generate_items(self.PROJECT_NAME, 50_001, name="a")