utils.py

import os
import json
import itertools
import random as rnd
import numpy as np
import torch
import torch.nn as nn
import torch.nn.functional as F
import matplotlib.pyplot as plt
from collections import Counter
from sklearn.metrics import confusion_matrix

############################################
# Training Utilities
############################################

def get_random_seed(seed):
    rnd.seed(seed)
    os.environ['PYTHONHASHSEED'] = str(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    torch.cuda.manual_seed(seed)
    torch.backends.cudnn.deterministic = True
    torch.backends.cudnn.benchmark = False

############################################
# Visualization
############################################

def plot_confusion_matrix(cm, classes, title='Confusion Matrix', cmap=plt.cm.Blues):
    plt.clf()
    plt.figure(figsize=(10, 10))
    plt.imshow(cm, interpolation='nearest', cmap=cmap)
    plt.title(title)
    plt.colorbar()
    tick_marks = np.arange(len(classes))
    plt.xticks(tick_marks, classes, rotation=45)
    plt.yticks(tick_marks, classes)
    thresh = cm.max() / 2.
    for i, j in itertools.product(range(cm.shape[0]), range(cm.shape[1])):
        plt.text(j, i, cm[i, j],
                 horizontalalignment="center",
                 color="white" if cm[i, j] > thresh else "black")

    plt.tight_layout()
    plt.ylabel('True label')
    plt.xlabel('Predicted label')

############################################
# Evaluation Utilities
############################################

def reportOnlineAccuracy(all_preds,all_targets,datarange,savename):
    #按照online的规则计算多个指标，前6个集成判断
    numofperson=datarange[1]-datarange[0]+1
    personsize = all_preds.shape[0]//numofperson
    perman_acc,perman_faults={},{}
    for j in range(numofperson):
        all_targets_sub=all_targets[j*personsize:(j+1)*personsize]
        all_preds_sub=all_preds[j*personsize:(j+1)*personsize]
        action=[]
        silent,silentindex=False,0
        for time in range(all_targets_sub.shape[0]):
            # 0 表示这一时刻没有动作输出
            add=0
            if silent:silentindex+=1
            if silentindex==10:
                silent=False
                silentindex=0
            #取当前和前5个元素
            counter=Counter(all_preds_sub[max(0,time-4):time+1])
            for key,value in counter.items():
                if value>=5 and key!=0 and not silent:
                    add=key
                    silent=True
            action.append(add)
        #不转化为array的话没办法判断每一位是否=0
        action=np.array(action)
        #判断每一个target是否有action在规定范围响应
        last,Startindex,Endindex,eventcount,rightcount,faultcount=0,0,0,0,0,0
        for time in range(all_targets_sub.shape[0]):
            if last==0 and all_targets_sub[time]!=0:
                Startindex=time
                #记录一共的target事件数
                eventcount+=1
                if Endindex < Startindex:
                    faultcount+=sum(action[Endindex+1:Startindex]!=0)
            elif last!=0 and (all_targets_sub[time]==0 or time==all_targets_sub.shape[0]-1):
                Endindex=time-1
                if Endindex > Startindex:
                    #减3意味着在动作结束的200ms及以前
                    if sum(action[Startindex:Endindex-3]==last)>0:
                        rightcount+=1
            last=all_targets_sub[time]
        #统计每个人的acc和faults
        perman_acc['subid_'+j]=rightcount/eventcount
        perman_faults['subid_'+j]=faultcount/sum(all_targets_sub==0)
        #保存每个人的acc和faults
        with open('hand/online_perman/'+savename+'.json','w+') as f:
            json.dump({'perman_acc':perman_acc,'perman_faults':perman_faults},f)
    return np.mean(perman_acc), np.mean(perman_faults)

def testslice(model,alltestset,device,subjectId_test):
    model.eval()
    testbatch=160
    all_preds=[]
    all_targets=[]
    with torch.no_grad():
        for batch in range(int(alltestset.x_data.shape[0]//testbatch+1)):
            x, y = alltestset.x_data[batch*testbatch:min((batch+1)*testbatch,alltestset.x_data.shape[0]),:,:,:], alltestset.y_data[batch*testbatch:min((batch+1)*testbatch,alltestset.x_data.shape[0])]
            x = x.to(device)
            y = y.to(device)
            #预测
            pred= model(x)
            prediction = torch.argsort(pred, dim=-1, descending=True)
            #all_preds为模型的总预测结果
            #all_targets为总的label
            predictions = np.array((np.vstack(prediction[:, 0:1].cpu())))
            predictions = predictions.reshape(predictions.shape[1],-1)[0]
            targets=np.array(y.cpu().numpy())
            all_targets.append(targets)
            all_preds.append(predictions)
    all_preds=np.concatenate(all_preds,axis=0)
    all_targets=np.concatenate(all_targets,axis=0)
    accperman={}
    #每6个target就投票算一个label
    for i in range(int(all_preds.shape[0]/6)):
        count=Counter(all_preds[i*6:(i+1)*6])
        maxkey=max(count, key=lambda key: count[key])
        maxvalue=count[maxkey]
        if maxvalue <=3:
            # 0表示休息
            vote=0
        else:
            vote=maxkey
        # 取一个中间值是target
        targetkey=all_targets[i*6+3]
        subject=subjectId_test[i*6+3]
        if subject not in accperman.keys():
            accperman[subject]=[]
        if targetkey==vote:
            accperman[subject].append(1)
        else:
            accperman[subject].append(0)
    for subject,sublist in accperman.items():
        accperman[subject]=np.mean(sublist)
    allacc=np.mean(list(accperman.values()))
    # record acc of every person
    return allacc, accperman

def testonline(model, testdata, device, datarange,savename):
    model.eval()
    with torch.no_grad():
        x, y = testdata.x_data, testdata.y_data
        x = x.to(device)
        y = y.to(device)
        #预测
        pred= model(x)
        prediction = torch.argsort(pred, dim=-1, descending=True)
    #all_preds为模型的总预测结果
    #all_targets为总的label
    all_preds=np.array((np.vstack(prediction[:, 0:1].cpu())))
    all_preds = all_preds.reshape(all_preds.shape[1],-1)[0]
    all_targets=np.array(y.cpu().numpy())
    accuracy, faults=reportOnlineAccuracy(all_preds,all_targets,datarange,savename)
    # record acc of every person
    return accuracy, faults

############################################
# Loss Functions
############################################

class LabelSmoothingLoss(nn.Module):
    def __init__(self, epsilon=0.1, num_classes=2):
        super(LabelSmoothingLoss, self).__init__()
        self.epsilon = epsilon
        self.num_classes = num_classes

    def forward(self, input, target):
        confidence = 1.0 - self.epsilon
        smoothing_value = self.epsilon / (self.num_classes - 1)

        one_hot = torch.full_like(input, smoothing_value)
        one_hot.scatter_(1, target.unsqueeze(1), confidence)

        return nn.CrossEntropyLoss()(input,one_hot)