Add assign images

Author: hovnatan

docs/source/superannotate.sdk.rst

@@ -42,9 +42,8 @@ ________
 .. autofunction:: superannotate.unshare_project
 .. autofunction:: superannotate.get_project_settings
 .. autofunction:: superannotate.set_project_settings
-.. 
-  ## .. autofunction:: superannotate.get_project_default_image_quality_in_editor
-  ## .. autofunction:: superannotate.set_project_default_image_quality_in_editor
+.. autofunction:: superannotate.get_project_default_image_quality_in_editor
+.. autofunction:: superannotate.set_project_default_image_quality_in_editor
 .. autofunction:: superannotate.get_project_workflow
 .. autofunction:: superannotate.set_project_workflow
 
@@ -77,6 +76,8 @@ ______
 .. autofunction:: superannotate.upload_annotations_from_json_to_image
 .. autofunction:: superannotate.copy_image
 .. autofunction:: superannotate.move_image
+.. autofunction:: superannotate.pin_image
+.. autofunction:: superannotate.assign_images
 .. autofunction:: superannotate.delete_image
 .. autofunction:: superannotate.add_annotation_bbox_to_image
 .. autofunction:: superannotate.add_annotation_polygon_to_image

install.sh

@@ -19,3 +19,6 @@ pip install -e .
 
 # for testing
 pip install pytest pytest-xdist
+
+# for docs
+pip install sphinx sphinx_rtd_theme

superannotate/__init__.py

@@ -45,7 +45,9 @@
     set_image_annotation_status, upload_annotations_from_json_to_image
 )
 from .db.project import get_project_metadata, search_projects
