Merge pull request #250 from superannotateai/develop

Develop

Author: VaghinakDev

src/superannotate/__init__.py

@@ -302,4 +302,3 @@
 logging.config.fileConfig(
     os.path.join(WORKING_DIR, "logging.conf"), disable_existing_loggers=False
 )
-sys.tracebacklimit = 0

src/superannotate/lib/app/analytics/common.py

@@ -212,6 +212,8 @@ def aggregate_annotations_as_df(
         "updatorRole": [],
         "updatorEmail": [],
         "folderName": [],
+        "imageAnnotator": [],
+        "imageQA": [],
     }
 
     if include_comments:
@@ -255,6 +257,8 @@ def __get_image_metadata(image_name, annotations):
         image_metadata["imageWidth"] = annotations["metadata"].get("width")
         image_metadata["imageStatus"] = annotations["metadata"].get("status")
         image_metadata["imagePinned"] = annotations["metadata"].get("pinned")
+        image_metadata["imageAnnotator"] = annotations["metadata"].get("annotatorEmail")
+        image_metadata["imageQA"] = annotations["metadata"].get("qaEmail")
         return image_metadata
 
     def __get_user_metadata(annotation):

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

@@ -30,7 +30,6 @@
 from lib.app.interface.types import AnnotationType
 from lib.app.interface.types import ImageQualityChoices
 from lib.app.interface.types import NotEmptyStr
-from lib.app.interface.types import Status
 from lib.app.interface.types import validate_arguments
 from lib.app.mixp.decorators import Trackable
 from lib.app.serializers import BaseSerializers
