feat: handle annotations conversion to json reponse for geospatial projects

Author: Nicolas Hervé

src/kili/adapters/authentification.py

@@ -20,7 +20,7 @@ def is_api_key_valid(
     """
     response = http_client.post(
         url=api_endpoint,
-        data='{"query":"{ me { id email } }"}',
+        data='{"query":"{ viewer { id email } }"}',
         timeout=30,
         headers={
             "Authorization": f"X-API-Key: {api_key}",

src/kili/adapters/kili_api_gateway/label/annotation_to_json_response.py

@@ -7,6 +7,7 @@
 from kili.domain.annotation import (
     ClassicAnnotation,
     ClassificationAnnotation,
+    ObjectDetectionAnnotation,
     RankingAnnotation,
     TranscriptionAnnotation,
     Vertice,
@@ -53,7 +54,7 @@ def patch_label_json_response(
 
         Modifies the input label.
         """
-        if self._project_input_type in {"VIDEO", "LLM_RLHF"}:
+        if self._project_input_type in {"VIDEO", "LLM_RLHF", "GEOSPATIAL"}:
             if not annotations and self._label_has_json_response_data(label):
                 return
 
@@ -63,8 +64,11 @@ def patch_label_json_response(
                     annotations=annotations, json_interface=self._project_json_interface
                 )
             else:
+                print("converting annotations")
                 annotations = cast(List[ClassicAnnotation], annotations)
-                converted_json_resp = _classic_annotations_to_json_response(annotations=annotations)
+                converted_json_resp = _classic_annotations_to_json_response(
+                    annotations=annotations, json_interface=self._project_json_interface
+                )
 
             label["jsonResponse"] = converted_json_resp
 
@@ -150,7 +154,7 @@ def _video_annotations_to_json_response(
 
 
 def _classic_annotations_to_json_response(
-    annotations: List[ClassicAnnotation],
+    annotations: List[ClassicAnnotation], json_interface: Dict
 ) -> Dict[str, Dict[JobName, Dict]]:
     """Convert label annotations to a json response."""
     json_resp = defaultdict(dict)
@@ -183,6 +187,18 @@ def _classic_annotations_to_json_response(
             for job_name, job_resp in ann_json_resp.items():
                 json_resp.setdefault(job_name, {}).setdefault("text", job_resp["text"])
 
+        elif ann["__typename"] == "ObjectDetectionAnnotation":
+            ann = cast(ObjectDetectionAnnotation, ann)
+            ann_json_resp = _object_detection_annotation_to_json_response(
+                ann,
+                other_annotations,
+                json_interface,
+            )
+            for job_name, job_resp in ann_json_resp.items():
+                json_resp.setdefault(job_name, {}).setdefault("annotations", []).extend(
+                    job_resp["annotations"]
+                )
+
         else:
             raise NotImplementedError(f"Cannot convert classic annotation to json response: {ann}")
 
@@ -350,6 +366,46 @@ def _transcription_annotation_to_json_response(
     return json_resp
 
 
+def _object_detection_annotation_to_json_response(
+    annotation: ObjectDetectionAnnotation,
+    other_annotations: List[ClassicAnnotation],
+    json_interface: Dict,
+) -> Dict[JobName, Dict]:
+    """Convert object detection annotation to a json response."""
+    child_annotations = _get_child_annotations(annotation, other_annotations)
+    json_resp_child_jobs = (
+        _compute_children_json_resp(child_annotations, other_annotations)
+        if child_annotations
+        else {}
+    )
+
+    annotation_dict = {
+        "children": json_resp_child_jobs,
+        "categories": [{"name": annotation["category"]}],
+        "mid": annotation["mid"],
+        "type": json_interface["jobs"][annotation["job"]]["tools"][0],
+    }
+
+    norm_vertices = annotation["annotationValue"]["vertices"]
+
+    if json_interface["jobs"][annotation["job"]]["tools"][0] == "marker":
+        annotation_dict["point"] = norm_vertices[0][0][0]
+
+    elif json_interface["jobs"][annotation["job"]]["tools"][0] in {"polygon", "rectangle"}:
+        annotation_dict["boundingPoly"] = [{"normalizedVertices": norm_vertices[0][0]}]
+
+    elif json_interface["jobs"][annotation["job"]]["tools"][0] == "semantic":
+        annotation_dict["boundingPoly"] = [
+            {"normalizedVertices": norm_vert} for norm_vert in norm_vertices[0]
+        ]
+
+    return {
+        annotation["job"]: {
+            "annotations": [annotation_dict],
+        },
+    }
+
+
 def _video_transcription_annotation_to_json_response(
     annotation: VideoTranscriptionAnnotation,
 ) -> Dict[str, Dict[JobName, Dict]]:

src/kili/domain/annotation.py

@@ -93,6 +93,21 @@ class Annotation(TypedDict):
     path: List[List[str]]
 
 
+class ObjectDetectionAnnotation(TypedDict):
+    """Object detection annotation."""
+
+    # pylint: disable=unused-private-member
+    __typename: Literal["ObjectDetectionAnnotation"]
+    id: AnnotationId
+    labelId: LabelId
+    job: JobName
+    path: List[List[str]]
+    annotationValue: ObjectDetectionAnnotationValue
+    name: Optional[str]
+    mid: str
+    category: str
+
+
 class FrameInterval(TypedDict):
     """Frame interval."""
 
@@ -174,6 +189,7 @@ class VideoTranscriptionAnnotation(TypedDict):
 
 ClassicAnnotation = Union[
     ClassificationAnnotation,
+    ObjectDetectionAnnotation,
     RankingAnnotation,
     TranscriptionAnnotation,
 ]

tests/unit/adapters/kili_api_gateway/label/test_label_utils.py

@@ -173,7 +173,7 @@ def test_given_classic_label_annotations_when_converting_to_json_resp_it_works(
     _ = annotations
 
     # When
-    json_resp = _classic_annotations_to_json_response(annotations)
+    json_resp = _classic_annotations_to_json_response(annotations, {})
 
     # Then
     assert json_resp == expected_json_resp