feat: add FastLabel 3D to T4 format

config/convert_fastlabel_to_t4.yaml

@@ -0,0 +1,24 @@
+task: convert_fastlabel_to_t4
+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_NARROW: 0
+    CAM_FRONT_WIDE: 1
+    CAM_FRONT_RIGHT: 2
+    CAM_BACK_RIGHT: 3
+    CAM_BACK_NARROW: 4
+    CAM_BACK_WIDE: 5
+    CAM_BACK_LEFT: 6
+    CAM_FRONT_LEFT: 7
+
+conversion:
+  make_t4_dataset_dir: false # If true, the output directory includes t4_dataset directory (such as "scene_dir"/t4_dataset/data|annotation). If false, "scene_dir"/data|annotation.
+  input_base: ./data/non_annotated_t4_format
+  input_anno_base: ./data/fastlabel/pcd_annotation
+  output_base: ./data/t4_format
+  input_bag_base: null  #optional
+  topic_list: null  #necessary if input_bag_base is not null

config/label/object.yaml

@@ -26,6 +26,8 @@ pedestrian:
     pedestrian.adult,
     pedestrian.child,
   ]
+other_pedestrian: [other_pedestrian]
+other_vehicle: [other_vehicle]
 truck: [truck, TRUCK, vehicle.truck]
 trailer: [trailer, TRAILER, vehicle.trailer]
 ambulance: [ambulance, AMBULANCE, vehicle.ambulance]

perception_dataset/convert.py

@@ -387,6 +387,38 @@ def main():
         converter.convert()
         logger.info(f"[Done] Merging T4 dataset ({input_base}) into T4 dataset ({output_base})")
 
+    elif task == "convert_fastlabel_to_t4":
+        from perception_dataset.fastlabel_to_t4.fastlabel_to_t4_converter import (
+            FastLabelToT4Converter,
+        )
+
+        make_t4_dataset_dir = config_dict["conversion"]["make_t4_dataset_dir"]
+        input_base = config_dict["conversion"]["input_base"]
+        input_anno_base = config_dict["conversion"]["input_anno_base"]
+        output_base = config_dict["conversion"]["output_base"]
+        description = config_dict["description"]
+        input_bag_base = config_dict["conversion"]["input_bag_base"]
+        if input_bag_base is not None:
+            topic_list_yaml_path = config_dict["conversion"]["topic_list"]
+            with open(topic_list_yaml_path) as f:
+                topic_list_yaml = yaml.safe_load(f)
+        else:
+            topic_list_yaml = None
+
+        converter = FastLabelToT4Converter(
+            input_base=input_base,
+            output_base=output_base,
+            input_anno_base=input_anno_base,
+            overwrite_mode=args.overwrite,
+            description=description,
+            make_t4_dataset_dir=make_t4_dataset_dir,
+            input_bag_base=input_bag_base,
+            topic_list=topic_list_yaml,
+        )
+        logger.info(f"[BEGIN] Converting Fastlabel data ({input_base}) to T4 data ({output_base})")
+        converter.convert()
+        logger.info(f"[END] Converting Fastlabel data ({input_base}) to T4 data ({output_base})")
+
     else:
         raise NotImplementedError()
 

perception_dataset/fastlabel_to_t4/fastlabel_2d_to_t4_converter.py

@@ -88,12 +88,27 @@ def convert(self):
                 dataset_name=t4dataset_name,
             )
 
-    def _load_annotation_jsons(self):
+    def _load_annotation_jsons(
+        self, t4_datasets: Optional[List[str]] = None, delimiter: Optional[str] = None
+    ) -> Dict[str, List[dict[str, Any]]]:
         """Load annotations from all JSON files in the input directory and return as a dictionary."""
-        anno_dict = {}
+        anno_dict = defaultdict(list)
+        if t4_datasets is None:
+            for file in self._input_anno_files:
+                with open(file) as f:
+                    anno_dict[file.name] = json.load(f)
+                    return anno_dict
+
         for file in self._input_anno_files:
+            t4_dataset_name = file.name.split(delimiter)[0]
+            for dataset in t4_datasets:
+                if dataset in t4_dataset_name:
+                    break
+            else:
+                continue
             with open(file) as f:
-                anno_dict[file.name] = json.load(f)
+                one_label = json.load(f)
+                anno_dict[dataset].extend(one_label)
         return anno_dict
 
     def _format_fastlabel_annotation(self, annotations: Dict[str, List[Dict[str, Any]]]):

perception_dataset/fastlabel_to_t4/fastlabel_to_t4_converter.py

