metrics.py

import time
import multiprocessing
import numpy as np
from copy import deepcopy
from tqdm.auto import tqdm
from scipy.ndimage import gaussian_filter
from collections import defaultdict

from skimage import measure
from sklearn.metrics import auc, roc_auc_score, average_precision_score, precision_recall_curve, \
                            roc_curve, accuracy_score, f1_score, recall_score, precision_score

import torch
from torch.nn import functional as F

from adeval import EvalAccumulatorCuda


def safe_roc_auc(y_true, y_score, context=''):
    """Compute ROC AUC but guard against single-class y_true.

    Returns np.nan and prints a warning if y_true contains only one class.
    """
    try:
        if np.unique(y_true).size <= 1:
            print(f"Warning: ROC-AUC undefined for single-class ground truth. Context: {context}")
            return float('nan')
        return roc_auc_score(y_true, y_score)
    except Exception as e:
        print(f"Warning: roc_auc computation failed ({e}). Context: {context}")
        return float('nan')


def safe_average_precision(y_true, y_score, context=''):
    try:
        if np.unique(y_true).size <= 1:
            print(f"Warning: average_precision undefined for single-class ground truth. Context: {context}")
            return float('nan')
        return average_precision_score(y_true, y_score)
    except Exception as e:
        print(f"Warning: average_precision computation failed ({e}). Context: {context}")
        return float('nan')

def get_timepc():
    if torch.cuda.is_available():
        torch.cuda.synchronize()
    return time.perf_counter()


def log_msg(logger, msg, level='info'):
    if logger is not None:
        if msg is not None and level == 'info':
            logger.info(msg)
            
            
def topk_percent_average(tensor, k: float = 0.01):
    """
    Calculate the average of the top 1% values in each H x W matrix of an N x H x W tensor.

    Parameters:
    tensor (numpy.ndarray): A 3D tensor of shape (N, H, W).

    Returns:
    numpy.ndarray: A 1D array of length N containing the top 1% averages for each H x W.
    """
    N, H, W = tensor.shape
    top_1_percent_means = np.zeros(N)

    flattened_size = H * W
    threshold = max(1, int(flattened_size * k))  # Ensure at least one element is included

    for i in range(N):
        flattened = tensor[i].ravel()  # Flatten to a 1D array
        top_indices = np.argpartition(flattened, -threshold)[-threshold:]  # Indices of top 1% elements
        top_values = flattened[top_indices]  # Get top 1% values
        top_1_percent_means[i] = top_values.mean()  # Calculate the mean of the top 1%

    return top_1_percent_means


