coding = UTF-8

import sys import os

class Logger(object): def init(self, filename='default.log', stream=sys.stdout): # 构造函数，接受一个可选的文件名和一个可选的输出流，默认为将输出写入标准输出流(sys.stdout)。 self.terminal = stream # 将给定的输出流保存到self.terminal变量中。 self.log = open(filename, 'w') #打开一个文件，将文件对象保存到self.log变量中，以便将输出写入该文件。

def write(self, message):   #  自定义的write方法，用于将消息写入控制台和日志文件。它先将消息写入控制台，然后再写入日志文件。
    self.terminal.write(message)
    self.log.write(message)

def flush(self):  #自定义的flush方法，不执行任何操作。这个方法是为了兼容性，因为在某些情况下，需要调用flush方法来确保输出被完全刷新。
    pass

sys.stdout = Logger('result.log', sys.stdout) # 这行代码将标准输出重定向到一个名为'result.log'的日志文件中。 import argparse import numpy as np import time from sklearn.metrics import accuracy_score, confusion_matrix from net.LET1D_CNN import CNN # 导入 LET1D_CNN 模块中的 CNN 类 from mydataset import MyDataset import torch from torch.utils.data import DataLoader import torch.nn as nn import matplotlib.pyplot as plt import pandas as pd import seaborn as sns # µ¼Èë°ü from tqdm import *

def test(model, dataset, criterion): #该函数用于对模型进行测试，输入参数包括模型、数据集和损失函数。函数的功能包括计算模型在测试集上的准确率、平均损失，并返回特征列表和混淆矩阵。 model.eval() #将模型设置为评估模式（model.eval()） total_batch_num =0. #初始化总批次数为0，验证损失为0，预测结果列表和标签列表为空，特征列表为空的Tensor。 val_loss = 0 prediction = [] labels = [] feature_list = torch.tensor([]) # test×ÜÊý£¬ÌØÕ÷Êý #if torch.cuda.is_available(): #这段代码用于判断当前系统是否支持CUDA（一个用于深度学习的计算平台），如果支持，则将feature_list转移到GPU上进行计算。1 # feature_list = feature_list.cuda() for (step, i) in enumerate(dataset): #这段代码是一个for循环，用来遍历一个名为dataset的数据集。在每次循环中，使用enumerate函数来获取数据集中每个元素的索引step和对应的元素i。 total_batch_num = total_batch_num + 1 #是计算总批次数的代码，每次循环都会将total_batch_num增加1 batch_x = i['data'] #将当前数据项的"data"键对应的值赋给batch_x。 batch_y = i['label'] # 将当前数据项的"label"键对应的值赋给batch_y。 batch_x = torch.unsqueeze(batch_x, dim=1) # (50, 1, 10000, 12) #batch_x张量的维度扩展为(50, 1, 10000, 12)，在第2维度上添加一个维度。 batch_x = batch_x.float() #将batch_x张量转换为float类型。 if torch.cuda.is_available(): #检查是否有可用的CUDA设备（即GPU）。3 batch_x = batch_x.cuda() batch_y = batch_y.cuda() feature, probs = model(batch_x) # feature 20,400 #使用模型对输入数据进行前向传播，得到特征和预测概率。 feature = feature.cpu().detach()#cpu batch_label = batch_y.unsqueeze(1).float() #feature将批次标签转换为浮点型，并添加一个维度。 batch_label = batch_label.cpu().detach() #cpu feature_label = torch.cat((feature, batch_label), dim=1) #将特征和标签按列拼接在一起。 feature_list = torch.cat((feature_list, feature_label), dim=0) # feature_list #将特征列表与当前特征和标签拼接在一起。 loss = criterion(probs, batch_y) #计算预测概率和真实标签之间的损失。 _, pred = torch.max(probs, dim=1) #找出预测概率最大的类别作为预测结果。 predi = pred.tolist() #将预测结果转换为Python列表。 label = batch_y.tolist() #将真实标签转换为Python列表。 val_loss += loss.item() #累加验证集上的损失值。 x prediction.extend(predi) #将当前批次的预测结果添加到整体预测结果列表中。 labels.extend(label) #将当前批次的真实标签添加到整体标签列表中。 accuracy = accuracy_score(labels, prediction) #计算整体预测准确率。 C = confusion_matrix(labels, prediction) #计算整体预测结果的混淆矩阵。 return accuracy, val_loss / total_batch_num, feature_list, C #返回准确率、平均损失、特征列表和混淆矩阵。