@@ -0,0 +1,227 @@
+from collections import defaultdict
+import os.path as osp
+from pathlib import Path
+import shutil
+from typing import Any, Dict, List, Optional, Union
+
+from perception_dataset.constants import LABEL_PATH_ENUM
+from perception_dataset.fastlabel_to_t4.fastlabel_2d_to_t4_converter import (
+    FastLabel2dToT4Converter,
+)
+from perception_dataset.t4_dataset.annotation_files_generator import AnnotationFilesGenerator
+from perception_dataset.t4_dataset.keyframe_consistency_resolver import KeyFrameConsistencyResolver
+from perception_dataset.utils.label_converter import LabelConverter
+from perception_dataset.utils.logger import configure_logger
+from perception_dataset.utils.transform import rotation_to_quaternion
+
+logger = configure_logger(modname=__name__)
+
+
+class FastLabelToT4Converter(FastLabel2dToT4Converter):
+    def __init__(
+        self,
+        input_base: str,
+        output_base: str,
+        input_anno_base: str,
+        overwrite_mode: bool,
+        description: Dict[str, Dict[str, str]],
+        make_t4_dataset_dir: bool = True,
+        input_bag_base: Optional[str] = None,
+        topic_list: Optional[Union[Dict[str, List[str]], List[str]]] = None,
+    ):
+        super().__init__(
+            input_base,
+            output_base,
+            input_anno_base,
+            None,
+            overwrite_mode,
+            description,
+            input_bag_base,
+            topic_list,
+        )
+        self._make_t4_dataset_dir = make_t4_dataset_dir
+        self._label_converter = LabelConverter(
+            label_path=LABEL_PATH_ENUM.OBJECT_LABEL,
+            attribute_path=LABEL_PATH_ENUM.ATTRIBUTE,
+        )
+
+    def convert(self) -> None:
+        t4_datasets = sorted([d.name for d in self._input_base.iterdir() if d.is_dir()])
+        anno_jsons_dict = self._load_annotation_jsons(t4_datasets, ".pcd")
+        fl_annotations = self._format_fastlabel_3d_annotation(anno_jsons_dict)
+
+        for t4dataset_name in t4_datasets:
+            # 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
+            if self._make_t4_dataset_dir:
+                output_dir = output_dir / "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.")
+
+            if t4dataset_name not in fl_annotations.keys():
+                logger.warning(f"No annotation for {t4dataset_name}")
+                continue
+
+            # Start updating annotations
+            annotation_files_generator = AnnotationFilesGenerator(description=self._description)
+            annotation_files_generator.convert_one_scene(
+                input_dir=input_dir,
+                output_dir=output_dir,
+                scene_anno_dict=fl_annotations[t4dataset_name],
+                dataset_name=t4dataset_name,
+            )
+
+            # fix non-keyframe (no-labeled frame) in t4 dataset
+            modifier = KeyFrameConsistencyResolver()
+            modifier.inspect_and_fix_t4_segment(Path(output_dir))
+
+    def _format_fastlabel_3d_annotation(self, annotations: Dict[str, List[Dict[str, Any]]]):
+        """
+        e.g. of input_anno_file(fastlabel):
+        [
+            {
+                "id": "675f15cb-f3c1-45df-b8e1-6daaa36402bd",
+                "name": "r7ZRDFWf_2024-08-21T15-13-16+0900_10/00011.pcd",
+                "status": "completed",
+                "externalStatus": "approved",
+                "url": "https://annotations.fastlabel.ai/workspaces/......",
+                "annotations": [
+                    {
+                        "id": "9feb60dc-6170-4c2c-95d7-165b7862d12a",
+                        "type": "cuboid",
+                        "title": "truck",
+                        "value": "truck",
+                        "color": "#D10069",
+                        "attributes": [
+                            {
+                                "type": "radio",
+                                "name": "status",
+                                "key": "status",
+                                "title": "approval",
+                                "value": "approval"
+                            },
+                            {
+                                "type": "radio",
+                                "name": "occlusion_state",
+                                "key": "occlusion_state",
+                                "title": "none",
+                                "value": "none"
+                            },
+                            {
+                                "type": "radio",
+                                "name": "vehicle_state",
+                                "key": "vehicle_state",
+                                "title": "driving",
+                                "value": "driving"
+                            }
+                        ],
+                        "points": [
+                            8.76,   // coordinate x
+                            3.87,   // coordinate y
+                            1.71,   // coordinate z
+                            0.00,   // rotation x
+                            0.00,   // rotation y
+                            -0.03,  // rotation z
+                            8.27,   // length x
+                            2.47,   // length y
+                            3.17    // length z
+                        ],
+                        "rotation": 0,
+                        "keypoints": [],
+                        "confidenceScore": -1
+                    },
+                    ....
+                ]
+            },
+            ...
+        ]
+        """
+        fl_annotations: Dict[str, Dict[int, List[Dict[str, Any]]]] = {}
+
+        for filename, ann_list in sorted(annotations.items()):
+            dataset_name: str = Path(filename).stem
+            for ann in ann_list:
+                filename: str = ann["name"].split("/")[-1]
+                file_id: int = int(filename.split(".")[0])
+
+                if dataset_name not in fl_annotations:
+                    fl_annotations[dataset_name] = defaultdict(list)
+
+                for a in ann["annotations"]:
+                    visibility: str = "Not available"
+                    instance_id = a["id"]
+
+                    if "attributes" not in a or a["attributes"] is None:
+                        logger.error(f"No attributes in {a}")
+                    attributes = []
+                    for att in a["attributes"]:
+                        if att["name"] == "status":
+                            continue
+                        attributes.append(f"{att['name'].lower()}.{att['value']}")
+                        if att["key"] == "occlusion_state":
+                            visibility = self._convert_occlusion_to_visibility(att["value"])
+                    category = self._label_converter.convert_label(a["title"])
+                    label_t4_dict: Dict[str, Any] = {
+                        "category_name": category,
+                        "instance_id": instance_id,
+                        "attribute_names": attributes,
+                        "visibility_name": visibility,
+                    }
+                    points = a["points"]
+                    q = rotation_to_quaternion(a["points"][3:6])
+                    label_t4_dict.update(
+                        {
+                            "three_d_bbox": {
+                                "translation": {
+                                    "x": points[0],
+                                    "y": points[1],
+                                    "z": points[2],
+                                },
+                                "velocity": None,
+                                "acceleration": None,
+                                "size": {
+                                    "length": points[6],
+                                    "width": points[7],
+                                    "height": points[8],
+                                },
+                                "rotation": {
+                                    "x": q[0],
+                                    "y": q[1],
+                                    "z": q[2],
+                                    "w": q[3],
+                                },
+                            },
+                            "num_lidar_pts": 0,
+                            "num_radar_pts": 0,
+                        }
+                    )
+                    fl_annotations[dataset_name][file_id].append(label_t4_dict)
+
+        return fl_annotations