-from .db.project_images import copy_image, move_image, upload_image_to_project
+from .db.project_images import (
+    assign_images, copy_image, move_image, pin_image, upload_image_to_project
+)
 from .db.projects import (
     create_project, create_project_like_project, delete_project,
     get_project_default_image_quality_in_editor, get_project_image_count,

superannotate/db/images.py

@@ -903,50 +903,49 @@ def upload_annotations_from_json_to_image(
         path=f'/image/{image_id}/annotation/getAnnotationUploadToken',
         params=params
     )
-    if response.ok:
-        res = response.json()
-        if project_type == 1:  # vector
-            res = res['objects']
-            s3_session = boto3.Session(
-                aws_access_key_id=res['accessKeyId'],
-                aws_secret_access_key=res['secretAccessKey'],
-                aws_session_token=res['sessionToken']
-            )
-            s3_resource = s3_session.resource('s3')
-            bucket = s3_resource.Bucket(res["bucket"])
-            bucket.put_object(
-                Key=res['filePath'], Body=json.dumps(annotation_json)
-            )
-        else:  # pixel
-            if mask is None:
-                raise SABaseException(0, "Pixel annotation should have mask.")
-            if not isinstance(mask, io.BytesIO):
-                with open(mask, "rb") as f:
-                    mask = io.BytesIO(f.read())
-            res_j = res['pixel']
-            s3_session = boto3.Session(
-                aws_access_key_id=res_j['accessKeyId'],
-                aws_secret_access_key=res_j['secretAccessKey'],
-                aws_session_token=res_j['sessionToken']
-            )
-            s3_resource = s3_session.resource('s3')
-            bucket = s3_resource.Bucket(res_j["bucket"])
-            bucket.put_object(
-                Key=res_j['filePath'], Body=json.dumps(annotation_json)
-            )
-            res_m = res['save']
-            s3_session = boto3.Session(
-                aws_access_key_id=res_m['accessKeyId'],
-                aws_secret_access_key=res_m['secretAccessKey'],
-                aws_session_token=res_m['sessionToken']
-            )
-            s3_resource = s3_session.resource('s3')
-            bucket = s3_resource.Bucket(res_m["bucket"])
-            bucket.put_object(Key=res_m['filePath'], Body=mask)
-    else:
+    if not response.ok:
         raise SABaseException(
             response.status_code, "Couldn't upload annotation. " + response.text
         )
+    res = response.json()
+    if project_type == 1:  # vector
+        res = res['objects']
+        s3_session = boto3.Session(
+            aws_access_key_id=res['accessKeyId'],
+            aws_secret_access_key=res['secretAccessKey'],
+            aws_session_token=res['sessionToken']
+        )
+        s3_resource = s3_session.resource('s3')
+        bucket = s3_resource.Bucket(res["bucket"])
+        bucket.put_object(
+            Key=res['filePath'], Body=json.dumps(annotation_json)
+        )
+    else:  # pixel
+        if mask is None:
+            raise SABaseException(0, "Pixel annotation should have mask.")
+        if not isinstance(mask, io.BytesIO):
+            with open(mask, "rb") as f:
+                mask = io.BytesIO(f.read())
+        res_j = res['pixel']
+        s3_session = boto3.Session(
+            aws_access_key_id=res_j['accessKeyId'],
+            aws_secret_access_key=res_j['secretAccessKey'],
+            aws_session_token=res_j['sessionToken']
+        )
+        s3_resource = s3_session.resource('s3')
+        bucket = s3_resource.Bucket(res_j["bucket"])
+        bucket.put_object(
+            Key=res_j['filePath'], Body=json.dumps(annotation_json)
+        )
+        res_m = res['save']
+        s3_session = boto3.Session(
+            aws_access_key_id=res_m['accessKeyId'],
+            aws_secret_access_key=res_m['secretAccessKey'],
+            aws_session_token=res_m['sessionToken']
+        )
+        s3_resource = s3_session.resource('s3')
+        bucket = s3_resource.Bucket(res_m["bucket"])
+        bucket.put_object(Key=res_m['filePath'], Body=mask)
 
 
 def create_fuse_image(

superannotate/db/project_images.py

@@ -11,7 +11,8 @@
 from ..exceptions import SABaseException
 from .images import (
     delete_image, get_image_annotations, get_image_bytes, get_image_metadata,
-    set_image_annotation_status, upload_annotations_from_json_to_image
+    search_images, set_image_annotation_status,
+    upload_annotations_from_json_to_image
 )
 from .projects import (
     __create_image, _get_project_image_quality_in_editor,
@@ -126,7 +127,8 @@ def copy_image(
     image_name,
     destination_project,
     include_annotations=False,
-    copy_annotation_status=False
+    copy_annotation_status=False,
+    copy_pin=False
 ):
     """Copy image to a project. The image's project is the same as destination
     project then the name will be changed to <image_name>_(<num>).<image_ext>,
@@ -142,6 +144,8 @@ def copy_image(
     :type include_annotations: bool
     :param copy_annotation_status: enables annotations status copy
     :type copy_annotation_status: bool
+    :param copy_pin: enables image pin status copy
+    :type copy_pin: bool
     """
     if not isinstance(source_project, dict):
         source_project = get_project_metadata(source_project)
@@ -190,6 +194,8 @@ def copy_image(
             destination_project, new_name,
             annotation_status_int_to_str(img_metadata["annotation_status"])
         )
+    if copy_pin:
+        pin_image(destination_project, new_name, img_metadata["is_pinned"])
 
     logger.info(
         "Copied image %s/%s to %s/%s.", source_project["name"], image_name,
@@ -201,8 +207,9 @@ def move_image(
     source_project,
     image_name,
     destination_project,
-    include_annotations=False,
-    copy_annotation_status=False
+    include_annotations=True,
+    copy_annotation_status=True,
+    copy_pin=True
 ):
     """Move image from source_project to destination_project. source_project
     and destination_project cannot be the same.
@@ -217,6 +224,8 @@ def move_image(
     :type include_annotations: bool
     :param copy_annotation_status: enables annotations status copy
     :type copy_annotation_status: bool
+    :param copy_pin: enables image pin status copy
+    :type copy_pin: bool
     """
     if not isinstance(source_project, dict):
         source_project = get_project_metadata(source_project)
@@ -228,7 +237,86 @@ def move_image(
         )
     copy_image(
         source_project, image_name, destination_project, include_annotations,
-        copy_annotation_status
+        copy_annotation_status, copy_pin
     )
     delete_image(source_project, image_name)
     logger.info("Deleted image %s/%s.", source_project["name"], image_name)
+
+
+def pin_image(project, image_name, pin=True):
+    """Pins (or unpins) image
+
+    :param project: project name or metadata of the project
+    :type project: str or dict
+    :param image_name: image name
+    :type image: str
+    :param pin: sets to pin if True, else unpins image
+    :type pin: bool
+    """
+    if not isinstance(project, dict):
+        project = get_project_metadata(project)
+    img_metadata = get_image_metadata(project, image_name)
+    team_id, project_id, image_id = project["team_id"], project[
+        "id"], img_metadata["id"]
+    params = {"team_id": team_id, "project_id": project_id}
+    json_req = {"is_pinned": int(pin)}
+    response = _api.send_request(
+        req_type='PUT',
+        path=f'/image/{image_id}',
+        params=params,
+        json_req=json_req
+    )
+    if not response.ok:
+        raise SABaseException(
+            response.status_code, "Couldn't pin image " + response.text
+        )
+
+
+def assign_images(project, image_names, user):
+    """Assigns images to a user. The assignment role, QA or Annotator, will
+    be deduced from the user's role in the project. With SDK, the user can be
+    assigned to a role in the project with the share_project function.
+
+    :param project: project name or metadata of the project
+    :type project: str or dict
+    :param image_names: list of image names to assign
+    :type image_names: list of str
+    :param user: user email
+    :type user: str
+    """
+    logger.info("Assign %s images to user %s", len(image_names), user)
+    if len(image_names) == 0:
+        return
+    if not isinstance(project, dict):
+        project = get_project_metadata(project)
+    folder_id = None
+    images = search_images(project, return_metadata=True)
+    image_dict = {}
+    for image in images:
+        image_dict[image["name"]] = image["id"]
+        if folder_id is None:
+            folder_id = image["folder_id"]
+        elif folder_id != image["folder_id"]:
+            raise SABaseException(0, "Folders not implemented yet")
+
+    image_ids = []
+    for image_name in image_names:
+        image_ids.append(image_dict[image_name])
+    team_id, project_id = project["team_id"], project["id"]
+    params = {
+        "team_id": team_id,
+        "project_id": project_id,
+        "folder_id": folder_id
+    }
+    json_req = {"user_id": user, "image_ids": image_ids}
+    response = _api.send_request(
+        req_type='POST',
+        path='/images/assign',
+        params=params,
+        json_req=json_req
+    )
+    if not response.ok:
+        raise SABaseException(
+            response.status_code, "Couldn't assign images " + response.text
+        )
+    # print(response.json())

tests/test_assign_images.py

@@ -0,0 +1,64 @@
+from pathlib import Path
+import time
+
+import pytest
+
+import superannotate as sa
+
+PROJECT_NAME_VECTOR = "test assign images"
+
+
+def test_assign_images(tmpdir):
+    tmpdir = Path(tmpdir)
+
+    projects = sa.search_projects(PROJECT_NAME_VECTOR, return_metadata=True)
+    for project in projects:
+        sa.delete_project(project)
+
+    project = sa.create_project(PROJECT_NAME_VECTOR, "test", "Vector")
+    sa.share_project(project, "hovnatan@superannotate.com", "QA")
+
+    sa.upload_images_from_folder_to_project(
+        project, "./tests/sample_project_vector"
+    )
+
+    sa.assign_images(
+        project, ["example_image_1.jpg", "example_image_2.jpg"],
+        "hovnatan@superannotate.com"
+    )
+
+    time.sleep(1)
+    im1_metadata = sa.get_image_metadata(project, "example_image_1.jpg")
+    im2_metadata = sa.get_image_metadata(project, "example_image_2.jpg")
+
+    assert im1_metadata["qa_id"] == "hovnatan@superannotate.com"
+    assert im2_metadata["qa_id"] == "hovnatan@superannotate.com"
+
+    sa.unshare_project(project, "hovnatan@superannotate.com")
+
+    time.sleep(1)
+
+    im1_metadata = sa.get_image_metadata(project, "example_image_1.jpg")
+    im2_metadata = sa.get_image_metadata(project, "example_image_2.jpg")
+
+    assert im1_metadata["qa_id"] is None
+    assert im2_metadata["qa_id"] is None
+    assert im1_metadata["annotator_id"] is None
+    assert im2_metadata["annotator_id"] is None
+
+    sa.share_project(project, "hovnatan@superannotate.com", "Annotator")
+
+    sa.assign_images(
+        project, ["example_image_1.jpg", "example_image_2.jpg"],
+        "hovnatan@superannotate.com"
+    )
+
+    time.sleep(1)
+    im1_metadata = sa.get_image_metadata(project, "example_image_1.jpg")
+    im2_metadata = sa.get_image_metadata(project, "example_image_2.jpg")
+
+    assert im1_metadata["annotator_id"] == "hovnatan@superannotate.com"
+    assert im2_metadata["annotator_id"] == "hovnatan@superannotate.com"
+    assert im1_metadata["qa_id"] is None
+    assert im2_metadata["qa_id"] is None
+    print(sa.get_project_metadata(PROJECT_NAME_VECTOR))

tests/test_image_copy_move.py

@@ -37,6 +37,7 @@ def test_image_copy_mult(tmpdir):
         "./tests/sample_project_vector/example_image_2.jpg",
         annotation_status="InProgress"
     )
+    sa.pin_image(project, "example_image_1.jpg")
 
     images = sa.search_images(project)
     assert len(images) == 2
@@ -48,7 +49,8 @@ def test_image_copy_mult(tmpdir):
             image,
             project,
             include_annotations=True,
-            copy_annotation_status=True
+            copy_annotation_status=True,
+            copy_pin=True
         )
     assert len(sa.search_images(project)) == 5
     images = sa.search_images(project)
@@ -57,6 +59,9 @@ def test_image_copy_mult(tmpdir):
     anns = sa.get_image_annotations(project, f"example_image_1_({i+1}).jpg")
     assert anns["annotation_json"] is not None
 
+    metadata = sa.get_image_metadata(project, f"example_image_1_({i+1}).jpg")
+    assert metadata["is_pinned"] == 1
+
 
 def test_image_copy(tmpdir):
     tmpdir = Path(tmpdir)