feat: add evaluator

Signed-off-by: ktro2828 <kotaro.uetake@tier4.jp>

Author: ktro2828

docs/tutorials/evaluation.md

@@ -0,0 +1,16 @@
+## Perception Evaluation
+
+```python
+from t4_devkit.evaluation import PerceptionEvaluator, PerceptionEvaluationConfig, EvaluationTask
+
+config = PerceptionEvaluationConfig(
+    dataset="<PATH_TO_DATASET>",
+    task=EvaluationTask.<TASK_ENUM>,
+)
+
+evaluator = PerceptionEvaluator(config)
+```
+
+## Sensing Evaluation
+
+TBD

mkdocs.yaml

@@ -20,6 +20,7 @@ nav:
       - Home: cli/index.md
       - t4viz: cli/t4viz.md
       - t4sanity: cli/t4sanity.md
+  - Evaluation: tutorials/evaluation.md
   - API References:
       - t4_devkit.tier4: apis/tier4.md
       - t4_devkit.helper: apis/helper.md

t4_devkit/evaluation/__init__.py

@@ -5,3 +5,5 @@
 from .result import *  # noqa
 from .metric import *  # noqa
 from .task import *  # noqa
+from .config import *  # noqa
+from .evaluator import *  # noqa

t4_devkit/evaluation/config.py

@@ -0,0 +1,20 @@
+from __future__ import annotations
+
+from attrs import define, field
+
+from t4_devkit.filtering import FilterParams
+
+from .matching import MatchingParams
+from .task import EvaluationTask
+
+__all__ = ["PerceptionEvaluationConfig"]
+
+
+@define
+class PerceptionEvaluationConfig:
+    """Evaluation configuration for perception tasks."""
+
+    dataset: str
+    task: EvaluationTask = field(converter=EvaluationTask)
+    filtering: FilterParams = field(default=FilterParams())
+    matching: MatchingParams = field(default=MatchingParams())

t4_devkit/evaluation/dataset.py

@@ -1,19 +1,27 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, TypeVar
 
 from attrs import define
 
 from t4_devkit import Tier4
-from t4_devkit.dataclass import HomogeneousMatrix, TransformBuffer
+from t4_devkit.dataclass import BoxLike, HomogeneousMatrix, SegmentationPointCloud, TransformBuffer
+from t4_devkit.typing import NDArrayU8
 
 from .task import EvaluationTask
 
 if TYPE_CHECKING:
-    from t4_devkit.dataclass import BoxLike
     from t4_devkit.schema import EgoPose, Sensor
 
 
+EvaluationObjectLike = TypeVar(
+    "EvaluationObjectLike",
+    list[BoxLike],  # boxes
+    SegmentationPointCloud,  # pointcloud
+    NDArrayU8,  # mask
+)
+
+
 __all__ = ["load_dataset", "FrameGroundTruth", "SceneGroundTruth"]
 
 
@@ -31,12 +39,16 @@ def load_dataset(data_root: str, task: EvaluationTask) -> SceneGroundTruth:
 
     frames: list[FrameGroundTruth] = []
     for i, sample in enumerate(t4.sample):
-        # annotation boxes
-        boxes = (
-            list(map(t4.get_box3d, sample.ann_3ds))
-            if task.is_3d()
-            else list(map(t4.get_box2d, sample.ann_2ds))
-        )
+        # annotations
+        if task.is_segmentation():
+            # TODO(ktro2828): add support of segmentation object
+            raise NotImplementedError("Segmentation task is under construction.")
+        else:
+            annotations = (
+                list(map(t4.get_box3d, sample.ann_3ds))
+                if task.is_3d()
+                else list(map(t4.get_box2d, sample.ann_2ds))
+            )
 
         # transformation matrix from ego to map
         ego_pose = _closest_ego_pose(t4, sample.timestamp)
@@ -51,7 +63,7 @@ def load_dataset(data_root: str, task: EvaluationTask) -> SceneGroundTruth:
             FrameGroundTruth(
                 unix_time=sample.timestamp,
                 frame_index=i,
-                boxes=boxes,
+                annotations=annotations,
                 ego2map=ego2map,
             )
         )
@@ -84,13 +96,13 @@ class FrameGroundTruth:
     Attributes:
         unix_time (int): Unix timestamp.
         frame_index (int): Index number of the frame.
