feat(evaluation): add support of evaluation

docs/apis/evaluation/dataset.md

@@ -0,0 +1,3 @@
+<!-- prettier-ignore-start -->
+::: t4_devkit.evaluation.dataset
+<!-- prettier-ignore-end -->

docs/apis/evaluation/index.md

@@ -0,0 +1,7 @@
+# `evaluation`
+
+- [Evaluation Tasks](./task.md)
+- [Load Dataset](./dataset.md)
+- [Matching Boxes](./matching.md)
+- [Matching Results](./result.md)
+- [Metrics](./metric.md)

docs/apis/evaluation/matching.md

@@ -0,0 +1,9 @@
+<!-- prettier-ignore-start -->
+::: t4_devkit.evaluation.matching.parameter
+
+::: t4_devkit.evaluation.matching.scorer
+
+::: t4_devkit.evaluation.matching.policy
+
+::: t4_devkit.evaluation.matching.algorithm
+<!-- prettier-ignore-end -->

docs/apis/evaluation/metric.md

@@ -0,0 +1,5 @@
+<!-- prettier-ignore-start -->
+::: t4_devkit.evaluation.metric.ap
+
+::: t4_devkit.evaluation.metric.clear
+<!-- prettier-ignore-end -->

docs/apis/evaluation/result.md

@@ -0,0 +1,5 @@
+<!-- prettier-ignore-start -->
+::: t4_devkit.evaluation.result.box
+
+::: t4_devkit.evaluation.result.status
+<!-- prettier-ignore-end -->

docs/apis/evaluation/task.md

@@ -0,0 +1,3 @@
+<!-- prettier-ignore-start -->
+::: t4_devkit.evaluation.task
+<!-- prettier-ignore-end -->

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
@@ -33,6 +34,13 @@ nav:
           - Serialize Schema: apis/schema/serialize.md
       - t4_devkit.dataclass: apis/dataclass.md
       - t4_devkit.filtering: apis/filtering.md
+      - t4_devkit.evaluation:
+          - Home: apis/evaluation/index.md
+          - Evaluation Tasks: apis/evaluation/task.md
+          - Load Dataset: apis/evaluation/dataset.md
+          - Matching Boxes: apis/evaluation/matching.md
+          - Matching Results: apis/evaluation/result.md
+          - Metrics: apis/evaluation/metric.md
       - t4_devkit.viewer: apis/viewer.md
       - t4_devkit.typing: apis/typing.md
       - t4_devkit.common: apis/common.md

t4_devkit/dataclass/box.py

@@ -231,6 +231,33 @@ def corners(self, box_scale: float = 1.0) -> NDArrayF64:
         # Rotate and translate
         return np.dot(self.rotation.rotation_matrix, corners).T + self.position
 
+    def diff_yaw(self, other: Box3D) -> float:
+        """Return the yaw difference between the two boxes.
+
+        Args:
+            other (Box3D): Another box.
+
+        Raises:
+            ValueError: Both boxes must have the same `frame_id`.
+
+        Returns:
+            Yaw difference in the range of [-pi, pi].
+        """
+        if self.frame_id != other.frame_id:
+            raise ValueError(f"Invalid frame comparison: {self.frame_id=} and {other.frame_id=}")
+
+        yaw1, *_ = self.rotation.yaw_pitch_roll
+        yaw2, *_ = other.rotation.yaw_pitch_roll
+
+        def _clip(diff: float) -> float:
+            if diff < -np.pi:
+                diff += 2 * np.pi
+            elif diff > np.pi:
+                diff -= 2 * np.pi
+            return diff
+
+        return _clip(yaw2 - yaw1)
+
 
 @define(eq=False)
 class Box2D(BaseBox):

t4_devkit/evaluation/__init__.py

@@ -0,0 +1,9 @@
+from .dataset import *  # noqa
+from .matching import *  # noqa
+from .result import *  # noqa
+from .matching import *  # noqa
+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

