Upload 3 files

misc/caculate_metric.py

@@ -0,0 +1,112 @@
+import torch
+import numpy as np
+
+def calcuate_confusion_matrix(num_class:int, gt:torch.tensor, pred:torch.tensor):
+    gt_vector = gt.flatten()
+    pred_vector = pred.flatten()
+    mask = (gt_vector >= 0) & (gt_vector < num_class)
+    cm = torch.bincount(num_class * gt_vector[mask].to(dtype=int) + pred_vector[mask], minlength=num_class ** 2).reshape(num_class, num_class)
+    return cm
+
+class segmengtion_metric(object):
+    def __init__(self, num_class:int, device:str):
+        self.num_class = num_class
+        self.device = device
+        self.confusion_matrix = torch.zeros((self.num_class, self.num_class)).to(self.device)
+
+
+    def clear(self):
+        self.confusion_matrix = torch.zeros((self.num_class, self.num_class)).to(self.device)
+
+
+    def update_confusion_matrix(self, gt, pred):
+        cm = calcuate_confusion_matrix(self.num_class, gt, pred)
+        self.confusion_matrix += cm
+
+    def get_matrix_per_batch(self, gt, pred):
+        confusion_matrix = calcuate_confusion_matrix(self.num_class, gt, pred)
+
+        tp = torch.diag(confusion_matrix)
+
+        sum_a1 = torch.sum(confusion_matrix, dim=1)
+
+        sum_a0 = torch.sum(confusion_matrix, dim=0)
+
+        acc = tp.sum() / (confusion_matrix.sum() + torch.finfo(type=torch.float32).eps)
+        recall = tp / (sum_a1 + torch.finfo(type=torch.float32).eps)
+        precision = tp / (sum_a0 + torch.finfo(type=torch.float32).eps)
+        f1 = (2 * recall * precision) / (recall + precision + torch.finfo(type=torch.float32).eps)
+        iou = tp / (sum_a1 + sum_a0 - tp + torch.finfo(type=torch.float32).eps)
+
+        cls_precision = dict(zip(['pre_class[{}]'.format(i) for i in range(self.num_class)], precision))
+        cls_recall = dict(zip(['rec_class[{}]'.format(i) for i in range(self.num_class)], recall))
+        cls_f1 = dict(zip(['f1_class[{}]'.format(i) for i in range(self.num_class)], f1))
+        cls_iou = dict(zip(['iou_class[{}]'.format(i) for i in range(self.num_class)], iou))
+
+        mean_precision = precision[precision != 0].mean()
+        mean_recall = recall[recall != 0].mean()
+        mean_iou = iou[iou != 0].mean()
+        mean_f1 = f1[f1 != 0].mean()
+
+        score_dict_batch = {'acc': acc, 'mean_pre': mean_precision, 'mean_rec': mean_recall, 'mIoU': mean_iou, 'mF1': mean_f1}
+        score_dict_batch.update(cls_precision)
+        score_dict_batch.update(cls_recall)
+        score_dict_batch.update(cls_iou)
+        score_dict_batch.update(cls_f1)
+
+        return score_dict_batch
+
+    def get_metric_dict_per_epoch(self):
+
+        tp = torch.diag(self.confusion_matrix)
+
+        sum_a1 = torch.sum(self.confusion_matrix, dim=1)
+
+        sum_a0 = torch.sum(self.confusion_matrix, dim=0)
+
+        acc = tp.sum() / (self.confusion_matrix.sum() + torch.finfo(type=torch.float32).eps)
+
+        recall = tp / (sum_a1 + torch.finfo(type=torch.float32).eps)
+
+        precision = tp / (sum_a0 + torch.finfo(type=torch.float32).eps)
+
+        f1 = (2 * recall * precision) / (recall + precision + torch.finfo(type=torch.float32).eps)
+
+        iou = tp / (sum_a1 + sum_a0 - tp + torch.finfo(type=torch.float32).eps)
+
+        cls_precision = dict(zip(['Precision_Class[{}]'.format(i) for i in range(self.num_class)], precision))
+        cls_recall = dict(zip(['Recall_Class[{}]'.format(i) for i in range(self.num_class)], recall))
+        cls_iou = dict(zip(['IoU_Class[{}]'.format(i) for i in range(self.num_class)], iou))
+        cls_f1 = dict(zip(['F1_Class[{}]'.format(i) for i in range(self.num_class)], f1))
+
+        mean_precision = precision.mean()
+        mean_recall = recall.mean()
+        mean_iou = iou.mean()
+        mean_f1 = f1.mean()
+        score_dict_epoch = {'Accuracy': acc, 'mean_Precision': mean_precision, 'mean_Recall': mean_recall,
+                            'mIoU': mean_iou, 'mF1': mean_f1}
+
+        score_dict_epoch.update(cls_precision)
+        score_dict_epoch.update(cls_recall)
+        score_dict_epoch.update(cls_iou)
+        score_dict_epoch.update(cls_f1)
+        return score_dict_epoch
+
+
+
+
+
+
+if __name__=="__main__":
+    gt_label = torch.tensor([[0, 1, 2, 3, 1],
+                         [1, 2, 2, 3, 4]])
+
+    pre_label = torch.tensor([[0, 1, 2, 3, 1],
+                          [5, 1, 2, 1, 4]])
+
+    num_class = 6
+    metric = segmengtion_metric(6, 'cuda:0')
+    res = metric.get_matrix_per_batch(gt_label, pre_label)
+    res1 = metric.get_metric_dict_per_epoch()
+    print(res)
+    print(res1)