def train(model, train_x, train_y, optimizer, criterion): model.train() #设置模型为训练模式。 model.zero_grad() #清空模型的梯度。 _,probs = model(train_x) #使用训练数据train_x输入模型得到预测结果probs。下划线_用于忽略第一个返回值。 loss = criterion(probs, train_y) #使用损失函数criterion计算预测结果probs与训练标签train_y之间的损失。 _, pred = torch.max(probs, dim=1) #在预测结果probs的第一维度上取最大值，并返回最大值和对应的索引pred。 labels = train_y.tolist() #将训练标签train_y转换为Python列表格式。 predi = pred.tolist() #将预测结果pred转换为Python列表格式。 loss.backward() #反向传播计算梯度。 optimizer.step() #使用优化器optimizer更新模型参数。 return labels, predi, loss.item() #返回训练标签、预测结果和损失的值。

def draw(train_acc, train_loss, test_acc, test_loss): #定义一个名为draw的函数，该函数接受四个参数：train_acc、train_loss、test_acc和test_loss。 x1 = range(len(train_acc)) #定义一个变量x1，用range函数生成一个包含从0到train_acc的长度的整数序列。 x2 = range(len(train_loss)) #义一个变量x2，用range函数生成一个包含从0到train_loss的长度的整数序列。 y1 = train_acc y2 = train_loss y3 = test_acc y4 = test_loss plt.subplot(2, 1, 1) plt.plot(x1, y1, 'o-', label="train", color='b') plt.plot(x1, y3, 'o-', label="test", color='r') plt.legend(loc='upper left') plt.title('accuracy & NAR vs. epochs') plt.ylabel('accuracy') plt.subplot(2, 1, 2) plt.plot(x2, y2, '.-', label="train", color='b') plt.plot(x2, y4, '.-', label="test", color='r') plt.legend(loc='upper left') plt.xlabel('loss vs. epochs') plt.ylabel('loss') plt.savefig("accuracy_loss.jpg") plt.show()

def draw_result(C): fig = plt.figure() ax = fig.add_subplot(111) df = pd.DataFrame(C) # f1 = fm.fontProperties('Times New Roman', size=15) sns.heatmap(df, fmt='g', annot=True, annot_kws={'size': 10}, xticklabels=['1', '2', '3', '4', '5', '6'], yticklabels=['1', '2', '3', '4', '5', '6'], cmap='Blues') ax.set_xlabel('Predicted label') # fontProperties=f1) # xÖá ax.set_ylabel('True label') # fontProperties=f1) # yÖá plt.savefig('./cnn_gmlp_matrix.jpg') plt.show() Acc = (C[0][0] + C[1][1] + C[2][2] + C[3][3] + C[4][4] + C[5][5]) / sum(C[0] + C[1] + C[2] + C[3] + C[4] + C[5]) print('acc: %.3f' % Acc) lie_he = sum(C, 1) - 1 for i in range(1, 7): Precision = C[i - 1][i - 1] / lie_he[i - 1] NAR = (sum(C[i - 1]) - C[i - 1][i - 1]) / sum(C[i - 1]) F1_score = 2 * C[i - 1][i - 1] / (lie_he[i - 1] + sum(C[i - 1])) print('precision_%d: %.3f' % (i, Precision)) print('NAR_%d: %.3f' % (i, NAR)) print('F1_score_%d: %.3f' % (i, F1_score))

def main(args): # »®·ÖѵÁ·¼¯¡¢²âÊÔ¼¯ train_dataset = MyDataset(args.root, args.txtpath, transform=None) train_loader = DataLoader(dataset=train_dataset, batch_size=args.batch_size, shuffle=True, num_workers=0) test_dataset = MyDataset(args.root2, args.txtpath2, transform=None) test_loader = DataLoader(dataset=test_dataset, batch_size=args.batch_size, shuffle=True, num_workers=0)

cnn_gmlp_model= {"CNN": CNN}



# ¶þÕßÈ¡ÆäÒ»
# model = torch.load('./gmlp-modelpth/40.pth')#,map_location=torch.device('cpu'))  # ¼ÓÔزÎÊý
model = cnn_gmlp_model[args.model]()

if torch.cuda.is_available():
    model = model.cuda()
