diff --git a/docs/apis/evaluation/dataset.md b/docs/apis/evaluation/dataset.md new file mode 100644 index 00000000..52d3660c --- /dev/null +++ b/docs/apis/evaluation/dataset.md @@ -0,0 +1,3 @@ + +::: t4_devkit.evaluation.dataset + diff --git a/docs/apis/evaluation/index.md b/docs/apis/evaluation/index.md new file mode 100644 index 00000000..429ca69b --- /dev/null +++ b/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) diff --git a/docs/apis/evaluation/matching.md b/docs/apis/evaluation/matching.md new file mode 100644 index 00000000..2d628646 --- /dev/null +++ b/docs/apis/evaluation/matching.md @@ -0,0 +1,9 @@ + +::: t4_devkit.evaluation.matching.parameter + +::: t4_devkit.evaluation.matching.scorer + +::: t4_devkit.evaluation.matching.policy + +::: t4_devkit.evaluation.matching.algorithm + diff --git a/docs/apis/evaluation/metric.md b/docs/apis/evaluation/metric.md new file mode 100644 index 00000000..034a203a --- /dev/null +++ b/docs/apis/evaluation/metric.md @@ -0,0 +1,5 @@ + +::: t4_devkit.evaluation.metric.ap + +::: t4_devkit.evaluation.metric.clear + diff --git a/docs/apis/evaluation/result.md b/docs/apis/evaluation/result.md new file mode 100644 index 00000000..beee5a28 --- /dev/null +++ b/docs/apis/evaluation/result.md @@ -0,0 +1,5 @@ + +::: t4_devkit.evaluation.result.box + +::: t4_devkit.evaluation.result.status + diff --git a/docs/apis/evaluation/task.md b/docs/apis/evaluation/task.md new file mode 100644 index 00000000..da4912f8 --- /dev/null +++ b/docs/apis/evaluation/task.md @@ -0,0 +1,3 @@ + +::: t4_devkit.evaluation.task + diff --git a/docs/tutorials/evaluation.md b/docs/tutorials/evaluation.md new file mode 100644 index 00000000..037990ab --- /dev/null +++ b/docs/tutorials/evaluation.md @@ -0,0 +1,16 @@ +## Perception Evaluation + +```python +from t4_devkit.evaluation import PerceptionEvaluator, PerceptionEvaluationConfig, EvaluationTask + +config = PerceptionEvaluationConfig( + dataset="", + task=EvaluationTask., +) + +evaluator = PerceptionEvaluator(config) +``` + +## Sensing Evaluation + +TBD diff --git a/mkdocs.yaml b/mkdocs.yaml index 942f1899..4b473ea7 100644 --- a/mkdocs.yaml +++ b/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 diff --git a/t4_devkit/dataclass/box.py b/t4_devkit/dataclass/box.py index e1fd1ec4..52ef4d64 100644 --- a/t4_devkit/dataclass/box.py +++ b/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): diff --git a/t4_devkit/evaluation/__init__.py b/t4_devkit/evaluation/__init__.py new file mode 100644 index 00000000..57993890 --- /dev/null +++ b/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 diff --git a/t4_devkit/evaluation/config.py b/t4_devkit/evaluation/config.py new file mode 100644 index 00000000..448d9e7d --- /dev/null +++ b/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()) diff --git a/t4_devkit/evaluation/dataset.py b/t4_devkit/evaluation/dataset.py new file mode 100644 index 00000000..ab2b1366 --- /dev/null +++ b/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 diff --git a/t4_devkit/evaluation/evaluator.py b/t4_devkit/evaluation/evaluator.py new file mode 100644 index 00000000..fbc83dc4 --- /dev/null +++ b/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, + ) diff --git a/t4_devkit/evaluation/matching/__init__.py b/t4_devkit/evaluation/matching/__init__.py new file mode 100644 index 00000000..af5ae063 --- /dev/null +++ b/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 diff --git a/t4_devkit/evaluation/matching/algorithm.py b/t4_devkit/evaluation/matching/algorithm.py new file mode 100644 index 00000000..8a7770a4 --- /dev/null +++ b/t4_devkit/evaluation/matching/algorithm.py @@ -0,0 +1,150 @@ +from __future__ import annotations + +from abc import ABC, abstractmethod +from copy import deepcopy +from typing import TYPE_CHECKING, Sequence, TypeVar + +import numpy as np + +from ..result import BoxMatch + +if TYPE_CHECKING: + from t4_devkit.dataclass import BoxLike + from t4_devkit.typing import NDArrayF64 + + from .policy import MatchingPolicyLike + from .scorer import MatchingScorerLike + +__all__ = ["GreedyMatcher", "MatchingAlgorithmLike"] + + +# ===== Base Class for Matching Algorithm ===== + + +class MatchingAlgorithmImpl(ABC): + """Abstract base class for matching algorithm class.""" + + def __init__( + self, + scorer: MatchingScorerLike, + policy: MatchingPolicyLike, + matchable_threshold: float, + ) -> None: + super().__init__() + self._scorer = scorer + self._policy = policy + self._matchable_threshold = matchable_threshold + + def __call__( + self, + estimations: Sequence[BoxLike], + ground_truths: Sequence[BoxLike], + ) -> list[BoxMatch]: + """Execute matching. + + Args: + estimations (Sequence[BoxLike]): Sequence of estimations. + ground_truths (Sequence[BoxLike]): Sequence of ground truths. + + Returns: + list[BoxMatch]: List of matches. + """ + score_table = self._score_table(estimations, ground_truths) + return self._do_matching(estimations, ground_truths, score_table) + + def _score_table( + self, + estimations: Sequence[BoxLike], + ground_truths: Sequence[BoxLike], + ) -> NDArrayF64: + """Create a score table. + + Args: + estimations (Sequence[BoxLike]): Sequence of estimations. + ground_truths (Sequence[BoxLike]): Sequence of ground truths. + + Returns: + NDArrayF64: Score table in the shape of (NumEst, NumGT). + """ + num_rows, num_cols = len(estimations), len(ground_truths) + + table: NDArrayF64 = np.full((num_rows, num_cols), fill_value=np.nan) + for i, box1 in enumerate(estimations): + for j, box2 in enumerate(ground_truths): + if box1.frame_id != box2.frame_id: + continue + + score = self._scorer(box1, box2) + + # check if boxes distance and label is matchable + if self._scorer.is_better_than( + score, self._matchable_threshold + ) and self._policy.is_matchable(box1, box2): + table[i, j] = score + + return table + + def _get_indices(self, score_table: NDArrayF64) -> tuple[int, int]: + """Return indices of estimation and ground truth in the score table at the best score. + + Args: + score_table (NDArrayF64): Score table in the shape of (NumEst, NumGt). + + Returns: + Estimation index and ground truth index. + """ + estimation_idx, ground_truth_idx = ( + np.unravel_index(np.nanargmin(score_table), score_table.shape) + if self._scorer.is_smaller_score_better() + else np.unravel_index(np.nanargmax(score_table), score_table.shape) + ) + return estimation_idx, ground_truth_idx + + @abstractmethod + def _do_matching( + self, + estimations: Sequence[BoxLike], + ground_truths: Sequence[BoxLike], + score_table: NDArrayF64, + ) -> list[BoxMatch]: + pass + + +MatchingAlgorithmLike = TypeVar("MatchingAlgorithmLike", bound=MatchingAlgorithmImpl) + + +# ===== Specific Matching Algorithms ===== + + +class GreedyMatcher(MatchingAlgorithmImpl): + def _do_matching( + self, + estimations: Sequence[BoxLike], + ground_truths: Sequence[BoxLike], + score_table: NDArrayF64, + ) -> list[BoxMatch]: + tmp_estimations = list(deepcopy(estimations)) + tmp_ground_truths = list(deepcopy(ground_truths)) + + output: list[BoxMatch] = [] + # 1. match the nearest matchable estimations and GTs + num_estimations, *_ = score_table.shape + for _ in range(num_estimations): + if np.isnan(score_table).all(): + break + + estimation_idx, ground_truth_idx = self._get_indices(score_table) + + estimation_picked = tmp_estimations.pop(estimation_idx) + ground_truth_picked = tmp_ground_truths.pop(ground_truth_idx) + output.append(BoxMatch(estimation_picked, ground_truth_picked)) + + # remove picked estimations and GTs + score_table = np.delete(score_table, estimation_idx, axis=0) + score_table = np.delete(score_table, ground_truth_idx, axis=1) + + # 2. assign remaining estimations(=FPs) and GTs(=FNs) + output += [BoxMatch(estimation=estimation) for estimation in tmp_estimations] + output += [BoxMatch(ground_truth=ground_truth) for ground_truth in tmp_ground_truths] + + return output diff --git a/t4_devkit/evaluation/matching/parameter.py b/t4_devkit/evaluation/matching/parameter.py new file mode 100644 index 00000000..d4caf4cf --- /dev/null +++ b/t4_devkit/evaluation/matching/parameter.py @@ -0,0 +1,123 @@ +from __future__ import annotations + +from enum import Enum + +import numpy as np +from attrs import define, field + +from .algorithm import GreedyMatcher, MatchingAlgorithmLike +from .policy import AllowAnyPolicy, AllowUnknownPolicy, MatchingPolicyLike, StrictPolicy +from .scorer import CenterDistance, Iou2D, Iou3D, MatchingScorerLike, PlaneDistance + +__all__ = [ + "build_matcher", + "build_matching_scorer", + "build_matching_policy", + "MatchingScorer", + "MatchingPolicy", + "MatchingAlgorithm", + "MatchingParams", +] + +# ===== Builder Functions ===== + + +def build_matcher(params: MatchingParams) -> MatchingAlgorithmLike: + """Build a matching algorithm from the parameter. + + Examples: + >>> params = MatchingParams(...) + >>> matcher = build_matcher(params) + """ + scorer = build_matching_scorer(params) + policy = build_matching_policy(params) + if params.algorithm == MatchingAlgorithm.GREEDY: + return GreedyMatcher( + scorer=scorer, + policy=policy, + matchable_threshold=params.matchable_distance, + ) + else: + raise ValueError(f"Unexpected algorithm name: {params.algorithm}") + + +def build_matching_scorer(params: MatchingParams) -> MatchingScorerLike: + """Build a matching scorer from the parameter. + + Examples: + >>> params = MatchingParams(...) + >>> scorer = build_matching_scorer(params) + """ + if params.scorer == MatchingScorer.CENTER_DISTANCE: + return CenterDistance() + elif params.scorer == MatchingScorer.PLANE_DISTANCE: + return PlaneDistance() + elif params.scorer == MatchingScorer.IOU2D: + return Iou2D() + elif params.scorer == MatchingScorer.IOU3D: + return Iou3D() + else: + raise ValueError(f"Unexpected scorer name: {params.scorer}") + + +def build_matching_policy(params: MatchingParams) -> MatchingPolicyLike: + """Build a matching policy from the parameter. + + Examples: + >>> params = MatchingParams(...) + >>> policy = build_matching_policy(params) + """ + if params.policy == MatchingPolicy.STRICT: + return StrictPolicy() + elif params.policy == MatchingPolicy.ALLOW_UNKNOWN: + return AllowUnknownPolicy() + elif params.policy == MatchingPolicy.ALLOW_ANY: + return AllowAnyPolicy() + else: + raise ValueError(f"Unexpected policy name: {params.policy}") + + +# ===== Parameters ===== + + +class MatchingScorer(str, Enum): + """An enum to represent matching scorer names.""" + + CENTER_DISTANCE = "CENTER_DISTANCE" + PLANE_DISTANCE = "PLANE_DISTANCE" + IOU2D = "IOU2D" + IOU3D = "IOU3D" + + +class MatchingPolicy(str, Enum): + """An enum to represent matching policy names.""" + + STRICT = "STRICT" + ALLOW_UNKNOWN = "ALLOW_UNKNOWN" + ALLOW_ANY = "ALLOW_ANY" + + +class MatchingAlgorithm(str, Enum): + """An enum to represent matching algorithm names.""" + + GREEDY = "GREEDY" + + +@define +class MatchingParams: + """A dataclass to represent matching parameters. + + Attributes: + scorer (MatchingScorer): Name of matching scorer. + policy (MatchingPolicy): Name of matching policy. + algorithm (MatchingAlgorithm): Name of matching algorithm. + matchable_distance (float): Max distance from a GT which a estimation can be matched. + """ + + scorer: MatchingScorer = field(default=MatchingScorer.CENTER_DISTANCE, converter=MatchingScorer) + policy: MatchingPolicy = field(default=MatchingPolicy.STRICT, converter=MatchingPolicy) + algorithm: MatchingAlgorithm = field( + default=MatchingAlgorithm.GREEDY, + converter=MatchingAlgorithm, + ) + matchable_distance: float = field(default=np.inf) diff --git a/t4_devkit/evaluation/matching/policy.py b/t4_devkit/evaluation/matching/policy.py new file mode 100644 index 00000000..32fa5ca2 --- /dev/null +++ b/t4_devkit/evaluation/matching/policy.py @@ -0,0 +1,66 @@ +from __future__ import annotations + +from abc import ABC, abstractmethod +from typing import TYPE_CHECKING, TypeVar + +if TYPE_CHECKING: + from t4_devkit.dataclass import BoxLike + + +__all__ = [ + "MatchingPolicyLike", + "StrictPolicy", + "AllowUnknownPolicy", + "AllowAnyPolicy", +] + +# ===== Base Class for Matching Policy ===== + + +class MatchingPolicyImpl(ABC): + """Abstract base class of the matching policy functions.""" + + @abstractmethod + def is_matchable(self, box1: BoxLike, box2: BoxLike) -> bool: + """Check if the input boxes satisfy the policy requirements. + + Args: + box1 (BoxLike): A box. + box2 (BoxLike): A box. + + Returns: + bool: Return `True` if the input boxes satisfied the policy requirements. + """ + pass + + +MatchingPolicyLike = TypeVar("MatchingPolicyLike", bound=MatchingPolicyImpl) + + +# ===== Specific Matching Policies ===== + + +class StrictPolicy(MatchingPolicyImpl): + """This policy allows the case only if the two boxes have the same label.""" + + def is_matchable(self, box1: BoxLike, box2: BoxLike) -> bool: + return box1.semantic_label == box2.semantic_label + + +class AllowUnknownPolicy(MatchingPolicyImpl): + """This policy allows the case if the two boxes have the same label + or one of boxes have the `UNKNOWN`.""" + + def is_matchable(self, box1: BoxLike, box2: BoxLike) -> bool: + return ( + box1.semantic_label == box2.semantic_label + or box1.semantic_label.name.upper() == "UNKNOWN" + or box2.semantic_label.name.upper() == "UNKNOWN" + ) + + +class AllowAnyPolicy(MatchingPolicyImpl): + """This policy allows any cases.""" + + def is_matchable(self, box1: BoxLike, box2: BoxLike) -> bool: + return True diff --git a/t4_devkit/evaluation/matching/scorer.py b/t4_devkit/evaluation/matching/scorer.py new file mode 100644 index 00000000..890b00df --- /dev/null +++ b/t4_devkit/evaluation/matching/scorer.py @@ -0,0 +1,247 @@ +from __future__ import annotations + +from abc import ABC, abstractmethod +from typing import TYPE_CHECKING, TypeVar + +import numpy as np + +from t4_devkit.dataclass import Box3D, HomogeneousMatrix, ShapeType + +from .utility import compute_area_intersection, compute_volume_intersection + +if TYPE_CHECKING: + from t4_devkit.dataclass import BoxLike + from t4_devkit.typing import Vector3Like + +__all__ = [ + "MatchingScorerLike", + "CenterDistance", + "PlaneDistance", + "Iou2D", + "Iou3D", +] + + +# ===== Base Class for Matching Scorer ===== + + +class MatchingScorerImpl(ABC): + """Abstract base class of the function of matching scorer.""" + + def __call__( + self, + box1: BoxLike, + box2: BoxLike, + ego2map: HomogeneousMatrix | None = None, + ) -> float: + self._validate(box1, box2) + return self._calculate_score(box1, box2, ego2map) + + def _validate(self, box1: BoxLike, box2: BoxLike) -> None: + """Validate the input two boxes. + + Args: + box1 (BoxLike): A box. + box2 (BoxLike): A box. + + Raises: + TypeError: Expecting two boxes have the same type. + """ + if not isinstance(box1, type(box2)): + raise TypeError( + f"Both boxes must be the same type, but got {type(box1)} and {type(box2)}." + ) + + def is_better_than(self, score: float, threshold: float) -> bool: + """Check if the matching score of the input boxes satisfies the threshold. + + Args: + score (float): Matching score. + threshold (float): Threshold value of the matching score. + + Returns: + Returns `True` the calculated matching score is better than the threshold. + """ + return score <= threshold if self.is_smaller_score_better() else threshold <= score + + @classmethod + @abstractmethod + def is_smaller_score_better(cls) -> bool: + """Indicate whether the smaller matching score is better. + + Returns: + Returns `True` if the smaller score is better. + """ + pass + + @abstractmethod + def _calculate_score( + self, + box1: BoxLike, + box2: BoxLike, + ego2map: HomogeneousMatrix | None = None, + ) -> float: + """Calculate the matching score of the input two boxes. + + Args: + box1 (BoxLike): A box. + box2 (BoxLike): A box. + ego2map (HomogeneousMatrix | None, optional): Transformation matrix from map to ego frame. + + Returns: + Calculated matching score. + """ + pass + + +MatchingScorerLike = TypeVar("MatchingScorer", bound=MatchingScorerImpl) + +# ===== Specific Matching Scorers ===== + + +class CenterDistance(MatchingScorerImpl): + @classmethod + def is_smaller_score_better(cls) -> bool: + return True + + def _calculate_score( + self, + box1: BoxLike, + box2: BoxLike, + _ego2map: HomogeneousMatrix | None = None, + ) -> float: + return ( + np.linalg.norm(box1.position - box2.position) + if isinstance(box1, Box3D) + else np.linalg.norm(box1.center - box2.center) + ) + + +class PlaneDistance(MatchingScorerImpl): + def _validate(self, box1, box2): + super()._validate(box1, box2) + + if not isinstance(box1, Box3D): + raise TypeError("For PlaneDistance, input boxes must be 3D.") + + @classmethod + def is_smaller_score_better(cls) -> bool: + return True + + def _calculate_score( + self, + box1: Box3D, + box2: Box3D, + ego2map: HomogeneousMatrix | None = None, + ) -> float: + box1 = self._transform2ego(box1, ego2map) + box2 = self._transform2ego(box2, ego2map) + + if ( + box1.shape.shape_type == ShapeType.BOUNDING_BOX + and box2.shape.shape_type == ShapeType.BOUNDING_BOX + ): + footprint1 = np.array(box1.footprint.exterior.coords)[:-1] + footprint2 = np.array(box2.footprint.exterior.coords)[:-1] + + if box1.diff_yaw(box2) > 0.5 * np.pi: + footprint1 = footprint1[[1, 0, 3, 2, 1]] + + sort_idx = np.argsort(np.linalg.norm(footprint2[:, :2], axis=1)) + + sorted_footprint1 = footprint1[sort_idx] + sorted_footprint2 = footprint2[sort_idx] + + left1, right1 = self._point_left_and_right(sorted_footprint1[0], sorted_footprint1[1]) + left2, right2 = self._point_left_and_right(sorted_footprint2[0], sorted_footprint2[1]) + + distance_left = np.linalg.norm(left1[:2] - left2[:2]) + distance_right = np.linalg.norm(right1[:2] - right2[:2]) + + distance = round(np.sqrt(0.5 * (distance_left**2 + distance_right**2)), 10) + else: + distance = np.linalg.norm(box1.position[:2] - box2.position[:2]) + + return distance + + def _transform2ego(self, box: Box3D, ego2map: HomogeneousMatrix | None = None) -> Box3D: + """Transform the box to base link frame if it is not. + + Todo: + This method should be function. + """ + if box.frame_id == "base_link": + return box + + if ego2map is None: + raise ValueError(f"For {box.frame_id}, `ego2map` must be specified.") + + matrix = HomogeneousMatrix( + position=box.position, + rotation=box.rotation, + src=box.frame_id, + dst="base_link", + ) + + tf = ego2map.inv().transform(matrix=matrix) + + return Box3D( + unix_time=box.unix_time, + frame_id="base_link", + semantic_label=box.semantic_label, + position=tf.position, + rotation=tf.rotation, + shape=box.shape, + velocity=box.velocity, + num_points=box.num_points, + future=box.future, + confidence=box.confidence, + uuid=box.uuid, + ) + + def _point_left_and_right( + self, + point1: Vector3Like, + point2: Vector3Like, + ) -> tuple[Vector3Like, Vector3Like]: + """Determine the input point is left or right.""" + cross_product = np.round(np.cross(point1[:2], point2[:2]), 10) + return (point1, point2) if cross_product < 0 else (point2, point1) + + +class Iou2D(MatchingScorerImpl): + @classmethod + def is_smaller_score_better(cls) -> bool: + return False + + def _calculate_score( + self, + box1: BoxLike, + box2: BoxLike, + _ego2map: HomogeneousMatrix | None = None, + ) -> float: + intersection = compute_area_intersection(box1, box2) + union = box1.area + box2.area - intersection + return intersection / union + + +class Iou3D(MatchingScorerImpl): + def _validate(self, box1, box2): + super()._validate(box1, box2) + + if not isinstance(box1, Box3D): + raise TypeError("For Iou3D, input boxes must be 3D.") + + @classmethod + def is_smaller_score_better(cls) -> bool: + return False + + def _calculate_score( + self, + box1: Box3D, + box2: Box3D, + _ego2map: HomogeneousMatrix | None = None, + ) -> float: + intersection = compute_volume_intersection(box1, box2) + union = box1.volume + box2.volume - intersection + return intersection / union diff --git a/t4_devkit/evaluation/matching/utility.py b/t4_devkit/evaluation/matching/utility.py new file mode 100644 index 00000000..736fe3ea --- /dev/null +++ b/t4_devkit/evaluation/matching/utility.py @@ -0,0 +1,42 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from t4_devkit.dataclass import Box3D + +__all__ = ["compute_area_intersection", "compute_volume_intersection"] + + +def compute_area_intersection(box1: Box3D, box2: Box3D) -> float: + """Return the area intersection between two boxes. + + Args: + box1 (Box3D): A box. + box2 (Box3D): A box. + + Returns: + float: Intersection of area. + + """ + # TODO(ktro2828): add support of 2D boxes + return box1.footprint.intersection(box2.footprint).area + + +def compute_volume_intersection(box1: Box3D, box2: Box3D) -> float: + """Return the volume intersection between two 3D boxes. + + Args: + box1 (Box3D): A box. + box2 (Box3D): A box. + + Returns: + float: Intersection of volume. + """ + area = compute_area_intersection(box1, box2) + + min_z = max(box1.position[2] - 0.5 * box1.size[2], box2.position[2] - 0.5 * box2.size[2]) + max_z = min(box1.position[2] + 0.5 * box1.size[2], box2.position[2] + 0.5 * box2.size[2]) + height = max(0, max_z - min_z) + + return area * height diff --git a/t4_devkit/evaluation/metric/__init__.py b/t4_devkit/evaluation/metric/__init__.py new file mode 100644 index 00000000..75a02ff6 --- /dev/null +++ b/t4_devkit/evaluation/metric/__init__.py @@ -0,0 +1,2 @@ +from .ap import * # noqa +from .clear import * # noqa diff --git a/t4_devkit/evaluation/metric/ap.py b/t4_devkit/evaluation/metric/ap.py new file mode 100644 index 00000000..9d47ddd6 --- /dev/null +++ b/t4_devkit/evaluation/metric/ap.py @@ -0,0 +1,111 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +import numpy as np +from attrs import define, field + +from .base import BaseMetric + +if TYPE_CHECKING: + from t4_devkit.evaluation import BoxMatch, FrameBoxMatch, MatchingScorerLike + +__all__ = ["Ap", "ApH"] + + +class Ap(BaseMetric): + num_recall_point = 101 + min_precision = 0.1 + min_recall = 0.1 + + @define + class ApBuffer: + """Buffer to compute AP.""" + + num_gt: int = field(init=False, default=0) + tp_list: list[float] = field(init=False, factory=list) + fp_list: list[float] = field(init=False, factory=list) + confidences: list[float] = field(init=False, factory=list) + + def compute_ap(self) -> float: + """Compute average precision.""" + if self.num_gt == 0: + return 0.0 + + sorted_idx = np.argsort(self.confidences)[::-1] + tp_sorted = np.array(self.tp_list)[sorted_idx] + fp_sorted = np.array(self.fp_list)[sorted_idx] + + tps = np.cumsum(tp_sorted) + fps = np.cumsum(fp_sorted) + + precisions = [] + recalls = [] + for tp, fp in zip(tps, fps, strict=True): + denominator = tp + fp + precisions.append(0.0) if denominator == 0.0 else precisions.append( + tp / denominator + ) + recalls.append(tp / self.num_gt) + + precision_envelope = np.maximum.accumulate(precisions[::-1])[::-1] + + recall_interp = np.linspace(0.0, 1.0, Ap.num_recall_point) + + precision_interp = np.interp(recall_interp, recalls, precision_envelope, right=0) + + first_idx = int(round(100 * Ap.min_recall)) + + filtered_precision = precision_interp[first_idx:] - Ap.min_precision + filtered_precision = np.clip(filtered_precision, 0.0, None) + + return float(np.mean(filtered_precision)) / (1.0 - Ap.min_precision) + + def __init__(self, scorer: MatchingScorerLike, threshold: float) -> None: + self.scorer = scorer + self.threshold = threshold + + def __call__(self, frames: list[FrameBoxMatch]) -> float: + buffer = self._update_buffer(frames) + return buffer.compute_ap() + + def _compute_tp(self, box_match: BoxMatch) -> float: + return 1.0 + + def _update_buffer(self, frames: list[FrameBoxMatch]) -> ApBuffer: + buffer = self.ApBuffer() + for frame in frames: + self._update_buffer_frame(frame, buffer) + return buffer + + def _update_buffer_frame(self, frame: FrameBoxMatch, buffer: ApBuffer) -> None: + buffer.num_gt += frame.num_gt + for box_match in frame.matches: + if box_match.estimation is None: + continue + + buffer.confidences.append(box_match.estimation.confidence) + if box_match.is_tp( + scorer=self.scorer, + threshold=self.threshold, + ego2map=frame.ego2map, + ): + buffer.tp_list.append(self._compute_tp(box_match)) + buffer.fp_list.append(0.0) + else: + buffer.tp_list.append(0.0) + buffer.fp_list.append(1.0) + + +class ApH(Ap): + def __init__(self, scorer: MatchingScorerLike, threshold: float) -> None: + super().__init__(scorer=scorer, threshold=threshold) + + def _compute_tp(self, box_match: BoxMatch) -> float: + if not box_match.is_matched(): + return 0.0 + + diff_yaw = box_match.estimation.diff_yaw(box_match.ground_truth) + if diff_yaw > np.pi: + diff_yaw = 2.0 * np.pi - diff_yaw + return min(1.0, max(0.0, 1.0 - diff_yaw / np.pi)) diff --git a/t4_devkit/evaluation/metric/base.py b/t4_devkit/evaluation/metric/base.py new file mode 100644 index 00000000..4e371400 --- /dev/null +++ b/t4_devkit/evaluation/metric/base.py @@ -0,0 +1,14 @@ +from __future__ import annotations + +from abc import ABC, abstractmethod +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from t4_devkit.evaluation import FrameBoxMatch + + +class BaseMetric(ABC): + @abstractmethod + def __call__(self, frames: list[FrameBoxMatch]) -> float: + """Compute metric score.""" + ... diff --git a/t4_devkit/evaluation/metric/clear.py b/t4_devkit/evaluation/metric/clear.py new file mode 100644 index 00000000..ecb0a9a6 --- /dev/null +++ b/t4_devkit/evaluation/metric/clear.py @@ -0,0 +1,138 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +from attrs import define, field + +from .base import BaseMetric + +if TYPE_CHECKING: + from t4_devkit.evaluation import BoxMatch, FrameBoxMatch, MatchingScorerLike + +__all__ = ["Mota", "Motp"] + + +class Mota(BaseMetric): + @define + class ClearBuffer: + num_gt: int = field(init=False, default=0) + num_tp: int = field(init=False, default=0) + num_fp: int = field(init=False, default=0) + num_id_switch: int = field(init=False, default=0) + score: float = field(init=False, default=0.0) + + def compute_mota(self) -> float: + return ( + max((self.num_tp - self.num_fp - self.num_id_switch) / self.num_gt, 0.0) + if self.num_gt > 0 + else 0.0 + ) + + def compute_motp(self) -> float: + return max(self.score / self.num_tp, 0.0) if self.num_tp > 0 else 0.0 + + def __init__(self, scorer: MatchingScorerLike, threshold: float) -> None: + self.scorer = scorer + self.threshold = threshold + + def __call__(self, frames: list[FrameBoxMatch]) -> float: + buffer = self._update_buffer(frames) + return buffer.compute_mota() + + def _update_buffer(self, frames: list[FrameBoxMatch]) -> ClearBuffer: + buffer = self.ClearBuffer() + num_frame = len(frames) + for i in range(1, num_frame): + current_frame = frames[i] + previous_frame = frames[i - 1] + self._update_buffer_frame(current_frame, previous_frame, buffer) + return buffer + + def _update_buffer_frame( + self, + current_frame: FrameBoxMatch, + previous_frame: FrameBoxMatch, + buffer: ClearBuffer, + ) -> None: + buffer.num_gt += current_frame.num_gt + for current_match in current_frame.matches: + is_id_switch = False + is_same_match = False + for previous_match in previous_frame.matches: + if not previous_match.is_tp( + self.scorer, + self.threshold, + previous_frame.ego2map, + ): + continue + + is_id_switch = self._is_id_switched(current_match, previous_match) + if is_id_switch: + break + + is_same_match = self._is_same_match(current_match, previous_match) + if is_same_match: + buffer.num_tp += current_match.is_tp( + scorer=self.scorer, + threshold=self.threshold, + ego2map=current_frame.ego2map, + ) + buffer.score += self.scorer( + previous_match.estimation, + previous_match.ground_truth, + ego2map=previous_frame.ego2map, + ) + break + + if is_same_match: + continue + + if current_match.is_tp(self.scorer, self.threshold, current_frame.ego2map): + buffer.num_tp += 1 + buffer.score += self.scorer( + current_match.estimation, + current_match.ground_truth, + current_frame.ego2map, + ) + if is_id_switch: + buffer.num_id_switch += 1 + else: + buffer.num_fp += 1 + + def _is_id_switched(self, current_match: BoxMatch, previous_match: BoxMatch) -> bool: + if (not current_match.is_matched()) and (not previous_match.is_matched()): + return False + + is_same_estimated_uuid, is_same_estimated_label, is_same_gt_uuid = self._check_match( + current_match, previous_match + ) + + if is_same_estimated_uuid and is_same_estimated_label: + return not is_same_gt_uuid + else: + return is_same_gt_uuid + + def _is_same_match(self, current_match: BoxMatch, previous_match: BoxMatch) -> bool: + is_same_estimated_uuid, is_same_estimated_label, is_same_gt_uuid = self._check_match( + current_match, previous_match + ) + return is_same_estimated_uuid and is_same_estimated_label and is_same_gt_uuid + + def _check_match( + self, + current_match: BoxMatch, + previous_match: BoxMatch, + ) -> tuple[bool, bool, bool]: + is_same_estimated_uuid = current_match.estimation.uuid == previous_match.estimation.uuid + is_same_estimated_label = ( + current_match.estimation.semantic_label == previous_match.estimation.semantic_label + ) + is_same_gt_uuid = current_match.ground_truth.uuid == previous_match.ground_truth.uuid + + return is_same_estimated_uuid, is_same_estimated_label, is_same_gt_uuid + + +class Motp(Mota): + def __call__(self, frames: list[FrameBoxMatch]) -> float: + buffer = self._update_buffer(frames) + return buffer.compute_motp() diff --git a/t4_devkit/evaluation/result/__init__.py b/t4_devkit/evaluation/result/__init__.py new file mode 100644 index 00000000..269d138c --- /dev/null +++ b/t4_devkit/evaluation/result/__init__.py @@ -0,0 +1,2 @@ +from .box import * # noqa +from .status import * # noqa diff --git a/t4_devkit/evaluation/result/box.py b/t4_devkit/evaluation/result/box.py new file mode 100644 index 00000000..6f06a1dc --- /dev/null +++ b/t4_devkit/evaluation/result/box.py @@ -0,0 +1,106 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +from attrs import define, field + +from .status import MatchingStatus + +if TYPE_CHECKING: + from t4_devkit.dataclass import BoxLike, HomogeneousMatrix + from t4_devkit.evaluation import MatchingScorerLike + +__all__ = ["BoxMatch", "FrameBoxMatch"] + + +@define +class BoxMatch: + """A class to represent the match of an estimation and ground truth. + + Raises: + ValueError: Construction raises error if both `estimation` and `ground_truth` is `None`. + + Attributes: + estimation (BoxLike | None): Estimation. + ground_truth (BoxLike | None): Ground truth. + """ + + estimation: BoxLike | None = field(default=None) + ground_truth: BoxLike | None = field(default=None) + + def __attrs_post_init__(self) -> None: + if self.estimation is None and self.ground_truth is None: + raise ValueError("At least one of `estimation` or `ground_truth` must be set.") + + def is_matched(self) -> bool: + return self.estimation is not None and self.ground_truth is not None + + def is_label_ok(self) -> bool: + return ( + False + if not self.is_matched() + else self.estimation.semantic_label == self.ground_truth.semantic_label + ) + + def is_tp( + self, + scorer: MatchingScorerLike, + threshold: float, + ego2map: HomogeneousMatrix | None = None, + ) -> bool: + if not self.is_matched(): + return False + + score = scorer(self.estimation, self.ground_truth, ego2map) + return scorer.is_better_than(score, threshold) + + def to_status(self, scorer: MatchingScorerLike, threshold: float) -> MatchingStatus: + if self.estimation is None: + return MatchingStatus.FN + elif self.ground_truth is None: + return MatchingStatus.FP + else: + return MatchingStatus.TP if self.is_tp(scorer, threshold) else MatchingStatus.FP + + +@define +class FrameBoxMatch: + """A container of box matches at a single frame. + + Attributes: + unix_time (int): Unix timestamp. + frame_index (int): Index number of the frame. + matches (list[BoxMatch]): List of box matches. + ego2map (HomogeneousMatrix): Transformation matrix from ego to map coordinate. + """ + + unix_time: int + frame_index: int + matches: list[BoxMatch] + ego2map: HomogeneousMatrix + + @property + def num_estimation(self) -> int: + """Return the total number of estimations.""" + return sum(m.estimation is not None for m in self.matches) + + @property + def num_gt(self) -> int: + """Return the total number of ground truths.""" + return sum(m.ground_truth is not None for m in self.matches) + + def num_tp(self, scorer: MatchingScorerLike, threshold: float) -> int: + """Return the number of TPs.""" + return sum(1 for m in self.matches if m.to_status(scorer, threshold) == MatchingStatus.TP) + + def num_fp(self, scorer: MatchingScorerLike, threshold: float) -> int: + """Return the number of FPs.""" + return sum(1 for m in self.matches if m.to_status(scorer, threshold) == MatchingStatus.FP) + + def num_fn(self, scorer: MatchingScorerLike, threshold: float) -> int: + """Return the number of FNs.""" + return sum(1 for m in self.matches if m.to_status(scorer, threshold) == MatchingStatus.FN) + + def num_tn(self, scorer: MatchingScorerLike, threshold: float) -> int: + """Return the number of TNs.""" + return sum(1 for m in self.matches if m.to_status(scorer, threshold) == MatchingStatus.TN) diff --git a/t4_devkit/evaluation/result/status.py b/t4_devkit/evaluation/result/status.py new file mode 100644 index 00000000..5bc53b1e --- /dev/null +++ b/t4_devkit/evaluation/result/status.py @@ -0,0 +1,24 @@ +from __future__ import annotations + +from enum import Enum + +__all__ = ["MatchingStatus"] + + +class MatchingStatus(str, Enum): + TP = "TP" + FP = "FP" + FN = "FN" + TN = "TN" + + def is_positive(self) -> bool: + return self in ("TP", "FP") + + def is_negative(self) -> bool: + return self in ("TN", "FN") + + def is_true(self) -> bool: + return self in ("TP", "TN") + + def is_false(self) -> bool: + return self in ("FP", "FN") diff --git a/t4_devkit/evaluation/task.py b/t4_devkit/evaluation/task.py new file mode 100644 index 00000000..afc67cdd --- /dev/null +++ b/t4_devkit/evaluation/task.py @@ -0,0 +1,51 @@ +from __future__ import annotations + +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 2D objects. + + Returns: + Return `True` if the task is in [detection2d, tracking2d, segmentation2d]. + """ + return self in ( + EvaluationTask.DETECTION2D, + EvaluationTask.TRACKING2D, + EvaluationTask.SEGMENTATION2D, + ) + + 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) diff --git a/tests/conftest.py b/tests/conftest.py index 5a56b364..dcdaba89 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -313,3 +313,8 @@ def dummy_camera_calibration() -> tuple[tuple[int, int], NDArrayFloat]: @pytest.fixture(scope="session") def dummy_lanelet_path() -> str: return "tests/sample/t4dataset/map/lanelet2_map.osm" + + +@pytest.fixture(scope="session") +def dummy_dataset_root() -> str: + return "tests/sample/t4dataset" diff --git a/tests/evaluation/conftest.py b/tests/evaluation/conftest.py new file mode 100644 index 00000000..7978fbfe --- /dev/null +++ b/tests/evaluation/conftest.py @@ -0,0 +1,90 @@ +from __future__ import annotations + +import pytest +from pyquaternion import Quaternion + +from t4_devkit.dataclass import Box2D, Box3D, SemanticLabel, Shape, ShapeType + + +@pytest.fixture(scope="module") +def dummy_evaluation_box3ds() -> tuple[list[Box3D], list[Box3D]]: + estimations = [ + Box3D( + unix_time=100, + frame_id="base_link", + semantic_label=SemanticLabel("car"), + position=(1.0, 1.0, 1.0), + rotation=Quaternion([0.0, 0.0, 0.0, 1.0]), + shape=Shape(shape_type=ShapeType.BOUNDING_BOX, size=(1.0, 1.0, 1.0)), + velocity=(1.0, 1.0, 1.0), + confidence=1.0, + uuid="estimation_car3d_1", + ), + Box3D( + unix_time=100, + frame_id="base_link", + semantic_label=SemanticLabel("bicycle"), + position=(-1.0, -1.0, 1.0), + rotation=Quaternion([0.0, 0.0, 0.0, 1.0]), + shape=Shape(shape_type=ShapeType.BOUNDING_BOX, size=(1.0, 1.0, 1.0)), + velocity=(1.0, 1.0, 1.0), + confidence=1.0, + uuid="estimation_bicycle3d_1", + ), + Box3D( + unix_time=100, + frame_id="base_link", + semantic_label=SemanticLabel("car"), + position=(-1.0, 1.0, 1.0), + rotation=Quaternion([0.0, 0.0, 0.0, 1.0]), + shape=Shape(shape_type=ShapeType.BOUNDING_BOX, size=(1.0, 1.0, 1.0)), + velocity=(1.0, 1.0, 1.0), + confidence=1.0, + uuid="estimation_car3d_1", + ), + ] + + ground_truths = [ + Box3D( + unix_time=100, + frame_id="base_link", + semantic_label=SemanticLabel("car"), + position=(1.0, 1.0, 1.0), + rotation=Quaternion([0.0, 0.0, 0.0, 1.0]), + shape=Shape(shape_type=ShapeType.BOUNDING_BOX, size=(1.0, 1.0, 1.0)), + velocity=(1.0, 1.0, 1.0), + num_points=10, + confidence=1.0, + uuid="gt_car3d_1", + ), + Box3D( + unix_time=100, + frame_id="base_link", + semantic_label=SemanticLabel("bicycle"), + position=(-1.0, -1.0, 1.0), + rotation=Quaternion([0.0, 0.0, 0.0, 1.0]), + shape=Shape(shape_type=ShapeType.BOUNDING_BOX, size=(1.0, 1.0, 1.0)), + velocity=(1.0, 1.0, 1.0), + num_points=10, + confidence=1.0, + uuid="gt_bicycle3d_1", + ), + Box3D( + unix_time=100, + frame_id="base_link", + semantic_label=SemanticLabel("pedestrian"), + position=(-1.0, 1.0, 1.0), + rotation=Quaternion([0.0, 0.0, 0.0, 1.0]), + shape=Shape(shape_type=ShapeType.BOUNDING_BOX, size=(1.0, 1.0, 1.0)), + velocity=(1.0, 1.0, 1.0), + confidence=1.0, + uuid="gt_pedestrian3d_1", + ), + ] + + return estimations, ground_truths + + +@pytest.fixture(scope="module") +def dummy_evaluation_box2ds() -> tuple[list[Box2D], list[Box2D]]: + pass diff --git a/tests/evaluation/matching/test_matching_params.py b/tests/evaluation/matching/test_matching_params.py new file mode 100644 index 00000000..a16cb8a1 --- /dev/null +++ b/tests/evaluation/matching/test_matching_params.py @@ -0,0 +1,24 @@ +from __future__ import annotations + +import pytest + +from t4_devkit.evaluation import MatchingAlgorithm, MatchingParams, MatchingPolicy, MatchingScorer + + +def test_matching_params_with_objects() -> None: + _ = MatchingParams( + scorer=MatchingScorer.CENTER_DISTANCE, + policy=MatchingPolicy.STRICT, + algorithm=MatchingAlgorithm.GREEDY, + ) + + +def test_matching_param_with_str() -> None: + _ = MatchingParams(scorer="CENTER_DISTANCE", policy="STRICT", algorithm="GREEDY") + + +def test_matching_param_with_lower_str() -> None: + """Lower case is not supported to initialize matching params.""" + + with pytest.raises(ValueError): + _ = MatchingParams(scorer="center_distance", policy="strict", algorithm="greedy") diff --git a/tests/evaluation/matching/test_policy.py b/tests/evaluation/matching/test_policy.py new file mode 100644 index 00000000..27f30fb0 --- /dev/null +++ b/tests/evaluation/matching/test_policy.py @@ -0,0 +1,56 @@ +from __future__ import annotations + +import pytest +from pyquaternion import Quaternion + +from t4_devkit.dataclass import Box3D, SemanticLabel, Shape, ShapeType +from t4_devkit.evaluation import AllowAnyPolicy, AllowUnknownPolicy, StrictPolicy + + +@pytest.fixture(scope="module") +def estimation() -> Box3D: + return Box3D( + unix_time=100, + frame_id="base_link", + semantic_label=SemanticLabel("car"), + position=(1.0, 1.0, 1.0), + rotation=Quaternion([0.0, 0.0, 0.0, 1.0]), + shape=Shape(shape_type=ShapeType.BOUNDING_BOX, size=(1.0, 1.0, 1.0)), + velocity=(1.0, 1.0, 1.0), + confidence=1.0, + uuid="estimation_car3d_1", + ) + + +@pytest.fixture(scope="module") +def ground_truth() -> Box3D: + return Box3D( + unix_time=100, + frame_id="base_link", + semantic_label=SemanticLabel("unknown"), + position=(1.0, 1.0, 1.0), + rotation=Quaternion([0.0, 0.0, 0.0, 1.0]), + shape=Shape(shape_type=ShapeType.BOUNDING_BOX, size=(1.0, 1.0, 1.0)), + velocity=(1.0, 1.0, 1.0), + num_points=10, + confidence=1.0, + uuid="gt_car3d_1", + ) + + +def test_strict_policy(estimation, ground_truth) -> None: + policy = StrictPolicy() + + assert policy.is_matchable(estimation, ground_truth) is False + + +def test_allow_unknown_policy(estimation, ground_truth) -> None: + policy = AllowUnknownPolicy() + + assert policy.is_matchable(estimation, ground_truth) + + +def test_allow_any_policy(estimation, ground_truth) -> None: + policy = AllowAnyPolicy() + + assert policy.is_matchable(estimation, ground_truth) diff --git a/tests/evaluation/matching/test_scorer.py b/tests/evaluation/matching/test_scorer.py new file mode 100644 index 00000000..a8199e6b --- /dev/null +++ b/tests/evaluation/matching/test_scorer.py @@ -0,0 +1,72 @@ +from __future__ import annotations + +import math + +import pytest +from pyquaternion import Quaternion + +from t4_devkit.dataclass import Box3D, SemanticLabel, Shape, ShapeType +from t4_devkit.evaluation import CenterDistance, Iou2D, Iou3D, PlaneDistance + + +@pytest.fixture(scope="module") +def estimation() -> Box3D: + return Box3D( + unix_time=100, + frame_id="base_link", + semantic_label=SemanticLabel("car"), + position=(1.0, 1.0, 1.0), + rotation=Quaternion([0.0, 0.0, 0.0, 1.0]), + shape=Shape(shape_type=ShapeType.BOUNDING_BOX, size=(1.0, 1.0, 1.0)), + velocity=(1.0, 1.0, 1.0), + confidence=1.0, + uuid="estimation_car3d_1", + ) + + +@pytest.fixture(scope="module") +def ground_truth() -> Box3D: + return Box3D( + unix_time=100, + frame_id="base_link", + semantic_label=SemanticLabel("car"), + position=(1.0, 1.0, 1.0), + rotation=Quaternion([0.0, 0.0, 0.0, 1.0]), + shape=Shape(shape_type=ShapeType.BOUNDING_BOX, size=(1.0, 1.0, 1.0)), + velocity=(1.0, 1.0, 1.0), + num_points=10, + confidence=1.0, + uuid="gt_car3d_1", + ) + + +def test_center_distance(estimation, ground_truth) -> None: + scorer = CenterDistance() + + score = scorer(estimation, ground_truth) + + assert math.isclose(score, 0.0) + + +def test_plane_distance(estimation, ground_truth) -> None: + scorer = PlaneDistance() + + score = scorer(estimation, ground_truth) + + assert math.isclose(score, 0.0) + + +def test_iou2d(estimation, ground_truth) -> None: + scorer = Iou2D() + + score = scorer(estimation, ground_truth) + + assert math.isclose(score, 1.0) + + +def test_iou3d(estimation, ground_truth) -> None: + scorer = Iou3D() + + score = scorer(estimation, ground_truth) + + assert math.isclose(score, 1.0) diff --git a/tests/evaluation/metric/test_ap.py b/tests/evaluation/metric/test_ap.py new file mode 100644 index 00000000..5f075ba7 --- /dev/null +++ b/tests/evaluation/metric/test_ap.py @@ -0,0 +1,49 @@ +from __future__ import annotations + +import math + +import pytest + +from t4_devkit.dataclass import HomogeneousMatrix +from t4_devkit.evaluation import ( + Ap, + ApH, + CenterDistance, + FrameBoxMatch, + MatchingParams, + build_matcher, +) + + +@pytest.fixture(scope="module") +def dummy_frame3d(dummy_evaluation_box3ds) -> FrameBoxMatch: + estimations, ground_truths = dummy_evaluation_box3ds + + matcher = build_matcher(params=MatchingParams()) + + matches = matcher(estimations, ground_truths) + + ego2map = HomogeneousMatrix( + position=(1, 0, 0), + rotation=(1, 0, 0, 0), + src="base_link", + dst="map", + ) + + return FrameBoxMatch(unix_time=100, frame_index=0, matches=matches, ego2map=ego2map) + + +def test_ap(dummy_frame3d) -> None: + ap = Ap(scorer=CenterDistance(), threshold=1.0) + + score = ap(frames=[dummy_frame3d]) + + assert math.isclose(score, 0.394, rel_tol=1e-3) + + +def test_aph(dummy_frame3d) -> None: + aph = ApH(scorer=CenterDistance(), threshold=1.0) + + score = aph(frames=[dummy_frame3d]) + + assert math.isclose(score, 0.394, rel_tol=1e-3) diff --git a/tests/evaluation/metric/test_clear.py b/tests/evaluation/metric/test_clear.py new file mode 100644 index 00000000..cb0f0cc7 --- /dev/null +++ b/tests/evaluation/metric/test_clear.py @@ -0,0 +1,49 @@ +from __future__ import annotations + +import math + +import pytest + +from t4_devkit.dataclass import HomogeneousMatrix +from t4_devkit.evaluation import ( + CenterDistance, + FrameBoxMatch, + MatchingParams, + Mota, + Motp, + build_matcher, +) + + +@pytest.fixture(scope="module") +def dummy_frame3d(dummy_evaluation_box3ds) -> FrameBoxMatch: + estimations, ground_truths = dummy_evaluation_box3ds + + matcher = build_matcher(params=MatchingParams()) + + matches = matcher(estimations, ground_truths) + + ego2map = HomogeneousMatrix( + position=(1, 0, 0), + rotation=(1, 0, 0, 0), + src="base_link", + dst="map", + ) + + return FrameBoxMatch(unix_time=100, frame_index=0, matches=matches, ego2map=ego2map) + + +def test_ap(dummy_frame3d) -> None: + mota = Mota(scorer=CenterDistance(), threshold=1.0) + + score = mota(frames=[dummy_frame3d]) + + assert math.isclose(score, 0.0) + + +def test_aph(dummy_frame3d) -> None: + motp = Motp(scorer=CenterDistance(), threshold=1.0) + + score = motp(frames=[dummy_frame3d]) + + assert math.isclose(score, 0.0) diff --git a/tests/evaluation/result/test_box_match.py b/tests/evaluation/result/test_box_match.py new file mode 100644 index 00000000..5ea186c5 --- /dev/null +++ b/tests/evaluation/result/test_box_match.py @@ -0,0 +1,49 @@ +from __future__ import annotations + +from t4_devkit.dataclass import HomogeneousMatrix +from t4_devkit.evaluation import ( + BoxMatch, + FrameBoxMatch, + MatchingParams, + build_matcher, + build_matching_scorer, +) + + +def test_box_match(dummy_evaluation_box3ds) -> None: + estimations, ground_truths = dummy_evaluation_box3ds + + matcher = build_matcher(params=MatchingParams()) + + matches = matcher(estimations, ground_truths) + + assert isinstance(matches, list) + assert all(isinstance(m, BoxMatch) for m in matches) + assert len(matches) == 4 + + +def test_frame_box_match(dummy_evaluation_box3ds) -> None: + estimations, ground_truths = dummy_evaluation_box3ds + + matcher = build_matcher(params=MatchingParams()) + + matches = matcher(estimations, ground_truths) + + ego2map = HomogeneousMatrix( + position=(1, 0, 0), + rotation=(1, 0, 0, 0), + src="base_link", + dst="map", + ) + + frame = FrameBoxMatch(unix_time=100, frame_index=0, matches=matches, ego2map=ego2map) + + scorer = build_matching_scorer(params=MatchingParams()) + threshold = 1.0 + + assert frame.num_estimation == 3 + assert frame.num_gt == 3 + assert frame.num_tp(scorer=scorer, threshold=threshold) == 2 + assert frame.num_fp(scorer=scorer, threshold=threshold) == 1 + assert frame.num_fn(scorer=scorer, threshold=threshold) == 1 + assert frame.num_tn(scorer=scorer, threshold=threshold) == 0 diff --git a/tests/evaluation/result/test_status.py b/tests/evaluation/result/test_status.py new file mode 100644 index 00000000..93daf03d --- /dev/null +++ b/tests/evaluation/result/test_status.py @@ -0,0 +1,21 @@ +from __future__ import annotations + +from t4_devkit.evaluation import MatchingStatus + + +def test_matching_status() -> None: + # TP + assert MatchingStatus.TP.is_true() and MatchingStatus.TP.is_positive() + assert (not MatchingStatus.TP.is_false()) and (not MatchingStatus.TP.is_negative()) + + # FP + assert MatchingStatus.FP.is_false() and MatchingStatus.FP.is_positive() + assert (not MatchingStatus.FP.is_true()) and (not MatchingStatus.FP.is_negative()) + + # FN + assert MatchingStatus.FN.is_false() and MatchingStatus.FN.is_negative() + assert (not MatchingStatus.FN.is_true()) and (not MatchingStatus.FN.is_positive()) + + # TN + assert MatchingStatus.TN.is_true() and MatchingStatus.TN.is_negative() + assert (not MatchingStatus.TN.is_false()) and (not MatchingStatus.TN.is_positive()) diff --git a/tests/evaluation/test_dataset.py b/tests/evaluation/test_dataset.py new file mode 100644 index 00000000..1db504ee --- /dev/null +++ b/tests/evaluation/test_dataset.py @@ -0,0 +1,13 @@ +from __future__ import annotations + +import pytest + +from t4_devkit.evaluation import EvaluationTask, load_dataset + + +def test_load_dataset(dummy_dataset_root) -> None: + _ = load_dataset(dummy_dataset_root, task=EvaluationTask.DETECTION3D) + + with pytest.raises(NotImplementedError): + _ = load_dataset(dummy_dataset_root, task=EvaluationTask.SEGMENTATION3D) + _ = load_dataset(dummy_dataset_root, task=EvaluationTask.SEGMENTATION2D) diff --git a/tests/evaluation/test_task.py b/tests/evaluation/test_task.py new file mode 100644 index 00000000..eff80670 --- /dev/null +++ b/tests/evaluation/test_task.py @@ -0,0 +1,31 @@ +from __future__ import annotations + +from t4_devkit.evaluation import EvaluationTask + + +def test_task() -> None: + task_names = { + "detection3d", + "tracking3d", + "prediction3d", + "segmentation3d", + "detection2d", + "tracking2d", + "segmentation2d", + } + + assert task_names == {e.value for e in EvaluationTask} + + for name in task_names: + task = EvaluationTask(name) + + assert task == name + if name in ("detection3d", "tracking3d", "prediction3d", "segmentation3d"): + assert task.is_3d() + elif name in ("detection2d", "tracking2d", "segmentation2d"): + assert task.is_2d() + else: + raise ValueError(f"{name} doesn't have an indicator") + + if name in ("segmentation3d", "segmentation2d"): + assert task.is_segmentation()