feat: add T4 dataset updater with the specifeid FastLabel annotations

config/update_t4_with_fastlabel_sample.yaml

@@ -0,0 +1,22 @@
+task: update_t4_with_fastlabel
+description:
+  visibility:
+    full: "No occlusion of the object."
+    most: "Object is occluded, but by less than 50%."
+    partial: "The object is occluded by more than 50% (but not completely)."
+    none: "The object is 90-100% occluded and no points/pixels are visible in the label."
+  camera_index:
+    CAM_FRONT: 0
+    CAM_FRONT_RIGHT: 1
+    CAM_BACK_RIGHT: 2
+    CAM_BACK: 3
+    CAM_BACK_LEFT: 4
+    CAM_FRONT_LEFT: 5
+
+conversion:
+  input_base: ./data/input_t4_format # could be non_annotated_t4_format or t4_format_3d_annotated
+  input_anno_base: ./data/fastlabel
+  output_base: ./data/output_t4_format # currently, this only includes the 2D annotations
+  dataset_corresponding:
+    # input t4dataset_name: FastLabel json file name
+    T4DatasetName: FastLabelAnnotationFile

perception_dataset/convert.py

@@ -370,6 +370,33 @@ def main():
         converter.convert()
         logger.info(f"[END] Converting Fastlabel data ({input_base}) to T4 data ({output_base})")
 
+    elif task == "update_t4_with_fastlabel":
+        from perception_dataset.fastlabel_to_t4.fastlabel_2d_to_t4_updater import (
+            FastLabel2dToT4Updater,
+        )
+
+        input_base = config_dict["conversion"]["input_base"]
+        output_base = config_dict["conversion"]["output_base"]
+        input_anno_base = config_dict["conversion"]["input_anno_base"]
+        dataset_corresponding = config_dict["conversion"]["dataset_corresponding"]
+        description = config_dict["description"]
+
+        converter = FastLabel2dToT4Updater(
+            input_base=input_base,
+            output_base=output_base,
+            input_anno_base=input_anno_base,
+            dataset_corresponding=dataset_corresponding,
+            overwrite_mode=args.overwrite,
+            description=description,
+        )
+        logger.info(
+            f"[BEGIN] Updating T4 dataset ({input_base}) with FastLabel {input_anno_base} into T4 data ({output_base})"
+        )
+        converter.convert()
+        logger.info(
+            f"[DONE] Updating T4 dataset ({input_base}) with FastLabel {input_anno_base} into T4 data ({output_base})"
+        )
+
     elif task == "merge_2d_t4dataset_to_3d":
         from perception_dataset.t4_dataset.t4_dataset_2d3d_merger import T4dataset2D3DMerger
 

perception_dataset/fastlabel_to_t4/fastlabel_2d_to_t4_converter.py

@@ -1,16 +1,21 @@
+import base64
 from collections import defaultdict
 import json
 import os.path as osp
 from pathlib import Path
 import shutil
 from typing import Any, Dict, List, Optional, Union
 
+import pycocotools.mask as cocomask
+
 from perception_dataset.deepen.deepen_to_t4_converter import DeepenToT4Converter
 from perception_dataset.t4_dataset.annotation_files_generator import AnnotationFilesGenerator
 from perception_dataset.utils.logger import configure_logger
 
 logger = configure_logger(modname=__name__)
 
+Points2DLike = list[list[list[float]]]
+
 
 class FastLabel2dToT4Converter(DeepenToT4Converter):
     def __init__(
@@ -146,18 +151,14 @@ def _format_fastlabel_annotation(self, annotations: Dict[str, List[Dict[str, Any
                     ]
                 }
                 ],
-                "annotations": [
-                {
-                    "points": [
+                "points": [
                     1221.25,
                     488.44,
                     1275.38,
                     570.47
-                    ],
-                    "rotation": 0,
-                    "autogenerated": false
-                }
-                ]
+                ],
+                "rotation": 0,
+                "autogenerated": false
             },
         },
         ....
