from typing import Dict, List, Union
from pathlib import Path
import datasets
import torch
import evaluate
import json
from tqdm import tqdm
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
__author__ = 'tsungyi'
# 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]
try:
iou = _mask.iou
merge = _mask.merge
frPyObjects = _mask.frPyObjects
except:
pass
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]
# 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 maskUtils(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)
# 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
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
# 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
# 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()
import contextlib
import copy
import os
from typing import Dict, List, Union
import numpy as np
import torch
_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
_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