Metrics changes + tests

This commit is contained in:
LinasKo 2024-08-12 11:16:50 +03:00
parent b061f4fd84
commit 4c903cf7f2
4 changed files with 440 additions and 202 deletions

View File

@ -0,0 +1,2 @@
from supervision.metrics.core import UnsupportedMetricTargetError, Metric, MetricTarget
from supervision.metrics.intersection_over_union import IntersectionOverUnion

View File

@ -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})"
)

View File

@ -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(

282
test/metrics/test_core.py Normal file
View File

@ -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)