@@ -280,7 +279,7 @@ def clone_project(
 def search_images(
     project: Union[NotEmptyStr, dict],
     image_name_prefix: Optional[NotEmptyStr] = None,
-    annotation_status: Optional[Status] = None,
+    annotation_status: Optional[AnnotationStatuses] = None,
     return_metadata: Optional[StrictBool] = False,
 ):
     """Search images by name_prefix (case-insensitive) and annotation status
@@ -333,10 +332,6 @@ def create_folder(project: NotEmptyStr, folder_name: NotEmptyStr):
     res = controller.create_folder(project=project, folder_name=folder_name)
     if res.data:
         folder = res.data
-        if folder and folder.name != folder_name:
-            logger.warning(
-                f"Created folder has name {folder.name}, since folder with name {folder_name} already existed.",
-            )
         logger.info(f"Folder {folder.name} created in project {project}")
         return folder.to_dict()
     if res.errors:
@@ -603,7 +598,7 @@ def upload_images_from_public_urls_to_project(
     project: Union[NotEmptyStr, dict],
     img_urls: List[NotEmptyStr],
     img_names: Optional[List[NotEmptyStr]] = None,
-    annotation_status: Optional[Status] = "NotStarted",
+    annotation_status: Optional[AnnotationStatuses] = "NotStarted",
     image_quality_in_editor: Optional[NotEmptyStr] = None,
 ):
     """Uploads all images given in the list of URL strings in img_urls to the project.
@@ -1641,7 +1636,7 @@ def upload_videos_from_folder_to_project(
     target_fps: Optional[int] = None,
     start_time: Optional[float] = 0.0,
     end_time: Optional[float] = None,
-    annotation_status: Optional[Status] = "NotStarted",
+    annotation_status: Optional[AnnotationStatuses] = "NotStarted",
     image_quality_in_editor: Optional[str] = None,
 ):
     """Uploads image frames from all videos with given extensions from folder_path to the project.
@@ -1705,11 +1700,7 @@ def upload_videos_from_folder_to_project(
     project_folder_name = project_name + (f"/{folder_name}" if folder_name else "")
 
     logger.info(
-        "Uploading all videos with extensions %s from %s to project %s. Excluded file patterns are: %s.",
-        extensions,
-        str(folder_path),
-        project_name,
-        exclude_file_patterns,
+        f"Uploading all videos with extensions {extensions} from {str(folder_path)} to project {project_name}. Excluded file patterns are: {[*exclude_file_patterns]}.",
     )
     uploaded_paths = []
     for path in video_paths:
@@ -1778,7 +1769,7 @@ def upload_video_to_project(
     target_fps: Optional[int] = None,
     start_time: Optional[float] = 0.0,
     end_time: Optional[float] = None,
-    annotation_status: Optional[Status] = "NotStarted",
+    annotation_status: Optional[AnnotationStatuses] = "NotStarted",
     image_quality_in_editor: Optional[NotEmptyStr] = None,
 ):
     """Uploads image frames from video to platform. Uploaded images will have
@@ -2235,7 +2226,7 @@ def download_image(
 def attach_image_urls_to_project(
     project: Union[NotEmptyStr, dict],
     attachments: Union[str, Path],
-    annotation_status: Optional[Status] = "NotStarted",
+    annotation_status: Optional[AnnotationStatuses] = "NotStarted",
 ):
     """Link images on external storage to SuperAnnotate.
 
@@ -2298,7 +2289,7 @@ def attach_image_urls_to_project(
 def attach_video_urls_to_project(
     project: Union[NotEmptyStr, dict],
     attachments: Union[str, Path],
-    annotation_status: Optional[Status] = "NotStarted",
+    annotation_status: Optional[AnnotationStatuses] = "NotStarted",
 ):
     """Link videos on external storage to SuperAnnotate.
 
@@ -3314,7 +3305,7 @@ def upload_image_to_project(
     project: NotEmptyStr,
     img,
     image_name: Optional[NotEmptyStr] = None,
-    annotation_status: Optional[Status] = "NotStarted",
+    annotation_status: Optional[AnnotationStatuses] = "NotStarted",
     from_s3_bucket=None,
     image_quality_in_editor: Optional[NotEmptyStr] = None,
 ):
@@ -3389,7 +3380,7 @@ def search_models(
 def upload_images_to_project(
     project: NotEmptyStr,
     img_paths: List[NotEmptyStr],
-    annotation_status: Optional[Status] = "NotStarted",
+    annotation_status: Optional[AnnotationStatuses] = "NotStarted",
     from_s3_bucket=None,
     image_quality_in_editor: Optional[ImageQualityChoices] = None,
 ):
@@ -3523,7 +3514,7 @@ def delete_annotations(
 def attach_document_urls_to_project(
     project: Union[NotEmptyStr, dict],
     attachments: Union[Path, NotEmptyStr],
-    annotation_status: Optional[Status] = "NotStarted",
+    annotation_status: Optional[AnnotationStatuses] = "NotStarted",
 ):
     """Link documents on external storage to SuperAnnotate.
 

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

@@ -17,7 +17,7 @@ def validate(cls, value: Union[str]) -> Union[str]:
         if cls.curtail_length and len(value) > cls.curtail_length:
             value = value[: cls.curtail_length]
         if value.lower() not in AnnotationStatus.values():
-            raise TypeError(f"Available statuses is {', '.join(AnnotationStatus)}. ")
+            raise TypeError(f"Available statuses is {', '.join(AnnotationStatus.titles())}. ")
         return value
 
 
@@ -80,6 +80,5 @@ def wrapped(*args, **kwargs):
                         field, " " * (48 - len(field)), f"\n {' ' * 48}".join(text)
                     )
                 )
-            raise Exception("\n".join(texts))
-
+            raise Exception("\n".join(texts)) from None
     return wrapped

src/superannotate/lib/core/helpers.py

@@ -0,0 +1,116 @@
+from collections import defaultdict
+from typing import List
+
+
+def map_annotation_classes_name(annotation_classes, logger=None) -> dict:
+    classes_data = defaultdict(dict)
+    for annotation_class in annotation_classes:
+        class_info = {"id": annotation_class.uuid}
+        if annotation_class.attribute_groups:
+            for attribute_group in annotation_class.attribute_groups:
+                attribute_group_data = defaultdict(dict)
+                for attribute in attribute_group["attributes"]:
+                    if logger and attribute["name"] in attribute_group_data.keys():
+                        logger.warning(
+                            f"Duplicate annotation class attribute name {attribute['name']}"
+                            f" in attribute group {attribute_group['name']}. "
+                            "Only one of the annotation class attributes will be used. "
+                            "This will result in errors in annotation upload."
+                        )
+                    attribute_group_data[attribute["name"]] = attribute["id"]
+                if logger and attribute_group["name"] in class_info.keys():
+                    logger.warning(
+                        f"Duplicate annotation class attribute group name {attribute_group['name']}."
+                        " Only one of the annotation class attribute groups will be used."
+                        " This will result in errors in annotation upload."
+                    )
+                class_info["attribute_groups"] = {
+                    attribute_group["name"]: {
+                        "id": attribute_group["id"],
+                        "attributes": attribute_group_data,
+                    }
+                }
+        if logger and annotation_class.name in classes_data.keys():
+            logger.warning(
+                f"Duplicate annotation class name {annotation_class.name}."
+                f" Only one of the annotation classes will be used."
+                " This will result in errors in annotation upload.",
+            )
+        classes_data[annotation_class.name] = class_info
+    return classes_data
+
+
+def fill_annotation_ids(annotations: dict, annotation_classes_name_maps: dict, templates: List[dict], logger=None):
+    annotation_classes_name_maps = annotation_classes_name_maps
+    if "instances" not in annotations:
+        return
+    missing_classes = set()
+    missing_attribute_groups = set()
+    missing_attributes = set()
+    unknown_classes = dict()
+    report = {
+        "missing_classes": missing_classes,
+        "missing_attribute_groups": missing_attribute_groups,
+        "missing_attributes": missing_attributes,
+    }
+    for annotation in [i for i in annotations["instances"] if "className" in i]:
+        if "className" not in annotation:
+            return
+        annotation_class_name = annotation["className"]
+        if annotation_class_name not in annotation_classes_name_maps.keys():
+            if annotation_class_name not in unknown_classes:
+                missing_classes.add(annotation_class_name)
+                unknown_classes[annotation_class_name] = {
+                    "id": -(len(unknown_classes) + 1),
+                    "attribute_groups": {},
+                }
+    annotation_classes_name_maps.update(unknown_classes)
+    template_name_id_map = {template["name"]: template["id"] for template in templates}
+    for annotation in (
+            i for i in annotations["instances"] if i.get("type", None) == "template"
+    ):
+        annotation["templateId"] = template_name_id_map.get(
+            annotation.get("templateName", ""), -1
+        )
+
+    for annotation in [i for i in annotations["instances"] if "className" in i]:
+        annotation_class_name = annotation["className"]
+        if annotation_class_name not in annotation_classes_name_maps.keys():
+            if logger:
+                logger.warning(
+                    f"Couldn't find annotation class {annotation_class_name}"
+                )
+            continue
+        annotation["classId"] = annotation_classes_name_maps[annotation_class_name]["id"]
+        for attribute in annotation["attributes"]:
+            if (
+                    attribute["groupName"]
+                    not in annotation_classes_name_maps[annotation_class_name]["attribute_groups"]
+            ):
+                if logger:
+                    logger.warning(
+                        f"Couldn't find annotation group {attribute['groupName']}."
+                    )
+                missing_attribute_groups.add(attribute["groupName"])
+                continue
+            attribute["groupId"] = annotation_classes_name_maps[annotation_class_name][
+                "attribute_groups"
+            ][attribute["groupName"]]["id"]
+            if (
+                    attribute["name"]
+                    not in annotation_classes_name_maps[annotation_class_name][
+                "attribute_groups"
+            ][attribute["groupName"]]["attributes"]
+            ):
+                del attribute["groupId"]
+                if logger:
+                    logger.warning(
+                        f"Couldn't find annotation name {attribute['name']} in"
+                        f" annotation group {attribute['groupName']}",
+                    )
+                missing_attributes.add(attribute["name"])
+                continue
+            attribute["id"] = annotation_classes_name_maps[annotation_class_name][
+                "attribute_groups"
+            ][attribute["groupName"]]["attributes"][attribute["name"]]
+    return report

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

@@ -27,6 +27,7 @@ def __init__(
         self._project = project
         self._folder = folder
         self._folders = folders
+        self._origin_name = folder.name
 
     def validate_folder(self):
         if not self._folder.name:
@@ -53,6 +54,11 @@ def execute(self):
         if self.is_valid():
             self._folder.project_id = self._project.uuid
             self._response.data = self._folders.insert(self._folder)
+            if self._response.data.name not in (self._origin_name, self._folder.name):
+                logger.warning(
+                    f"Created folder has name {self._response.data.name},"
+                    f" since folder with name {self._folder.name} already existed."
+                )
         return self._response