@@ -0,0 +1,137 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, TypeVar
+
+from attrs import define
+
+from t4_devkit import Tier4
+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.schema import EgoPose, Sensor
+
+
+EvaluationObjectLike = TypeVar(
+    "EvaluationObjectLike",
+    list[BoxLike],  # boxes
+    SegmentationPointCloud,  # pointcloud
+    NDArrayU8,  # mask
+)
+
+
+__all__ = ["load_dataset", "FrameGroundTruth", "SceneGroundTruth"]
+
+
+def load_dataset(data_root: str, task: EvaluationTask) -> SceneGroundTruth:
+    """Load dataset.
+
+    Args:
+        data_root (str): Root directory path to the dataset.
+        task (EvaluationTask): Evaluation task.
+
+    Returns:
+        SceneGroundTruth: Loaded container of ground truths.
+    """
+    t4 = Tier4(data_root, verbose=False)
+
+    frames: list[FrameGroundTruth] = []
+    for i, sample in enumerate(t4.sample):
+        # annotations
+        if task.is_segmentation():
+            # TODO(ktro2828): add support of segmentation object
+            raise NotImplementedError("Segmentation task is under construction.")
+        else:
+            # TODO(ktro2828): add support of prediction future
+            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)
+        ego2map = HomogeneousMatrix(
+            position=ego_pose.translation,
+            rotation=ego_pose.rotation,
+            src="base_link",
+            dst="map",
+        )
+
+        frames.append(
+            FrameGroundTruth(
+                unix_time=sample.timestamp,
+                frame_index=i,
+                annotations=annotations,
+                ego2map=ego2map,
+            )
+        )
+
+    # transformation matrices from ego to each sensor
+    ego2sensors = TransformBuffer()
+    for cs_record in t4.calibrated_sensor:
+        sensor: Sensor = t4.get("sensor", cs_record.sensor_token)
+        matrix = HomogeneousMatrix(
+            position=cs_record.translation,
+            rotation=cs_record.rotation,
+            src="base_link",
+            dst=sensor.channel,
+        )
+
+        ego2sensors.set_transform(matrix)
+
+    return SceneGroundTruth(data_root=data_root, frames=frames, ego2sensors=ego2sensors)
+
+
+def _closest_ego_pose(t4: Tier4, timestamp: int) -> EgoPose:
+    """Lookup the ego pose record at the closest timestamp."""
+    return min(t4.ego_pose, key=lambda e: abs(e.timestamp - timestamp))
+
+
+@define
+class FrameGroundTruth:
+    """A container of boxes at a single frame.
+
+    Attributes:
+        unix_time (int): Unix timestamp.
+        frame_index (int): Index number of the frame.
+        annotations (EvaluationObjectLike): Set of ground truth instances.
+        ego2map (HomogeneousMatrix): Transformation matrix from ego to map coordinate.
+    """
+
+    unix_time: int
+    frame_index: int
+    annotations: EvaluationObjectLike
+    ego2map: HomogeneousMatrix
+
+
+@define
+class SceneGroundTruth:
+    """A container of frame ground truths.
+
+    Attributes:
+        data_root (str): Root directory path to the dataset.
+        frames (list[FrameGroundTruth]): List of frame ground truths.
+        ego2sensors (TransformBuffer): Buffer of transformation matrices from ego to each sensor coordinates.
+    """
+
+    data_root: str
+    frames: list[FrameGroundTruth]
+    ego2sensors: TransformBuffer
+
+    def lookup_frame(self, unix_time: int, tolerance: int) -> FrameGroundTruth | None:
+        """Lookup the closest set of ground truth frame.
+
+        Return None if the minimum time difference exceeds `tolerance`.
+
+        Args:
+            unix_time (int): Unix timestamp.
+            tolerance (int): Time difference tolerance in micro seconds.
+
+        Returns:
+            Return frame ground truth if succeeded, otherwise None.
+        """
+        closest = min(self.frames, key=lambda f: abs(unix_time - f.unix_time))
+        return closest if abs(unix_time - closest.unix_time) <= tolerance else None

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/matching/__init__.py

@@ -0,0 +1,4 @@
+from .algorithm import *  # noqa
+from .parameter import *  # noqa
+from .policy import *  # noqa
+from .scorer import *  # noqa