class Evaluator(object):
    def __init__(self, metrics=[], pooling_ks=None, max_step_aupro=100, mp=False, use_adeval=True):
        if len(metrics) == 0:
            self.metrics = [
                'mAUROC_sp_max', 'mAUROC_sp_top1', 'mAUROC_px', 'mAUPRO_px',
                'mAP_sp_max', 'mAP_sp_top1', 'mAP_px',
                'mF1_max_sp_max', 'mF1_max_sp_top1',
                'mF1_px_0.2_0.8_0.1', 'mAcc_px_0.2_0.8_0.1', 'mIoU_px_0.2_0.8_0.1',
                'mF1_max_px', 'mIoU_max_px',
            ]
        else:
            self.metrics = metrics

        self.pooling_ks = pooling_ks
        self.max_step_aupro = max_step_aupro
        self.mp = mp

        self.eps = 1e-8
        self.beta = 1.0

        self.boundary = 1e-7
        self.use_adeval = use_adeval

    def run(self, results, cls_name, cls_index: list = [], logger=None):        
        if cls_name != 'all':
            idxes = results['cls_names'] == cls_name  
            gt_px = results['imgs_masks'][idxes].squeeze(1)
            pr_px = results['anomaly_maps'][idxes]
        else: 
            if len(cls_index) > 0:
                gt_px = results['imgs_masks'][cls_index].squeeze(1)
                pr_px = results['anomaly_maps'][cls_index]
            else:
                gt_px = results['imgs_masks'].squeeze(1)
                pr_px = results['anomaly_maps']
        
        if 'smp_pre' in results:
            pr_sample = results['smp_pre'][idxes]
            subarrays = np.split(pr_sample, len(pr_sample)//5)
            pr_sa_max = np.array([np.max(subarray) for subarray in subarrays])
            subarray2 = np.split(idxes, len(idxes) // 5)
            sample_idxes = np.array([subarr[0] for subarr in subarray2])
            gt_sa = results['smp_masks'][sample_idxes]
        if len(gt_px.shape) == 4:
            gt_px = gt_px.squeeze(1)
        if len(pr_px.shape) == 4:
            pr_px = pr_px.squeeze(1)
            
        # normalization for pixel-level evaluations
        pr_px_norm = (pr_px - pr_px.min()) / (pr_px.max() - pr_px.min())
        gt_sp = gt_px.max(axis=(1, 2))
        if self.pooling_ks is not None:
            pr_px_pooling = F.avg_pool2d(torch.tensor(pr_px).unsqueeze(1), self.pooling_ks, stride=1).numpy().squeeze(1)
            pr_sp_max = pr_px_pooling.max(axis=(1, 2))
            pr_sp_mean = pr_px_pooling.mean(axis=(1, 2))
        else:
            pr_sp_max = pr_px.max(axis=(1, 2))
            pr_sp_top1 = topk_percent_average(pr_px)
            pr_sp_mean = pr_px.mean(axis=(1, 2))

        if self.use_adeval:
            score_min = min(pr_sp_max) - self.boundary
            score_max = max(pr_sp_max) + self.boundary
            anomap_min = pr_px.min()
            anomap_max = pr_px.max()
            accum = EvalAccumulatorCuda(score_min, score_max, anomap_min, anomap_max, skip_pixel_aupro=False, nstrips=50)
            accum.add_anomap_batch(torch.tensor(pr_px).cuda(non_blocking=True),
                                   torch.tensor(gt_px.astype(np.uint8)).cuda(non_blocking=True))
            for i in range(torch.tensor(pr_px).size(0)):
                accum.add_image(torch.tensor(pr_sp_max[i]), torch.tensor(gt_sp[i]))
            metrics = accum.summary()

        metric_str = f'==> Metric Time for {cls_name:<15}: '
        metric_results = {}
        for metric in self.metrics:
            t0 = get_timepc()
            if metric.startswith('mAUROC_sp_max'):
                auroc_sp = safe_roc_auc(gt_sp, pr_sp_max, context=f"{cls_name} - mAUROC_sp_max")
                metric_results[metric] = auroc_sp
            elif metric.startswith('mAUROC_sp_top1'):
                auroc_sp = safe_roc_auc(gt_sp, pr_sp_top1, context=f"{cls_name} - mAUROC_sp_top1")
                metric_results[metric] = auroc_sp
            elif metric.startswith('mAUROC_sp_mean'):
                auroc_sp = safe_roc_auc(gt_sp, pr_sp_mean, context=f"{cls_name} - mAUROC_sp_mean")
                metric_results[metric] = auroc_sp
            elif metric.startswith('mAUROC_sa_max'):
                auroc_sp = safe_roc_auc(gt_sa, pr_sa_max, context=f"{cls_name} - mAUROC_sa_max")
                metric_results[metric] = auroc_sp
            elif metric.startswith('mAUROC_px'):
                if not self.use_adeval:
                    auroc_px = safe_roc_auc(gt_px.ravel(), pr_px.ravel(), context=f"{cls_name} - mAUROC_px")
                    metric_results[metric] = auroc_px
                else:
                    metric_results[metric] = metrics['p_auroc']
            elif metric.startswith('mAUPRO_px'):
                if not self.use_adeval:
                    aupro_px = self.cal_pro_score(gt_px, pr_px, max_step=self.max_step_aupro, mp=self.mp)
                    metric_results[metric] = aupro_px
                else:
                    metric_results[metric] = metrics['p_aupro']
            elif metric.startswith('mAP_sp_max'):
                ap_sp = safe_average_precision(gt_sp, pr_sp_max, context=f"{cls_name} - mAP_sp_max")
                metric_results[metric] = ap_sp
            elif metric.startswith('mAP_sp_top1'):
                ap_sp = safe_average_precision(gt_sp, pr_sp_top1, context=f"{cls_name} - mAP_sp_top1")
                metric_results[metric] = ap_sp
            elif metric.startswith('AP_sp_mean'):
                ap_sp = safe_average_precision(gt_sp, pr_sp_mean, context=f"{cls_name} - AP_sp_mean")
                metric_results[metric] = ap_sp
            elif metric.startswith('mAP_px'):
                if not self.use_adeval:
                    ap_px = safe_average_precision(gt_px.ravel(), pr_px.ravel(), context=f"{cls_name} - mAP_px")
                    metric_results[metric] = ap_px
                else:
                    metric_results[metric] = metrics['p_aupr']
            elif metric.startswith('mAP_sa_max'):
                ap_sp = safe_average_precision(gt_sa, pr_sa_max, context=f"{cls_name} - mAP_sa_max")
                metric_results[metric] = ap_sp
            elif metric.startswith('mF1_max_sp_max'):
                precisions, recalls, thresholds = precision_recall_curve(gt_sp, pr_sp_max)
                f1_scores = (2 * precisions * recalls) / (precisions + recalls)
                best_f1_score = np.max(f1_scores[np.isfinite(f1_scores)])
                best_f1_score_index = np.argmax(f1_scores[np.isfinite(f1_scores)])
                best_threshold = thresholds[best_f1_score_index]
                metric_results[metric] = best_f1_score
            elif metric.startswith('mF1_max_sp_top1'):
                precisions, recalls, thresholds = precision_recall_curve(gt_sp, pr_sp_top1)
                f1_scores = (2 * precisions * recalls) / (precisions + recalls)
                best_f1_score = np.max(f1_scores[np.isfinite(f1_scores)])
                best_f1_score_index = np.argmax(f1_scores[np.isfinite(f1_scores)])
                best_threshold = thresholds[best_f1_score_index]
                metric_results[metric] = best_f1_score
            elif metric.startswith('mF1_max_sa_max'):
                precisions, recalls, thresholds = precision_recall_curve(gt_sa, pr_sa_max)
                f1_scores = (2 * precisions * recalls) / (precisions + recalls)
                best_f1_score = np.max(f1_scores[np.isfinite(f1_scores)])
                best_f1_score_index = np.argmax(f1_scores[np.isfinite(f1_scores)])
                best_threshold = thresholds[best_f1_score_index]
                metric_results[metric] = best_f1_score
            elif metric.startswith('F1_max_sp_mean'):
                precisions, recalls, thresholds = precision_recall_curve(gt_sp, pr_sp_mean)
                f1_scores = (2 * precisions * recalls) / (precisions + recalls)
                best_f1_score = np.max(f1_scores[np.isfinite(f1_scores)])
                best_f1_score_index = np.argmax(f1_scores[np.isfinite(f1_scores)])
                best_threshold = thresholds[best_f1_score_index]
                metric_results[metric] = best_f1_score
            elif metric.startswith('mF1_px') or metric.startswith('mDice_px') or metric.startswith('mAcc_px') or metric.startswith('mIoU_px'):  # example: F1_sp_0.3_0.8
                # F1_px equals Dice_px
                coms = metric.split('_')
                assert len(coms) == 5, f"{metric} should contain parameters 'score_l', 'score_h', and 'score_step' "
                score_l, score_h, score_step = float(metric.split('_')[-3]), float(metric.split('_')[-2]), float(metric.split('_')[-1])
                gt = gt_px.astype(np.bool_)
                metric_scores = []
                for score in np.arange(score_l, score_h + 1e-3, score_step):
                    pr = pr_px_norm > score
                    total_area_intersect = np.logical_and(gt, pr).sum(axis=(0, 1, 2))
                    total_area_union = np.logical_or(gt, pr).sum(axis=(0, 1, 2))
                    total_area_pred_label = pr.sum(axis=(0, 1, 2))
                    total_area_label = gt.sum(axis=(0, 1, 2))
                    if metric.startswith('mF1_px'):
                        precision = total_area_intersect / (total_area_pred_label + self.eps)
                        recall = total_area_intersect / (total_area_label + self.eps)
                        f1_px = (1 + self.beta ** 2) * precision * recall / (self.beta ** 2 * precision + recall + self.eps)
                        metric_scores.append(f1_px)
                    elif metric.startswith('mDice_px'):
                        dice_px = 2 * total_area_intersect / (total_area_pred_label + total_area_label + self.eps)
                        metric_scores.append(dice_px)
                    elif metric.startswith('mAcc_px'):
                        acc_px = total_area_intersect / (total_area_label + self.eps)
                        metric_scores.append(acc_px)
                    elif metric.startswith('mIoU_px'):
                        iou_px = total_area_intersect / (total_area_union + self.eps)
                        metric_scores.append(iou_px)
                    else:
                        raise f'invalid metric: {metric}'
                metric_results[metric] = np.array(metric_scores).mean()
            elif metric.startswith('mF1_max_px') or metric.startswith('mDice_max_px') or metric.startswith('mAcc_max_px') or metric.startswith('mIoU_max_px'):
                # F1_px equals Dice_px
                score_l, score_h, score_step = 0.0, 1.0, 0.05
                gt = gt_px.astype(np.bool_)
                metric_scores = []
                for score in np.arange(score_l, score_h + 1e-3, score_step):
                    pr = pr_px_norm > score
                    total_area_intersect = np.logical_and(gt, pr).sum(axis=(0, 1, 2))
                    total_area_union = np.logical_or(gt, pr).sum(axis=(0, 1, 2))
                    total_area_pred_label = pr.sum(axis=(0, 1, 2))
                    total_area_label = gt.sum(axis=(0, 1, 2))
                    if metric.startswith('mF1_max_px'):
                        precision = total_area_intersect / (total_area_pred_label + self.eps)
                        recall = total_area_intersect / (total_area_label + self.eps)
                        f1_px = (1 + self.beta ** 2) * precision * recall / (
                                    self.beta ** 2 * precision + recall + self.eps)
                        metric_scores.append(f1_px)
                    elif metric.startswith('mDice_max_px'):
                        dice_px = 2 * total_area_intersect / (total_area_pred_label + total_area_label + self.eps)
                        metric_scores.append(dice_px)
                    elif metric.startswith('mAcc_max_px'):
                        acc_px = total_area_intersect / (total_area_label + self.eps)
                        metric_scores.append(acc_px)
                    elif metric.startswith('mIoU_max_px'):
                        iou_px = total_area_intersect / (total_area_union + self.eps)
                        metric_scores.append(iou_px)
                    else:
                        raise f'invalid metric: {metric}'
                metric_results[metric] = np.array(metric_scores).max()
            t1 = get_timepc()
            metric_str += f'{t1 - t0:7.3f} ({metric})\t'
        log_msg(logger, metric_str)
        return metric_results

    @staticmethod
    def cal_anomaly_map(ft_list, fs_list, out_size=[224, 224], uni_am=False, use_cos=True, amap_mode='add', gaussian_sigma=0, weights=None):
        # ft_list = [f.cpu() for f in ft_list]
        # fs_list = [f.cpu() for f in fs_list]
        bs = ft_list[0].shape[0]
        weights = weights if weights else [1] * len(ft_list)
        anomaly_map = np.ones([bs] + out_size) if amap_mode == 'mul' else np.zeros([bs] + out_size)
        a_map_list = []
        if uni_am:
            size = (ft_list[0].shape[2], ft_list[0].shape[3])
            for i in range(len(ft_list)):
                ft_list[i] = F.interpolate(F.normalize(ft_list[i], p=2), size=size, mode='bilinear', align_corners=True)
                fs_list[i] = F.interpolate(F.normalize(fs_list[i], p=2), size=size, mode='bilinear', align_corners=True)
            ft_map, fs_map = torch.cat(ft_list, dim=1), torch.cat(fs_list, dim=1)
            if use_cos:
                a_map = 1 - F.cosine_similarity(ft_map, fs_map, dim=1)
                a_map = a_map.unsqueeze(dim=1)
            else:
                a_map = torch.sqrt(torch.sum((ft_map - fs_map) ** 2, dim=1, keepdim=True))
            a_map = F.interpolate(a_map, size=out_size, mode='bilinear', align_corners=True)
            a_map = a_map.squeeze(dim=1).cpu().detach().numpy()
            anomaly_map = a_map
            a_map_list.append(a_map)
        else:
            for i in range(len(ft_list)):
                ft = ft_list[i]
                fs = fs_list[i]
                if use_cos:
                    a_map = 1 - F.cosine_similarity(ft, fs, dim=1)
                    a_map = a_map.unsqueeze(dim=1)
                else:
                    a_map = torch.sqrt(torch.sum((ft - fs) ** 2, dim=1, keepdim=True))
                a_map = F.interpolate(a_map, size=out_size, mode='bilinear', align_corners=True)
                a_map = a_map.squeeze(dim=1)
                a_map = a_map.cpu().detach().numpy()
                a_map_list.append(a_map)
                if amap_mode == 'add':
                    anomaly_map += a_map * weights[i]
                else:
                    anomaly_map *= a_map
            if amap_mode == 'add':
                anomaly_map /= (len(ft_list) * sum(weights))
        if gaussian_sigma > 0:
            for idx in range(anomaly_map.shape[0]):
                anomaly_map[idx] = gaussian_filter(anomaly_map[idx], sigma=gaussian_sigma)
        return anomaly_map, a_map_list

    @staticmethod
    def cal_pro_thr(results, th, amaps, masks):
        binary_amaps = np.zeros_like(amaps, dtype=np.bool_)
        binary_amaps[amaps <= th], binary_amaps[amaps > th] = 0, 1
        pro = []
        for binary_amap, mask in zip(binary_amaps, masks):
            for region in measure.regionprops(measure.label(mask)):
                tp_pixels = binary_amap[region.coords[:, 0], region.coords[:, 1]].sum()
                pro.append(tp_pixels / region.area)
        inverse_masks = 1 - masks
        fp_pixels = np.logical_and(inverse_masks, binary_amaps).sum()
        fpr = fp_pixels / inverse_masks.sum()
        results.append([np.array(pro).mean(), fpr, th])
        # return [np.array(pro).mean(), fpr, th]

    @staticmethod
    def cal_pro_score(masks, amaps, max_step=200, expect_fpr=0.3, mp=False):
        # ref: https://github.com/gudovskiy/cflow-ad/blob/master/train.py
        min_th, max_th = amaps.min(), amaps.max()
        delta = (max_th - min_th) / max_step
        pros, fprs, ths = [], [], []
        if mp:

            # multiprocessing.Process has no `get` function
            results = multiprocessing.Manager().list()
            jobs = []
            for th in np.arange(min_th, max_th, delta):
                job = multiprocessing.Process(target=Evaluator.cal_pro_thr, args=(results, th, amaps, masks,))
                job.start()
                jobs.append(job)
            for job in jobs:
                job.join()
            results = list(results)
            results.sort(key=lambda x: float(x[2]))
            for result in results:
                pros.append(result[0])
                fprs.append(result[1])
                ths.append(result[2])
        else:
            binary_amaps = np.zeros_like(amaps, dtype=np.bool_)
            for th in np.arange(min_th, max_th, delta):
                binary_amaps[amaps <= th], binary_amaps[amaps > th] = 0, 1
                pro = []
                for binary_amap, mask in zip(binary_amaps, masks):
                    for region in measure.regionprops(measure.label(mask)):
                        tp_pixels = binary_amap[region.coords[:, 0], region.coords[:, 1]].sum()
                        pro.append(tp_pixels / region.area)
                inverse_masks = 1 - masks
                fp_pixels = np.logical_and(inverse_masks, binary_amaps).sum()
                fpr = fp_pixels / inverse_masks.sum()
                pros.append(np.array(pro).mean())
                fprs.append(fpr)
                ths.append(th)
        pros, fprs, ths = np.array(pros), np.array(fprs), np.array(ths)
        idxes = fprs < expect_fpr
        fprs = fprs[idxes]
        fprs = (fprs - fprs.min()) / (fprs.max() - fprs.min())
        pro_auc = auc(fprs, pros[idxes])
        return pro_auc
    
    
def score_scaling(scores):
    if not isinstance(scores, np.ndarray):
        scores = np.array(scores)
        
    # score scaling
    min_scores = scores.min(axis=-1).reshape(-1, 1)
    max_scores = scores.max(axis=-1).reshape(-1, 1)
    scores = (scores - min_scores) / (max_scores - min_scores)
    scores = np.mean(scores, axis=0)
    
    return scores

def segmentation_scaling(segmentations):
    if not isinstance(segmentations, np.ndarray):
        segmentations = np.array(segmentations)
        
    # segmentation scaling
    min_scores = (
        segmentations.reshape(len(segmentations), -1)
        .min(axis=-1)
        .reshape(-1, 1, 1, 1)
    )
    max_scores = (
        segmentations.reshape(len(segmentations), -1)
        .max(axis=-1)
        .reshape(-1, 1, 1, 1)
    )
    
    segmentations = (segmentations - min_scores) / (max_scores - min_scores)
    segmentations = np.mean(segmentations, axis=0)
    
    return segmentations


def get_best_threshold_scores(y_true, y_score, **kwargs):
    precisions, recalls, thresholds = precision_recall_curve(y_true, y_score)
    f1_scores = 2 * (precisions * recalls) / (precisions + recalls + 1e-9)  # F1-score 계산
    best_index = np.argmax(f1_scores)  # F1-score가 최대인 인덱스
    best_threshold = thresholds[best_index]
    
    # fpr, tpr, thresholds = roc_curve(y_true, y_score)
    # best_idx = np.argmax(tpr-fpr)
    # return thresholds[best_idx]
    
    return best_threshold

def get_best_threshold_maps(y_true, y_score): 
    score_l, score_h, score_step = 0.0, 1.0, 0.01
    
    best_f1_px = 0
    best_threshold = 0.5  # default threshold

    # If there's no anomaly in GT, return default threshold
    if y_true.sum() == 0:
        return best_threshold

    for score in np.arange(score_l, score_h + 1e-3, score_step):
        pr = y_score > score
        total_area_intersect = np.logical_and(y_true, pr).sum()
        total_area_pred_label = pr.sum()
        total_area_label = y_true.sum()
        
        if total_area_pred_label == 0 or total_area_label == 0:
            continue
            
        precision = total_area_intersect / total_area_pred_label
        recall = total_area_intersect / total_area_label
        f1_px = (2 * precision * recall) / (precision + recall + 1e-9)
        
        if best_f1_px < f1_px:
            best_threshold = score 
            best_f1_px = f1_px
    
    return best_threshold
        

def get_fpr_ratio_threhsold(y_true, y_score, fpr_ratio: float = 0.1):
    fpr, tpr, thresholds = roc_curve(y_true, y_score)    
    fpr_ratio_idx = np.argmin(np.abs(fpr-fpr_ratio))
    return thresholds[fpr_ratio_idx]


def calc_metrics_scores(y_true, y_pred):
    acc = accuracy_score(y_true, y_pred)
    recall = recall_score(y_true, y_pred)
    precision = precision_score(y_true, y_pred)
    f1 = f1_score(y_true, y_pred)
    
    return {
        'Accuracy': acc,
        'Recall': recall,
        'Precision': precision,
        'F1-score': f1
    }
    
def calc_metrics_maps(y_true, y_pred):
    total_area_intersect = np.logical_and(y_true, y_pred).sum()
    total_area_union = np.logical_or(y_true, y_pred).sum()
    total_area_pred_label = y_pred.sum()
    total_area_label = y_true.sum()
    
    precision = total_area_intersect / total_area_pred_label
    recall = total_area_intersect / total_area_label
    f1_px = (2 * precision * recall) / (precision + recall)
    acc_px = total_area_intersect / total_area_label
    iou_px = total_area_intersect / total_area_union
    
    return {
        'Accuracy'  : acc_px,
        'Recall'    : recall,
        'Precision' : precision,
        'F1-score'  : f1_px,
        'IoU'       : iou_px
    }
    
def _binary_results(eval_args, **kwargs):
    results = {'all': defaultdict(dict), 'cls': defaultdict(dict)}
    
    # image-level
    all_threshold = get_best_threshold_scores(
        eval_args['labels'], 
        eval_args['anomaly_scores'], 
        **kwargs
    )

    for cls in np.unique(eval_args['cls_names']):
        cls_idx = np.where(eval_args['cls_names'] == cls)[0]
        
        # threshold for all
        if 'anomaly_scores_pred' in eval_args:
            all_preds = eval_args['anomaly_scores_pred'][cls_idx]
        else:
            all_preds = [1 if s > all_threshold else 0 for s in eval_args['anomaly_scores'][cls_idx]]
        results['all']['image'][cls] = calc_metrics_scores(
            y_true = eval_args['labels'][cls_idx], 
            y_pred = all_preds
        )
        
        # threshold for each class
        cls_threshold = get_best_threshold_scores(
            eval_args['labels'][cls_idx], 
            eval_args['anomaly_scores'][cls_idx], 
            **kwargs
        )
        cls_preds = [1 if s > cls_threshold else 0 for s in eval_args['anomaly_scores'][cls_idx]]
        results['cls']['image'][cls] = calc_metrics_scores(
            y_true = eval_args['labels'][cls_idx], 
            y_pred = cls_preds
        )
        
    # pixel-level
    anomaly_maps = eval_args['anomaly_maps']
    anomaly_masks = eval_args['imgs_masks']
    all_threshold = get_best_threshold_maps(
        anomaly_masks.ravel(), 
        anomaly_maps.ravel(), 
        **kwargs
    )

    for cls in np.unique(eval_args['cls_names']):
        cls_idx = np.where(eval_args['cls_names'] == cls)[0]
        
        # threshold for all
        if 'anomaly_maps_pred' in eval_args:
            all_preds = eval_args['anomaly_maps_pred'][cls_idx].ravel()
        else:
            all_preds = np.zeros_like(anomaly_maps[cls_idx].ravel())
            all_preds[anomaly_maps[cls_idx].ravel() > all_threshold] = 1
            
        results['all']['pixel'][cls] = calc_metrics_maps(
            y_true = anomaly_masks[cls_idx].ravel(), 
            y_pred = all_preds
        )
        
        # threshold for each class
        cls_threshold = get_best_threshold_maps(
            anomaly_masks[cls_idx].ravel(), 
            anomaly_maps[cls_idx].ravel(), 
            **kwargs
        )
        cls_preds = np.zeros_like(anomaly_maps[cls_idx].ravel())
        cls_preds[anomaly_maps[cls_idx].ravel() > cls_threshold] = 1
        
        results['cls']['pixel'][cls] = calc_metrics_maps(
            y_true = anomaly_masks[cls_idx].ravel(), 
            y_pred = cls_preds
        )
    
    return results
    

def _get_prediction_semantic_match(eval_args, semantic_match_indices):
    anomaly_scores_pred_sem = np.zeros_like(eval_args['anomaly_scores'])
    anomaly_maps_preds_sem = np.zeros_like(eval_args['anomaly_maps'])
    
    scaled_anomaly_maps = segmentation_scaling([eval_args['anomaly_maps']])
    
    for idx in np.unique(semantic_match_indices):
        match_idx = np.where(semantic_match_indices==idx)[0]
        
        anomaly_scores_pred_sem_i = np.zeros_like(eval_args['anomaly_scores'][match_idx])
        anomaly_maps_preds_sem_i = np.zeros_like(eval_args['anomaly_maps'][match_idx])
        
        best_score_threshold = get_best_threshold_scores(
            y_true  = eval_args['labels'][match_idx], 
            y_score = eval_args['anomaly_scores'][match_idx]
        )
        best_map_threshold = get_best_threshold_maps(
            y_true  = eval_args['imgs_masks'][match_idx].ravel(), 
            y_score = scaled_anomaly_maps[match_idx].ravel()
        )
        
        anomaly_scores_pred_sem_i[eval_args['anomaly_scores'][match_idx] > best_score_threshold] = 1
        anomaly_maps_preds_sem_i[scaled_anomaly_maps[match_idx] > best_map_threshold] = 1
        
        anomaly_scores_pred_sem[match_idx] = anomaly_scores_pred_sem_i
        anomaly_maps_preds_sem[match_idx] = anomaly_maps_preds_sem_i
    
    eval_args['anomaly_scores_pred'] = anomaly_scores_pred_sem
    eval_args['anomaly_maps_pred'] = anomaly_maps_preds_sem
    
    return eval_args

    
def get_binary_results(eval_args, semantic_match_indices: list = []):
    eval_values = deepcopy(eval_args)
    
    eval_values['anomaly_scores'] = eval_values['anomaly_maps'].max(axis=(1,2))
    eval_values['labels'] = eval_values['imgs_masks'].max(axis=(1,2,3))

    # get prediction for semantic matching indices
    if len(semantic_match_indices) > 0:
        eval_values = _get_prediction_semantic_match(
            eval_args              = eval_values, 
            semantic_match_indices = semantic_match_indices
        )
        
    # scaling
    eval_values['anomaly_maps'] = segmentation_scaling([eval_values['anomaly_maps']])
    eval_values['anomaly_scores'] = score_scaling(eval_values['anomaly_scores'])    

    results = {}
    results['best_threshold'] = _binary_results(eval_args=eval_values)
        
    return results