perception_dataset/t4_dataset/classes/object_ann.py

@@ -1,4 +1,4 @@
-from typing import Dict, List
+from typing import Any, Dict, List
 
 from perception_dataset.constants import EXTENSION_ENUM
 from perception_dataset.t4_dataset.classes.abstract_class import AbstractRecord, AbstractTable
@@ -12,7 +12,7 @@ def __init__(
         category_token: str,
         attribute_tokens: str,
         bbox: List[float],
-        mask: Dict[str, any],
+        mask: Dict[str, Any],
     ):
         super().__init__()
 
@@ -23,7 +23,7 @@ def __init__(
         self._category_token: str = category_token
         self._attribute_tokens: List[str] = attribute_tokens
         self._bbox: List[float] = bbox
-        self._mask: Dict[str, any] = mask
+        self._mask: Dict[str, Any] = mask
 
     def to_dict(self):
         d = {
@@ -58,7 +58,7 @@ def _to_record(
         category_token: str,
         attribute_tokens: str,
         bbox: List[float],
-        mask: Dict[str, any],
+        mask: Dict[str, Any],
     ):
         record = ObjectAnnRecord(
             sample_data_token=sample_data_token,

perception_dataset/t4_dataset/classes/surface_ann.py

@@ -1,4 +1,4 @@
-from typing import Dict
+from typing import Any, Dict
 
 from perception_dataset.constants import EXTENSION_ENUM
 from perception_dataset.t4_dataset.classes.abstract_class import AbstractRecord, AbstractTable
@@ -8,13 +8,13 @@ class SurfaceAnnRecord(AbstractRecord):
     def __init__(
         self,
         category_token: str,
-        mask: Dict[str, any],
+        mask: Dict[str, Any],
         sample_data_token: str,
     ):
         super().__init__()
 
         self._category_token: str = category_token
-        self._mask: Dict[str, any] = mask
+        self._mask: Dict[str, Any] = mask
         self._sample_data_token: str = sample_data_token
 
     def to_dict(self):
@@ -38,7 +38,7 @@ def __init__(self):
     def _to_record(
         self,
         category_token: str,
-        mask: Dict[str, any],
+        mask: Dict[str, Any],
         sample_data_token: str,
     ):
         record = SurfaceAnnRecord(