-        boxes (list[BoxLike]): List of ground truth instances.
+        annotations (EvaluationObjectLike): Set of ground truth instances.
         ego2map (HomogeneousMatrix): Transformation matrix from ego to map coordinate.
     """
 
     unix_time: int
     frame_index: int
-    boxes: list[BoxLike]
+    annotations: EvaluationObjectLike
     ego2map: HomogeneousMatrix
 
 

t4_devkit/evaluation/evaluator.py

@@ -0,0 +1,89 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from .dataset import load_dataset
+from .matching import build_matcher
+from .result import FrameBoxMatch
+
+if TYPE_CHECKING:
+    from t4_devkit.dataclass import BoxLike
+
+    from .config import PerceptionEvaluationConfig
+    from .dataset import EvaluationObjectLike, FrameGroundTruth
+
+
+class PerceptionEvaluator:
+    """Evaluation manager for perception tasks."""
+
+    def __init__(self, config: PerceptionEvaluationConfig) -> None:
+        self.config = config
+
+        self.scene_ground_truth = load_dataset(data_root=self.config.dataset, task=self.config.task)
+        self.matcher = build_matcher(params=self.config.matching)
+
+        self.frames: list[FrameBoxMatch] = []
+
+    def add_frame(
+        self,
+        unix_time: int,
+        estimations: EvaluationObjectLike,
+    ) -> FrameBoxMatch | None:
+        """Add frame result.
+
+        Returns None if it failed to find the closest timestamp ground truth.
+
+        Args:
+            unix_time (int): Current unix time.
+            estimations (EvaluationObjectLike): Set of estimations at the current frame.
+
+        Returns:
+            Return frame result only if it succeeded to find the closest timestamp ground truth,
+                otherwise None.
+        """
+        frame_ground_truth = self.scene_ground_truth.lookup_frame(
+            unix_time=unix_time,
+            tolerance=7500,
+        )
+
+        if frame_ground_truth is None:
+            return None
+
+        # TODO(ktro2828): add support of segmentation object
+        if self.config.task.is_segmentation():
+            raise NotImplementedError("Segmentation task is under construction.")
+        else:
+            frame = self._to_frame_box(
+                unix_time=unix_time,
+                estimations=estimations,
+                frame_ground_truth=frame_ground_truth,
+            )
+
+        self.frames.append(frame)
+
+        return frame
+
+    def _to_frame_box(
+        self,
+        unix_time: int,
+        estimations: list[BoxLike],
+        frame_ground_truth: FrameGroundTruth,
+    ) -> FrameBoxMatch:
+        """Match estimations to ground truths and convert to frame result.
+
+        Args:
+            unix_time (int): Current unix time associated with estimations.
+            estimations (list[BoxLike]): List of estimations.
+            frame_ground_truth (FrameGroundTruth): Frame ground truth.
+
+        Returns:
+            Frame result.
+        """
+        matches = self.matcher(estimations, frame_ground_truth.annotations)
+
+        return FrameBoxMatch(
+            unix_time=unix_time,
+            frame_index=frame_ground_truth.frame_index,
+            matches=matches,
+            ego2map=frame_ground_truth.ego2map,
+        )

t4_devkit/evaluation/task.py

@@ -1,25 +1,47 @@
 from __future__ import annotations
 
-from enum import Enum
+from enum import Enum, unique
 
 __all__ = ["EvaluationTask"]
 
 
+@unique
 class EvaluationTask(str, Enum):
     """Enumeration of evaluation tasks."""
 
     DETECTION3D = "detection3d"
     TRACKING3D = "tracking3d"
     PREDICTION3D = "prediction3d"
+    SEGMENTATION3D = "segmentation3d"
     DETECTION2D = "detection2d"
     TRACKING2D = "tracking2d"
+    SEGMENTATION2D = "segmentation2d"
 
     def is_3d(self) -> bool:
+        """Indicate whether the task considers 3D objects.
+
+        Returns:
+            Return `True` if the task is in [detection3d, tracking3d, prediction3d, segmentation3d].
+        """
         return self in (
             EvaluationTask.DETECTION3D,
             EvaluationTask.TRACKING3D,
             EvaluationTask.PREDICTION3D,
+            EvaluationTask.SEGMENTATION3D,
         )
 
     def is_2d(self) -> bool:
+        """Indicate whether the task considers 3D objects.
+
+        Returns:
+            Return `True` if the task is in [detection2d, tracking2d, segmentation2d].
+        """
         return self in (EvaluationTask.DETECTION2D, EvaluationTask.TRACKING2D)
+
+    def is_segmentation(self) -> bool:
+        """Indicate whether the task deals with segmentation.
+
+        Returns:
+            Return `True` if the task is in [segmentation3d, segmentation2d].
+        """
+        return self in (EvaluationTask.SEGMENTATION3D, EvaluationTask.SEGMENTATION2D)

tests/evaluation/test_task.py

@@ -4,7 +4,15 @@
 
 
 def test_task() -> None:
-    task_names = {"detection3d", "tracking3d", "prediction3d", "detection2d", "tracking2d"}
+    task_names = {
+        "detection3d",
+        "tracking3d",
+        "segmentation3d",
+        "prediction3d",
+        "detection2d",
+        "tracking2d",
+        "segmentation2d",
+    }
 
     assert task_names == {e.value for e in EvaluationTask}