Add files

README.md

@@ -1,10 +1,168 @@
 ---
 title: Detection Metrics
-emoji: 🌖
+emoji: 📈
 colorFrom: green
-colorTo: pink
+colorTo: indigo
 sdk: static
-pinned: false
+app_file: README.md
+pinned: true
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+![alt text](https://huggingface.co/spaces/rafaelpadilla/detection_metrics/resolve/main/assets/metrics_small.png)
+
+This project implements object detection **Average Precision** metrics using COCO style.  
+
+With `Detection Metrics` you can easily compute all 12 COCO metrics given the bounding boxes output by your object detection model:
+
+### Average Precision (AP):
+1. **AP**: AP at IoU=.50:.05:.95 
+2. **AP<sup>IoU=.50</sup>**: AP at IoU=.50 (similar to mAP PASCAL VOC metric)  
+3. **AP<sup>IoU=.75%</sup>**: AP at IoU=.75 (strict metric)  
+
+### AP Across Scales:
+4. **AP<sup>small</sup>**: AP for small objects: area < 322  
+5. **AP<sup>medium</sup>**: AP for medium objects: 322 < area < 962  
+6. **AP<sup>large</sup>**: AP for large objects: area > 962  
+
+### Average Recall (AR):
+7. **AR<sup>max=1</sup>**: AR given 1 detection per image  
+8. **AR<sup>max=10</sup>**: AR given 10 detections per image  
+9. **AR<sup>max=100</sup>**: AR given 100 detections per image 
+
+### AR Across Scales:
+10. **AR<sup>small</sup>**: AR for small objects: area < 322 
+11. **AR<sup>medium</sup>**: AR for medium objects: 322 < area < 962 
+12. **AR<sup>large</sup>**: AR for large objects: area > 962
+
+## How to use detection metrics?
+
+Basically, you just need to create your ground-truth data and prepare your evaluation loop to output the boxes, confidences and classes in the required format. Follow these steps:  
+
+
+### Step 1: Prepare your ground-truth dataset
+
+Convert your ground-truth annotations in JSON following the COCO format.
+COCO ground-truth annotations are represented in a dictionary containing 3 elements: "images", "annotations" and "categories". 
+The snippet below shows an example of the dictionary, and you can find [here](https://towardsdatascience.com/how-to-work-with-object-detection-datasets-in-coco-format-9bf4fb5848a4).
+
+```
+{
+    "images": [
+        {
+            "id": 212226,
+            "width": 500,
+            "height": 335
+        },
+        ...
+    ],
+    "annotations": [
+        {
+            "id": 489885,
+            "category_id": 1,
+            "iscrowd": 0,
+            "image_id": 212226,
+            "area": 12836,
+            "bbox": [
+                235.6300048828125, # x
+                84.30999755859375, # y
+                158.08999633789062, # w
+                185.9499969482422 # h
+            ]
+        },
+        ....
+    ],
+    "categories": [
+        {
+            "supercategory": "none",
+            "id": 1,
+            "name": "person"
+        },
+        ...
+    ]
+}
+```
+You do not need to save the JSON in disk, you can keep it in memory as a dictionary.
+
+### Step 2: Load the object detection evaluator:
+
+Install Hugging Face's `Evaluate` module (`pip install evaluate`) to load the evaluator. More instructions [here](https://huggingface.co/docs/evaluate/installation).
+
+Load the object detection evaluator passing the JSON created on the previous step through the argument `json_gt`:
+`evaluator = evaluate.load("rafaelpadilla/detection_metrics", json_gt=ground_truth_annotations, iou_type="bbox")`
+
+### Step 3: Loop through your dataset samples to obtain the predictions:
+
+```python
+    # Loop through your dataset
+    for batch in dataloader_train:   
+
+        # Get the image(s) from the batch
+        images = batch["images"]
+        # Get the image ids of the image
+        image_ids = batch["image_ids"]
+
+        # Pass the image(s) to your model to obtain bounding boxes, scores and labels
+        predictions = model.predict_boxes(images)
+        # Pass the predictions and image id to the evaluator
+        evaluator.add(prediction=predictions, reference=image_ids)
+    
+    # Call compute to obtain your results
+    results = evaluator.compute()
+    print(results)
+
+```
+
+Regardless your model's architecture, your predictions must be converted to a dictionary containing 3 fields as shown below:
+
+```python
+predictions: [
+    {
+        "scores": [0.55, 0.95, 0.87], 
+        "labels": [6, 1, 1],
+        "boxes": [[100, 30, 40, 28], [40, 32, 50, 28], [128, 44, 23, 69]]
+    },
+    ...
+]
+```
+* `scores`: List or torch tensor containing the confidences of your detections. A confidence is a value between 0 and 1.  
+* `labels`: List or torch tensor with the indexes representing the labels of your detections.  
+* `boxes`: List or torch tensors with the detected bounding boxes in the format `x,y,w,h`.  
+
+The `reference` added to the evaluator in each loop is represented by a list of dictionaries containing the image id of the image in that batch.
+
+For example, in a batch containing two images, with ids 508101 and 1853, the `reference` argument must receive `image_ids` in the following format:
+
+```python
+image_ids = [ {'image_id': [508101]}, {'image_id': [1853]} ]
+```
+
+After the loop, you have to call `evaluator.compute()` to obtain your results in the format of a dictionary. The metrics can also be seen in the prompt as:
+
+```
+IoU metric: bbox
+ Average Precision  (AP) @[ IoU=0.50:0.95 | area=   all | maxDets=100 ] = 0.415
+ Average Precision  (AP) @[ IoU=0.50      | area=   all | maxDets=100 ] = 0.613
+ Average Precision  (AP) @[ IoU=0.75      | area=   all | maxDets=100 ] = 0.436
+ Average Precision  (AP) @[ IoU=0.50:0.95 | area= small | maxDets=100 ] = 0.209
+ Average Precision  (AP) @[ IoU=0.50:0.95 | area=medium | maxDets=100 ] = 0.449
+ Average Precision  (AP) @[ IoU=0.50:0.95 | area= large | maxDets=100 ] = 0.601
+ Average Recall     (AR) @[ IoU=0.50:0.95 | area=   all | maxDets=  1 ] = 0.333
+ Average Recall     (AR) @[ IoU=0.50:0.95 | area=   all | maxDets= 10 ] = 0.531
+ Average Recall     (AR) @[ IoU=0.50:0.95 | area=   all | maxDets=100 ] = 0.572
+ Average Recall     (AR) @[ IoU=0.50:0.95 | area= small | maxDets=100 ] = 0.321
+ Average Recall     (AR) @[ IoU=0.50:0.95 | area=medium | maxDets=100 ] = 0.624
+ Average Recall     (AR) @[ IoU=0.50:0.95 | area= large | maxDets=100 ] = 0.794
+```
+
+The scheme below illustrates how your `for` loop should look like:
+
+![alt text](https://huggingface.co/spaces/rafaelpadilla/detection_metrics/resolve/main/assets/scheme_coco_evaluate.png)
+
+-----------------------
+
+## References and further readings:
+
+1. [COCO Evaluation Metrics](https://cocodataset.org/#detection-eval)  
+2. [A Survey on performance metrics for object-detection algorithms](https://www.researchgate.net/profile/Rafael-Padilla/publication/343194514_A_Survey_on_Performance_Metrics_for_Object-Detection_Algorithms/links/5f1b5a5e45851515ef478268/A-Survey-on-Performance-Metrics-for-Object-Detection-Algorithms.pdf)  
+3. [A Comparative Analysis of Object Detection Metrics with a Companion Open-Source Toolkit](https://www.mdpi.com/2079-9292/10/3/279/pdf)  
+4. [COCO ground-truth annotations for your datasets in JSON](https://towardsdatascience.com/how-to-work-with-object-detection-datasets-in-coco-format-9bf4fb5848a4)

detection_metrics.py

@@ -0,0 +1,203 @@
+from typing import Dict, List, Union
+from pathlib import Path
+import datasets
+import torch
+import evaluate
+import json
+from tqdm import tqdm
+from detection_metrics.pycocotools.coco import COCO 
+from detection_metrics.coco_evaluate import COCOEvaluator 
+from detection_metrics.utils import _TYPING_PREDICTION, _TYPING_REFERENCE
+
+_DESCRIPTION = "This class evaluates object detection models using the COCO dataset \
+    and its evaluation metrics."
+_HOMEPAGE = "https://cocodataset.org"
+_CITATION = """
+    @misc{lin2015microsoft, \
+      title={Microsoft COCO: Common Objects in Context},
+      author={Tsung-Yi Lin and Michael Maire and Serge Belongie and Lubomir Bourdev and \
+          Ross Girshick and James Hays and Pietro Perona and Deva Ramanan and C. Lawrence Zitnick \
+              and Piotr Dollár},
+      year={2015},
+      eprint={1405.0312},
+      archivePrefix={arXiv},
+      primaryClass={cs.CV}
+}
+"""
+_REFERENCE_URLS = [
+    "https://ieeexplore.ieee.org/abstract/document/9145130",
+    "https://www.mdpi.com/2079-9292/10/3/279",
+    "https://cocodataset.org/#detection-eval",
+]
+_KWARGS_DESCRIPTION = """\
+Computes COCO metrics for object detection: AP(mAP) and its variants.
+
+Args:
+    coco (COCO): COCO Evaluator object for evaluating predictions.
+    **kwargs: Additional keyword arguments forwarded to evaluate.Metrics.
+"""
+
+class EvaluateObjectDetection(evaluate.Metric):
+    """
+    Class for evaluating object detection models.
+    """
+
+    def __init__(self, json_gt: Union[Path, Dict], iou_type: str = "bbox", **kwargs):
+        """
+        Initializes the EvaluateObjectDetection class.
+
+        Args:
+            json_gt: JSON with ground-truth annotations in COCO format.
+            # coco_groundtruth (COCO): COCO Evaluator object for evaluating predictions.
+            **kwargs: Additional keyword arguments forwarded to evaluate.Metrics.
+        """
+        super().__init__(**kwargs)
+
+        # Create COCO object from ground-truth annotations
+        if isinstance(json_gt, Path):
+            assert json_gt.exists(), f"Path {json_gt} does not exist."
+            with open(json_gt) as f:
+                json_data = json.load(f)
+        elif isinstance(json_gt, dict):
+            json_data = json_gt
+        coco = COCO(json_data)
+
+        self.coco_evaluator = COCOEvaluator(coco, [iou_type])
+
+    def remove_classes(self, classes_to_remove: List[str]):
+        to_remove = [c.upper() for c in classes_to_remove]
+        cats = {}
+        for id, cat in self.coco_evaluator.coco_eval["bbox"].cocoGt.cats.items():
+            if cat["name"].upper() not in to_remove:
+                cats[id] = cat
+        self.coco_evaluator.coco_eval["bbox"].cocoGt.cats = cats
+        self.coco_evaluator.coco_gt.cats = cats
+        self.coco_evaluator.coco_gt.dataset["categories"] = list(cats.values())
+        self.coco_evaluator.coco_eval["bbox"].params.catIds = [c["id"] for c in cats.values()]
+        
+    def _info(self):
+        """
+        Returns the MetricInfo object with information about the module.
+
+        Returns:
+            evaluate.MetricInfo: Metric information object.
+        """
+        return evaluate.MetricInfo(
+            module_type="metric",
+            description=_DESCRIPTION,
+            citation=_CITATION,
+            inputs_description=_KWARGS_DESCRIPTION,
+            # This defines the format of each prediction and reference
+            features=datasets.Features(
+                {
+                    "predictions": [
+                        datasets.Features(
+                            {
+                                "scores": datasets.Sequence(datasets.Value("float")),
+                                "labels": datasets.Sequence(datasets.Value("int64")),
+                                "boxes": datasets.Sequence(
+                                    datasets.Sequence(datasets.Value("float"))
+                                ),
+                            }
+                        )
+                    ],
+                    "references": [
+                        datasets.Features(
+                            {
+                                "image_id": datasets.Sequence(datasets.Value("int64")),
+                            }
+                        )
+                    ],
+                }
+            ),
+            # Homepage of the module for documentation
+            homepage=_HOMEPAGE,
+            # Additional links to the codebase or references
+            reference_urls=_REFERENCE_URLS,
+        )
+
+    def _preprocess(
+        self, predictions: List[Dict[str, torch.Tensor]]
+    ) -> List[_TYPING_PREDICTION]:
+        """
+        Preprocesses the predictions before computing the scores.
+
+        Args:
+            predictions (List[Dict[str, torch.Tensor]]): A list of prediction dicts.
+
+        Returns:
+            List[_TYPING_PREDICTION]: A list of preprocessed prediction dicts.
+        """
+        processed_predictions = []
+        for pred in predictions:
+            processed_pred: _TYPING_PREDICTION = {}
+            for k, val in pred.items():
+                if isinstance(val, torch.Tensor):
+                    val = val.detach().cpu().tolist()
+                if k == "labels":
+                    val = list(map(int, val))
+                processed_pred[k] = val
+            processed_predictions.append(processed_pred)
+        return processed_predictions
+
+    def _clear_predictions(self, predictions):
+        # Remove unnecessary keys from predictions
+        required = ["scores", "labels", "boxes"]
+        ret = []
+        for prediction in predictions:
+            ret.append({k: v for k, v in prediction.items() if k in required})
+        return ret
+    
+    def _clear_references(self, references):
+        required = [""]
+        ret = []
+        for ref in references:
+            ret.append({k: v for k, v in ref.items() if k in required})
+        return ret
+                
+    def add(self, *, prediction = None, reference = None, **kwargs):
+        """
+        Preprocesses the predictions and references and calls the parent class function.
+
+        Args:
+            prediction: A list of prediction dicts.
+            reference: A list of reference dicts.
+            **kwargs: Additional keyword arguments.
+        """
+        if prediction is not None:
+            prediction = self._clear_predictions(prediction)
+            prediction = self._preprocess(prediction)
+        
+        res = {}  # {image_id} : prediction
+        for output, target in zip(prediction, reference):
+            res[target["image_id"][0]] = output
+        self.coco_evaluator.update(res)
+
+        super(evaluate.Metric, self).add(prediction=prediction, references=reference, **kwargs)
+
+    def _compute(
+        self,
+        predictions: List[List[_TYPING_PREDICTION]],
+        references: List[List[_TYPING_REFERENCE]],
+    ) -> Dict[str, Dict[str, float]]:
+        """
+        Returns the evaluation scores.
+
+        Args:
+            predictions (List[List[_TYPING_PREDICTION]]): A list of predictions.
+            references (List[List[_TYPING_REFERENCE]]): A list of references.
+
+        Returns:
+            Dict: A dictionary containing evaluation scores.
+        """
+        print("Synchronizing processes")
+        self.coco_evaluator.synchronize_between_processes()
+        
+        print("Accumulating values")
+        self.coco_evaluator.accumulate()
+        
+        print("Summarizing results")
+        self.coco_evaluator.summarize()
+        
+        stats = self.coco_evaluator.get_results()
+        return stats

detection_metrics/.gitignore

@@ -0,0 +1,5 @@
+setup.cfg
+bin/
+__pycache__/
+build/
+*.egg-*/

detection_metrics/__init__.py

@@ -0,0 +1 @@
+__version__ = "0.0.3"

detection_metrics/coco_evaluate.py

@@ -0,0 +1,225 @@
+import contextlib
+import copy
+import os
+from typing import Dict, List, Union
+
+import numpy as np
+import torch
+
+from detection_metrics.pycocotools.coco import COCO
+from detection_metrics.pycocotools.cocoeval import COCOeval
+from detection_metrics.utils import (_TYPING_BOX, _TYPING_PREDICTIONS, convert_to_xywh,
+                       create_common_coco_eval)
+
+_SUPPORTED_TYPES = ["bbox"]
+
+
+class COCOEvaluator(object):
+    """
+    Class to perform evaluation for the COCO dataset.
+    """
+
+    def __init__(self, coco_gt: COCO, iou_types: List[str] = ["bbox"]):
+        """
+        Initializes COCOEvaluator with the ground truth COCO dataset and IoU types.
+
+        Args:
+            coco_gt: The ground truth COCO dataset.
+            iou_types: Intersection over Union (IoU) types for evaluation (Supported: "bbox").
+        """
+        self.coco_gt = copy.deepcopy(coco_gt)
+
+        self.coco_eval = {}
+        for iou_type in iou_types:
+            assert iou_type in _SUPPORTED_TYPES, ValueError(
+                f"IoU type not supported {iou_type}"
+            )
+            self.coco_eval[iou_type] = COCOeval(self.coco_gt, iouType=iou_type)
+
+        self.iou_types = iou_types
+        self.img_ids = []
+        self.eval_imgs = {k: [] for k in iou_types}
+
+    def update(self, predictions: _TYPING_PREDICTIONS) -> None:
+        """
+        Update the evaluator with new predictions.
+
+        Args:
+            predictions: The predictions to update.
+        """
+        img_ids = list(np.unique(list(predictions.keys())))
+        self.img_ids.extend(img_ids)
+
+        for iou_type in self.iou_types:
+            results = self.prepare(predictions, iou_type)
+
+            # suppress pycocotools prints
+            with open(os.devnull, "w") as devnull:
+                with contextlib.redirect_stdout(devnull):
+                    coco_dt = COCO.loadRes(self.coco_gt, results) if results else COCO()
+            coco_eval = self.coco_eval[iou_type]
+
+            coco_eval.cocoDt = coco_dt
+            coco_eval.params.imgIds = list(img_ids)
+            eval_imgs = coco_eval.evaluate()
+            self.eval_imgs[iou_type].append(eval_imgs)
+
+    
+    def synchronize_between_processes(self) -> None:
+        """
+        Synchronizes evaluation images between processes.
+        """
+        for iou_type in self.iou_types:
+            self.eval_imgs[iou_type] = np.concatenate(self.eval_imgs[iou_type], 2)
+            create_common_coco_eval(
+                self.coco_eval[iou_type], self.img_ids, self.eval_imgs[iou_type]
+            )
+
+    def accumulate(self) -> None:
+        """
+        Accumulates the evaluation results.
+        """
+        for coco_eval in self.coco_eval.values():
+            coco_eval.accumulate()
+
+    def summarize(self) -> None:
+        """
+        Prints the IoU metric and summarizes the evaluation results.
+        """
+        for iou_type, coco_eval in self.coco_eval.items():
+            print("IoU metric: {}".format(iou_type))
+            coco_eval.summarize()
+
+    def prepare(
+        self, predictions: _TYPING_PREDICTIONS, iou_type: str
+    ) -> List[Dict[str, Union[int, _TYPING_BOX, float]]]:
+        """
+        Prepares the predictions for COCO detection.
+
+        Args:
+            predictions: The predictions to prepare.
+            iou_type: The Intersection over Union (IoU) type for evaluation.
+
+        Returns:
+            A dictionary with the prepared predictions.
+        """
+        if iou_type == "bbox":
+            return self.prepare_for_coco_detection(predictions)
+        else:
+            raise ValueError(f"IoU type not supported {iou_type}")
+
+    def _post_process_stats(
+        self, stats, coco_eval_object, iou_type="bbox"
+    ) -> Dict[str, float]:
+        """
+        Prepares the predictions for COCO detection.
+
+        Args:
+            predictions: The predictions to prepare.
+            iou_type: The Intersection over Union (IoU) type for evaluation.
+
+        Returns:
+            A dictionary with the prepared predictions.
+        """
+        if iou_type not in _SUPPORTED_TYPES:
+            raise ValueError(f"iou_type '{iou_type}' not supported")
+
+        current_max_dets = coco_eval_object.params.maxDets
+
+        index_to_title = {
+            "bbox": {
+                0: f"AP-IoU=0.50:0.95-area=all-maxDets={current_max_dets[2]}",
+                1: f"AP-IoU=0.50-area=all-maxDets={current_max_dets[2]}",
+                2: f"AP-IoU=0.75-area=all-maxDets={current_max_dets[2]}",
+                3: f"AP-IoU=0.50:0.95-area=small-maxDets={current_max_dets[2]}",
+                4: f"AP-IoU=0.50:0.95-area=medium-maxDets={current_max_dets[2]}",
+                5: f"AP-IoU=0.50:0.95-area=large-maxDets={current_max_dets[2]}",
+                6: f"AR-IoU=0.50:0.95-area=all-maxDets={current_max_dets[0]}",
+                7: f"AR-IoU=0.50:0.95-area=all-maxDets={current_max_dets[1]}",
+                8: f"AR-IoU=0.50:0.95-area=all-maxDets={current_max_dets[2]}",
+                9: f"AR-IoU=0.50:0.95-area=small-maxDets={current_max_dets[2]}",
+                10: f"AR-IoU=0.50:0.95-area=medium-maxDets={current_max_dets[2]}",
+                11: f"AR-IoU=0.50:0.95-area=large-maxDets={current_max_dets[2]}",
+            },
+            "keypoints": {
+                0: "AP-IoU=0.50:0.95-area=all-maxDets=20",
+                1: "AP-IoU=0.50-area=all-maxDets=20",
+                2: "AP-IoU=0.75-area=all-maxDets=20",
+                3: "AP-IoU=0.50:0.95-area=medium-maxDets=20",
+                4: "AP-IoU=0.50:0.95-area=large-maxDets=20",
+                5: "AR-IoU=0.50:0.95-area=all-maxDets=20",
+                6: "AR-IoU=0.50-area=all-maxDets=20",
+                7: "AR-IoU=0.75-area=all-maxDets=20",
+                8: "AR-IoU=0.50:0.95-area=medium-maxDets=20",
+                9: "AR-IoU=0.50:0.95-area=large-maxDets=20",
+            },
+        }
+
+        output_dict: Dict[str, float] = {}
+        for index, stat in enumerate(stats):
+            output_dict[index_to_title[iou_type][index]] = stat
+
+        return output_dict
+
+    def get_results(self) -> Dict[str, Dict[str, float]]:
+        """
+        Gets the results of the COCO evaluation.
+
+        Returns:
+            A dictionary with the results of the COCO evaluation.
+        """
+        output_dict = {}
+
+        for iou_type, coco_eval in self.coco_eval.items():
+            if iou_type == "segm":
+                iou_type = "bbox"
+            output_dict[f"iou_{iou_type}"] = self._post_process_stats(
+                coco_eval.stats, coco_eval, iou_type
+            )
+        return output_dict
+
+    def prepare_for_coco_detection(
+        self, predictions: _TYPING_PREDICTIONS
+    ) -> List[Dict[str, Union[int, _TYPING_BOX, float]]]:
+        """
+        Prepares the predictions for COCO detection.
+
+        Args:
+            predictions: The predictions to prepare.
+
+        Returns:
+            A list of dictionaries with the prepared predictions.
+        """
+        coco_results = []
+        for original_id, prediction in predictions.items():
+            if len(prediction) == 0:
+                continue
+
+            boxes = prediction["boxes"]
+            if len(boxes) == 0:
+                continue
+
+            if not isinstance(boxes, torch.Tensor):
+                boxes = torch.as_tensor(boxes)
+                boxes = boxes.tolist()
+
+            scores = prediction["scores"]
+            if not isinstance(scores, list):
+                scores = scores.tolist()
+
+            labels = prediction["labels"]
+            if not isinstance(labels, list):
+                labels = prediction["labels"].tolist()
+
+            coco_results.extend(
+                [
+                    {
+                        "image_id": original_id,
+                        "category_id": labels[k],
+                        "bbox": box,
+                        "score": scores[k],
+                    }
+                    for k, box in enumerate(boxes)
+                ]
+            )
+        return coco_results

detection_metrics/pycocotools/coco.py

@@ -0,0 +1,491 @@
+# This code is basically a copy and paste from the original cocoapi file:
+# https://github.com/cocodataset/cocoapi/blob/master/PythonAPI/pycocotools/coco.py
+# with the following changes:
+# * Instead of receiving the path to the annotation file, it receives a json object.
+# * Commented out all parts of code that depends on maskUtils, which is not needed
+#       for bounding box evaluation.
+
+__author__ = "tylin"
+__version__ = "2.0"
+# Interface for accessing the Microsoft COCO dataset.
+
+# Microsoft COCO is a large image dataset designed for object detection,
+# segmentation, and caption generation. pycocotools is a Python API that
+# assists in loading, parsing and visualizing the annotations in COCO.
+# Please visit http://mscoco.org/ for more information on COCO, including
+# for the data, paper, and tutorials. The exact format of the annotations
+# is also described on the COCO website. For example usage of the pycocotools
+# please see pycocotools_demo.ipynb. In addition to this API, please download both
+# the COCO images and annotations in order to run the demo.
+
+# An alternative to using the API is to load the annotations directly
+# into Python dictionary
+# Using the API provides additional utility functions. Note that this API
+# supports both *instance* and *caption* annotations. In the case of
+# captions not all functions are defined (e.g. categories are undefined).
+
+# The following API functions are defined:
+#  COCO       - COCO api class that loads COCO annotation file and prepare data structures.
+#  decodeMask - Decode binary mask M encoded via run-length encoding.
+#  encodeMask - Encode binary mask M using run-length encoding.
+#  getAnnIds  - Get ann ids that satisfy given filter conditions.
+#  getCatIds  - Get cat ids that satisfy given filter conditions.
+#  getImgIds  - Get img ids that satisfy given filter conditions.
+#  loadAnns   - Load anns with the specified ids.
+#  loadCats   - Load cats with the specified ids.
+#  loadImgs   - Load imgs with the specified ids.
+#  annToMask  - Convert segmentation in an annotation to binary mask.
+#  showAnns   - Display the specified annotations.
+#  loadRes    - Load algorithm results and create API for accessing them.
+#  download   - Download COCO images from mscoco.org server.
+# Throughout the API "ann"=annotation, "cat"=category, and "img"=image.
+# Help on each functions can be accessed by: "help COCO>function".
+
+# See also COCO>decodeMask,
+# COCO>encodeMask, COCO>getAnnIds, COCO>getCatIds,
+# COCO>getImgIds, COCO>loadAnns, COCO>loadCats,
+# COCO>loadImgs, COCO>annToMask, COCO>showAnns
+
+# Microsoft COCO Toolbox.      version 2.0
+# Data, paper, and tutorials available at:  http://mscoco.org/
+# Code written by Piotr Dollar and Tsung-Yi Lin, 2014.
+# Licensed under the Simplified BSD License [see bsd.txt]
+
+import copy
+import itertools
+import json
+# from . import mask as maskUtils
+import os
+import sys
+import time
+from collections import defaultdict
+
+import matplotlib.pyplot as plt
+import numpy as np
+from matplotlib.collections import PatchCollection
+from matplotlib.patches import Polygon
+
+PYTHON_VERSION = sys.version_info[0]
+if PYTHON_VERSION == 2:
+    from urllib import urlretrieve
+elif PYTHON_VERSION == 3:
+    from urllib.request import urlretrieve
+
+
+def _isArrayLike(obj):
+    return hasattr(obj, "__iter__") and hasattr(obj, "__len__")
+
+
+class COCO:
+    def __init__(self, annotations=None):
+        """
+        Constructor of Microsoft COCO helper class for reading and visualizing annotations.
+        :param annotation_file (str): location of annotation file
+        :param image_folder (str): location to the folder that hosts images.
+        :return:
+        """
+        # load dataset
+        self.dataset, self.anns, self.cats, self.imgs = dict(), dict(), dict(), dict()
+        self.imgToAnns, self.catToImgs = defaultdict(list), defaultdict(list)
+        # Modified the original code to receive a json object instead of a path to a file
+        if annotations:
+            assert (
+                type(annotations) == dict
+            ), f"annotation file format {type(annotations)} not supported."
+            self.dataset = annotations
+            self.createIndex()
+
+    def createIndex(self):
+        # create index
+        print("creating index...")
+        anns, cats, imgs = {}, {}, {}
+        imgToAnns, catToImgs = defaultdict(list), defaultdict(list)
+        if "annotations" in self.dataset:
+            for ann in self.dataset["annotations"]:
+                imgToAnns[ann["image_id"]].append(ann)
+                anns[ann["id"]] = ann
+
+        if "images" in self.dataset:
+            for img in self.dataset["images"]:
+                imgs[img["id"]] = img
+
+        if "categories" in self.dataset:
+            for cat in self.dataset["categories"]:
+                cats[cat["id"]] = cat
+
+        if "annotations" in self.dataset and "categories" in self.dataset:
+            for ann in self.dataset["annotations"]:
+                catToImgs[ann["category_id"]].append(ann["image_id"])
+
+        print("index created!")
+
+        # create class members
+        self.anns = anns
+        self.imgToAnns = imgToAnns
+        self.catToImgs = catToImgs
+        self.imgs = imgs
+        self.cats = cats
+
+    def info(self):
+        """
+        Print information about the annotation file.
+        :return:
+        """
+        for key, value in self.dataset["info"].items():
+            print("{}: {}".format(key, value))
+
+    def getAnnIds(self, imgIds=[], catIds=[], areaRng=[], iscrowd=None):
+        """
+        Get ann ids that satisfy given filter conditions. default skips that filter
+        :param imgIds  (int array)     : get anns for given imgs
+               catIds  (int array)     : get anns for given cats
+               areaRng (float array)   : get anns for given area range (e.g. [0 inf])
+               iscrowd (boolean)       : get anns for given crowd label (False or True)
+        :return: ids (int array)       : integer array of ann ids
+        """
+        imgIds = imgIds if _isArrayLike(imgIds) else [imgIds]
+        catIds = catIds if _isArrayLike(catIds) else [catIds]
+
+        if len(imgIds) == len(catIds) == len(areaRng) == 0:
+            anns = self.dataset["annotations"]
+        else:
+            if not len(imgIds) == 0:
+                lists = [
+                    self.imgToAnns[imgId] for imgId in imgIds if imgId in self.imgToAnns
+                ]
+                anns = list(itertools.chain.from_iterable(lists))
+            else:
+                anns = self.dataset["annotations"]
+            anns = (
+                anns
+                if len(catIds) == 0
+                else [ann for ann in anns if ann["category_id"] in catIds]
+            )
+            anns = (
+                anns
+                if len(areaRng) == 0
+                else [
+                    ann
+                    for ann in anns
+                    if ann["area"] > areaRng[0] and ann["area"] < areaRng[1]
+                ]
+            )
+        if not iscrowd == None:
+            ids = [ann["id"] for ann in anns if ann["iscrowd"] == iscrowd]
+        else:
+            ids = [ann["id"] for ann in anns]
+        return ids
+
+    def getCatIds(self, catNms=[], supNms=[], catIds=[]):
+        """
+        filtering parameters. default skips that filter.
+        :param catNms (str array)  : get cats for given cat names
+        :param supNms (str array)  : get cats for given supercategory names
+        :param catIds (int array)  : get cats for given cat ids
+        :return: ids (int array)   : integer array of cat ids
+        """
+        catNms = catNms if _isArrayLike(catNms) else [catNms]
+        supNms = supNms if _isArrayLike(supNms) else [supNms]
+        catIds = catIds if _isArrayLike(catIds) else [catIds]
+
+        if len(catNms) == len(supNms) == len(catIds) == 0:
+            cats = self.dataset["categories"]
+        else:
+            cats = self.dataset["categories"]
+            cats = (
+                cats
+                if len(catNms) == 0
+                else [cat for cat in cats if cat["name"] in catNms]
+            )
+            cats = (
+                cats
+                if len(supNms) == 0
+                else [cat for cat in cats if cat["supercategory"] in supNms]
+            )
+            cats = (
+                cats
+                if len(catIds) == 0
+                else [cat for cat in cats if cat["id"] in catIds]
+            )
+        ids = [cat["id"] for cat in cats]
+        return ids
+
+    def getImgIds(self, imgIds=[], catIds=[]):
+        """
+        Get img ids that satisfy given filter conditions.
+        :param imgIds (int array) : get imgs for given ids
+        :param catIds (int array) : get imgs with all given cats
+        :return: ids (int array)  : integer array of img ids
+        """
+        imgIds = imgIds if _isArrayLike(imgIds) else [imgIds]
+        catIds = catIds if _isArrayLike(catIds) else [catIds]
+
+        if len(imgIds) == len(catIds) == 0:
+            ids = self.imgs.keys()
+        else:
+            ids = set(imgIds)
+            for i, catId in enumerate(catIds):
+                if i == 0 and len(ids) == 0:
+                    ids = set(self.catToImgs[catId])
+                else:
+                    ids &= set(self.catToImgs[catId])
+        return list(ids)
+
+    def loadAnns(self, ids=[]):
+        """
+        Load anns with the specified ids.
+        :param ids (int array)       : integer ids specifying anns
+        :return: anns (object array) : loaded ann objects
+        """
+        if _isArrayLike(ids):
+            return [self.anns[id] for id in ids]
+        elif type(ids) == int:
+            return [self.anns[ids]]
+
+    def loadCats(self, ids=[]):
+        """
+        Load cats with the specified ids.
+        :param ids (int array)       : integer ids specifying cats
+        :return: cats (object array) : loaded cat objects
+        """
+        if _isArrayLike(ids):
+            return [self.cats[id] for id in ids]
+        elif type(ids) == int:
+            return [self.cats[ids]]
+
+    def loadImgs(self, ids=[]):
+        """
+        Load anns with the specified ids.
+        :param ids (int array)       : integer ids specifying img
+        :return: imgs (object array) : loaded img objects
+        """
+        if _isArrayLike(ids):
+            return [self.imgs[id] for id in ids]
+        elif type(ids) == int:
+            return [self.imgs[ids]]
+
+    def showAnns(self, anns, draw_bbox=False):
+        """
+        Display the specified annotations.
+        :param anns (array of object): annotations to display
+        :return: None
+        """
+        if len(anns) == 0:
+            return 0
+        if "segmentation" in anns[0] or "keypoints" in anns[0]:
+            datasetType = "instances"
+        elif "caption" in anns[0]:
+            datasetType = "captions"
+        else:
+            raise Exception("datasetType not supported")
+        if datasetType == "instances":
+            ax = plt.gca()
+            ax.set_autoscale_on(False)
+            polygons = []
+            color = []
+            for ann in anns:
+                c = (np.random.random((1, 3)) * 0.6 + 0.4).tolist()[0]
+                if "segmentation" in ann:
+                    if type(ann["segmentation"]) == list:
+                        # polygon
+                        for seg in ann["segmentation"]:
+                            poly = np.array(seg).reshape((int(len(seg) / 2), 2))
+                            polygons.append(Polygon(poly))
+                            color.append(c)
+                    else:
+                        raise NotImplementedError(
+                            "This type is not is not supported yet."
+                        )
+                        # # mask
+                        # t = self.imgs[ann['image_id']]
+                        # if type(ann['segmentation']['counts']) == list:
+                        #     rle = maskUtils.frPyObjects([ann['segmentation']], t['height'], t['width'])
+                        # else:
+                        #     rle = [ann['segmentation']]
+                        # m = maskUtils.decode(rle)
+                        # img = np.ones( (m.shape[0], m.shape[1], 3) )
+                        # if ann['iscrowd'] == 1:
+                        #     color_mask = np.array([2.0,166.0,101.0])/255
+                        # if ann['iscrowd'] == 0:
+                        #     color_mask = np.random.random((1, 3)).tolist()[0]
+                        # for i in range(3):
+                        #     img[:,:,i] = color_mask[i]
+                        # ax.imshow(np.dstack( (img, m*0.5) ))
+                if "keypoints" in ann and type(ann["keypoints"]) == list:
+                    # turn skeleton into zero-based index
+                    sks = np.array(self.loadCats(ann["category_id"])[0]["skeleton"]) - 1
+                    kp = np.array(ann["keypoints"])
+                    x = kp[0::3]
+                    y = kp[1::3]
+                    v = kp[2::3]
+                    for sk in sks:
+                        if np.all(v[sk] > 0):
+                            plt.plot(x[sk], y[sk], linewidth=3, color=c)
+                    plt.plot(
+                        x[v > 0],
+                        y[v > 0],
+                        "o",
+                        markersize=8,
+                        markerfacecolor=c,
+                        markeredgecolor="k",
+                        markeredgewidth=2,
+                    )
+                    plt.plot(
+                        x[v > 1],
+                        y[v > 1],
+                        "o",
+                        markersize=8,
+                        markerfacecolor=c,
+                        markeredgecolor=c,
+                        markeredgewidth=2,
+                    )
+
+                if draw_bbox:
+                    [bbox_x, bbox_y, bbox_w, bbox_h] = ann["bbox"]
+                    poly = [
+                        [bbox_x, bbox_y],
+                        [bbox_x, bbox_y + bbox_h],
+                        [bbox_x + bbox_w, bbox_y + bbox_h],
+                        [bbox_x + bbox_w, bbox_y],
+                    ]
+                    np_poly = np.array(poly).reshape((4, 2))
+                    polygons.append(Polygon(np_poly))
+                    color.append(c)
+
+            p = PatchCollection(polygons, facecolor=color, linewidths=0, alpha=0.4)
+            ax.add_collection(p)
+            p = PatchCollection(
+                polygons, facecolor="none", edgecolors=color, linewidths=2
+            )
+            ax.add_collection(p)
+        elif datasetType == "captions":
+            for ann in anns:
+                print(ann["caption"])
+
+    def loadRes(self, resFile):
+        """
+        Load result file and return a result api object.
+        :param   resFile (str)     : file name of result file
+        :return: res (obj)         : result api object
+        """
+        res = COCO()
+        res.dataset["images"] = [img for img in self.dataset["images"]]
+
+        print("Loading and preparing results...")
+        tic = time.time()
+        if type(resFile) == str or (PYTHON_VERSION == 2 and type(resFile) == unicode):
+            anns = json.load(open(resFile))
+        elif type(resFile) == np.ndarray:
+            anns = self.loadNumpyAnnotations(resFile)
+        else:
+            anns = resFile
+        assert type(anns) == list, "results in not an array of objects"
+        annsImgIds = [ann["image_id"] for ann in anns]
+        assert set(annsImgIds) == (
+            set(annsImgIds) & set(self.getImgIds())
+        ), "Results do not correspond to current coco set"
+        if "caption" in anns[0]:
+            raise NotImplementedError("Evaluating caption is not supported yet.")
+        elif "bbox" in anns[0] and not anns[0]["bbox"] == []:
+            res.dataset["categories"] = copy.deepcopy(self.dataset["categories"])
+            for id, ann in enumerate(anns):
+                bb = ann["bbox"]
+                x1, x2, y1, y2 = [bb[0], bb[0] + bb[2], bb[1], bb[1] + bb[3]]
+                if not "segmentation" in ann:
+                    ann["segmentation"] = [[x1, y1, x1, y2, x2, y2, x2, y1]]
+                ann["area"] = bb[2] * bb[3]
+                ann["id"] = id + 1
+                ann["iscrowd"] = 0
+        elif "segmentation" in anns[0]:
+            raise NotImplementedError("Evaluating caption is not supported yet.")
+        elif "keypoints" in anns[0]:
+            raise NotImplementedError("Evaluating caption is not supported yet.")
+        print("DONE (t={:0.2f}s)".format(time.time() - tic))
+
+        res.dataset["annotations"] = anns
+        res.createIndex()
+        return res
+
+    def download(self, tarDir=None, imgIds=[]):
+        """
+        Download COCO images from mscoco.org server.
+        :param tarDir (str): COCO results directory name
+               imgIds (list): images to be downloaded
+        :return:
+        """
+        if tarDir is None:
+            print("Please specify target directory")
+            return -1
+        if len(imgIds) == 0:
+            imgs = self.imgs.values()
+        else:
+            imgs = self.loadImgs(imgIds)
+        N = len(imgs)
+        if not os.path.exists(tarDir):
+            os.makedirs(tarDir)
+        for i, img in enumerate(imgs):
+            tic = time.time()
+            fname = os.path.join(tarDir, img["file_name"])
+            if not os.path.exists(fname):
+                urlretrieve(img["coco_url"], fname)
+            print(
+                "downloaded {}/{} images (t={:0.1f}s)".format(i, N, time.time() - tic)
+            )
+
+    def loadNumpyAnnotations(self, data):
+        """
+        Convert result data from a numpy array [Nx7] where each row contains {imageID,x1,y1,w,h,score,class}
+        :param  data (numpy.ndarray)
+        :return: annotations (python nested list)
+        """
+        print("Converting ndarray to lists...")
+        assert type(data) == np.ndarray
+        print(data.shape)
+        assert data.shape[1] == 7
+        N = data.shape[0]
+        ann = []
+        for i in range(N):
+            if i % 1000000 == 0:
+                print("{}/{}".format(i, N))
+            ann += [
+                {
+                    "image_id": int(data[i, 0]),
+                    "bbox": [data[i, 1], data[i, 2], data[i, 3], data[i, 4]],
+                    "score": data[i, 5],
+                    "category_id": int(data[i, 6]),
+                }
+            ]
+        return ann
+
+    def annToRLE(self, ann):
+        """
+        Convert annotation which can be polygons, uncompressed RLE to RLE.
+        :return: binary mask (numpy 2D array)
+        """
+        t = self.imgs[ann["image_id"]]
+        h, w = t["height"], t["width"]
+        segm = ann["segmentation"]
+        if type(segm) == list:
+            raise NotImplementedError("This type is not is not supported yet.")
+            # polygon -- a single object might consist of multiple parts
+            # we merge all parts into one mask rle code
+            # rles = maskUtils.frPyObjects(segm, h, w)
+            # rle = maskUtils.merge(rles)
+        elif type(segm["counts"]) == list:
+            raise NotImplementedError("This type is not is not supported yet.")
+            # uncompressed RLE
+            # rle = maskUtils.frPyObjects(segm, h, w)
+        else:
+            # rle
+            rle = ann["segmentation"]
+        return rle
+
+    def annToMask(self, ann):
+        """
+        Convert annotation which can be polygons, uncompressed RLE, or RLE to binary mask.
+        :return: binary mask (numpy 2D array)
+        """
+        rle = self.annToRLE(ann)
+        # m = maskUtils.decode(rle)
+        raise NotImplementedError("This type is not is not supported yet.")
+        return m

detection_metrics/pycocotools/cocoeval.py

@@ -0,0 +1,631 @@
+# This code is basically a copy and paste from the original cocoapi repo:
+# https://github.com/cocodataset/cocoapi/blob/master/PythonAPI/pycocotools/cocoeval.py
+# with the following changes have been made:
+# * Replace the usage of mask (maskUtils) by MaskEvaluator.
+# * Comment out prints in the evaluate() function.
+# * Include a return of the function evaluate. Inspired
+#      by @ybelkada (https://huggingface.co/spaces/ybelkada/cocoevaluate/)
+
+__author__ = "tsungyi"
+
+import copy
+import datetime
+import time
+from collections import defaultdict
+from packaging import version
+
+import numpy as np
+
+if version.parse(np.__version__) < version.parse("1.24"):
+    dtype_float = np.float
+else:
+    dtype_float = np.float32
+    
+from .mask_utils import MaskEvaluator as maskUtils
+
+class COCOeval:
+    # Interface for evaluating detection on the Microsoft COCO dataset.
+    #
+    # The usage for CocoEval is as follows:
+    #  cocoGt=..., cocoDt=...       # load dataset and results
+    #  E = CocoEval(cocoGt,cocoDt); # initialize CocoEval object
+    #  E.params.recThrs = ...;      # set parameters as desired
+    #  E.evaluate();                # run per image evaluation
+    #  E.accumulate();              # accumulate per image results
+    #  E.summarize();               # display summary metrics of results
+    # For example usage see evalDemo.m and http://mscoco.org/.
+    #
+    # The evaluation parameters are as follows (defaults in brackets):
+    #  imgIds     - [all] N img ids to use for evaluation
+    #  catIds     - [all] K cat ids to use for evaluation
+    #  iouThrs    - [.5:.05:.95] T=10 IoU thresholds for evaluation
+    #  recThrs    - [0:.01:1] R=101 recall thresholds for evaluation
+    #  areaRng    - [...] A=4 object area ranges for evaluation
+    #  maxDets    - [1 10 100] M=3 thresholds on max detections per image
+    #  iouType    - ['segm'] set iouType to 'segm', 'bbox' or 'keypoints'
+    #  iouType replaced the now DEPRECATED useSegm parameter.
+    #  useCats    - [1] if true use category labels for evaluation
+    # Note: if useCats=0 category labels are ignored as in proposal scoring.
+    # Note: multiple areaRngs [Ax2] and maxDets [Mx1] can be specified.
+    #
+    # evaluate(): evaluates detections on every image and every category and
+    # concats the results into the "evalImgs" with fields:
+    #  dtIds      - [1xD] id for each of the D detections (dt)
+    #  gtIds      - [1xG] id for each of the G ground truths (gt)
+    #  dtMatches  - [TxD] matching gt id at each IoU or 0
+    #  gtMatches  - [TxG] matching dt id at each IoU or 0
+    #  dtScores   - [1xD] confidence of each dt
+    #  gtIgnore   - [1xG] ignore flag for each gt
+    #  dtIgnore   - [TxD] ignore flag for each dt at each IoU
+    #
+    # accumulate(): accumulates the per-image, per-category evaluation
+    # results in "evalImgs" into the dictionary "eval" with fields:
+    #  params     - parameters used for evaluation
+    #  date       - date evaluation was performed
+    #  counts     - [T,R,K,A,M] parameter dimensions (see above)
+    #  precision  - [TxRxKxAxM] precision for every evaluation setting
+    #  recall     - [TxKxAxM] max recall for every evaluation setting
+    # Note: precision and recall==-1 for settings with no gt objects.
+    #
+    # See also coco, mask, pycocoDemo, pycocoEvalDemo
+    #
+    # Microsoft COCO Toolbox.      version 2.0
+    # Data, paper, and tutorials available at:  http://mscoco.org/
+    # Code written by Piotr Dollar and Tsung-Yi Lin, 2015.
+    # Licensed under the Simplified BSD License [see coco/license.txt]
+    def __init__(self, cocoGt=None, cocoDt=None, iouType="segm"):
+        """
+        Initialize CocoEval using coco APIs for gt and dt
+        :param cocoGt: coco object with ground truth annotations
+        :param cocoDt: coco object with detection results
+        :return: None
+        """
+        if not iouType:
+            print("iouType not specified. use default iouType segm")
+        self.cocoGt = cocoGt  # ground truth COCO API
+        self.cocoDt = cocoDt  # detections COCO API
+        self.evalImgs = defaultdict(
+            list
+        )  # per-image per-category evaluation results [KxAxI] elements
+        self.eval = {}  # accumulated evaluation results
+        self._gts = defaultdict(list)  # gt for evaluation
+        self._dts = defaultdict(list)  # dt for evaluation
+        self.params = Params(iouType=iouType)  # parameters
+        self._paramsEval = {}  # parameters for evaluation
+        self.stats = []  # result summarization
+        self.ious = {}  # ious between all gts and dts
+        if not cocoGt is None:
+            self.params.imgIds = sorted(cocoGt.getImgIds())
+            self.params.catIds = sorted(cocoGt.getCatIds())
+
+    def _prepare(self):
+        """
+        Prepare ._gts and ._dts for evaluation based on params
+        :return: None
+        """
+
+        def _toMask(anns, coco):
+            # modify ann['segmentation'] by reference
+            for ann in anns:
+                rle = coco.annToRLE(ann)
+                ann["segmentation"] = rle
+
+        p = self.params
+        if p.useCats:
+            gts = self.cocoGt.loadAnns(
+                self.cocoGt.getAnnIds(imgIds=p.imgIds, catIds=p.catIds)
+            )
+            dts = self.cocoDt.loadAnns(
+                self.cocoDt.getAnnIds(imgIds=p.imgIds, catIds=p.catIds)
+            )
+        else:
+            gts = self.cocoGt.loadAnns(self.cocoGt.getAnnIds(imgIds=p.imgIds))
+            dts = self.cocoDt.loadAnns(self.cocoDt.getAnnIds(imgIds=p.imgIds))
+
+        # convert ground truth to mask if iouType == 'segm'
+        if p.iouType == "segm":
+            _toMask(gts, self.cocoGt)
+            _toMask(dts, self.cocoDt)
+        # set ignore flag
+        for gt in gts:
+            gt["ignore"] = gt["ignore"] if "ignore" in gt else 0
+            gt["ignore"] = "iscrowd" in gt and gt["iscrowd"]
+            if p.iouType == "keypoints":
+                gt["ignore"] = (gt["num_keypoints"] == 0) or gt["ignore"]
+        self._gts = defaultdict(list)  # gt for evaluation
+        self._dts = defaultdict(list)  # dt for evaluation
+        for gt in gts:
+            self._gts[gt["image_id"], gt["category_id"]].append(gt)
+        for dt in dts:
+            self._dts[dt["image_id"], dt["category_id"]].append(dt)
+        self.evalImgs = defaultdict(list)  # per-image per-category evaluation results
+        self.eval = {}  # accumulated evaluation results
+
+    def evaluate(self):
+        """
+        Run per image evaluation on given images and store results (a list of dict) in self.evalImgs
+        :return: None
+        """
+        # tic = time.time()
+        # print("Running per image evaluation...")
+        p = self.params
+        # add backward compatibility if useSegm is specified in params
+        if not p.useSegm is None:
+            p.iouType = "segm" if p.useSegm == 1 else "bbox"
+        #     print(
+        #         "useSegm (deprecated) is not None. Running {} evaluation".format(
+        #             p.iouType
+        #         )
+        #     )
+        # print("Evaluate annotation type *{}*".format(p.iouType))
+        p.imgIds = list(np.unique(p.imgIds))
+        if p.useCats:
+            p.catIds = list(np.unique(p.catIds))
+        p.maxDets = sorted(p.maxDets)
+        self.params = p
+
+        self._prepare()
+        # loop through images, area range, max detection number
+        catIds = p.catIds if p.useCats else [-1]
+
+        if p.iouType == "segm" or p.iouType == "bbox":
+            computeIoU = self.computeIoU
+        elif p.iouType == "keypoints":
+            computeIoU = self.computeOks
+        self.ious = {
+            (imgId, catId): computeIoU(imgId, catId)
+            for imgId in p.imgIds
+            for catId in catIds
+        }
+
+        evaluateImg = self.evaluateImg
+        maxDet = p.maxDets[-1]
+        self.evalImgs = [
+            evaluateImg(imgId, catId, areaRng, maxDet)
+            for catId in catIds
+            for areaRng in p.areaRng
+            for imgId in p.imgIds
+        ]
+        self._paramsEval = copy.deepcopy(self.params)
+        ret_evalImgs = np.asarray(self.evalImgs).reshape(
+            len(catIds), len(p.areaRng), len(p.imgIds)
+        )
+        # toc = time.time()
+        # print("DONE (t={:0.2f}s).".format(toc - tic))
+        return ret_evalImgs
+
+    def computeIoU(self, imgId, catId):
+        p = self.params
+        if p.useCats:
+            gt = self._gts[imgId, catId]
+            dt = self._dts[imgId, catId]
+        else:
+            gt = [_ for cId in p.catIds for _ in self._gts[imgId, cId]]
+            dt = [_ for cId in p.catIds for _ in self._dts[imgId, cId]]
+        if len(gt) == 0 and len(dt) == 0:
+            return []
+        inds = np.argsort([-d["score"] for d in dt], kind="mergesort")
+        dt = [dt[i] for i in inds]
+        if len(dt) > p.maxDets[-1]:
+            dt = dt[0 : p.maxDets[-1]]
+
+        if p.iouType == "segm":
+            g = [g["segmentation"] for g in gt]
+            d = [d["segmentation"] for d in dt]
+        elif p.iouType == "bbox":
+            g = [g["bbox"] for g in gt]
+            d = [d["bbox"] for d in dt]
+        else:
+            raise Exception("unknown iouType for iou computation")
+
+        # compute iou between each dt and gt region
+        iscrowd = [int(o["iscrowd"]) for o in gt]
+        ious = maskUtils.iou(d, g, iscrowd)
+        return ious
+
+    def computeOks(self, imgId, catId):
+        p = self.params
+        # dimention here should be Nxm
+        gts = self._gts[imgId, catId]
+        dts = self._dts[imgId, catId]
+        inds = np.argsort([-d["score"] for d in dts], kind="mergesort")
+        dts = [dts[i] for i in inds]
+        if len(dts) > p.maxDets[-1]:
+            dts = dts[0 : p.maxDets[-1]]
+        # if len(gts) == 0 and len(dts) == 0:
+        if len(gts) == 0 or len(dts) == 0:
+            return []
+        ious = np.zeros((len(dts), len(gts)))
+        sigmas = p.kpt_oks_sigmas
+        vars = (sigmas * 2) ** 2
+        k = len(sigmas)
+        # compute oks between each detection and ground truth object
+        for j, gt in enumerate(gts):
+            # create bounds for ignore regions(double the gt bbox)
+            g = np.array(gt["keypoints"])
+            xg = g[0::3]
+            yg = g[1::3]
+            vg = g[2::3]
+            k1 = np.count_nonzero(vg > 0)
+            bb = gt["bbox"]
+            x0 = bb[0] - bb[2]
+            x1 = bb[0] + bb[2] * 2
+            y0 = bb[1] - bb[3]
+            y1 = bb[1] + bb[3] * 2
+            for i, dt in enumerate(dts):
+                d = np.array(dt["keypoints"])
+                xd = d[0::3]
+                yd = d[1::3]
+                if k1 > 0:
+                    # measure the per-keypoint distance if keypoints visible
+                    dx = xd - xg
+                    dy = yd - yg
+                else:
+                    # measure minimum distance to keypoints in (x0,y0) & (x1,y1)
+                    z = np.zeros((k))
+                    dx = np.max((z, x0 - xd), axis=0) + np.max((z, xd - x1), axis=0)
+                    dy = np.max((z, y0 - yd), axis=0) + np.max((z, yd - y1), axis=0)
+                e = (dx**2 + dy**2) / vars / (gt["area"] + np.spacing(1)) / 2
+                if k1 > 0:
+                    e = e[vg > 0]
+                ious[i, j] = np.sum(np.exp(-e)) / e.shape[0]
+        return ious
+
+    def evaluateImg(self, imgId, catId, aRng, maxDet):
+        """
+        perform evaluation for single category and image
+        :return: dict (single image results)
+        """
+        p = self.params
+        if p.useCats:
+            gt = self._gts[imgId, catId]
+            dt = self._dts[imgId, catId]
+        else:
+            gt = [_ for cId in p.catIds for _ in self._gts[imgId, cId]]
+            dt = [_ for cId in p.catIds for _ in self._dts[imgId, cId]]
+        if len(gt) == 0 and len(dt) == 0:
+            return None
+
+        for g in gt:
+            if g["ignore"] or (g["area"] < aRng[0] or g["area"] > aRng[1]):
+                g["_ignore"] = 1
+            else:
+                g["_ignore"] = 0
+
+        # sort dt highest score first, sort gt ignore last
+        gtind = np.argsort([g["_ignore"] for g in gt], kind="mergesort")
+        gt = [gt[i] for i in gtind]
+        dtind = np.argsort([-d["score"] for d in dt], kind="mergesort")
+        dt = [dt[i] for i in dtind[0:maxDet]]
+        iscrowd = [int(o["iscrowd"]) for o in gt]
+        # load computed ious
+        ious = (
+            self.ious[imgId, catId][:, gtind]
+            if len(self.ious[imgId, catId]) > 0
+            else self.ious[imgId, catId]
+        )
+
+        T = len(p.iouThrs)
+        G = len(gt)
+        D = len(dt)
+        gtm = np.zeros((T, G))
+        dtm = np.zeros((T, D))
+        gtIg = np.array([g["_ignore"] for g in gt])
+        dtIg = np.zeros((T, D))
+        if not len(ious) == 0:
+            for tind, t in enumerate(p.iouThrs):
+                for dind, d in enumerate(dt):
+                    # information about best match so far (m=-1 -> unmatched)
+                    iou = min([t, 1 - 1e-10])
+                    m = -1
+                    for gind, g in enumerate(gt):
+                        # if this gt already matched, and not a crowd, continue
+                        if gtm[tind, gind] > 0 and not iscrowd[gind]:
+                            continue
+                        # if dt matched to reg gt, and on ignore gt, stop
+                        if m > -1 and gtIg[m] == 0 and gtIg[gind] == 1:
+                            break
+                        # continue to next gt unless better match made
+                        if ious[dind, gind] < iou:
+                            continue
+                        # if match successful and best so far, store appropriately
+                        iou = ious[dind, gind]
+                        m = gind
+                    # if match made store id of match for both dt and gt
+                    if m == -1:
+                        continue
+                    dtIg[tind, dind] = gtIg[m]
+                    dtm[tind, dind] = gt[m]["id"]
+                    gtm[tind, m] = d["id"]
+        # set unmatched detections outside of area range to ignore
+        a = np.array([d["area"] < aRng[0] or d["area"] > aRng[1] for d in dt]).reshape(
+            (1, len(dt))
+        )
+        dtIg = np.logical_or(dtIg, np.logical_and(dtm == 0, np.repeat(a, T, 0)))
+        # store results for given image and category
+        return {
+            "image_id": imgId,
+            "category_id": catId,
+            "aRng": aRng,
+            "maxDet": maxDet,
+            "dtIds": [d["id"] for d in dt],
+            "gtIds": [g["id"] for g in gt],
+            "dtMatches": dtm,
+            "gtMatches": gtm,
+            "dtScores": [d["score"] for d in dt],
+            "gtIgnore": gtIg,
+            "dtIgnore": dtIg,
+        }
+
+    def accumulate(self, p=None):
+        """
+        Accumulate per image evaluation results and store the result in self.eval
+        :param p: input params for evaluation
+        :return: None
+        """
+        print("Accumulating evaluation results...")
+        tic = time.time()
+        if not self.evalImgs:
+            print("Please run evaluate() first")
+        # allows input customized parameters
+        if p is None:
+            p = self.params
+        p.catIds = p.catIds if p.useCats == 1 else [-1]
+        T = len(p.iouThrs)
+        R = len(p.recThrs)
+        K = len(p.catIds) if p.useCats else 1
+        A = len(p.areaRng)
+        M = len(p.maxDets)
+        precision = -np.ones(
+            (T, R, K, A, M)
+        )  # -1 for the precision of absent categories
+        recall = -np.ones((T, K, A, M))
+        scores = -np.ones((T, R, K, A, M))
+
+        # create dictionary for future indexing
+        _pe = self._paramsEval
+        catIds = _pe.catIds if _pe.useCats else [-1]
+        setK = set(catIds)
+        setA = set(map(tuple, _pe.areaRng))
+        setM = set(_pe.maxDets)
+        setI = set(_pe.imgIds)
+        # get inds to evaluate
+        k_list = [n for n, k in enumerate(p.catIds) if k in setK]
+        m_list = [m for n, m in enumerate(p.maxDets) if m in setM]
+        a_list = [
+            n for n, a in enumerate(map(lambda x: tuple(x), p.areaRng)) if a in setA
+        ]
+        i_list = [n for n, i in enumerate(p.imgIds) if i in setI]
+        I0 = len(_pe.imgIds)
+        A0 = len(_pe.areaRng)
+        # retrieve E at each category, area range, and max number of detections
+        for k, k0 in enumerate(k_list):
+            Nk = k0 * A0 * I0
+            for a, a0 in enumerate(a_list):
+                Na = a0 * I0
+                for m, maxDet in enumerate(m_list):
+                    E = [self.evalImgs[Nk + Na + i] for i in i_list]
+                    E = [e for e in E if not e is None]
+                    if len(E) == 0:
+                        continue
+                    dtScores = np.concatenate([e["dtScores"][0:maxDet] for e in E])
+
+                    # different sorting method generates slightly different results.
+                    # mergesort is used to be consistent as Matlab implementation.
+                    inds = np.argsort(-dtScores, kind="mergesort")
+                    dtScoresSorted = dtScores[inds]
+
+                    dtm = np.concatenate(
+                        [e["dtMatches"][:, 0:maxDet] for e in E], axis=1
+                    )[:, inds]
+                    dtIg = np.concatenate(
+                        [e["dtIgnore"][:, 0:maxDet] for e in E], axis=1
+                    )[:, inds]
+                    gtIg = np.concatenate([e["gtIgnore"] for e in E])
+                    npig = np.count_nonzero(gtIg == 0)
+                    if npig == 0:
+                        continue
+                    tps = np.logical_and(dtm, np.logical_not(dtIg))
+                    fps = np.logical_and(np.logical_not(dtm), np.logical_not(dtIg))
+
+                    tp_sum = np.cumsum(tps, axis=1).astype(dtype=dtype_float)
+                    fp_sum = np.cumsum(fps, axis=1).astype(dtype=dtype_float)
+                    for t, (tp, fp) in enumerate(zip(tp_sum, fp_sum)):
+                        tp = np.array(tp)
+                        fp = np.array(fp)
+                        nd = len(tp)
+                        rc = tp / npig
+                        pr = tp / (fp + tp + np.spacing(1))
+                        q = np.zeros((R,))
+                        ss = np.zeros((R,))
+
+                        if nd:
+                            recall[t, k, a, m] = rc[-1]
+                        else:
+                            recall[t, k, a, m] = 0
+
+                        # numpy is slow without cython optimization for accessing elements
+                        # use python array gets significant speed improvement
+                        pr = pr.tolist()
+                        q = q.tolist()
+
+                        for i in range(nd - 1, 0, -1):
+                            if pr[i] > pr[i - 1]:
+                                pr[i - 1] = pr[i]
+
+                        inds = np.searchsorted(rc, p.recThrs, side="left")
+                        try:
+                            for ri, pi in enumerate(inds):
+                                q[ri] = pr[pi]
+                                ss[ri] = dtScoresSorted[pi]
+                        except:
+                            pass
+                        precision[t, :, k, a, m] = np.array(q)
+                        scores[t, :, k, a, m] = np.array(ss)
+        self.eval = {
+            "params": p,
+            "counts": [T, R, K, A, M],
+            "date": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+            "precision": precision,
+            "recall": recall,
+            "scores": scores,
+        }
+        toc = time.time()
+        print("DONE (t={:0.2f}s).".format(toc - tic))
+
+    def summarize(self):
+        """
+        Compute and display summary metrics for evaluation results.
+        Note this functin can *only* be applied on the default parameter setting
+        """
+
+        def _summarize(ap=1, iouThr=None, areaRng="all", maxDets=100):
+            p = self.params
+            iStr = " {:<18} {} @[ IoU={:<9} | area={:>6s} | maxDets={:>3d} ] = {:0.3f}"
+            titleStr = "Average Precision" if ap == 1 else "Average Recall"
+            typeStr = "(AP)" if ap == 1 else "(AR)"
+            iouStr = (
+                "{:0.2f}:{:0.2f}".format(p.iouThrs[0], p.iouThrs[-1])
+                if iouThr is None
+                else "{:0.2f}".format(iouThr)
+            )
+
+            aind = [i for i, aRng in enumerate(p.areaRngLbl) if aRng == areaRng]
+            mind = [i for i, mDet in enumerate(p.maxDets) if mDet == maxDets]
+            if ap == 1:
+                # dimension of precision: [TxRxKxAxM]
+                s = self.eval["precision"]
+                # IoU
+                if iouThr is not None:
+                    t = np.where(iouThr == p.iouThrs)[0]
+                    s = s[t]
+                s = s[:, :, :, aind, mind]
+            else:
+                # dimension of recall: [TxKxAxM]
+                s = self.eval["recall"]
+                if iouThr is not None:
+                    t = np.where(iouThr == p.iouThrs)[0]
+                    s = s[t]
+                s = s[:, :, aind, mind]
+            if len(s[s > -1]) == 0:
+                mean_s = -1
+            else:
+                mean_s = np.mean(s[s > -1])
+            print(iStr.format(titleStr, typeStr, iouStr, areaRng, maxDets, mean_s))
+            return mean_s
+
+        def _summarizeDets():
+            stats = np.zeros((12,))
+            stats[0] = _summarize(1)
+            stats[1] = _summarize(1, iouThr=0.5, maxDets=self.params.maxDets[2])
+            stats[2] = _summarize(1, iouThr=0.75, maxDets=self.params.maxDets[2])
+            stats[3] = _summarize(1, areaRng="small", maxDets=self.params.maxDets[2])
+            stats[4] = _summarize(1, areaRng="medium", maxDets=self.params.maxDets[2])
+            stats[5] = _summarize(1, areaRng="large", maxDets=self.params.maxDets[2])
+            stats[6] = _summarize(0, maxDets=self.params.maxDets[0])
+            stats[7] = _summarize(0, maxDets=self.params.maxDets[1])
+            stats[8] = _summarize(0, maxDets=self.params.maxDets[2])
+            stats[9] = _summarize(0, areaRng="small", maxDets=self.params.maxDets[2])
+            stats[10] = _summarize(0, areaRng="medium", maxDets=self.params.maxDets[2])
+            stats[11] = _summarize(0, areaRng="large", maxDets=self.params.maxDets[2])
+            return stats
+
+        def _summarizeKps():
+            stats = np.zeros((10,))
+            stats[0] = _summarize(1, maxDets=20)
+            stats[1] = _summarize(1, maxDets=20, iouThr=0.5)
+            stats[2] = _summarize(1, maxDets=20, iouThr=0.75)
+            stats[3] = _summarize(1, maxDets=20, areaRng="medium")
+            stats[4] = _summarize(1, maxDets=20, areaRng="large")
+            stats[5] = _summarize(0, maxDets=20)
+            stats[6] = _summarize(0, maxDets=20, iouThr=0.5)
+            stats[7] = _summarize(0, maxDets=20, iouThr=0.75)
+            stats[8] = _summarize(0, maxDets=20, areaRng="medium")
+            stats[9] = _summarize(0, maxDets=20, areaRng="large")
+            return stats
+
+        if not self.eval:
+            raise Exception("Please run accumulate() first")
+        iouType = self.params.iouType
+        if iouType == "segm" or iouType == "bbox":
+            summarize = _summarizeDets
+        elif iouType == "keypoints":
+            summarize = _summarizeKps
+        self.stats = summarize()
+
+    def __str__(self):
+        self.summarize()
+
+
+class Params:
+    """
+    Params for coco evaluation api
+    """
+
+    def setDetParams(self):
+        self.imgIds = []
+        self.catIds = []
+        # np.arange causes trouble.  the data point on arange is slightly larger than the true value
+        self.iouThrs = np.linspace(
+            0.5, 0.95, int(np.round((0.95 - 0.5) / 0.05)) + 1, endpoint=True
+        )
+        self.recThrs = np.linspace(
+            0.0, 1.00, int(np.round((1.00 - 0.0) / 0.01)) + 1, endpoint=True
+        )
+        self.maxDets = [1, 10, 100]
+        self.areaRng = [
+            [0**2, 1e5**2],
+            [0**2, 32**2],
+            [32**2, 96**2],
+            [96**2, 1e5**2],
+        ]
+        self.areaRngLbl = ["all", "small", "medium", "large"]
+        self.useCats = 1
+
+    def setKpParams(self):
+        self.imgIds = []
+        self.catIds = []
+        # np.arange causes trouble.  the data point on arange is slightly larger than the true value
+        self.iouThrs = np.linspace(
+            0.5, 0.95, int(np.round((0.95 - 0.5) / 0.05)) + 1, endpoint=True
+        )
+        self.recThrs = np.linspace(
+            0.0, 1.00, int(np.round((1.00 - 0.0) / 0.01)) + 1, endpoint=True
+        )
+        self.maxDets = [20]
+        self.areaRng = [[0**2, 1e5**2], [32**2, 96**2], [96**2, 1e5**2]]
+        self.areaRngLbl = ["all", "medium", "large"]
+        self.useCats = 1
+        self.kpt_oks_sigmas = (
+            np.array(
+                [
+                    0.26,
+                    0.25,
+                    0.25,
+                    0.35,
+                    0.35,
+                    0.79,
+                    0.79,
+                    0.72,
+                    0.72,
+                    0.62,
+                    0.62,
+                    1.07,
+                    1.07,
+                    0.87,
+                    0.87,
+                    0.89,
+                    0.89,
+                ]
+            )
+            / 10.0
+        )
+
+    def __init__(self, iouType="segm"):
+        if iouType == "bbox":
+            self.setDetParams()
+        else:
+            raise Exception("iouType not supported")
+        self.iouType = iouType
+        # useSegm is deprecated
+        self.useSegm = None

detection_metrics/pycocotools/mask.py

@@ -0,0 +1,103 @@
+__author__ = 'tsungyi'
+
+from detection_metrics.pycocotools import _mask
+
+# Interface for manipulating masks stored in RLE format.
+#
+# RLE is a simple yet efficient format for storing binary masks. RLE
+# first divides a vector (or vectorized image) into a series of piecewise
+# constant regions and then for each piece simply stores the length of
+# that piece. For example, given M=[0 0 1 1 1 0 1] the RLE counts would
+# be [2 3 1 1], or for M=[1 1 1 1 1 1 0] the counts would be [0 6 1]
+# (note that the odd counts are always the numbers of zeros). Instead of
+# storing the counts directly, additional compression is achieved with a
+# variable bitrate representation based on a common scheme called LEB128.
+#
+# Compression is greatest given large piecewise constant regions.
+# Specifically, the size of the RLE is proportional to the number of
+# *boundaries* in M (or for an image the number of boundaries in the y
+# direction). Assuming fairly simple shapes, the RLE representation is
+# O(sqrt(n)) where n is number of pixels in the object. Hence space usage
+# is substantially lower, especially for large simple objects (large n).
+#
+# Many common operations on masks can be computed directly using the RLE
+# (without need for decoding). This includes computations such as area,
+# union, intersection, etc. All of these operations are linear in the
+# size of the RLE, in other words they are O(sqrt(n)) where n is the area
+# of the object. Computing these operations on the original mask is O(n).
+# Thus, using the RLE can result in substantial computational savings.
+#
+# The following API functions are defined:
+#  encode         - Encode binary masks using RLE.
+#  decode         - Decode binary masks encoded via RLE.
+#  merge          - Compute union or intersection of encoded masks.
+#  iou            - Compute intersection over union between masks.
+#  area           - Compute area of encoded masks.
+#  toBbox         - Get bounding boxes surrounding encoded masks.
+#  frPyObjects    - Convert polygon, bbox, and uncompressed RLE to encoded RLE mask.
+#
+# Usage:
+#  Rs     = encode( masks )
+#  masks  = decode( Rs )
+#  R      = merge( Rs, intersect=false )
+#  o      = iou( dt, gt, iscrowd )
+#  a      = area( Rs )
+#  bbs    = toBbox( Rs )
+#  Rs     = frPyObjects( [pyObjects], h, w )
+#
+# In the API the following formats are used:
+#  Rs      - [dict] Run-length encoding of binary masks
+#  R       - dict Run-length encoding of binary mask
+#  masks   - [hxwxn] Binary mask(s) (must have type np.ndarray(dtype=uint8) in column-major order)
+#  iscrowd - [nx1] list of np.ndarray. 1 indicates corresponding gt image has crowd region to ignore
+#  bbs     - [nx4] Bounding box(es) stored as [x y w h]
+#  poly    - Polygon stored as [[x1 y1 x2 y2...],[x1 y1 ...],...] (2D list)
+#  dt,gt   - May be either bounding boxes or encoded masks
+# Both poly and bbs are 0-indexed (bbox=[0 0 1 1] encloses first pixel).
+#
+# Finally, a note about the intersection over union (iou) computation.
+# The standard iou of a ground truth (gt) and detected (dt) object is
+#  iou(gt,dt) = area(intersect(gt,dt)) / area(union(gt,dt))
+# For "crowd" regions, we use a modified criteria. If a gt object is
+# marked as "iscrowd", we allow a dt to match any subregion of the gt.
+# Choosing gt' in the crowd gt that best matches the dt can be done using
+# gt'=intersect(dt,gt). Since by definition union(gt',dt)=dt, computing
+#  iou(gt,dt,iscrowd) = iou(gt',dt) = area(intersect(gt,dt)) / area(dt)
+# For crowd gt regions we use this modified criteria above for the iou.
+#
+# To compile run "python setup.py build_ext --inplace"
+# Please do not contact us for help with compiling.
+#
+# Microsoft COCO Toolbox.      version 2.0
+# Data, paper, and tutorials available at:  http://mscoco.org/
+# Code written by Piotr Dollar and Tsung-Yi Lin, 2015.
+# Licensed under the Simplified BSD License [see coco/license.txt]
+
+iou         = _mask.iou
+merge       = _mask.merge
+frPyObjects = _mask.frPyObjects
+
+def encode(bimask):
+    if len(bimask.shape) == 3:
+        return _mask.encode(bimask)
+    elif len(bimask.shape) == 2:
+        h, w = bimask.shape
+        return _mask.encode(bimask.reshape((h, w, 1), order='F'))[0]
+
+def decode(rleObjs):
+    if type(rleObjs) == list:
+        return _mask.decode(rleObjs)
+    else:
+        return _mask.decode([rleObjs])[:,:,0]
+
+def area(rleObjs):
+    if type(rleObjs) == list:
+        return _mask.area(rleObjs)
+    else:
+        return _mask.area([rleObjs])[0]
+
+def toBbox(rleObjs):
+    if type(rleObjs) == list:
+        return _mask.toBbox(rleObjs)
+    else:
+        return _mask.toBbox([rleObjs])[0]

detection_metrics/pycocotools/mask_utils.py

@@ -0,0 +1,76 @@
+# This code is a copy and paste with small modifications of the code:
+# https://github.com/rafaelpadilla/review_object_detection_metrics/blob/main/src/evaluators/coco_evaluator.py
+
+from typing import List
+import numpy as np
+
+class MaskEvaluator(object):
+    @staticmethod
+    def iou(
+        dt: List[List[float]], gt: List[List[float]], iscrowd: List[bool]
+    ) -> np.ndarray:
+        """
+        Calculate the intersection over union (IoU) between detection bounding boxes (dt) and \
+            ground truth bounding boxes (gt).
+        Reference: https://github.com/rafaelpadilla/review_object_detection_metrics
+
+        Args:
+            dt (List[List[float]]): List of detection bounding boxes in the \
+                format [x, y, width, height].
+            gt (List[List[float]]): List of ground-truth bounding boxes in the \
+                format [x, y, width, height].
+            iscrowd (List[bool]): List indicating if each ground-truth bounding box \
+                is a crowd region or not.
+
+        Returns:
+            np.ndarray: Array of IoU values of shape (len(dt), len(gt)).
+        """
+        assert len(iscrowd) == len(gt), "iou(iscrowd=) must have the same length as gt"
+        if len(dt) == 0 or len(gt) == 0:
+            return []
+        ious = np.zeros((len(dt), len(gt)), dtype=np.float64)
+        for g_idx, g in enumerate(gt):
+            for d_idx, d in enumerate(dt):
+                ious[d_idx, g_idx] = _jaccard(d, g, iscrowd[g_idx])
+        return ious
+
+
+def _jaccard(a: List[float], b: List[float], iscrowd: bool) -> float:
+    """
+    Calculate the Jaccard index (intersection over union) between two bounding boxes.
+    For "crowd" regions, we use a modified criteria. If a gt object is
+    marked as "iscrowd", we allow a dt to match any subregion of the gt.
+    Choosing gt' in the crowd gt that best matches the dt can be done using
+    gt'=intersect(dt,gt). Since by definition union(gt',dt)=dt, computing
+        iou(gt,dt,iscrowd) = iou(gt',dt) = area(intersect(gt,dt)) / area(dt)
+    For crowd gt regions we use this modified criteria above for the iou.
+
+    Args:
+        a (List[float]): Bounding box coordinates in the format [x, y, width, height].
+        b (List[float]): Bounding box coordinates in the format [x, y, width, height].
+        iscrowd (bool): Flag indicating if the second bounding box is a crowd region or not.
+
+    Returns:
+        float: Jaccard index between the two bounding boxes.
+    """  
+    eps = 4e-12
+    xa, ya, x2a, y2a = a[0], a[1], a[0] + a[2], a[1] + a[3]
+    xb, yb, x2b, y2b = b[0], b[1], b[0] + b[2], b[1] + b[3]
+
+    # innermost left x
+    xi = max(xa, xb)
+    # innermost right x
+    x2i = min(x2a, x2b)
+    # same for y
+    yi = max(ya, yb)
+    y2i = min(y2a, y2b)
+
+    # calculate areas
+    Aa = max(x2a - xa, 0.) * max(y2a - ya, 0.)
+    Ab = max(x2b - xb, 0.) * max(y2b - yb, 0.)
+    Ai = max(x2i - xi, 0.) * max(y2i - yi, 0.)
+
+    if iscrowd:
+        return Ai / (Aa + eps)
+
+    return Ai / (Aa + Ab - Ai + eps)

detection_metrics/utils.py

@@ -0,0 +1,156 @@
+import copy
+import pickle
+from typing import Dict, List, Tuple, Union
+from tqdm import tqdm
+import numpy as np
+import torch
+import torch.distributed as dist
+from datasets import Dataset
+
+from detection_metrics.pycocotools.cocoeval import COCOeval
+
+# Typings
+_TYPING_BOX = Tuple[float, float, float, float]
+_TYPING_SCORES = List[float]
+_TYPING_LABELS = List[int]
+_TYPING_BOXES = List[_TYPING_BOX]
+_TYPING_PRED_REF = Union[_TYPING_SCORES, _TYPING_LABELS, _TYPING_BOXES]
+_TYPING_PREDICTION = Dict[str, _TYPING_PRED_REF]
+_TYPING_REFERENCE = Dict[str, _TYPING_PRED_REF]
+_TYPING_PREDICTIONS = Dict[int, _TYPING_PREDICTION]
+
+def convert_to_xywh(boxes: torch.Tensor) -> torch.Tensor:
+    """
+    Convert bounding boxes from (xmin, ymin, xmax, ymax) format to (x, y, width, height) format.
+
+    Args:
+        boxes (torch.Tensor): Tensor of shape (N, 4) representing bounding boxes in \
+            (xmin, ymin, xmax, ymax) format.
+
+    Returns:
+        torch.Tensor: Tensor of shape (N, 4) representing bounding boxes in (x, y, width, height) \
+            format.
+    """
+    xmin, ymin, xmax, ymax = boxes.unbind(1)
+    return torch.stack((xmin, ymin, xmax - xmin, ymax - ymin), dim=1)
+
+
+def create_common_coco_eval(
+    coco_eval: COCOeval, img_ids: List[int], eval_imgs: np.ndarray
+) -> None:
+    """
+    Create a common COCO evaluation by merging image IDs and evaluation images into the \
+        coco_eval object.
+
+    Args:
+        coco_eval: COCOeval evaluation object.
+        img_ids (List[int]): Tensor of image IDs.
+        eval_imgs (torch.Tensor): Tensor of evaluation images.
+    """
+    img_ids, eval_imgs = merge(img_ids, eval_imgs)
+    img_ids = list(img_ids)
+    eval_imgs = list(eval_imgs.flatten())
+
+    coco_eval.evalImgs = eval_imgs
+    coco_eval.params.imgIds = img_ids
+    coco_eval._paramsEval = copy.deepcopy(coco_eval.params)
+
+
+def merge(img_ids: List[int], eval_imgs: np.ndarray) -> Tuple[np.ndarray, np.ndarray]:
+    """
+    Merge image IDs and evaluation images from different processes.
+
+    Args:
+        img_ids (List[int]): List of image ID arrays from different processes.
+        eval_imgs (np.ndarray): Evaluation images from different processes.
+
+    Returns:
+        Tuple[np.ndarray, np.ndarray]: Merged image IDs and evaluation images.
+    """
+    all_img_ids = all_gather(img_ids)
+    all_eval_imgs = all_gather(eval_imgs)
+
+    merged_img_ids = []
+    for p in all_img_ids:
+        merged_img_ids.extend(p)
+
+    merged_eval_imgs = []
+    for p in all_eval_imgs:
+        merged_eval_imgs.append(p)
+
+    merged_img_ids = np.array(merged_img_ids)
+    merged_eval_imgs = np.concatenate(merged_eval_imgs, 2)
+
+    # keep only unique (and in sorted order) images
+    merged_img_ids, idx = np.unique(merged_img_ids, return_index=True)
+    merged_eval_imgs = merged_eval_imgs[..., idx]
+
+    return merged_img_ids, merged_eval_imgs
+
+
+def all_gather(data: List[int]) -> List[List[int]]:
+    """
+    Run all_gather on arbitrary picklable data (not necessarily tensors).
+
+    Args:
+        data (List[int]): any picklable object
+    Returns:
+        List[List[int]]: list of data gathered from each rank
+    """
+    world_size = get_world_size()
+    if world_size == 1:
+        return [data]
+
+    # serialized to a Tensor
+    buffer = pickle.dumps(data)
+    storage = torch.ByteStorage.from_buffer(buffer)
+    tensor = torch.ByteTensor(storage).to("cuda")
+
+    # obtain Tensor size of each rank
+    local_size = torch.tensor([tensor.numel()], device="cuda")
+    size_list = [torch.tensor([0], device="cuda") for _ in range(world_size)]
+    dist.all_gather(size_list, local_size)
+    size_list = [int(size.item()) for size in size_list]
+    max_size = max(size_list)
+
+    # receiving Tensor from all ranks
+    # we pad the tensor because torch all_gather does not support
+    # gathering tensors of different shapes
+    tensor_list = []
+    for _ in size_list:
+        tensor_list.append(torch.empty((max_size,), dtype=torch.uint8, device="cuda"))
+    if local_size != max_size:
+        padding = torch.empty(
+            size=(max_size - local_size,), dtype=torch.uint8, device="cuda"
+        )
+        tensor = torch.cat((tensor, padding), dim=0)
+    dist.all_gather(tensor_list, tensor)
+
+    data_list = []
+    for size, tensor in zip(size_list, tensor_list):
+        buffer = tensor.cpu().numpy().tobytes()[:size]
+        data_list.append(pickle.loads(buffer))
+
+    return data_list
+
+
+def get_world_size() -> int:
+    """
+    Get the number of processes in the distributed environment.
+
+    Returns:
+        int: Number of processes.
+    """
+    if not is_dist_avail_and_initialized():
+        return 1
+    return dist.get_world_size()
+
+
+def is_dist_avail_and_initialized() -> bool:
+    """
+    Check if distributed environment is available and initialized.
+
+    Returns:
+        bool: True if distributed environment is available and initialized, False otherwise.
+    """
+    return dist.is_available() and dist.is_initialized()

requirements.txt

@@ -0,0 +1,2 @@
+torch
+torchvision

setup.py

@@ -0,0 +1,42 @@
+from detection_metrics import __version__
+import subprocess
+
+from setuptools import setup
+from setuptools.command.develop import develop
+from setuptools.command.egg_info import egg_info
+from setuptools.command.install import install
+
+
+def custom_command():
+    subprocess.call(["pip", "install", "numpy", "cython"])
+    subprocess.call(["pip", "install", "-r", "requirements.txt", "--user"])
+
+class CustomInstallCommand(install):
+    def run(self):
+        install.run(self)
+        custom_command()
+
+class CustomDevelopCommand(develop):
+    def run(self):
+        develop.run(self)
+        custom_command()
+
+
+class CustomEggInfoCommand(egg_info):
+    def run(self):
+        egg_info.run(self)
+        custom_command()
+
+setup(
+    name="detection_metrics",
+    description="COCO Metrics for Object Detection and Instance Segmentation",
+    version=__version__,
+    zip_safe=True,
+    packages=["detection_metrics", "detection_metrics.pycocotools"],
+    include_package_data=True,
+    cmdclass={
+        "install": CustomInstallCommand,
+        "develop": CustomDevelopCommand,
+        "egg_info": CustomEggInfoCommand,
+    },
+)