misc/metric_tools.py

@@ -0,0 +1,230 @@
+import numpy as np
+from scipy import stats
+import math
+
+class AverageMeter(object):
+
+    def __init__(self):
+        self.initialized = False
+        self.val = None
+        self.avg = None
+        self.sum = None
+        self.count = None
+
+    def initialize(self, val, weight):
+        self.val = val
+        self.avg = val
+        self.sum = val * weight
+        self.count = weight
+        self.initialized = True
+
+    def update(self, val, weight=1):
+        if not self.initialized:
+            self.initialize(val, weight)
+        else:
+            self.add(val, weight)
+
+    def add(self, val, weight):
+        self.val = val
+        self.sum += val * weight
+        self.count += weight
+        self.avg = self.sum / self.count
+
+    def value(self):
+        return self.val
+
+    def average(self):
+        return self.avg
+
+    def get_scores(self):
+        scores_dict = cm2score(self.sum)
+        return scores_dict
+
+    def clear(self):
+        self.initialized = False
+
+
+class ConfuseMatrixMeter(AverageMeter):
+
+    def __init__(self, n_class):
+        super(ConfuseMatrixMeter, self).__init__()
+        self.n_class = n_class
+
+    def update_cm(self, pr, gt, weight=1):
+
+        val = get_confuse_matrix(num_classes=self.n_class, label_gts=gt, label_preds=pr)
+        self.update(val, weight)
+        current_score = cm2F1(val)
+        return current_score
+
+    def get_scores(self):
+        scores_dict = cm2score(self.sum)
+        return scores_dict
+
+
+
+def harmonic_mean(xs):
+    harmonic_mean = len(xs) / sum((x+1e-6)**-1 for x in xs)
+    return harmonic_mean
+
+
+def cm2F1(confusion_matrix):
+    hist = confusion_matrix
+    n_class = hist.shape[0]
+    tp = np.diag(hist)
+    sum_a1 = hist.sum(axis=1)
+    sum_a0 = hist.sum(axis=0)
+
+    acc = tp.sum() / (hist.sum() + np.finfo(np.float32).eps)
+
+
+    recall = tp / (sum_a1 + np.finfo(np.float32).eps)
+
+    precision = tp / (sum_a0 + np.finfo(np.float32).eps)
+
+    F1 = 2 * recall * precision / (recall + precision + np.finfo(np.float32).eps)
+    mean_F1 = np.nanmean(F1)
+    return mean_F1
+
+
+def cm2score(confusion_matrix):
+    hist = confusion_matrix
+    n_class = hist.shape[0]
+
+    if n_class > 2:
+
+        hist_fg = hist[1:, 1:]
+        c2hist = np.zeros((2, 2))
+        c2hist[0][0] = hist[0][0]
+        c2hist[0][1] = hist.sum(1)[0] - hist[0][0]
+        c2hist[1][0] = hist.sum(0)[0] - hist[0][0]
+        c2hist[1][1] = hist_fg.sum()
+        hist_n0 = hist.copy()
+        hist_n0[0][0] = 0
+        kappa_n0 = cal_kappa(hist_n0)
+        iu_scd = np.nan_to_num(np.diag(c2hist) / (c2hist.sum(1) + c2hist.sum(0) - np.diag(c2hist)))
+        IoU_fg = iu_scd[1]
+        IoU_mean = (iu_scd[0] + iu_scd[1]) / 2
+        Sek = (kappa_n0 * math.exp(IoU_fg)) / math.e
+        pixel_sum = hist.sum()
+        change_pred_sum = pixel_sum - hist.sum(1)[0].sum()
+        change_label_sum = pixel_sum - hist.sum(0)[0].sum()
+        change_ratio = change_label_sum / pixel_sum
+        SC_TP = np.diag(hist[1:, 1:]).sum()
+        SC_Precision = np.nan_to_num(SC_TP / change_pred_sum) + np.finfo(np.float32).eps
+        SC_Recall = np.nan_to_num(SC_TP / change_label_sum) + np.finfo(np.float32).eps
+        Fscd = stats.hmean([SC_Precision, SC_Recall])
+
+
+    tp = np.diag(hist)
+    sum_a1 = hist.sum(axis=1)
+    sum_a0 = hist.sum(axis=0)
+
+    acc = tp.sum() / (hist.sum() + np.finfo(np.float32).eps)
+
+
+    recall = tp / (sum_a1 + np.finfo(np.float32).eps)
+
+    precision = tp / (sum_a0 + np.finfo(np.float32).eps)
+
+    F1 = 2*recall * precision / (recall + precision + np.finfo(np.float32).eps)
+
+    mean_F1 = np.nanmean(F1)
+
+    iu = tp / (sum_a1 + hist.sum(axis=0) - tp + np.finfo(np.float32).eps)
+    mean_iu = np.nanmean(iu)
+
+    freq = sum_a1 / (hist.sum() + np.finfo(np.float32).eps)
+    fwavacc = (freq[freq > 0] * iu[freq > 0]).sum()
+
+    cls_iou = dict(zip(['iou_'+str(i) for i in range(n_class)], iu))
+
+    cls_precision = dict(zip(['precision_'+str(i) for i in range(n_class)], precision))
+    cls_recall = dict(zip(['recall_'+str(i) for i in range(n_class)], recall))
+    cls_F1 = dict(zip(['F1_'+str(i) for i in range(n_class)], F1))
+
+    if n_class > 2:
+        score_dict = {'acc': acc, 'miou': mean_iu, 'mf1':mean_F1, 'SCD_Sek':Sek, 'Fscd':Fscd, 'SCD_IoU_mean':IoU_mean}
+    else:
+        score_dict = {'acc': acc, 'miou': mean_iu, 'mf1':mean_F1}
+    score_dict.update(cls_iou)
+    score_dict.update(cls_F1)
+    score_dict.update(cls_precision)
+    score_dict.update(cls_recall)
+    return score_dict
+
+
+def get_confuse_matrix(num_classes, label_gts, label_preds):
+
+    def __fast_hist(label_gt, label_pred):
+
+        mask = (label_gt >= 0) & (label_gt < num_classes)
+        hist = np.bincount(num_classes * label_gt[mask].astype(int) + label_pred[mask],
+                           minlength=num_classes**2).reshape(num_classes, num_classes)
+        return hist
+    confusion_matrix = np.zeros((num_classes, num_classes))
+    for lt, lp in zip(label_gts, label_preds):
+        confusion_matrix += __fast_hist(lt.flatten(), lp.flatten())
+    return confusion_matrix
+
+
+def get_mIoU(num_classes, label_gts, label_preds):
+    confusion_matrix = get_confuse_matrix(num_classes, label_gts, label_preds)
+    score_dict = cm2score(confusion_matrix)
+    return score_dict['miou']
+
+def fast_hist(a, b, n):
+    k = (a >= 0) & (a < n)
+    return np.bincount(n * a[k].astype(int) + b[k], minlength=n ** 2).reshape(n, n)
+
+def get_hist(image, label, num_class):
+    hist = np.zeros((num_class, num_class))
+    hist += fast_hist(image.flatten(), label.flatten(), num_class)
+    return hist
+
+def cal_kappa(hist):
+    if hist.sum() == 0:
+        po = 0
+        pe = 1
+        kappa = 0
+    else:
+        po = np.diag(hist).sum() / hist.sum()
+        pe = np.matmul(hist.sum(1), hist.sum(0).T) / hist.sum() ** 2
+        if pe == 1:
+            kappa = 0
+        else:
+            kappa = (po - pe) / (1 - pe)
+    return kappa
+
+def SCDD_eval_all(preds, labels, num_class):
+    hist = np.zeros((num_class, num_class))
+    for pred, label in zip(preds, labels):
+        infer_array = np.array(pred)
+        unique_set = set(np.unique(infer_array))
+        assert unique_set.issubset(set([0, 1, 2, 3, 4, 5, 6])), "unrecognized label number"
+        label_array = np.array(label)
+        assert infer_array.shape == label_array.shape, "The size of prediction and target must be the same"
+        hist += get_hist(infer_array, label_array, num_class)
+
+    hist_fg = hist[1:, 1:]
+    c2hist = np.zeros((2, 2))
+    c2hist[0][0] = hist[0][0]
+    c2hist[0][1] = hist.sum(1)[0] - hist[0][0]
+    c2hist[1][0] = hist.sum(0)[0] - hist[0][0]
+    c2hist[1][1] = hist_fg.sum()
+    hist_n0 = hist.copy()
+    hist_n0[0][0] = 0
+    kappa_n0 = cal_kappa(hist_n0)
+    iu = np.diag(c2hist) / (c2hist.sum(1) + c2hist.sum(0) - np.diag(c2hist))
+    IoU_fg = iu[1]
+    IoU_mean = (iu[0] + iu[1]) / 2
+    Sek = (kappa_n0 * math.exp(IoU_fg)) / math.e
+
+    pixel_sum = hist.sum()
+    change_pred_sum = pixel_sum - hist.sum(1)[0].sum()
+    change_label_sum = pixel_sum - hist.sum(0)[0].sum()
+    SC_TP = np.diag(hist[1:, 1:]).sum()
+    SC_Precision = SC_TP / change_pred_sum
+    SC_Recall = SC_TP / change_label_sum
+    Fscd = stats.hmean([SC_Precision, SC_Recall])
+    return Fscd, IoU_mean, Sek

