import torch import numpy as np import torch.nn.functional as F from utils import get_msg_mgr def cuda_dist(x, y, metric='euc'): x = torch.from_numpy(x).cuda() y = torch.from_numpy(y).cuda() if metric == 'cos': x = F.normalize(x, p=2, dim=2) # n p c y = F.normalize(y, p=2, dim=2) # n p c num_bin = x.size(1) n_x = x.size(0) n_y = y.size(0) dist = torch.zeros(n_x, n_y).cuda() for i in range(num_bin): _x = x[:, i, ...] _y = y[:, i, ...] if metric == 'cos': dist += torch.matmul(_x, _y.transpose(0, 1)) else: _dist = torch.sum(_x ** 2, 1).unsqueeze(1) + torch.sum(_y ** 2, 1).unsqueeze( 1).transpose(0, 1) - 2 * torch.matmul(_x, _y.transpose(0, 1)) dist += torch.sqrt(F.relu(_dist)) return 1 - dist/num_bin if metric == 'cos' else dist / num_bin # Exclude identical-view cases def de_diag(acc, each_angle=False): dividend = acc.shape[1] - 1. result = np.sum(acc - np.diag(np.diag(acc)), 1) / dividend if not each_angle: result = np.mean(result) return result # Modified From https://github.com/AbnerHqC/GaitSet/blob/master/model/utils/evaluator.py def identification(data, dataset, metric='euc'): msg_mgr = get_msg_mgr() feature, label, seq_type, view = data['embeddings'], data['labels'], data['types'], data['views'] label = np.array(label) view_list = list(set(view)) view_list.sort() view_num = len(view_list) # sample_num = len(feature) probe_seq_dict = {'CASIA-B': [['nm-05', 'nm-06'], ['bg-01', 'bg-02'], ['cl-01', 'cl-02']], 'OUMVLP': [['00']]} gallery_seq_dict = {'CASIA-B': [['nm-01', 'nm-02', 'nm-03', 'nm-04']], 'OUMVLP': [['01']]} if dataset not in (probe_seq_dict or gallery_seq_dict): raise KeyError("DataSet %s hasn't been supported !" % dataset) num_rank = 5 acc = np.zeros([len(probe_seq_dict[dataset]), view_num, view_num, num_rank]) - 1. for (p, probe_seq) in enumerate(probe_seq_dict[dataset]): for gallery_seq in gallery_seq_dict[dataset]: for (v1, probe_view) in enumerate(view_list): for (v2, gallery_view) in enumerate(view_list): gseq_mask = np.isin(seq_type, gallery_seq) & np.isin( view, [gallery_view]) gallery_x = feature[gseq_mask, :] gallery_y = label[gseq_mask] pseq_mask = np.isin(seq_type, probe_seq) & np.isin( view, [probe_view]) probe_x = feature[pseq_mask, :] probe_y = label[pseq_mask] dist = cuda_dist(probe_x, gallery_x, metric) idx = dist.sort(1)[1].cpu().numpy() acc[p, v1, v2, :] = np.round( np.sum(np.cumsum(np.reshape(probe_y, [-1, 1]) == gallery_y[idx[:, 0:num_rank]], 1) > 0, 0) * 100 / dist.shape[0], 2) result_dict = {} np.set_printoptions(precision=3, suppress=True) if 'OUMVLP' not in dataset: for i in range(1): msg_mgr.log_info( '===Rank-%d (Include identical-view cases)===' % (i + 1)) msg_mgr.log_info('NM: %.3f,\tBG: %.3f,\tCL: %.3f' % ( np.mean(acc[0, :, :, i]), np.mean(acc[1, :, :, i]), np.mean(acc[2, :, :, i]))) for i in range(1): msg_mgr.log_info( '===Rank-%d (Exclude identical-view cases)===' % (i + 1)) msg_mgr.log_info('NM: %.3f,\tBG: %.3f,\tCL: %.3f' % ( de_diag(acc[0, :, :, i]), de_diag(acc[1, :, :, i]), de_diag(acc[2, :, :, i]))) result_dict["scalar/test_accuracy/NM"] = de_diag(acc[0, :, :, i]) result_dict["scalar/test_accuracy/BG"] = de_diag(acc[1, :, :, i]) result_dict["scalar/test_accuracy/CL"] = de_diag(acc[2, :, :, i]) np.set_printoptions(precision=2, floatmode='fixed') for i in range(1): msg_mgr.log_info( '===Rank-%d of each angle (Exclude identical-view cases)===' % (i + 1)) msg_mgr.log_info('NM: {}'.format(de_diag(acc[0, :, :, i], True))) msg_mgr.log_info('BG: {}'.format(de_diag(acc[1, :, :, i], True))) msg_mgr.log_info('CL: {}'.format(de_diag(acc[2, :, :, i], True))) else: msg_mgr.log_info('===Rank-1 (Include identical-view cases)===') msg_mgr.log_info('NM: %.3f ' % (np.mean(acc[0, :, :, 0]))) msg_mgr.log_info('===Rank-1 (Exclude identical-view cases)===') msg_mgr.log_info('NM: %.3f ' % (de_diag(acc[0, :, :, 0]))) msg_mgr.log_info( '===Rank-1 of each angle (Exclude identical-view cases)===') msg_mgr.log_info('NM: {}'.format(de_diag(acc[0, :, :, 0], True))) result_dict["scalar/test_accuracy/NM"] = de_diag(acc[0, :, :, 0]) return result_dict def identification_real_scene(data, dataset, metric='euc'): msg_mgr = get_msg_mgr() feature, label, seq_type = data['embeddings'], data['labels'], data['types'] label = np.array(label) gallery_seq_type = {'0001-1000': ['1', '2'], "HID2021": ['0'], '0001-1000-test': ['0']} probe_seq_type = {'0001-1000': ['3', '4', '5', '6'], "HID2021": ['1'], '0001-1000-test': ['1']} num_rank = 5 acc = np.zeros([num_rank]) - 1. gseq_mask = np.isin(seq_type, gallery_seq_type[dataset]) gallery_x = feature[gseq_mask, :] gallery_y = label[gseq_mask] pseq_mask = np.isin(seq_type, probe_seq_type[dataset]) probe_x = feature[pseq_mask, :] probe_y = label[pseq_mask] dist = cuda_dist(probe_x, gallery_x, metric) idx = dist.cpu().sort(1)[1].numpy() acc = np.round(np.sum(np.cumsum(np.reshape(probe_y, [-1, 1]) == gallery_y[idx[:, 0:num_rank]], 1) > 0, 0) * 100 / dist.shape[0], 2) msg_mgr.log_info('==Rank-1==') msg_mgr.log_info('%.3f' % (np.mean(acc[0]))) msg_mgr.log_info('==Rank-5==') msg_mgr.log_info('%.3f' % (np.mean(acc[4]))) return {"scalar/test_accuracy/Rank-1": np.mean(acc[0]), "scalar/test_accuracy/Rank-5": np.mean(acc[4])} def evaluate_HID(data, dataset, metric='euc'): msg_mgr = get_msg_mgr() msg_mgr.log_info("Evaluating HID") feature, label, seq_type = data['embeddings'], data['labels'], data['types'] label = np.array(label) seq_type = np.array(seq_type) probe_mask = (label == "probe") gallery_mask = (label != "probe") gallery_x = feature[gallery_mask, :] gallery_y = label[gallery_mask] probe_x = feature[probe_mask, :] probe_y = seq_type[probe_mask] dist = cuda_dist(probe_x, gallery_x, metric) idx = dist.cpu().sort(1)[1].numpy() import os from time import strftime, localtime save_path = os.path.join( "HID_result/"+strftime('%Y-%m%d-%H%M%S', localtime())+".csv") os.makedirs("HID_result", exist_ok=True) with open(save_path, "w") as f: f.write("videoID,label\n") for i in range(len(idx)): f.write("{},{}\n".format(probe_y[i], gallery_y[idx[i, 0]])) print("HID result saved to {}/{}".format(os.getcwd(), save_path)) return