#model = torch.load('./cnn-gmlp-modelpth/40.pth')

optimizer = torch.optim.Adam(model.parameters(), lr=1e-4, weight_decay=1e-5)
# optimizer.load_state_dict(torch.load('save_optim.pt'))
criterion = nn.CrossEntropyLoss()
batches_per_epoch = int(len(train_dataset) / args.batch_size)
train_loss_list = []
train_acc_list = []
test_loss_list = []
test_acc_list = []
train_time = 0
model_save_path = 'cnn-gmlp-modelpth2/'
os.makedirs(model_save_path, exist_ok=True)
for epoch in range(args.epochs):
    tic = time.time()
    train_predict = []
    train_label = []
    runloss = 0
    for (cnt, i) in enumerate(tqdm(train_loader)):
        batch_x = i['data']
        batch_y = i['label']
        batch_x = torch.unsqueeze(batch_x, dim=1)
        batch_x = batch_x.float()

        if torch.cuda.is_available():
            batch_x = batch_x.cuda()
            batch_y = batch_y.cuda()
        tlabels, tpredi, tloss = train(model, batch_x, batch_y, optimizer, criterion)
        runloss = runloss + tloss
        train_label.extend(tlabels)
        train_predict.extend(tpredi)
    per_epoch_train_time = time.time() - tic
    train_time = per_epoch_train_time + train_time
    taccuracy = accuracy_score(train_label, train_predict)
    train_acc_list.append(taccuracy)
    loss = runloss / batches_per_epoch
    train_loss_list.append(loss)
    acc_score, loss_score, feature_list, C = test(model, test_loader, criterion)
    print("Epoch %d Val_accuracy %.3f Val_loss %.3f" % (epoch, acc_score, loss_score))
    with open('1d_cnn_train_log.csv', 'a+') as f:
        f.write(f'{epoch}, {round(taccuracy, 4)}, {round(loss, 4)}

') f.close() with open('1d_cnn_log.csv', 'a+')as f: f.write(f'{epoch}, {round(acc_score, 4)}, {round(loss_score, 4)} ') f.close() test_acc_list.append(acc_score) test_loss_list.append(loss_score)

    if epoch % 1 == 0:
        torch.save(model, f'cnn-gmlp-modelpth2/{epoch}.pth')
    if epoch == args.epochs - 1:  # ѵÁ·µ½×îºóÒ»ÂÖ
        torch.save(model, 'cnn-gmlp-modelpth2/final.pth')
        the_model = torch.load('./cnn-gmlp-modelpth2/final.pth')
        acc_score, loss_score, NAR, featucre_list = test(the_model, test_loader, criterion)
        #print("Epoch %d Train_accuracy %.3f Train_loss %.3f Val_accuracy %.3f Val_loss %.3f NAR %.3f " % (epoch, taccuracy, loss, acc_score, loss_score, NAR))
        print(f'Epoch {epoch} Train_accuracy {taccuracy} Train_loss {loss} Val_accuracy {acc_score} Val_loss {loss_score} NAR {NAR}')
        feature_list = feature_list.detach().cpu().numpy()
        np.savetxt('1d_feature_data.csv', feature_list, delimiter=',')  # ±£´æÌØÕ÷ÏòÁ¿
        draw_result(C)
        print('train totally using %.3f seconds ', train_time)
draw(train_acc_list, train_loss_list, test_acc_list, test_loss_list)

if name == "main": parser = argparse.ArgumentParser(description="CNN fou classification") '''save model''' parser.add_argument("--save", type=str, default="__", help="path to save model") '''model parameters''' rootpath = 'das_data' parser.add_argument("--root", type=str, default=rootpath + '/train', help="rootpath of traindata") parser.add_argument("--root2", type=str, default=rootpath + '/test', help="rootpath of valdata") parser.add_argument("--txtpath", type=str, default=rootpath + '/train/label.txt', help="path of train_list") parser.add_argument("--txtpath2", type=str, default=rootpath + '/test/label.txt', help="path pf val_list") parser.add_argument("--model", type=str, default="CNN", help="type of model to use for classification") parser.add_argument("--lr", type=float, default=0.01, help="learning rate") parser.add_argument("--epochs", type=int, default=70 , help="number of training epochs") parser.add_argument("--batch_size", type=int, default=100, help="batch size") my_args = parser.parse_args()

main(my_args)