fix(LAB-4125): handle all the formats with latestLabels

Author: FannyGaudin

src/kili/services/export/format/coco/__init__.py

@@ -51,9 +51,13 @@ def images_folder(self) -> Path:
     def process_and_save(self, assets: list[dict], output_filename: Path):
         """Extract formatted annotations from labels."""
         clean_assets = self.preprocess_assets(assets)
+        # Expand assets with latestLabels into multiple assets
+        expanded_assets = self._expand_assets_with_multiple_labels(clean_assets)
         try:
             self._save_assets_export(
-                clean_assets, self.export_root_folder, annotation_modifier=self.annotation_modifier
+                expanded_assets,
+                self.export_root_folder,
+                annotation_modifier=self.annotation_modifier,
             )
         except ImportError as e:
             raise ImportError("Install with `pip install kili[coco]` to use this feature.") from e
@@ -62,6 +66,43 @@ def process_and_save(self, assets: list[dict], output_filename: Path):
 
         self.logger.warning(output_filename)
 
+    def _expand_assets_with_multiple_labels(self, assets: list[dict]) -> list[dict]:
+        """Expand assets with latestLabels into multiple asset entries with latestLabel.
+
+        When an asset has multiple labels (latestLabels), create separate asset entries
+        for each label with a unique externalId suffix (_label0, _label1, etc.).
+        """
+        expanded_assets = []
+        for asset in assets:
+            # Collect all labels to process (handle both latestLabel and latestLabels)
+            labels_to_process = []
+            if "latestLabel" in asset and asset["latestLabel"]:
+                labels_to_process.append(asset["latestLabel"])
+            if "latestLabels" in asset and asset["latestLabels"]:
+                for label in asset["latestLabels"]:
+                    if label is not None:
+                        labels_to_process.append(label)
+
+            if not labels_to_process:
+                continue
+
+            # Create asset copy for each label
+            for label_idx, latest_label in enumerate(labels_to_process, start=1):
+                asset_copy = asset.copy()
+                # Add label suffix if we have multiple labels
+                if len(labels_to_process) > 1:
+                    label_suffix = f"_label{label_idx}"
+                    asset_copy["externalId"] = f"{asset['externalId']}{label_suffix}"
+
+                # Set latestLabel and remove latestLabels
+                asset_copy["latestLabel"] = latest_label
+                if "latestLabels" in asset_copy:
+                    del asset_copy["latestLabels"]
+
+                expanded_assets.append(asset_copy)
+
+        return expanded_assets
+
     def _save_assets_export(
         self,
         assets: list[dict],

src/kili/services/export/format/geojson/__init__.py

@@ -99,10 +99,27 @@ def _process_asset(
     json_interface: dict | None = None,
     flatten_properties: bool = False,
 ) -> None:
-    geojson_feature_collection = convert_from_kili_to_geojson_format(
-        asset["latestLabel"]["jsonResponse"], json_interface, flatten_properties
-    )
-    filepath = labels_folder / f"{asset['externalId']}.geojson"
-    filepath.parent.mkdir(parents=True, exist_ok=True)
-    with open(filepath, "w", encoding="utf-8") as file:
-        json.dump(geojson_feature_collection, file)
+    # Collect all labels to process (handle both latestLabel and latestLabels)
+    labels_to_process = []
+    if "latestLabel" in asset and asset["latestLabel"]:
+        labels_to_process.append(asset["latestLabel"])
+    if "latestLabels" in asset and asset["latestLabels"]:
+        for label in asset["latestLabels"]:
+            if label is not None:
+                labels_to_process.append(label)
+
+    if not labels_to_process:
+        return
+
+    # Process each label
+    for label_idx, latest_label in enumerate(labels_to_process, start=1):
+        # Add label suffix if we have multiple labels
+        label_suffix = f"_label{label_idx}" if len(labels_to_process) > 1 else ""
+
+        geojson_feature_collection = convert_from_kili_to_geojson_format(
+            latest_label["jsonResponse"], json_interface, flatten_properties
+        )
+        filepath = labels_folder / f"{asset['externalId']}{label_suffix}.geojson"
+        filepath.parent.mkdir(parents=True, exist_ok=True)
+        with open(filepath, "w", encoding="utf-8") as file:
+            json.dump(geojson_feature_collection, file)

src/kili/services/export/format/voc/__init__.py

@@ -74,60 +74,98 @@ def process_and_save(self, assets: list[dict], output_filename: Path) -> None:
 
 
 # pylint: disable=too-many-locals
-def _process_asset(
-    asset: dict, labels_folder: Path, project_input_type: str, valid_jobs: Sequence[str]
+def _process_video_asset(
+    asset: dict,
+    latest_label: dict,
+    labels_folder: Path,
+    label_suffix: str,
+    valid_jobs: Sequence[str],
 ) -> None:
-    """Process an asset."""
-    if project_input_type == "VIDEO":
-        nbr_frames = len(asset.get("latestLabel", {}).get("jsonResponse", {}))
-        if nbr_frames < 1:
-            return
-        leading_zeros = len(str(nbr_frames))
-
-        width = height = 0
-        frame_ext = ""
-        # jsonContent with frames
-        if isinstance(asset["jsonContent"], list) and Path(asset["jsonContent"][0]).is_file():
-            width, height = get_frame_dimensions(asset)
-            frame_ext = Path(asset["jsonContent"][0]).suffix
-
-        # video with shouldUseNativeVideo set to True (no frames available)
-        elif Path(asset["content"]).is_file():
-            try:
-                width, height = get_video_dimensions(asset)
-                cut_video(asset["content"], asset, leading_zeros, Path(asset["content"]).parent)
-                frame_ext = ".jpg"
-            except ImportError as e:
-                raise ImportError(
-                    "Install with `pip install kili[video]` to use this feature."
-                ) from e
-
-        else:
-            raise FileNotFoundError(f"Could not find frames or video for asset {asset}")
-
-        for frame_id, json_response in asset["latestLabel"]["jsonResponse"].items():
-            frame_name = f'{asset["externalId"]}_{str(int(frame_id)+1).zfill(leading_zeros)}'
-            parameters = {"filename": f"{frame_name}{frame_ext}"}
-            annotations = convert_from_kili_to_voc_format(
-                json_response, width, height, parameters, valid_jobs
-            )
-            filepath = labels_folder / f"{frame_name}.xml"
-            filepath.parent.mkdir(parents=True, exist_ok=True)
-            with open(filepath, "wb") as fout:
-                fout.write(f"{annotations}\n".encode())
-
-    elif project_input_type == "IMAGE":
-        json_response = asset["latestLabel"]["jsonResponse"]
-        width, height = get_image_dimensions(asset)
-        filename = (
-            Path(asset["content"]).name if Path(asset["content"]).is_file() else asset["externalId"]
+    """Process a video asset and save annotations."""
+    nbr_frames = len(latest_label.get("jsonResponse", {}))
+    if nbr_frames < 1:
+        return
+    leading_zeros = len(str(nbr_frames))
+
+    width = height = 0
+    frame_ext = ""
+    # jsonContent with frames
+    if isinstance(asset["jsonContent"], list) and Path(asset["jsonContent"][0]).is_file():
+        width, height = get_frame_dimensions(asset)
+        frame_ext = Path(asset["jsonContent"][0]).suffix
+
+    # video with shouldUseNativeVideo set to True (no frames available)
+    elif Path(asset["content"]).is_file():
+        try:
+            width, height = get_video_dimensions(asset)
+            cut_video(asset["content"], asset, leading_zeros, Path(asset["content"]).parent)
+            frame_ext = ".jpg"
+        except ImportError as e:
+            raise ImportError("Install with `pip install kili[video]` to use this feature.") from e
+
+    else:
+        raise FileNotFoundError(f"Could not find frames or video for asset {asset}")
+
+    for frame_id, json_response in latest_label["jsonResponse"].items():
+        frame_name = (
+            f'{asset["externalId"]}_{str(int(frame_id)+1).zfill(leading_zeros)}{label_suffix}'
         )
-        parameters = {"filename": filename}
+        parameters = {"filename": f"{frame_name}{frame_ext}"}
         annotations = convert_from_kili_to_voc_format(
             json_response, width, height, parameters, valid_jobs
         )
-        xml_filename = f'{asset["externalId"]}.xml'
-        filepath = labels_folder / xml_filename
+        filepath = labels_folder / f"{frame_name}.xml"
         filepath.parent.mkdir(parents=True, exist_ok=True)
         with open(filepath, "wb") as fout:
             fout.write(f"{annotations}\n".encode())
+
+
+def _process_image_asset(
+    asset: dict,
+    latest_label: dict,
+    labels_folder: Path,
+    label_suffix: str,
+    valid_jobs: Sequence[str],
+) -> None:
+    """Process an image asset and save annotations."""
+    json_response = latest_label["jsonResponse"]
+    width, height = get_image_dimensions(asset)
+    filename = (
+        Path(asset["content"]).name if Path(asset["content"]).is_file() else asset["externalId"]
+    )
+    parameters = {"filename": filename}
+    annotations = convert_from_kili_to_voc_format(
+        json_response, width, height, parameters, valid_jobs
+    )
+    xml_filename = f'{asset["externalId"]}{label_suffix}.xml'
+    filepath = labels_folder / xml_filename
+    filepath.parent.mkdir(parents=True, exist_ok=True)
+    with open(filepath, "wb") as fout:
+        fout.write(f"{annotations}\n".encode())
+
+
+def _process_asset(
+    asset: dict, labels_folder: Path, project_input_type: str, valid_jobs: Sequence[str]
+) -> None:
+    """Process an asset."""
+    # Collect all labels to process (handle both latestLabel and latestLabels)
+    labels_to_process = []
+    if "latestLabel" in asset and asset["latestLabel"]:
+        labels_to_process.append(asset["latestLabel"])
+    if "latestLabels" in asset and asset["latestLabels"]:
+        for label in asset["latestLabels"]:
+            if label is not None:
+                labels_to_process.append(label)
+
+    if not labels_to_process:
+        return
+
+    # Process each label
+    for label_idx, latest_label in enumerate(labels_to_process, start=1):
+        # Add label suffix if we have multiple labels
+        label_suffix = f"_label{label_idx}" if len(labels_to_process) > 1 else ""
+
+        if project_input_type == "VIDEO":
+            _process_video_asset(asset, latest_label, labels_folder, label_suffix, valid_jobs)
+        elif project_input_type == "IMAGE":
+            _process_image_asset(asset, latest_label, labels_folder, label_suffix, valid_jobs)