diff --git a/supervision/metrics/__init__.py b/supervision/metrics/__init__.py index e69de29b..c8b01dbc 100644 --- a/supervision/metrics/__init__.py +++ b/supervision/metrics/__init__.py @@ -0,0 +1,2 @@ +from supervision.metrics.core import UnsupportedMetricTargetError, Metric, MetricTarget +from supervision.metrics.intersection_over_union import IntersectionOverUnion \ No newline at end of file diff --git a/supervision/metrics/core.py b/supervision/metrics/core.py index 50da29fd..1011ab5b 100644 --- a/supervision/metrics/core.py +++ b/supervision/metrics/core.py @@ -1,9 +1,16 @@ from __future__ import annotations - from abc import ABC, abstractmethod from enum import Enum -from typing import Any +from typing import Any, Dict, Iterator, Optional, Tuple, Union +import numpy as np +import numpy.typing as npt + +from supervision import config +from supervision.detection.core import Detections + +"""Used by metrics module as class ID, when none is present""" +CLASS_ID_NONE = -1 class Metric(ABC): """ @@ -36,14 +43,15 @@ class Metric(ABC): Compute the metric from the internal state and return the result. """ raise NotImplementedError - - @abstractmethod - def to_pandas(self, *args, **kwargs) -> Any: - """ - Return a pandas DataFrame representation of the metric. - """ - self._ensure_pandas_installed() - raise NotImplementedError + + # TODO: determine if this is necessary. + # @abstractmethod + # def to_pandas(self, *args, **kwargs) -> Any: + # """ + # Return a pandas DataFrame representation of the metric. + # """ + # self._ensure_pandas_installed() + # raise NotImplementedError def _ensure_pandas_installed(self): try: @@ -73,3 +81,113 @@ class UnsupportedMetricTargetError(Exception): def __init__(self, metric: Metric, target: MetricTarget): super().__init__(f"Metric {metric} does not support target {target}") + + +class InternalMetricDataStore: + """ + Stores internal data of IntersectionOverUnion metric: + * Stores the basic data: boxes, masks, or oriented bounding boxes + * Validates data: ensures data types and shape are consistent + * Provides iteration by class + """ + + def __init__( + self, + metric_target: MetricTarget, + class_agnostic: bool + ): + self._metric_target = metric_target + self._class_agnostic = class_agnostic + self._data_1: Dict[int, npt.NDArray] + self._data_2: Dict[int, npt.NDArray] + self._datapoint_shape: Optional[Tuple[int, ...]] + self.reset() + + def reset(self) -> None: + self._data_1 = {} + self._data_2 = {} + if self._metric_target == MetricTarget.BOXES: + self._datapoint_shape = (4,) + elif self._metric_target == MetricTarget.MASKS: + # Determined when adding data + self._datapoint_shape = None + elif self._metric_target == MetricTarget.ORIENTED_BOUNDING_BOXES: + self._datapoint_shape = (8,) + + def update( + self, + data_1: Union[npt.NDArray, Detections], + data_2: Union[npt.NDArray, Detections], + ) -> None: + content_1 = self._get_content(data_1) + content_2 = self._get_content(data_2) + class_ids_1 = self._get_class_ids(data_1) + class_ids_2 = self._get_class_ids(data_2) + self._validate_class_ids(class_ids_1) + self._validate_class_ids(class_ids_2) + + if content_1 is not None and len(content_1) > 0: + assert len(content_1) == len(class_ids_1) + for class_id in set(class_ids_1): + content_of_class = content_1[class_ids_1 == class_id] + if class_id not in self._data_1: + self._data_1[class_id] = content_of_class + continue + self._data_1[class_id] = np.vstack((self._data_1[class_id], content_of_class)) + + if content_2 is not None and len(content_2) > 0: + assert len(content_2) == len(class_ids_2) + for class_id in set(class_ids_2): + content_of_class = content_2[class_ids_2 == class_id] + if class_id not in self._data_2: + self._data_2[class_id] = content_of_class + continue + self._data_2[class_id] = np.vstack((self._data_2[class_id], content_of_class)) + + def __iter__(self) -> Iterator[Tuple[int, Optional[npt.NDArray], Optional[npt.NDArray]]]: + class_ids = sorted(set.union( + set(self._data_1.keys()), + set(self._data_2.keys()) + )) + for class_id in class_ids: + yield class_id, self._data_1.get(class_id, None), self._data_2.get(class_id, None) + + def _get_content(self, data: Union[npt.NDArray, Detections]) -> Optional[npt.NDArray]: + """Return boxes, masks or oriented bounding boxes from the data.""" + if isinstance(data, np.ndarray): + return data + assert isinstance(data, Detections) + + if self._metric_target == MetricTarget.BOXES: + return data.xyxy + if self._metric_target == MetricTarget.MASKS: + return data.mask + if self._metric_target == MetricTarget.ORIENTED_BOUNDING_BOXES: + obb = data.data.get(config.ORIENTED_BOX_COORDINATES, None) + if isinstance(obb, list): + obb = np.array(obb, dtype=np.float32) + return obb + raise ValueError(f"Invalid metric target: {self._metric_target}") + + def _get_class_ids(self, data: Union[npt.NDArray, Detections]) -> npt.NDArray[np.int_]: + if self._class_agnostic or isinstance(data, np.ndarray): + return np.array([CLASS_ID_NONE] * len(data), dtype=int) + assert isinstance(data, Detections) + if data.class_id is None: + return np.array([CLASS_ID_NONE] * len(data), dtype=int) + return data.class_id + + def _validate_class_ids(self, class_id: npt.NDArray[np.int_]) -> None: + class_set = set(class_id) + if len(class_set) >= 2 and -1 in class_set: + raise ValueError("Metrics store received results with partially defined classes.") + + def _validate_shape(self, data: npt.NDArray) -> None: + if self._datapoint_shape is None: + assert self._metric_target == MetricTarget.MASKS + self._datapoint_shape = data.shape[1:] + return + if data.shape[1:] != self._datapoint_shape: + raise ValueError( + f"Invalid data shape: {data.shape}. Expected: (N, {self._datapoint_shape})" + ) diff --git a/supervision/metrics/intersection_over_union.py b/supervision/metrics/intersection_over_union.py index d8850b58..322329dd 100644 --- a/supervision/metrics/intersection_over_union.py +++ b/supervision/metrics/intersection_over_union.py @@ -1,194 +1,16 @@ -from typing import TYPE_CHECKING, Dict, Iterator, Optional, Tuple, Union +from typing import Dict, Union, TYPE_CHECKING import numpy as np import numpy.typing as npt import supervision.config as config from supervision.detection.core import Detections -from supervision.metrics.core import Metric, MetricTarget +from supervision.metrics.core import InternalMetricDataStore, Metric, MetricTarget if TYPE_CHECKING: import pandas as pd -CLASS_ID_NONE = -1 - -Data = Union[npt.NDArray, Detections] - - -class InternalMetricDataStore: - """ - Stores internal data of IntersectionOverUnion metric: - * Stores the basic data: boxes, masks, or oriented bounding boxes - * Validates data: ensures data types and shape are consistent - * Provides iteration by class - """ - - def __init__(self, metric_target: MetricTarget, class_agnostic: bool): - self._metric_target = metric_target - self._class_agnostic = class_agnostic - self._data_1: Dict[int, npt.NDArray] - self._data_2: Dict[int, npt.NDArray] - self._datapoint_shape: Optional[Tuple[int]] - self.reset() - - def reset(self) -> None: - self._data_1 = {} - self._data_2 = {} - if self._metric_target == MetricTarget.BOXES: - self._datapoint_shape = (4,) - elif self._metric_target == MetricTarget.MASKS: - # Determined when adding data - self._datapoint_shape = None - elif self._metric_target == MetricTarget.ORIENTED_BOUNDING_BOXES: - self._datapoint_shape = (8,) - - # def update( - # self, - # data_1: Union[npt.NDArray, Detections], - # data_2: Union[npt.NDArray, Detections], - # ) -> None: - # # This class dispatches to helper self._add_to_list, which does more validation - # # Then calls self._vstack which generates the final result - # if len(data_1) == 0 and len(data_2) == 0: - # return - - # if type(data_1) != type(data_2): - # raise ValueError( - # f"Data types must match. Got {type(data_1)=} and {type(data_2)=}." - # ) - - # if isinstance(data_1, npt.NDArray): - # assert isinstance(data_2, npt.NDArray) - # self._update(data_1, class_id=CLASS_ID_NONE, data_id=1) - # self._update(data_2, class_id=CLASS_ID_NONE, data_id=2) - # return - # assert isinstance(data_1, Detections) - # assert isinstance(data_2, Detections) - - # if self._class_agnostic: - # self._update(self._get_detections_content(data_1), class_id=CLASS_ID_NONE, data_id=1) - # self._update(self._get_detections_content(data_2), class_id=CLASS_ID_NONE, data_id=2) - # return - - # if data_1.class_id is None: - # self._update(self._get_detections_content(data_1), class_id=CLASS_ID_NONE, data_id=1) - # else: - # for class_id in set(data_1.class_id): - # data_1_of_class = data_1[data_1.class_id == class_id] - # assert isinstance(data_1_of_class, Detections) - # self._update(self._get_detections_content(data_1_of_class), class_id=class_id, data_id=1) - - # if data_2.class_id is None: - # self._update(self._get_detections_content(data_2), class_id=CLASS_ID_NONE, data_id=2) - # else: - # for class_id in set(data_2.class_id): - # data_2_of_class = data_2[data_2.class_id == class_id] - # assert isinstance(data_2_of_class, Detections) - # self._update(self._get_detections_content(data_2_of_class), class_id=class_id, data_id=2) - - def update( - self, - data_1: Union[npt.NDArray, Detections], - data_2: Union[npt.NDArray, Detections], - ) -> None: - content_1 = self._get_content(data_1) - content_2 = self._get_content(data_2) - class_ids_1 = self._get_class_ids(data_1) - class_ids_2 = self._get_class_ids(data_2) - self._validate_class_ids(class_ids_1) - self._validate_class_ids(class_ids_2) - - if content_1 is not None and len(content_1) > 0: - assert len(content_1) == len(class_ids_1) - for class_id in set(class_ids_1): - content_of_class = content_1[class_ids_1 == class_id] - if class_id not in self._data_1: - self._data_1[class_id] = content_of_class - continue - self._data_1[class_id] = np.vstack( - (self._data_1[class_id], content_of_class) - ) - - if content_2 is not None and len(content_2) > 0: - assert len(content_2) == len(class_ids_2) - for class_id in set(class_ids_2): - content_of_class = content_2[class_ids_2 == class_id] - if class_id not in self._data_2: - self._data_2[class_id] = content_of_class - continue - self._data_2[class_id] = np.vstack( - (self._data_2[class_id], content_of_class) - ) - - def __iter__( - self, - ) -> Iterator[Tuple[int, Optional[npt.NDArray], Optional[npt.NDArray]]]: - class_ids = sorted( - set.union(set(self._data_1.keys()), set(self._data_2.keys())) - ) - for class_id in class_ids: - yield ( - class_id, - self._data_1.get(class_id, None), - self._data_2.get(class_id, None), - ) - - def _get_content(self, data: Data) -> Optional[npt.NDArray]: - """Return boxes, masks or oriented bounding boxes from the data.""" - if isinstance(data, npt.NDArray): - return data - assert isinstance(data, Detections) - - if self._metric_target == MetricTarget.BOXES: - return data.xyxy - if self._metric_target == MetricTarget.MASKS: - return data.mask - if self._metric_target == MetricTarget.ORIENTED_BOUNDING_BOXES: - obb = data.data.get(config.ORIENTED_BOX_COORDINATES, None) - if isinstance(obb, list): - obb = np.array(obb, dtype=np.float32) - return obb - raise ValueError(f"Invalid metric target: {self._metric_target}") - - def _get_class_ids(self, data: Data) -> npt.NDArray[np.int_]: - if self._class_agnostic or isinstance(data, npt.NDArray): - return np.array([CLASS_ID_NONE] * len(data), dtype=int) - assert isinstance(data, Detections) - if data.class_id is None: - return np.array([CLASS_ID_NONE] * len(data), dtype=int) - return data.class_id - - # def _get_detections_content(self, data: Detections) -> Optional[npt.NDArray]: - # if self._metric_target == MetricTarget.BOXES: - # return data.xyxy - # if self._metric_target == MetricTarget.MASKS: - # return data.mask - # if self._metric_target == MetricTarget.ORIENTED_BOUNDING_BOXES: - # obb = data.data.get(config.ORIENTED_BOX_COORDINATES, None) - # if isinstance(obb, list): - # obb = np.array(obb, dtype=np.float32) - # return obb - # raise ValueError(f"Invalid metric target: {self._metric_target}") - - def _validate_class_ids(self, class_id: npt.NDArray[np.int_]) -> None: - class_set = set(class_id) - if len(class_set) >= 2 and -1 in class_set: - raise ValueError( - "Metrics store received results with partially defined classes." - ) - - def _validate_shape(self, data: npt.NDArray) -> None: - if self._datapoint_shape is None: - assert self._metric_target == MetricTarget.MASKS - self._datapoint_shape = data.shape[1:] - return - if data.shape[1:] != self._datapoint_shape: - raise ValueError( - f"Invalid data shape: {data.shape}. Expected: (N, {self._datapoint_shape})" - ) - - class IntersectionOverUnion(Metric): def __init__( self, @@ -261,22 +83,36 @@ class IntersectionOverUnion(Metric): ious[class_id] = iou return ious - def to_pandas(self) -> pd.DataFrame: - """ - Return a pandas DataFrame representation of the metric. - """ - self._ensure_pandas_installed() - import pandas as pd + # TODO: This would return dict[int, pd.DataFrame]. Do we want that? + # It'd be cleaner if it returned a single DataFrame, but the sizes + # differ if class_agnostic=False. - # TODO: use cache results instead - ious = self.compute() + # def to_pandas(self) -> 'pd.DataFrame': + # """ + # Return a pandas DataFrame representation of the metric. + # """ + # self._ensure_pandas_installed() + # import pandas as pd - # TODO: continue + # # TODO: use cached results + # ious = self.compute() + # print(len(ious)) - # data_frame = pd.DataFrame() + # class_names = [] + # arrays = [] - # return s - return pd.DataFrame() + # for class_id, array in ious.items(): + # print(array.shape) + # class_names.append(np.full(array.shape[0], class_id)) + # arrays.append(array) + # stacked_class_ids = np.concatenate(class_names) + # stacked_ious = np.vstack(arrays) + # combined = np.column_stack((stacked_class_ids, stacked_ious)) + + # column_names = ['class_id'] + [f'col_{i+1}' for i in range(stacked_ious.shape[1])] + # result = pd.DataFrame(combined, columns=column_names) + + # return result @staticmethod def _compute_box_iou( diff --git a/test/metrics/test_core.py b/test/metrics/test_core.py new file mode 100644 index 00000000..0b5ec8b6 --- /dev/null +++ b/test/metrics/test_core.py @@ -0,0 +1,282 @@ +from contextlib import ExitStack as DoesNotRaise +from test.test_utils import mock_detections +from typing import List, Optional, Tuple, Union + +import numpy as np +import numpy.typing as npt +import pytest + +from supervision.detection.core import Detections +from supervision.metrics.core import ( + InternalMetricDataStore, + MetricTarget, + CLASS_ID_NONE +) + +# Boxes, class-agnostic +@pytest.mark.parametrize( + "data_1, data_2, expected_result, exception", + [ + # Empty + ( + Detections.empty(), + Detections.empty(), + [], + DoesNotRaise() + ), + ( + np.empty((0, 4), dtype=np.float32), + np.empty((0, 4), dtype=np.float32), + [], + DoesNotRaise() + ), + ( + Detections.empty(), + np.empty((0, 4), dtype=np.float32), + [], + DoesNotRaise() + ), + + # Single box + Empty + ( + Detections( + xyxy=np.array([[0, 0, 1, 1]], dtype=np.float32) + ), + Detections.empty(), + [ + ( + CLASS_ID_NONE, + np.array([[0, 0, 1, 1]], dtype=np.float32), + None + ) + ], + DoesNotRaise() + ), + ( + Detections.empty(), + Detections( + xyxy=np.array([[0, 0, 1, 1]], dtype=np.float32) + ), + [ + ( + CLASS_ID_NONE, + None, + np.array([[0, 0, 1, 1]], dtype=np.float32) + ) + ], + DoesNotRaise() + ), + + # Multiple boxes + ( + Detections( + xyxy=np.array([[0, 0, 1, 1], [0, 0, 2, 2]], dtype=np.float32) + ), + Detections.empty(), + [ + ( + CLASS_ID_NONE, + np.array([[0, 0, 1, 1], [0, 0, 2, 2]], dtype=np.float32), + None + ) + ], + DoesNotRaise() + ), + ( + Detections( + xyxy=np.array([[0, 0, 1, 1], [0, 0, 2, 2]], dtype=np.float32) + ), + Detections( + xyxy=np.array([[0, 0, 1, 1], [0, 0, 2, 2]], dtype=np.float32) + ), + [ + ( + CLASS_ID_NONE, + np.array([[0, 0, 1, 1], [0, 0, 2, 2]], dtype=np.float32), + np.array([[0, 0, 1, 1], [0, 0, 2, 2]], dtype=np.float32) + ) + ], + DoesNotRaise() + ), + ( + Detections( + xyxy=np.array([[0, 0, 1, 1], [0, 0, 2, 2]], dtype=np.float32) + ), + np.array([[0, 0, 1, 1], [0, 0, 2, 2]], dtype=np.float32), + [ + ( + CLASS_ID_NONE, + np.array([[0, 0, 1, 1], [0, 0, 2, 2]], dtype=np.float32), + np.array([[0, 0, 1, 1], [0, 0, 2, 2]], dtype=np.float32) + ) + ], + DoesNotRaise() + ), + + # with classes - should be ignored. + ( + Detections( + xyxy=np.array([[0, 0, 1, 1], [0, 0, 2, 2]], dtype=np.float32), + class_id=np.array([1, 2], dtype=int) + ), + np.array([[0, 0, 1, 1], [0, 0, 2, 2]], dtype=np.float32), + [ + ( + CLASS_ID_NONE, + np.array([[0, 0, 1, 1], [0, 0, 2, 2]], dtype=np.float32), + np.array([[0, 0, 1, 1], [0, 0, 2, 2]], dtype=np.float32) + ) + ], + DoesNotRaise() + ), + ] +) +def test_store_boxes_class_agnostic( + data_1: Union[npt.NDArray, Detections], + data_2: Union[npt.NDArray, Detections], + expected_result: List[Tuple[int, Optional[npt.NDArray], Optional[npt.NDArray]]], + exception: Exception +) -> None: + store = InternalMetricDataStore(MetricTarget.BOXES, class_agnostic=True) + store.update(data_1, data_2) + result = [result for result in store] + assert len(result) == len(expected_result) + with exception: + for (class_id, content_1, content_2), (expected_class_id, expected_content_1, expected_content_2) in zip(result, expected_result): + assert class_id == expected_class_id + assert (content_1 is None and expected_content_1 is None) or np.array_equal(content_1, expected_content_1) + assert (content_2 is None and expected_content_2 is None) or np.array_equal(content_2, expected_content_2) + +# Boxes, by-class +@pytest.mark.parametrize( + "data_1, data_2, expected_result, exception", + [ + # Single box + Empty + ( + Detections( + xyxy=np.array([[0, 0, 1, 1]], dtype=np.float32), + class_id=np.array([1], dtype=int) + ), + Detections.empty(), + [ + ( + 1, + np.array([[0, 0, 1, 1]], dtype=np.float32), + None + ) + ], + DoesNotRaise() + ), + ( + Detections.empty(), + Detections( + xyxy=np.array([[0, 0, 1, 1]], dtype=np.float32), + class_id=np.array([1], dtype=int) + ), + [ + ( + 1, + None, + np.array([[0, 0, 1, 1]], dtype=np.float32) + ) + ], + DoesNotRaise() + ), + + # Multiple classes + ( + Detections( + xyxy=np.array([[0, 0, 1, 1], [0, 0, 2, 2]], dtype=np.float32), + class_id=np.array([1, 2], dtype=int) + ), + Detections.empty(), + [ + ( + 1, + np.array([[0, 0, 1, 1]], dtype=np.float32), + None + ), + ( + 2, + np.array([[0, 0, 2, 2]], dtype=np.float32), + None + ) + ], + DoesNotRaise() + ), + ( + Detections( + xyxy=np.array([[0, 0, 1, 1], [0, 0, 2, 2]], dtype=np.float32), + class_id=np.array([1, 2], dtype=int) + ), + Detections( + xyxy=np.array([[0, 0, 1, 1], [0, 0, 2, 2]], dtype=np.float32), + class_id=np.array([2, 3], dtype=int) + ), + [ + ( + 1, + np.array([[0, 0, 1, 1]], dtype=np.float32), + None + ), + ( + 2, + np.array([[0, 0, 2, 2]], dtype=np.float32), + np.array([[0, 0, 1, 1]], dtype=np.float32) + ), + ( + 3, + None, + np.array([[0, 0, 2, 2]], dtype=np.float32) + ) + ], + DoesNotRaise() + ), + ( + Detections( + xyxy=np.array([[0, 0, 1, 1], [0, 0, 2, 2]], dtype=np.float32) + ), + np.array([[0, 0, 1, 1], [0, 0, 2, 2]], dtype=np.float32), + [ + ( + CLASS_ID_NONE, + np.array([[0, 0, 1, 1], [0, 0, 2, 2]], dtype=np.float32), + np.array([[0, 0, 1, 1], [0, 0, 2, 2]], dtype=np.float32) + ) + ], + DoesNotRaise() + ), + + # with classes - should be ignored. + ( + Detections( + xyxy=np.array([[0, 0, 1, 1], [0, 0, 2, 2]], dtype=np.float32), + class_id=np.array([1, 2], dtype=int) + ), + np.array([[0, 0, 1, 1], [0, 0, 2, 2]], dtype=np.float32), + [ + ( + CLASS_ID_NONE, + np.array([[0, 0, 1, 1], [0, 0, 2, 2]], dtype=np.float32), + np.array([[0, 0, 1, 1], [0, 0, 2, 2]], dtype=np.float32) + ) + ], + DoesNotRaise() + ), + ] +) +def test_store_boxes_by_class( + data_1: Union[npt.NDArray, Detections], + data_2: Union[npt.NDArray, Detections], + expected_result: List[Tuple[int, Optional[npt.NDArray], Optional[npt.NDArray]]], + exception: Exception +) -> None: + store = InternalMetricDataStore(MetricTarget.BOXES, class_agnostic=False) + store.update(data_1, data_2) + result = [result for result in store] + assert len(result) == len(expected_result) + with exception: + for (class_id, content_1, content_2), (expected_class_id, expected_content_1, expected_content_2) in zip(result, expected_result): + assert class_id == expected_class_id + assert (content_1 is None and expected_content_1 is None) or np.array_equal(content_1, expected_content_1) + assert (content_2 is None and expected_content_2 is None) or np.array_equal(content_2, expected_content_2)