@@ -177,6 +178,9 @@ def _format_fastlabel_annotation(self, annotations: Dict[str, List[Dict[str, Any
                 if dataset_name not in fl_annotations:
                     fl_annotations[dataset_name] = defaultdict(list)
 
+                width: int = ann["width"]
+                height: int = ann["height"]
+
                 for a in ann["annotations"]:
                     occlusion_state: str = "occlusion_state.none"
                     visibility: str = "Not available"
@@ -199,12 +203,55 @@ def _format_fastlabel_annotation(self, annotations: Dict[str, List[Dict[str, Any
                         "attribute_names": [occlusion_state],
                         "visibility_name": visibility,
                     }
-                    label_t4_dict.update(
-                        {
-                            "two_d_box": a["annotations"][0]["points"],
-                            "sensor_id": self._camera2idx[camera],
-                        }
-                    )
+                    if a["type"] == "bbox":
+                        label_t4_dict.update(
+                            {
+                                "two_d_box": a["points"],
+                                "sensor_id": self._camera2idx[camera],
+                            }
+                        )
+                    elif a["type"] == "segmentation":
+                        label_t4_dict.update(
+                            {
+                                "two_d_segmentation": _rle_from_points(a["points"], width, height),
+                                "sensor_id": self._camera2idx[camera],
+                            }
+                        )
                     fl_annotations[dataset_name][file_id].append(label_t4_dict)
 
         return fl_annotations
+
+
+def _rle_from_points(points: Points2DLike, width: int, height: int) -> Dict[str, Any]:
+    """Encode points to RLE format mask.
+
+    Points format of 2D segmentation in FastLabel:
+    ```
+    "points": [
+        [
+            [...], // outer points (1)
+            [...], // hollowed out points (1)
+        ],
+        [
+            [...], // outer points (2)
+            [...], // hollowed out points (2)
+        ]
+    ],
+    ```
+
+    Args:
+        points (Points2DLike): 2D points, such as `[[[o1, o1, o2, o2, ...], [ho1, ho1, ho2, ho2, ...], ...]]`.
+        width (int): Image width.
+        height (int): Image height.
+
+    Returns:
+        Dict[str, Any]: RLE format mask.
+    """
+    flattened = [[coord for p in point for coord in p] for point in points]
+
+    rle_objects = cocomask.frPyObjects(flattened, height, width)
+    rle = cocomask.merge(rle_objects)
+
+    rle["counts"] = base64.b64encode(rle["counts"]).decode("ascii")
+
+    return rle

perception_dataset/fastlabel_to_t4/fastlabel_2d_to_t4_updater.py

@@ -0,0 +1,97 @@
+from __future__ import annotations
+
+import json
+import os.path as osp
+from pathlib import Path
+import shutil
+from typing import Dict
+
+from perception_dataset.fastlabel_to_t4.fastlabel_2d_to_t4_converter import (
+    FastLabel2dToT4Converter,
+)
+from perception_dataset.t4_dataset.annotation_files_updater import AnnotationFilesUpdater
+from perception_dataset.utils.logger import configure_logger
+
+logger = configure_logger(modname=__name__)
+
+
+# TODO: add support of 3D annotation format
+class FastLabel2dToT4Updater(FastLabel2dToT4Converter):
+    def __init__(
+        self,
+        input_base: str,
+        output_base: str,
+        input_anno_base: str,
+        dataset_corresponding: Dict[str, int],
+        overwrite_mode: bool,
+        description: Dict[str, Dict[str, str]],
+    ):
+        super().__init__(
+            input_base,
+            output_base,
+            input_anno_base,
+            dataset_corresponding,
+            overwrite_mode,
+            description,
+            input_bag_base=None,
+            topic_list=None,
+        )
+
+    def convert(self) -> None:
+        anno_jsons_dict = self._load_annotation_jsons()
+        fl_annotations = self._format_fastlabel_annotation(anno_jsons_dict)
+
+        for t4dataset_name in self._t4dataset_name_to_merge:
+            # Check if input directory exists
+            input_dir = self._input_base / t4dataset_name
+            input_annotation_dir = input_dir / "annotation"
+            if not osp.exists(input_annotation_dir):
+                logger.warning(f"input_dir {input_dir} not exists.")
+                continue
+
+            # Check if output directory already exists
+            output_dir = self._output_base / t4dataset_name / "t4_dataset"
+            if self._input_bag_base is not None:
+                input_bag_dir = Path(self._input_bag_base) / t4dataset_name
+
+            if osp.exists(output_dir):
+                logger.error(f"{output_dir} already exists.")
+                is_dir_exist = True
+            else:
+                is_dir_exist = False
+
+            if self._overwrite_mode or not is_dir_exist:
+                # Remove existing output directory
+                shutil.rmtree(output_dir, ignore_errors=True)
+                # Copy input data to output directory
+                self._copy_data(input_dir, output_dir)
+                # Make rosbag
+                if self._input_bag_base is not None and not osp.exists(
+                    osp.join(output_dir, "input_bag")
+                ):
+                    self._find_start_end_time(input_dir)
+                    self._make_rosbag(str(input_bag_dir), str(output_dir))
+            else:
+                raise ValueError("If you want to overwrite files, use --overwrite option.")
+
+            # Start updating annotations
+            annotation_files_updater = AnnotationFilesUpdater(description=self._description)
+            annotation_files_updater.convert_one_scene(
+                input_dir=input_dir,
+                output_dir=output_dir,
+                scene_anno_dict=fl_annotations[t4dataset_name],
+                dataset_name=t4dataset_name,
+            )
+
+    def _load_annotation_jsons(self):
+        anno_dict = {}
+        for file in self._input_anno_files:
+            t4_dataset_name = None
+            for name, ann_filename in self._t4dataset_name_to_merge.items():
+                if ann_filename == file.name:
+                    t4_dataset_name = name
+
+            assert t4_dataset_name is not None
+            with open(file) as f:
+                anno_dict[t4_dataset_name] = json.load(f)
+        return anno_dict

perception_dataset/t4_dataset/annotation_files_updater.py

@@ -0,0 +1,94 @@
+import json
+import os.path as osp
+from typing import Any, Dict
+
+from perception_dataset.t4_dataset.annotation_files_generator import AnnotationFilesGenerator
+from perception_dataset.t4_dataset.classes import (
+    AttributeTable,
+    CategoryTable,
+    InstanceTable,
+    ObjectAnnTable,
+    SampleAnnotationTable,
+    SurfaceAnnTable,
+    VisibilityTable,
+)
+
+
+def _load_json(filepath: str) -> Any:
+    with open(filepath) as f:
+        data = json.load(f)
+    return data
+
+
+class AnnotationFilesUpdater(AnnotationFilesGenerator):
+    def __init__(self, with_camera: bool = True, description: Dict[str, Dict[str, str]] = ...):
+        super().__init__(with_camera, description)
+        self.description = description
+
+    def convert_one_scene(
+        self,
+        input_dir: str,
+        output_dir: str,
+        scene_anno_dict: dict[str, list[dict[str, Any]]],
+        dataset_name: str,
+    ) -> None:
+        anno_dir = osp.join(input_dir, "annotation")
+        if not osp.exists(anno_dir):
+            raise ValueError(f"Annotations files doesn't exist in {anno_dir}")
+
+        # Load existence annotation files
+        self._init_table_from_json(anno_dir=anno_dir)
+
+        super().convert_one_scene(
+            input_dir=input_dir,
+            output_dir=output_dir,
+            scene_anno_dict=scene_anno_dict,
+            dataset_name=dataset_name,
+        )
+
+    def _init_table_from_json(self, anno_dir: str) -> None:
+        self._attribute_table = AttributeTable.from_json(
+            filepath=osp.join(anno_dir, AttributeTable.FILENAME),
+            name_to_description={},
+            default_value="",
+        )
+
+        self._category_table = CategoryTable.from_json(
+            filepath=osp.join(anno_dir, CategoryTable.FILENAME),
+            name_to_description={},
+            default_value="",
+        )
+
+        self._instance_table = InstanceTable.from_json(
+            filepath=osp.join(anno_dir, InstanceTable.FILENAME)
+        )
+
+        self._visibility_table = VisibilityTable.from_json(
+            filepath=osp.join(anno_dir, VisibilityTable.FILENAME),
+            level_to_description=self.description.get(
+                "visibility",
+                {
+                    "v0-40": "visibility of whole object is between 0 and 40%",
+                    "v40-60": "visibility of whole object is between 40 and 60%",
+                    "v60-80": "visibility of whole object is between 60 and 80%",
+                    "v80-100": "visibility of whole object is between 80 and 100%",
+                    "none": "visibility isn't available",
+                },
+            ),
+            default_value="",
+        )
+
+        if osp.exists(osp.join(anno_dir, SampleAnnotationTable.FILENAME)):
+            self._sample_annotation_table = SampleAnnotationTable.from_json(
+                osp.join(anno_dir, SampleAnnotationTable.FILENAME)
+            )
+
+        if osp.exists(osp.join(anno_dir, ObjectAnnTable.FILENAME)):
+            self._object_ann_table = ObjectAnnTable.from_json(
+                osp.join(anno_dir, ObjectAnnTable.FILENAME)
+            )
+
+        if osp.exists(osp.join(anno_dir, SurfaceAnnTable.FILENAME)):
+            self._surface_ann_table = SurfaceAnnTable.from_json(
+                osp.join(anno_dir, SurfaceAnnTable.FILENAME)
+            )

perception_dataset/t4_dataset/classes/abstract_class.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from abc import ABCMeta, abstractmethod
 import json
 import os.path as osp
@@ -14,6 +16,10 @@ def __init__(self):
     def token(self) -> str:
         return self._token
 
+    @token.setter
+    def token(self, token: str):
+        self._token = token
+
     @abstractmethod
     def to_dict(self) -> Dict[str, Any]:
         raise NotImplementedError()
@@ -71,3 +77,8 @@ def save_json(self, output_dir: str):
         table_data = self.to_data()
         with open(osp.join(output_dir, self.FILENAME), "w") as f:
             json.dump(table_data, f, indent=4)
+
+    @classmethod
+    @abstractmethod
+    def from_json(cls, filepath: str):
+        raise NotImplementedError()