Source code for mmaction.evaluation.metrics.anet_metric
# Copyright (c) OpenMMLab. All rights reserved.
import os
import os.path as osp
from collections import OrderedDict
from typing import Any, Optional, Sequence, Tuple
import mmcv
import mmengine
import numpy as np
from mmengine.evaluator import BaseMetric
from mmaction.evaluation import average_recall_at_avg_proposals
from mmaction.registry import METRICS
from mmaction.utils import ConfigType
[docs]@METRICS.register_module()
class ANetMetric(BaseMetric):
"""ActivityNet dataset evaluation metric."""
def __init__(self,
metric_type: str = 'TEM',
collect_device: str = 'cpu',
prefix: Optional[str] = None,
metric_options: dict = {},
dump_config: ConfigType = dict(out='')):
super().__init__(collect_device=collect_device, prefix=prefix)
self.metric_type = metric_type
assert 'out' in dump_config
self.output_format = dump_config.pop('output_format', 'csv')
self.out = dump_config['out']
self.metric_options = metric_options
if self.metric_type == 'AR@AN':
self.ground_truth = {}
[docs] def process(self, data_batch: Sequence[Tuple[Any, dict]],
predictions: Sequence[dict]) -> None:
"""Process one batch of data samples and predictions. 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[Tuple[Any, dict]]): A batch of data
from the dataloader.
predictions (Sequence[dict]): A batch of outputs from
the model.
"""
for pred in predictions:
self.results.append(pred)
if self.metric_type == 'AR@AN':
data_batch = data_batch['data_samples']
for data_sample in data_batch:
video_info = data_sample.metainfo
video_id = video_info['video_name'][2:]
this_video_gt = []
for ann in video_info['annotations']:
t_start, t_end = ann['segment']
label = ann['label']
this_video_gt.append([t_start, t_end, label])
self.ground_truth[video_id] = np.array(this_video_gt)
[docs] def compute_metrics(self, results: list) -> dict:
"""Compute the metrics from processed results.
If `metric_type` is 'TEM', only dump middle results and do not compute
any metrics.
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.
"""
self.dump_results(results)
if self.metric_type == 'AR@AN':
return self.compute_ARAN(results)
return OrderedDict()
[docs] def compute_ARAN(self, results: list) -> dict:
"""AR@AN evaluation metric."""
temporal_iou_thresholds = self.metric_options.setdefault(
'AR@AN', {}).setdefault('temporal_iou_thresholds',
np.linspace(0.5, 0.95, 10))
max_avg_proposals = self.metric_options.setdefault(
'AR@AN', {}).setdefault('max_avg_proposals', 100)
if isinstance(temporal_iou_thresholds, list):
temporal_iou_thresholds = np.array(temporal_iou_thresholds)
eval_results = OrderedDict()
proposal, num_proposals = self._import_proposals(results)
recall, _, _, auc = average_recall_at_avg_proposals(
self.ground_truth,
proposal,
num_proposals,
max_avg_proposals=max_avg_proposals,
temporal_iou_thresholds=temporal_iou_thresholds)
eval_results['auc'] = auc
eval_results['AR@1'] = np.mean(recall[:, 0])
eval_results['AR@5'] = np.mean(recall[:, 4])
eval_results['AR@10'] = np.mean(recall[:, 9])
eval_results['AR@100'] = np.mean(recall[:, 99])
return eval_results
[docs] def dump_results(self, results, version='VERSION 1.3'):
"""Save middle or final results to disk."""
if self.output_format == 'json':
result_dict = self.proposals2json(results)
output_dict = {
'version': version,
'results': result_dict,
'external_data': {}
}
mmengine.dump(output_dict, self.out)
elif self.output_format == 'csv':
os.makedirs(self.out, exist_ok=True)
header = 'action,start,end,tmin,tmax'
for result in results:
video_name, outputs = result
output_path = osp.join(self.out, video_name + '.csv')
np.savetxt(
output_path,
outputs,
header=header,
delimiter=',',
comments='')
else:
raise ValueError(
f'The output format {self.output_format} is not supported.')
[docs] @staticmethod
def proposals2json(results, show_progress=False):
"""Convert all proposals to a final dict(json) format.
Args:
results (list[dict]): All proposals.
show_progress (bool): Whether to show the progress bar.
Defaults: False.
Returns:
dict: The final result dict. E.g.
.. code-block:: Python
dict(video-1=[dict(segment=[1.1,2.0]. score=0.9),
dict(segment=[50.1, 129.3], score=0.6)])
"""
result_dict = {}
print('Convert proposals to json format')
if show_progress:
prog_bar = mmcv.ProgressBar(len(results))
for result in results:
video_name = result['video_name']
result_dict[video_name[2:]] = result['proposal_list']
if show_progress:
prog_bar.update()
return result_dict
@staticmethod
def _import_proposals(results):
"""Read predictions from results."""
proposals = {}
num_proposals = 0
for result in results:
video_id = result['video_name'][2:]
this_video_proposals = []
for proposal in result['proposal_list']:
t_start, t_end = proposal['segment']
score = proposal['score']
this_video_proposals.append([t_start, t_end, score])
num_proposals += 1
proposals[video_id] = np.array(this_video_proposals)
return proposals, num_proposals