misc/torchutils.py

@@ -0,0 +1,83 @@
+import torch
+import torch.nn as nn
+from torch import Tensor
+from torch.optim import lr_scheduler
+from typing import Iterable, Set, Tuple
+import logging
+import os
+
+logger = logging.getLogger('base')
+
+def simplex(t: Tensor, axis=1) -> bool:
+    _sum = t.sum(axis).float()
+    _ones = torch.ones_like(_sum, dtype=torch.float32)
+    return torch.allclose(_sum, _ones)
+
+def one_hot(t: Tensor, axis=1) -> bool:
+    return simplex(t, axis) and sset(t, [0, 1])
+
+def uniq(a: Tensor) -> Set:
+    return set(torch.unique(a.cpu()).numpy())
+
+def sset(a: Tensor, sub: Iterable) -> bool:
+    return uniq(a).issubset(sub)
+
+def class2one_hot(seg: Tensor, C: int) -> Tensor:
+    if len(seg.shape) == 2:  # (H, W) 的情况
+        seg = seg.unsqueeze(dim=0)
+    assert sset(seg, list(range(C))), "输入 Tensor 中的类别索引超出范围！"
+
+    if seg.ndim == 4:
+        seg = seg.squeeze(dim=1)
+
+    b, w, h = seg.shape  # 获取 batch 维度、宽度、高度
+    res = torch.stack([seg == c for c in range(C)], dim=1).int()
+    assert res.shape == (b, C, w, h)
+    assert one_hot(res), "转换后的 Tensor 不是 one-hot 编码！"
+
+    return res
+
+def get_scheduler(optimizer, args):
+    """返回学习率调度器"""
+    if args['scheduler']['lr_policy'] == 'linear':
+        def lambda_rule(epoch):
+            return 1.0 - epoch / float(args['n_epoch'] + 1)
+        return lr_scheduler.LambdaLR(optimizer, lr_lambda=lambda_rule)
+
+    elif args['scheduler']['lr_policy'] == 'step':
+        step_size = args['n_epoch'] // args['scheduler']['n_steps']
+        return lr_scheduler.StepLR(optimizer, step_size=step_size, gamma=args['scheduler']['gamma'])
+
+    else:
+        raise NotImplementedError(f"学习率策略 [{args['scheduler']['lr_policy']}] 未实现！")
+
+def save_network(opt, epoch, cd_model, optimizer, is_best_model=False):
+    """ 保存当前 epoch 的模型和优化器参数 """
+
+    os.makedirs(opt['path_cd']['checkpoint'], exist_ok=True)
+
+    cd_gen_path = os.path.join(opt['path_cd']['checkpoint'], f'cd_model_E{epoch}_gen.pth')
+    cd_opt_path = os.path.join(opt['path_cd']['checkpoint'], f'cd_model_E{epoch}_opt.pth')
+
+    best_cd_gen_path = os.path.join(opt['path_cd']['checkpoint'], 'best_cd_model_gen.pth')
+    best_cd_opt_path = os.path.join(opt['path_cd']['checkpoint'], 'best_cd_model_opt.pth')
+
+    network = cd_model.module if isinstance(cd_model, nn.DataParallel) else cd_model
+    state_dict = {key: param.cpu() for key, param in network.state_dict().items()}
+
+    torch.save(state_dict, cd_gen_path)
+    if is_best_model:
+        torch.save(state_dict, best_cd_gen_path)
+
+    opt_state = {
+        'epoch': epoch,
+        'scheduler': None,
+        'optimizer': optimizer.state_dict()
+    }
+    torch.save(opt_state, cd_opt_path)
+    if is_best_model:
+        torch.save(opt_state, best_cd_opt_path)
+
+    logger.info(f'✅ 当前模型已保存至 [{cd_gen_path}]')
+    if is_best_model:
+        logger.info(f'🏆 最佳模型已更新至 [{best_cd_gen_path}]')