Source code for mmaction.evaluation.metrics.acc_metric
# Copyright (c) OpenMMLab. All rights reserved.
import copy
from collections import OrderedDict
from itertools import product
from typing import Any, Dict, List, Optional, Sequence, Tuple, Union
import mmengine
import numpy as np
import torch
from mmengine.evaluator import BaseMetric
from mmaction.evaluation import (get_weighted_score, mean_average_precision,
mean_class_accuracy,
mmit_mean_average_precision, top_k_accuracy)
from mmaction.registry import METRICS
def to_tensor(value):
"""Convert value to torch.Tensor."""
if isinstance(value, np.ndarray):
value = torch.from_numpy(value)
elif isinstance(value, Sequence) and not mmengine.is_str(value):
value = torch.tensor(value)
elif not isinstance(value, torch.Tensor):
raise TypeError(f'{type(value)} is not an available argument.')
return value
[docs]@METRICS.register_module()
class AccMetric(BaseMetric):
"""Accuracy evaluation metric."""
default_prefix: Optional[str] = 'acc'
def __init__(self,
metric_list: Optional[Union[str, Tuple[str]]] = (
'top_k_accuracy', 'mean_class_accuracy'),
collect_device: str = 'cpu',
metric_options: Optional[Dict] = dict(
top_k_accuracy=dict(topk=(1, 5))),
prefix: Optional[str] = None) -> None:
# TODO: fix the metric_list argument with a better one.
# `metrics` is not a safe argument here with mmengine.
# we have to replace it with `metric_list`.
super().__init__(collect_device=collect_device, prefix=prefix)
if not isinstance(metric_list, (str, tuple)):
raise TypeError('metric_list must be str or tuple of str, '
f'but got {type(metric_list)}')
if isinstance(metric_list, str):
metrics = (metric_list, )
else:
metrics = metric_list
# coco evaluation metrics
for metric in metrics:
assert metric in [
'top_k_accuracy', 'mean_class_accuracy',
'mmit_mean_average_precision', 'mean_average_precision'
]
self.metrics = metrics
self.metric_options = metric_options
[docs] def process(self, data_batch: Sequence[Tuple[Any, Dict]],
data_samples: Sequence[Dict]) -> None:
"""Process one batch of data samples and data_samples. The processed
results should be stored in ``self.results``, which will be used to
compute the metrics when all batches have been processed.
Args:
data_batch (Sequence[dict]): A batch of data from the dataloader.
data_samples (Sequence[dict]): A batch of outputs from the model.
"""
data_samples = copy.deepcopy(data_samples)
for data_sample in data_samples:
result = dict()
pred = data_sample['pred_score']
label = data_sample['gt_label']
# Ad-hoc for RGBPoseConv3D
if isinstance(pred, dict):
for item_name, score in pred.items():
pred[item_name] = score.cpu().numpy()
else:
pred = pred.cpu().numpy()
result['pred'] = pred
if label.size(0) == 1:
# single-label
result['label'] = label.item()
else:
# multi-label
result['label'] = label.cpu().numpy()
self.results.append(result)
[docs] def compute_metrics(self, results: List) -> Dict:
"""Compute the metrics from processed results.
Args:
results (list): The processed results of each batch.
Returns:
dict: The computed metrics. The keys are the names of the metrics,
and the values are corresponding results.
"""
labels = [x['label'] for x in results]
eval_results = dict()
# Ad-hoc for RGBPoseConv3D
if isinstance(results[0]['pred'], dict):
for item_name in results[0]['pred'].keys():
preds = [x['pred'][item_name] for x in results]
eval_result = self.calculate(preds, labels)
eval_results.update(
{f'{item_name}_{k}': v
for k, v in eval_result.items()})
if len(results[0]['pred']) == 2 and \
'rgb' in results[0]['pred'] and \
'pose' in results[0]['pred']:
rgb = [x['pred']['rgb'] for x in results]
pose = [x['pred']['pose'] for x in results]
preds = {
'1:1': get_weighted_score([rgb, pose], [1, 1]),
'2:1': get_weighted_score([rgb, pose], [2, 1]),
'1:2': get_weighted_score([rgb, pose], [1, 2])
}
for k in preds:
eval_result = self.calculate(preds[k], labels)
eval_results.update({
f'RGBPose_{k}_{key}': v
for key, v in eval_result.items()
})
return eval_results
# Simple Acc Calculation
else:
preds = [x['pred'] for x in results]
return self.calculate(preds, labels)
[docs] def calculate(self, preds: List[np.ndarray],
labels: List[Union[int, np.ndarray]]) -> Dict:
"""Compute the metrics from processed results.
Args:
preds (list[np.ndarray]): List of the prediction scores.
labels (list[int | np.ndarray]): List of the labels.
Returns:
dict: The computed metrics. The keys are the names of the metrics,
and the values are corresponding results.
"""
eval_results = OrderedDict()
metric_options = copy.deepcopy(self.metric_options)
for metric in self.metrics:
if metric == 'top_k_accuracy':
topk = metric_options.setdefault('top_k_accuracy',
{}).setdefault(
'topk', (1, 5))
if not isinstance(topk, (int, tuple)):
raise TypeError('topk must be int or tuple of int, '
f'but got {type(topk)}')
if isinstance(topk, int):
topk = (topk, )
top_k_acc = top_k_accuracy(preds, labels, topk)
for k, acc in zip(topk, top_k_acc):
eval_results[f'top{k}'] = acc
if metric == 'mean_class_accuracy':
mean1 = mean_class_accuracy(preds, labels)
eval_results['mean1'] = mean1
if metric in [
'mean_average_precision',
'mmit_mean_average_precision',
]:
if metric == 'mean_average_precision':
mAP = mean_average_precision(preds, labels)
eval_results['mean_average_precision'] = mAP
elif metric == 'mmit_mean_average_precision':
mAP = mmit_mean_average_precision(preds, labels)
eval_results['mmit_mean_average_precision'] = mAP
return eval_results
[docs]@METRICS.register_module()
class ConfusionMatrix(BaseMetric):
r"""A metric to calculate confusion matrix for single-label tasks.
Args:
num_classes (int, optional): The number of classes. Defaults to None.
collect_device (str): Device name used for collecting results from
different ranks during distributed training. Must be 'cpu' or
'gpu'. Defaults to 'cpu'.
prefix (str, optional): The prefix that will be added in the metric
names to disambiguate homonymous metrics of different evaluators.
If prefix is not provided in the argument, self.default_prefix
will be used instead. Defaults to None.
Examples:
1. The basic usage.
>>> import torch
>>> from mmaction.evaluation import ConfusionMatrix
>>> y_pred = [0, 1, 1, 3]
>>> y_true = [0, 2, 1, 3]
>>> ConfusionMatrix.calculate(y_pred, y_true, num_classes=4)
tensor([[1, 0, 0, 0],
[0, 1, 0, 0],
[0, 1, 0, 0],
[0, 0, 0, 1]])
>>> # plot the confusion matrix
>>> import matplotlib.pyplot as plt
>>> y_score = torch.rand((1000, 10))
>>> y_true = torch.randint(10, (1000, ))
>>> matrix = ConfusionMatrix.calculate(y_score, y_true)
>>> ConfusionMatrix().plot(matrix)
>>> plt.show()
2. In the config file
.. code:: python
val_evaluator = dict(type='ConfusionMatrix')
test_evaluator = dict(type='ConfusionMatrix')
""" # noqa: E501
default_prefix = 'confusion_matrix'
def __init__(self,
num_classes: Optional[int] = None,
collect_device: str = 'cpu',
prefix: Optional[str] = None) -> None:
super().__init__(collect_device, prefix)
self.num_classes = num_classes
[docs] def process(self, data_batch, data_samples: Sequence[dict]) -> None:
for data_sample in data_samples:
pred_scores = data_sample.get('pred_score')
gt_label = data_sample['gt_label']
if pred_scores is not None:
pred_label = pred_scores.argmax(dim=0, keepdim=True)
self.num_classes = pred_scores.size(0)
else:
pred_label = data_sample['pred_label']
self.results.append({
'pred_label': pred_label,
'gt_label': gt_label
})
[docs] def compute_metrics(self, results: list) -> dict:
pred_labels = []
gt_labels = []
for result in results:
pred_labels.append(result['pred_label'])
gt_labels.append(result['gt_label'])
confusion_matrix = ConfusionMatrix.calculate(
torch.cat(pred_labels),
torch.cat(gt_labels),
num_classes=self.num_classes)
return {'result': confusion_matrix}
[docs] @staticmethod
def calculate(pred, target, num_classes=None) -> dict:
"""Calculate the confusion matrix for single-label task.
Args:
pred (torch.Tensor | np.ndarray | Sequence): The prediction
results. It can be labels (N, ), or scores of every
class (N, C).
target (torch.Tensor | np.ndarray | Sequence): The target of
each prediction with shape (N, ).
num_classes (Optional, int): The number of classes. If the ``pred``
is label instead of scores, this argument is required.
Defaults to None.
Returns:
torch.Tensor: The confusion matrix.
"""
pred = to_tensor(pred)
target_label = to_tensor(target).int()
assert pred.size(0) == target_label.size(0), \
f"The size of pred ({pred.size(0)}) doesn't match "\
f'the target ({target_label.size(0)}).'
assert target_label.ndim == 1
if pred.ndim == 1:
assert num_classes is not None, \
'Please specify the `num_classes` if the `pred` is labels ' \
'intead of scores.'
pred_label = pred
else:
num_classes = num_classes or pred.size(1)
pred_label = torch.argmax(pred, dim=1).flatten()
with torch.no_grad():
indices = num_classes * target_label + pred_label
matrix = torch.bincount(indices, minlength=num_classes**2)
matrix = matrix.reshape(num_classes, num_classes)
return matrix
[docs] @staticmethod
def plot(confusion_matrix: torch.Tensor,
include_values: bool = False,
cmap: str = 'viridis',
classes: Optional[List[str]] = None,
colorbar: bool = True,
show: bool = True):
"""Draw a confusion matrix by matplotlib.
Modified from `Scikit-Learn
<https://github.com/scikit-learn/scikit-learn/blob/dc580a8ef/sklearn/metrics/_plot/confusion_matrix.py#L81>`_
Args:
confusion_matrix (torch.Tensor): The confusion matrix to draw.
include_values (bool): Whether to draw the values in the figure.
Defaults to False.
cmap (str): The color map to use. Defaults to use "viridis".
classes (list[str], optional): The names of categories.
Defaults to None, which means to use index number.
colorbar (bool): Whether to show the colorbar. Defaults to True.
show (bool): Whether to show the figure immediately.
Defaults to True.
""" # noqa: E501
import matplotlib.pyplot as plt
fig, ax = plt.subplots(figsize=(10, 10))
num_classes = confusion_matrix.size(0)
im_ = ax.imshow(confusion_matrix, interpolation='nearest', cmap=cmap)
text_ = None
cmap_min, cmap_max = im_.cmap(0), im_.cmap(1.0)
if include_values:
text_ = np.empty_like(confusion_matrix, dtype=object)
# print text with appropriate color depending on background
thresh = (confusion_matrix.max() + confusion_matrix.min()) / 2.0
for i, j in product(range(num_classes), range(num_classes)):
color = cmap_max if confusion_matrix[i,
j] < thresh else cmap_min
text_cm = format(confusion_matrix[i, j], '.2g')
text_d = format(confusion_matrix[i, j], 'd')
if len(text_d) < len(text_cm):
text_cm = text_d
text_[i, j] = ax.text(
j, i, text_cm, ha='center', va='center', color=color)
display_labels = classes or np.arange(num_classes)
if colorbar:
fig.colorbar(im_, ax=ax)
ax.set(
xticks=np.arange(num_classes),
yticks=np.arange(num_classes),
xticklabels=display_labels,
yticklabels=display_labels,
ylabel='True label',
xlabel='Predicted label',
)
ax.invert_yaxis()
ax.xaxis.tick_top()
ax.set_ylim((num_classes - 0.5, -0.5))
# Automatically rotate the x labels.
fig.autofmt_xdate(ha='center')
if show:
plt.show()
return fig