Evaluation metrics (#4957)

Author: leoll2

application/backend/app/services/datumaro_converter.py

@@ -7,9 +7,9 @@
 
 import numpy as np
 import polars as pl
-from datumaro.experimental import Dataset, Sample, bbox_field, image_path_field, label_field
+from datumaro.experimental import Dataset, Sample, bbox_field, image_info_field, image_path_field, label_field
 from datumaro.experimental.categories import LabelCategories
-from datumaro.experimental.fields import polygon_field
+from datumaro.experimental.fields import ImageInfo, polygon_field
 
 from app.core.models.task_type import TaskType
 from app.db.schema import DatasetItemDB
@@ -25,22 +25,26 @@
 
 class DetectionSample(Sample):
     image: str = image_path_field()
+    image_info: ImageInfo = image_info_field()
     bboxes: np.ndarray[Any, Any] = bbox_field(dtype=pl.Int32)
     label: np.ndarray[Any, Any] = label_field(dtype=pl.Int32, is_list=True)
 
 
 class ClassificationSample(Sample):
     image: str = image_path_field()
+    image_info: ImageInfo = image_info_field()
     label: int = label_field(dtype=pl.Int32, is_list=False)
 
 
 class MultilabelClassificationSample(Sample):
     image: str = image_path_field()
-    label: np.ndarray[Any, Any] = label_field(dtype=pl.Int32, is_list=True)
+    image_info: ImageInfo = image_info_field()
+    label: np.ndarray[Any, Any] = label_field(dtype=pl.Int32, multi_label=True)
 
 
 class InstanceSegmentationSample(Sample):
     image: str = image_path_field()
+    image_info: ImageInfo = image_info_field()
     polygons: np.ndarray[Any, Any] = polygon_field(dtype=pl.Float32)
     label: np.ndarray[Any, Any] = label_field(dtype=pl.Int32, is_list=True)
 
@@ -137,6 +141,7 @@ def _convert_sample(
             return None
         return DetectionSample(
             image=image_path,
+            image_info=ImageInfo(width=dataset_item.width, height=dataset_item.height),
             bboxes=np.array(coords),
             label=np.array(labels_indexes),
         )
@@ -176,7 +181,11 @@ def _convert_sample(
             DatasetItemAnnotation.model_validate(annotation) for annotation in dataset_item.annotation_data
         )
         try:
-            return ClassificationSample(image=image_path, label=project_labels_ids.index(annotation.labels[0].id))
+            return ClassificationSample(
+                image=image_path,
+                image_info=ImageInfo(width=dataset_item.width, height=dataset_item.height),
+                label=project_labels_ids.index(annotation.labels[0].id),
+            )
         except ValueError:
             logger.error("Unable to find one of dataset item %s labels in project", dataset_item.id)
             return None
@@ -220,7 +229,11 @@ def _convert_sample(
         except ValueError:
             logger.error("Unable to find one of dataset item %s labels in project", dataset_item.id)
             return None
-        return MultilabelClassificationSample(image=image_path, label=np.array(labels_indexes))
+        return MultilabelClassificationSample(
+            image=image_path,
+            image_info=ImageInfo(width=dataset_item.width, height=dataset_item.height),
+            label=np.array(labels_indexes),
+        )
 
     return _convert_dataset(
         sample_type=MultilabelClassificationSample,
@@ -267,7 +280,10 @@ def _convert_sample(
             logger.error("Unable to find one of dataset item %s labels in project", dataset_item.id)
             return None
         return InstanceSegmentationSample(
-            image=image_path, polygons=np.array(polygons, dtype=np.float32), label=np.array(labels_indexes)
+            image=image_path,
+            image_info=ImageInfo(width=dataset_item.width, height=dataset_item.height),
+            polygons=np.array(polygons, dtype=np.float32),
+            label=np.array(labels_indexes),
         )
 
     return _convert_dataset(

application/backend/app/services/evaluation/__init__.py

@@ -0,0 +1,22 @@
+# Copyright (C) 2025 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+from .evaluators import (
+    AveragingMethod,
+    DetectionEvaluator,
+    Evaluator,
+    InstanceSegmentationEvaluator,
+    MultiClassClassificationEvaluator,
+    MultiLabelClassificationEvaluator,
+)
+from .factory import EvaluatorFactory
+
+__all__ = [
+    "AveragingMethod",
+    "DetectionEvaluator",
+    "Evaluator",
+    "EvaluatorFactory",
+    "InstanceSegmentationEvaluator",
+    "MultiClassClassificationEvaluator",
+    "MultiLabelClassificationEvaluator",
+]

application/backend/app/services/evaluation/evaluators.py

@@ -0,0 +1,254 @@
+# Copyright (C) 2025 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+from abc import ABCMeta, abstractmethod
+from enum import StrEnum
+
+import numpy as np
+from datumaro.experimental import Dataset
+from faster_coco_eval import COCO, COCOeval_faster
+from numpy.typing import NDArray
+from sklearn.metrics import accuracy_score, confusion_matrix, f1_score, precision_score, recall_score
+from sklearn.preprocessing import MultiLabelBinarizer
+
+from app.services.datumaro_converter import DetectionSample
+
+
+def datumaro_dataset_to_coco(dataset: Dataset) -> dict:
+    """
+    Convert Datumaro Dataset to COCO format.
+
+    Supports detection (DetectionSample) and instance segmentation (InstanceSegmentationSample) datasets.
+
+    Args:
+        dataset (Dataset): Datumaro Dataset to convert.
+    Returns:
+        dict: COCO formatted dataset.
+    """
+    coco_dataset_dict: dict[str, list] = {"images": [], "annotations": [], "categories": []}
+
+    # Add categories
+    for label_idx, label in enumerate(dataset.schema.attributes["label"].categories.labels):
+        coco_dataset_dict["categories"].append(
+            {
+                "id": label_idx,
+                "name": label,
+            }
+        )
+
+    annotation_id = 1  # COCOeval ignores annotation ID 0
+    for image_id, sample in enumerate(dataset):
+        # Add image entry
+        coco_dataset_dict["images"].append(
+            {
+                "id": image_id,
+                "file_name": sample.image,
+                "width": sample.image_info.width,
+                "height": sample.image_info.height,
+            }
+        )
+
+        # Detection
+        if hasattr(sample, "bboxes") and sample.bboxes is not None:
+            for bbox, label_idx in zip(sample.bboxes, sample.label):
+                x1, y1, x2, y2 = bbox
+                width = x2 - x1
+                height = y2 - y1
+                coco_dataset_dict["annotations"].append(
+                    {
+                        "id": annotation_id,
+                        "image_id": image_id,
+                        "category_id": int(label_idx),
+                        "bbox": [float(x1), float(y1), float(width), float(height)],
+                        "score": 1.0,
+                    }
+                )
+                annotation_id += 1
+
+        # Instance Segmentation
+        if hasattr(sample, "polygons") and sample.polygons is not None:
+            for polygon, label_idx in zip(sample.polygons, sample.label):
+                flattened_polygon = [coord for point in polygon for coord in point]
+                x_coords = [point[0] for point in polygon]
+                y_coords = [point[1] for point in polygon]
+                x_min, x_max = min(x_coords), max(x_coords)
+                y_min, y_max = min(y_coords), max(y_coords)
+                width = x_max - x_min
+                height = y_max - y_min
+                coco_dataset_dict["annotations"].append(
+                    {
+                        "id": annotation_id,
+                        "image_id": image_id,
+                        "category_id": int(label_idx),
+                        "segmentation": [flattened_polygon],
+                        "bbox": [float(x_min), float(y_min), float(width), float(height)],
+                        "score": 1.0,
+                    }
+                )
+                annotation_id += 1
+
+    return coco_dataset_dict
+
+
+class AveragingMethod(StrEnum):
+    MICRO = "micro"
+    MACRO = "macro"
+    WEIGHTED = "weighted"
+    SAMPLES = "samples"
+
+
+class EvaluatorBase(metaclass=ABCMeta):
+    """Base class for all evaluators."""
+
+    def __init__(self, predictions_dataset: Dataset, ground_truth_dataset: Dataset):
+        self.predictions_dataset = predictions_dataset
+        self.ground_truth_dataset = ground_truth_dataset
+
+
+class EvaluatorWithLabelArrays(EvaluatorBase):
+    """Base evaluator for tasks that use label arrays."""
+
+    def __init__(self, predictions_dataset: Dataset, ground_truth_dataset: Dataset):
+        super().__init__(predictions_dataset=predictions_dataset, ground_truth_dataset=ground_truth_dataset)
+        self.__pred_labels: NDArray[np.int_] | None = None
+        self.__gt_labels: NDArray[np.int_] | None = None
+
+    @abstractmethod
+    def _build_label_arrays(self) -> tuple[NDArray[np.int_], NDArray[np.int_]]:
+        """Set up the prediction and ground truth label arrays."""
+
+    @property
+    def _pred_labels(self) -> NDArray[np.int_]:
+        if self.__pred_labels is None:
+            self.__gt_labels, self.__pred_labels = self._build_label_arrays()
+        return self.__pred_labels
+
+    @property
+    def _gt_labels(self) -> NDArray[np.int_]:
+        if self.__gt_labels is None:
+            self.__gt_labels, self.__pred_labels = self._build_label_arrays()
+        return self.__gt_labels
+
+
+class AccuracyEvaluator(EvaluatorWithLabelArrays):
+    """Evaluator for accuracy, precision, recall, and F1 metrics."""
+
+    def __init__(self, predictions_dataset: Dataset, ground_truth_dataset: Dataset):
+        super().__init__(predictions_dataset=predictions_dataset, ground_truth_dataset=ground_truth_dataset)
+
+    def precision(self, averaging_method: AveragingMethod = AveragingMethod.MACRO) -> float:
+        return precision_score(y_true=self._gt_labels, y_pred=self._pred_labels, average=averaging_method.value)
+
+    def recall(self, averaging_method: AveragingMethod = AveragingMethod.MACRO) -> float:
+        return recall_score(y_true=self._gt_labels, y_pred=self._pred_labels, average=averaging_method.value)
+
+    def accuracy(self) -> float:
+        return accuracy_score(y_true=self._gt_labels, y_pred=self._pred_labels)
+
+    def f1_score(self, averaging_method: AveragingMethod = AveragingMethod.MACRO) -> float:
+        return f1_score(y_true=self._gt_labels, y_pred=self._pred_labels, average=averaging_method.value)
+
+
+class ConfusionMatrixEvaluator(EvaluatorWithLabelArrays):
+    """Evaluator for confusion matrix computation."""
+
+    def __init__(self, predictions_dataset: Dataset, ground_truth_dataset: Dataset):
+        super().__init__(predictions_dataset=predictions_dataset, ground_truth_dataset=ground_truth_dataset)
+
+    def confusion_matrix(self) -> np.ndarray:
+        """Compute the confusion matrix"""
+        return confusion_matrix(y_true=self._gt_labels, y_pred=self._pred_labels)
+
+
+class MeanAveragePrecisionEvaluator(EvaluatorBase):
+    """Evaluator for mean average precision (mAP) metrics."""
+
+    def __init__(self, predictions_dataset: Dataset, ground_truth_dataset: Dataset):
+        super().__init__(predictions_dataset=predictions_dataset, ground_truth_dataset=ground_truth_dataset)
+        self.__gt_coco_dict: dict | None = None
+        self.__pred_coco_dict: dict | None = None
+
+    @property
+    def _gt_coco_dict(self) -> dict:
+        if self.__gt_coco_dict is None:
+            self.__gt_coco_dict = datumaro_dataset_to_coco(self.ground_truth_dataset)
+        return self.__gt_coco_dict
+
+    @property
+    def _pred_coco_dict(self) -> dict:
+        if self.__pred_coco_dict is None:
+            self.__pred_coco_dict = datumaro_dataset_to_coco(self.predictions_dataset)
+        return self.__pred_coco_dict
+
+    def mean_average_precision(self) -> dict:
+        gt_coco = COCO(self._gt_coco_dict)
+        pred_coco = gt_coco.loadRes(self._pred_coco_dict["annotations"])
+        coco_evaluator = COCOeval_faster(
+            cocoGt=gt_coco,
+            cocoDt=pred_coco,
+            iouType="bbox" if self.predictions_dataset.dtype is DetectionSample else "segm",
+        )
+        coco_evaluator.run()
+        return coco_evaluator.stats_as_dict
+
+
+class MultiClassClassificationEvaluator(AccuracyEvaluator, ConfusionMatrixEvaluator):
+    """Evaluator for multi-class classification tasks."""
+
+    def __init__(self, predictions_dataset: Dataset, ground_truth_dataset: Dataset):
+        if (
+            predictions_dataset.schema.attributes["label"].annotation.multi_label
+            or ground_truth_dataset.schema.attributes["label"].annotation.multi_label
+        ):
+            raise ValueError(f"{self.__class__.__name__} should not be used for multi-label classification datasets")
+
+        AccuracyEvaluator.__init__(
+            self, predictions_dataset=predictions_dataset, ground_truth_dataset=ground_truth_dataset
+        )
+        ConfusionMatrixEvaluator.__init__(
+            self, predictions_dataset=predictions_dataset, ground_truth_dataset=ground_truth_dataset
+        )
+
+    def _build_label_arrays(self) -> tuple[NDArray[np.int_], NDArray[np.int_]]:
+        pred_labels = np.array([sample.label for sample in self.predictions_dataset], dtype=int)
+        gt_labels = np.array([sample.label for sample in self.ground_truth_dataset], dtype=int)
+        return gt_labels, pred_labels
+
+
+class MultiLabelClassificationEvaluator(AccuracyEvaluator):
+    """Evaluator for multi-label classification tasks."""
+
+    def __init__(self, predictions_dataset: Dataset, ground_truth_dataset: Dataset):
+        if not (
+            predictions_dataset.schema.attributes["label"].annotation.multi_label
+            and ground_truth_dataset.schema.attributes["label"].annotation.multi_label
+        ):
+            raise ValueError(f"{self.__class__.__name__} should only be used for multi-label classification datasets")
+
+        AccuracyEvaluator.__init__(
+            self, predictions_dataset=predictions_dataset, ground_truth_dataset=ground_truth_dataset
+        )
+
+    def _build_label_arrays(self) -> tuple[NDArray[np.int_], NDArray[np.int_]]:
+        mlb = MultiLabelBinarizer()
+        gt_labels_list = [s.label for s in self.ground_truth_dataset]
+        pred_labels_list = [s.label for s in self.predictions_dataset]
+        gt_labels = mlb.fit_transform(gt_labels_list)
+        pred_labels = mlb.transform(pred_labels_list)
+        return gt_labels, pred_labels
+
+
+class DetectionEvaluator(MeanAveragePrecisionEvaluator):
+    """Evaluator for object detection tasks."""
+
+
+class InstanceSegmentationEvaluator(MeanAveragePrecisionEvaluator):
+    """Evaluator for instance segmentation tasks."""
+
+
+Evaluator = (
+    MultiClassClassificationEvaluator
+    | MultiLabelClassificationEvaluator
+    | DetectionEvaluator
+    | InstanceSegmentationEvaluator
+)