Merge pull request #432 from superannotateai/attach_items

Attach items

Author: VaghinakDev

docs/source/superannotate.sdk.rst

@@ -74,6 +74,9 @@ ______
 
 .. autofunction:: superannotate.query
 .. autofunction:: superannotate.search_items
+.. autofunction:: superannotate.attach_items
+.. autofunction:: superannotate.copy_items
+.. autofunction:: superannotate.move_items
 .. autofunction:: superannotate.get_item_metadata
 
 ----------

src/superannotate/__init__.py

@@ -25,6 +25,7 @@
     attach_document_urls_to_project,
 )
 from superannotate.lib.app.interface.sdk_interface import attach_image_urls_to_project
+from superannotate.lib.app.interface.sdk_interface import attach_items
 from superannotate.lib.app.interface.sdk_interface import (
     attach_items_from_integrated_storage,
 )
@@ -34,6 +35,7 @@
 from superannotate.lib.app.interface.sdk_interface import consensus
 from superannotate.lib.app.interface.sdk_interface import copy_image
 from superannotate.lib.app.interface.sdk_interface import copy_images
+from superannotate.lib.app.interface.sdk_interface import copy_items
 from superannotate.lib.app.interface.sdk_interface import create_annotation_class
 from superannotate.lib.app.interface.sdk_interface import (
     create_annotation_classes_from_classes_json,
@@ -71,6 +73,7 @@
 from superannotate.lib.app.interface.sdk_interface import init
 from superannotate.lib.app.interface.sdk_interface import invite_contributors_to_team
 from superannotate.lib.app.interface.sdk_interface import move_images
+from superannotate.lib.app.interface.sdk_interface import move_items
 from superannotate.lib.app.interface.sdk_interface import pin_image
 from superannotate.lib.app.interface.sdk_interface import prepare_export
 from superannotate.lib.app.interface.sdk_interface import query
@@ -175,6 +178,9 @@
     "get_item_metadata",
     "search_items",
     "query",
+    "attach_items",
+    "copy_items",
+    "move_items",
     # Image Section
     "copy_images",
     "move_images",

src/superannotate/lib/app/helpers.py

@@ -8,6 +8,7 @@
 
 import boto3
 import pandas as pd
+from superannotate.lib.app.exceptions import AppException
 from superannotate.lib.app.exceptions import PathError
 from superannotate.lib.core import ATTACHED_VIDEO_ANNOTATION_POSTFIX
 from superannotate.lib.core import PIXEL_ANNOTATION_POSTFIX
@@ -168,3 +169,33 @@ def get_paths_and_duplicated_from_csv(csv_path):
         else:
             duplicate_images.append(temp)
     return images_to_upload, duplicate_images
+
+
+def get_name_url_duplicated_from_csv(csv_path):
+    image_data = pd.read_csv(csv_path, dtype=str)
+    if "url" not in image_data.columns:
+        raise AppException("Column 'url' is required")
+    image_data = image_data[~image_data["url"].isnull()]
+    if "name" in image_data.columns:
+        image_data["name"] = (
+            image_data["name"]
+            .fillna("")
+            .apply(lambda cell: cell if str(cell).strip() else str(uuid.uuid4()))
+        )
+    else:
+        image_data["name"] = [str(uuid.uuid4()) for _ in range(len(image_data.index))]
+
+    image_data = pd.DataFrame(image_data, columns=["name", "url"])
+    img_names_urls = image_data.to_dict(orient="records")
+    duplicate_images = []
+    seen = []
+    images_to_upload = []
+    for i in img_names_urls:
+        temp = i["name"]
+        i["name"] = i["name"].strip()
+        if i["name"] not in seen:
+            seen.append(i["name"])
+            images_to_upload.append(i)
+        else:
+            duplicate_images.append(temp)
+    return images_to_upload, duplicate_images

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

@@ -1,3 +1,4 @@
+import collections
 import io
 import json
 import os
@@ -17,10 +18,13 @@
 from lib.app.annotation_helpers import add_annotation_point_to_json
 from lib.app.helpers import extract_project_folder
 from lib.app.helpers import get_annotation_paths
+from lib.app.helpers import get_name_url_duplicated_from_csv
 from lib.app.helpers import get_paths_and_duplicated_from_csv
 from lib.app.interface.types import AnnotationStatuses
 from lib.app.interface.types import AnnotationType
 from lib.app.interface.types import AnnotatorRole
+from lib.app.interface.types import AttachmentArg
+from lib.app.interface.types import AttachmentDict
 from lib.app.interface.types import ClassType
 from lib.app.interface.types import EmailStr
 from lib.app.interface.types import ImageQualityChoices
@@ -36,6 +40,7 @@
 from lib.app.serializers import SettingsSerializer
 from lib.app.serializers import TeamSerializer
 from lib.core import LIMITED_FUNCTIONS
+from lib.core.entities import AttachmentEntity
 from lib.core.entities.integrations import IntegrationEntity
 from lib.core.entities.project_entities import AnnotationClassEntity
 from lib.core.enums import ImageQuality
@@ -588,15 +593,20 @@ def copy_images(
     :return: list of skipped image names
     :rtype: list of strs
     """
-
+    warning_msg = (
+        "We're deprecating the copy_images function. Please use copy_items instead. Learn more. \n"
+        "https://superannotate.readthedocs.io/en/stable/superannotate.sdk.html#superannotate.copy_items"
+    )
+    logger.warning(warning_msg)
+    warnings.warn(warning_msg, DeprecationWarning)
     project_name, source_folder_name = extract_project_folder(source_project)
 
     to_project_name, destination_folder_name = extract_project_folder(
         destination_project
     )
     if project_name != to_project_name:
         raise AppException(
-            "Source and destination projects should be the same for copy_images"
+            "Source and destination projects should be the same"
         )
     if not image_names:
         images = (
@@ -652,6 +662,12 @@ def move_images(
     :return: list of skipped image names
     :rtype: list of strs
     """
+    warning_msg = (
+        "We're deprecating the move_images function. Please use move_items instead. Learn more."
+        "https://superannotate.readthedocs.io/en/stable/superannotate.sdk.html#superannotate.move_items"
+    )
+    logger.warning(warning_msg)
+    warnings.warn(warning_msg, DeprecationWarning)
     project_name, source_folder_name = extract_project_folder(source_project)
 
     project = Controller.get_default().get_project_metadata(project_name).data
@@ -1811,6 +1827,12 @@ def attach_image_urls_to_project(
     :return: list of linked image names, list of failed image names, list of duplicate image names
     :rtype: tuple
     """
+    warning_msg = (
+        "We're deprecating the attach_image_urls_to_project function. Please use attach_items instead. Learn more."
+        "https://superannotate.readthedocs.io/en/stable/superannotate.sdk.html#superannotate.attach_items"
+    )
+    logger.warning(warning_msg)
+    warnings.warn(warning_msg, DeprecationWarning)
     project_name, folder_name = extract_project_folder(project)
     project = Controller.get_default().get_project_metadata(project_name).data
     project_folder_name = project_name + (f"/{folder_name}" if folder_name else "")
@@ -1878,6 +1900,12 @@ def attach_video_urls_to_project(
     :return: attached videos, failed videos, skipped videos
     :rtype: (list, list, list)
     """
+    warning_msg = (
+        "We're deprecating the attach_video_urls_to_project function. Please use attach_items instead. Learn more."
+        "https://superannotate.readthedocs.io/en/stable/superannotate.sdk.html#superannotate.attach_items"
+    )
+    logger.warning(warning_msg)
+    warnings.warn(warning_msg, DeprecationWarning)
     project_name, folder_name = extract_project_folder(project)
     project = Controller.get_default().get_project_metadata(project_name).data
     project_folder_name = project_name + (f"/{folder_name}" if folder_name else "")
@@ -2480,8 +2508,10 @@ def search_images_all_folders(
 
     :param project: project name
     :type project: str
+
     :param image_name_prefix: image name prefix for search
     :type image_name_prefix: str
+
     :param annotation_status: if not None, annotation statuses of images to filter,
                               should be one of NotStarted InProgress QualityCheck Returned Completed Skipped
     :type annotation_status: str
@@ -2741,6 +2771,12 @@ def attach_document_urls_to_project(
     :return: list of attached documents, list of not attached documents, list of skipped documents
     :rtype: tuple
     """
+    warning_msg = (
+        "We're deprecating the attach_document_urls_to_project function. Please use attach_items instead. Learn more."
+        "https://superannotate.readthedocs.io/en/stable/superannotate.sdk.html#superannotate.attach_items"
+    )
+    logger.warning(warning_msg)
+    warnings.warn(warning_msg, DeprecationWarning)
     project_name, folder_name = extract_project_folder(project)
     project = Controller.get_default().get_project_metadata(project_name).data
     project_folder_name = project_name + (f"/{folder_name}" if folder_name else "")
@@ -3093,3 +3129,127 @@ def search_items(
     if response.errors:
         raise AppException(response.errors)
     return BaseSerializer.serialize_iterable(response.data)
+
+
+@Trackable
+@validate_arguments
+def attach_items(
+        project: Union[NotEmptyStr, dict],
+        attachments: AttachmentArg,
+        annotation_status: Optional[AnnotationStatuses] = "NotStarted"
+):
+    attachments = attachments.data
+    project_name, folder_name = extract_project_folder(project)
+    if attachments and isinstance(attachments[0], AttachmentDict):
+        unique_attachments = set(attachments)
+        duplicate_attachments = [item for item, count in collections.Counter(attachments).items() if count > 1]
+    else:
+        unique_attachments, duplicate_attachments = get_name_url_duplicated_from_csv(attachments)
+    if duplicate_attachments:
+        logger.info("Dropping duplicates.")
+    unique_attachments = parse_obj_as(List[AttachmentEntity], unique_attachments)
+    uploaded, fails, duplicated = [], [], []
+    if unique_attachments:
+        logger.info(f"Attaching  {len(unique_attachments)} file(s) to project {project}.")
+        response = Controller.get_default().attach_items(
+            project_name=project_name,
+            folder_name=folder_name,
+            attachments=unique_attachments,
+            annotation_status=annotation_status,
+        )
+        if response.errors:
+            raise AppException(response.errors)
+        uploaded, duplicated = response.data
+        uploaded = [i["name"] for i in uploaded]
+        fails = [
+            attachment.name
+            for attachment in unique_attachments
+            if attachment.name not in uploaded and attachment.name not in duplicated
+        ]
+    return uploaded, fails, duplicated
+
+
+@Trackable
+@validate_arguments
+def copy_items(
+        source: Union[NotEmptyStr, dict],
+        destination: Union[NotEmptyStr, dict],
+        items: Optional[List[NotEmptyStr]] = None,
+        include_annotations: Optional[StrictBool] = True,
+):
+    """Copy images in bulk between folders in a project
+
+    :param source: project name or folder path to select items from (e.g., “project1/folder1”).
+    :type source: str
+
+    :param destination: project name (root) or folder path to place copied items.
+    :type destination: str
+
+    :param items: names of items to copy. If None, all items from the source directory will be copied.
+    :type itmes: list of str
+
+    :param include_annotations: enables annotations copy
+    :type include_annotations: bool
+
+    :return: list of skipped item names
+    :rtype: list of strs
+    """
+
+    project_name, source_folder = extract_project_folder(source)
+
+    to_project_name, destination_folder = extract_project_folder(destination)
+    if project_name != to_project_name:
+        raise AppException(
+            "Source and destination projects should be the same for copy_images"
+        )
+
+    response = Controller.get_default().copy_items(
+        project_name=project_name,
+        from_folder=source_folder,
+        to_folder=destination_folder,
+        items=items,
+        include_annotations=include_annotations,
+    )
+    if response.errors:
+        raise AppException(response.errors)
+
+    return response.data
+
+
+@Trackable
+@validate_arguments
+def move_items(
+        source: Union[NotEmptyStr, dict],
+        destination: Union[NotEmptyStr, dict],
+        items: Optional[List[NotEmptyStr]] = None,
+):
+    """Copy images in bulk between folders in a project
+
+    :param source: project name or folder path to pick items from (e.g., “project1/folder1”).
+    :type source: str
+
+    :param destination: project name (root) or folder path to move items to.
+    :type destination: str
+
+    :param items: names of items to move. If None, all items from the source directory will be moved.
+    :type items: list of str
+
+    :return: list of skipped item names
+    :rtype: list of strs
+    """
+
+    project_name, source_folder = extract_project_folder(source)
+    to_project_name, destination_folder = extract_project_folder(destination)
+    if project_name != to_project_name:
+        raise AppException(
+            "Source and destination projects should be the same"
+        )
+    response = Controller.get_default().move_items(
+        project_name=project_name,
+        from_folder=source_folder,
+        to_folder=destination_folder,
+        items=items,
+    )
+    if response.errors:
+        raise AppException(response.errors)
+    return response.data