|
|
import numpy as np |
|
|
from sklearn.metrics import confusion_matrix |
|
|
|
|
|
class _StreamMetrics(object): |
|
|
def __init__(self): |
|
|
""" Overridden by subclasses """ |
|
|
raise NotImplementedError() |
|
|
|
|
|
def update(self, gt, pred): |
|
|
""" Overridden by subclasses """ |
|
|
raise NotImplementedError() |
|
|
|
|
|
def get_results(self): |
|
|
""" Overridden by subclasses """ |
|
|
raise NotImplementedError() |
|
|
|
|
|
def to_str(self, metrics): |
|
|
""" Overridden by subclasses """ |
|
|
raise NotImplementedError() |
|
|
|
|
|
def reset(self): |
|
|
""" Overridden by subclasses """ |
|
|
raise NotImplementedError() |
|
|
|
|
|
class StreamSegMetrics(_StreamMetrics): |
|
|
""" |
|
|
Stream Metrics for Semantic Segmentation Task |
|
|
""" |
|
|
def __init__(self, n_classes): |
|
|
self.n_classes = n_classes |
|
|
self.confusion_matrix = np.zeros((n_classes, n_classes)) |
|
|
|
|
|
def update(self, label_trues, label_preds): |
|
|
for lt, lp in zip(label_trues, label_preds): |
|
|
self.confusion_matrix += self._fast_hist( lt.flatten(), lp.flatten() ) |
|
|
|
|
|
@staticmethod |
|
|
def to_str(results): |
|
|
string = "\n" |
|
|
for k, v in results.items(): |
|
|
if k!="Class IoU": |
|
|
string += "%s: %f\n"%(k, v) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
return string |
|
|
|
|
|
def _fast_hist(self, label_true, label_pred): |
|
|
mask = (label_true >= 0) & (label_true < self.n_classes) |
|
|
hist = np.bincount( |
|
|
self.n_classes * label_true[mask].astype(int) + label_pred[mask], |
|
|
minlength=self.n_classes ** 2, |
|
|
).reshape(self.n_classes, self.n_classes) |
|
|
return hist |
|
|
|
|
|
def get_results(self): |
|
|
"""Returns accuracy score evaluation result. |
|
|
- overall accuracy |
|
|
- mean accuracy |
|
|
- mean IU |
|
|
- fwavacc |
|
|
""" |
|
|
hist = self.confusion_matrix |
|
|
acc = np.diag(hist).sum() / hist.sum() |
|
|
acc_cls = np.diag(hist) / hist.sum(axis=1) |
|
|
acc_cls = np.nanmean(acc_cls) |
|
|
iu = np.diag(hist) / (hist.sum(axis=1) + hist.sum(axis=0) - np.diag(hist)) |
|
|
mean_iu = np.nanmean(iu) |
|
|
freq = hist.sum(axis=1) / hist.sum() |
|
|
fwavacc = (freq[freq > 0] * iu[freq > 0]).sum() |
|
|
cls_iu = dict(zip(range(self.n_classes), iu)) |
|
|
|
|
|
return { |
|
|
"Overall Acc": acc, |
|
|
"Mean Acc": acc_cls, |
|
|
"FreqW Acc": fwavacc, |
|
|
"Mean IoU": mean_iu, |
|
|
"Class IoU": cls_iu, |
|
|
} |
|
|
|
|
|
def reset(self): |
|
|
self.confusion_matrix = np.zeros((self.n_classes, self.n_classes)) |
|
|
|
|
|
class AverageMeter(object): |
|
|
"""Computes average values""" |
|
|
def __init__(self): |
|
|
self.book = dict() |
|
|
|
|
|
def reset_all(self): |
|
|
self.book.clear() |
|
|
|
|
|
def reset(self, id): |
|
|
item = self.book.get(id, None) |
|
|
if item is not None: |
|
|
item[0] = 0 |
|
|
item[1] = 0 |
|
|
|
|
|
def update(self, id, val): |
|
|
record = self.book.get(id, None) |
|
|
if record is None: |
|
|
self.book[id] = [val, 1] |
|
|
else: |
|
|
record[0]+=val |
|
|
record[1]+=1 |
|
|
|
|
|
def get_results(self, id): |
|
|
record = self.book.get(id, None) |
|
|
assert record is not None |
|
|
return record[0] / record[1] |
|
|
|
