<jupyter_start><jupyter_text>数据集UCI 人类活动识别数据集是以智能手机采集的传感器数据为基础的活动识别，创建于2012年，实验团队来自意大利热那亚大学。在2012年的论文《Human Activity Recognition on Smartphones using a Multiclass Hardware-Friendly Support Vector Machine》中，采用机器学习算法建模，提供了该数据集分类性能的baseline。在2013年的论文《A Public Domain Dataset for Human Activity Recognition Using Smartphones》中，对数据集进行了全面描述。数据集可以从UCI机器学习存储库免费下载：👉传送门。http://archive.ics.uci.edu/ml/machine-learning-databases/00240/<jupyter_code>'''
-*- coding: utf-8 -*-
@Time : 2021/8/7 20:55
@Author : Small_Volcano
@File : UCI_HAR_CNN.py
'''
import copy
import time
import torch
import torch.nn as nn
import torch.utils.data as Data
from torch.optim import Adam
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import os
def load_file(filepath):
    """
    加载文件
    :param filepath:
    :return:
    """
    dataframe = pd.read_csv(filepath, header=None, delim_whitespace=True)
    return dataframe.values
def load_dataset(data_rootdir, dirname, group):
    '''
    该函数实现将训练数据或测试数据文件列表堆叠为三维数组
    '''
    filename_list = []
    filepath_list = []
    X = []

    # os.walk() 方法是一个简单易用的文件、目录遍历器，可以高效的处理文件、目录。
    for rootdir, dirnames, filenames in os.walk(data_rootdir + dirname):
        for filename in filenames:
            filename_list.append(filename)
            filepath_list.append(os.path.join(rootdir, filename))
        #print(filename_list)
        #print(filepath_list)

    # 遍历根目录下的文件，并读取为DataFrame格式；
    for filepath in filepath_list:
        X.append(load_file(filepath))

    X = np.dstack(X) # dstack沿第三个维度叠加，两个二维数组叠加后，前两个维度尺寸不变，第三个维度增加；
    y = load_file(data_rootdir+'/y_'+group+'.txt')
    print('{}_X.shape:{},{}_y.shape:{}\n'.format(group,X.shape,group,y.shape))
    return X, y
train_rootdir = './UCI_HAR_Dataset/train/'
test_rootdir = './UCI_HAR_Dataset/test/'
data_dirname = '/Inertial Signals/'
trainX, trainy = load_dataset(train_rootdir, data_dirname, 'train')
testX, testy = load_dataset(test_rootdir, data_dirname, 'test')

# data_x_raw.reshape(-1, 1, data_x_raw.shape[1], data_x_raw.shape[2])  # (N, C, H, W) (7352, 1, 128, 9)

trainX = trainX.reshape(-1, 1, trainX.shape[1], trainX.shape[2])
testX = testX.reshape(-1, 1, testX.shape[1], testX.shape[2])
har_train_tensor = Data.TensorDataset(torch.from_numpy(trainX).to(torch.float32), torch.from_numpy(trainy) )
har_test_tensor =  Data.TensorDataset(torch.from_numpy(testX).to(torch.float32), torch.from_numpy(testy))
# class HAR(Data.Dataset):
#     def __init__(self, filename_x, filename_y):
#         self.filename_x = filename_x
#         self.filename_y = filename_y
#
#     def HAR_data(self):
#         """更改x的维度,加载x和y"""
#         data_x_raw = np.load(self.filename_x)
#         #print(data_x_raw.shape)                                                        #为什么是1通道 128*9
#         data_x = data_x_raw.reshape(-1, 1, data_x_raw.shape[1], data_x_raw.shape[2])  # (N, C, H, W) (7352, 1, 128, 9)
#         # data_x = np.expand_dims(data_x_raw, 1)
#         #print(data_x.shape)
#         data_y = np.load(self.filename_y)
#         print("datay{}".format(data_y))
#         torch_dataset = Data.TensorDataset(torch.from_numpy(data_x), torch.from_numpy(data_y))#造数据集
#         return torch_dataset
#
# data_train = HAR(train_x_list, train_y_list)# 这一步是做什么？
#   # 创造训练验证数据集
# #print(har_train_tensor)
# #测试集数据
# data_test = HAR(test_x_list, test_y_list)
# har_test_tensor = data_test.HAR_data()
train_loader = Data.DataLoader(dataset=har_train_tensor,
                               batch_size=128,
                               shuffle=True,
                               num_workers=0, )
#设置一个测试集加载器
test_loader = Data.DataLoader(dataset=har_test_tensor,
                               batch_size=1,
                               shuffle=True,
                               num_workers=0, )
#搭建卷积神经网络
class Net(nn.Module):
    def __init__(self):
        super(Net, self).__init__()
        #定义第一个卷积层
        self.conv1 = nn.Sequential(
            nn.Conv2d(in_channels=1,
                      out_channels=12,          #输出高度12
                      kernel_size=3,            #卷积核尺寸3*3
                      stride=1,
                      padding=1),               #(1*128*9)-->(12*128*9)
            nn.ReLU(),
            nn.MaxPool2d(kernel_size=2,stride=2) #(12*128*9)-->(12*64*4)
        )
        #定义第二个卷积层
        self.conv2 = nn.Sequential(
            nn.Conv2d(in_channels=12,
                      out_channels=32,
                      kernel_size=3,
                      stride=1,
                      padding=1),               #(12*64*4)-->(32*64*4)
            nn.ReLU(),
            nn.MaxPool2d(kernel_size=2,stride=2) #池化后：(32*32*2)
        )
        self.conv3 = nn.Sequential(
            nn.Conv2d(in_channels=32,
                      out_channels=64,
                      kernel_size=3,
                      stride=1,
                      padding=1),                #(32*32*2)-->(64*32*2)
            nn.ReLU()
        )
        #定义全连接层
        self.classifier = nn.Sequential(
            nn.Linear(64*32*2,128),              #长方体变平面
            nn.ReLU(),
            nn.Dropout(p = 0.5),
            nn.Linear(128,6)
        )

    #定义网络的前向传播路径
    def forward(self,x):
        x = self.conv1(x)
        print('conv1',x.shape)
        x = self.conv2(x)
        print('conv2',x.shape)
        x = self.conv3(x)
        print('conv3',x.shape)
        x = x.view(x.shape[0],-1) #展平多维的卷积图层
        output = self.classifier(x)
        print('output',output.shape)
        return output
#定义网络的训练过程函数
def train_model(model,traindataloader,train_rate,criterion,optimizer,num_epochs=25):
    #train_rate：训练集中训练数量的百分比
    #计算训练使用的batch数量
    batch_num = len(traindataloader)
    train_batch_num = round(batch_num * train_rate) #前train_rate（80%）的batch进行训练
    #复制最好模型的参数
    best_model_wts = copy.deepcopy(model.state_dict())
    best_acc = 0.0
    train_loss_all = []
    train_acc_all =    []
    val_loss_all = []
    val_acc_all = []
    since = time.time()
    for epoch in range(num_epochs):
        print('Epoch {}/{}'.format(epoch,num_epochs-1)) #格式化字符串
        print('-' * 10)
        #每个epoch有两个训练阶段
        train_loss = 0.0
        train_corrects = 0
        train_num = 0
        val_loss = 0.0
        val_corrects = 0
        val_num = 0
        for step,(b_x,b_y) in enumerate(traindataloader,1): #取标签和样本
            b_y = b_y.long()
            if step < train_batch_num:                      #前train_rate（80%）的batch进行训练
                model.train()                               #设置模型为训练模式，对Droopou有用
                output = model(b_x)
               # print(b_x)#取得模型预测结果
                pre_lab = torch.argmax(output,1)            #横向获得最大值位置
                b_y = b_y.squeeze(1)-1            #修改BUG
                print(b_y.shape, b_y)
                loss = criterion(output,b_y)                #每个样本的loss
                optimizer.zero_grad()
                loss.backward()
                optimizer.step()                            #修改权值
                train_loss += loss.item() * b_x.size(0)
                #print(pre_lab)
                #print(b_y.data)
                train_corrects += torch.sum(pre_lab == b_y.data) #训练正确个数
                train_num += b_x.size(0)
            else:
                model.eval()                                    #设置模型为验证模式
                output = model(b_x)
                pre_lab = torch.argmax(output,1)
                loss = criterion(output,b_y)
                val_loss += loss.item() * b_x.size(0)
                val_corrects += torch.sum(pre_lab == b_y.data)
                val_num += b_x.size(0)
        #计算训练集和验证集上的损失和精度
        train_loss_all.append(train_loss / train_num)           #一个epoch上的loss
        train_acc_all.append(train_corrects.double().item() / train_num)
        val_loss_all.append(val_loss / val_num)
        val_acc_all.append(val_corrects.double().item() / val_num)

        print('{} Train Loss: {:.4f} Train Acc: {:.4f}'.format(epoch,train_loss_all[-1],train_acc_all[-1])) #此处-1没搞明白
        print('{} Val Loss: {:.4f} Val Acc: {:.4f}'.format(epoch,val_loss_all[-1],val_acc_all[-1]))
        #拷贝模型最高精度下的参数
        if val_acc_all[-1] > best_acc:
            best_acc = val_acc_all[-1]
            best_model_wts = copy.deepcopy(model.state_dict())
            torch.save(model.state_dict(),"UCI_HAR_model")
            torch.save(optimizer.state_dict(),"UCI_HAR_optimizer")
        time_use = time.time() - since
        print("Train and val complete in {:.0f}m {:.0f}s".format(time_use // 60,time_use % 60)) #训练用时
    #使用最好模型的参数
    model.load_state_dict(best_model_wts)
    #组成数据表格train_process打印
    train_process = pd.DataFrame(data={"epoch":range(num_epochs),
                                       "train_loss_all":train_loss_all,
                                       "val_loss_all":val_loss_all,
                                       "train_acc_all":train_acc_all,
                                       "val_acc_all":val_acc_all})
    return model,train_process
#输出网络结构
net = Net()     #创建实例
print(net)
#对模型进行训练
optimizer = Adam(net.parameters(),lr=0.0003)            #优化器
criterion = nn.CrossEntropyLoss()                       #使用交叉熵作为损失函数
# 使用训练集的20%作为验证
net,train_process = train_model(net,train_loader,0.8, criterion,optimizer, num_epochs=25)
#可视化模型训练过程中
plt.figure(figsize=(12,4))
plt.subplot(1,2,1)
plt.plot(train_process.epoch,train_process.train_loss_all,"ro-",label="Train loss")
plt.plot(train_process.epoch,train_process.val_loss_all,"bs-",label="Val loss")
plt.legend()
plt.xlabel("epoch")
plt.ylabel("Loss")
plt.subplot(1,2,2)
plt.plot(train_process.epoch,train_process.train_acc_all,"ro-",label="Train acc")
plt.plot(train_process.epoch,train_process.val_acc_all,"bs-",label="Val acc")
plt.xlabel("epoch")
plt.ylabel("acc")
plt.legend()
plt.show()
#对测试集进行预测,计算模型的泛化能力
def test(model,testdataloader,criterion):
    test_loss_all = []
    test_acc_all = []
    test_loss = 0.0
    test_corrects = 0
    test_num = 0
    for step,(input,target) in enumerate(testdataloader):   #取标签和样本
        target = target.long()
        #target = torch.Tensor(target).long()
        model.eval()                                       #设置模型为训练模式，对Droopou有用
        output = model(input)
        # print(b_x)#取得模型预测结果
        pre_lab = torch.argmax(output,1)                    #横向获得最大值位置
        loss = criterion(output,target)                     #每个样本的loss
        test_loss += loss.item() * input.size(0)            #此处的b_x.size(0)=batch_size。此处相当于一个batch的loss？计算的是整体训练的loss
        #print(pre_lab)
        #print(input.data)
        test_corrects += torch.sum(pre_lab == target.data)  #测试正确个数
        test_num += input.size(0)
    test_loss_all.append(test_loss / test_num)
    test_acc_all.append(test_corrects.double().item() / test_num)
    print('Test all Loss: {:.4f} Test Acc: {:.4f}'.format(test_loss_all[-1], test_acc_all[-1]))

test = test(net,test_loader,criterion)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.12631098951208392, 'fraction_numerical': 0.023711810305517556, 'mean_word_length': 2.2167791141097095, 'pattern_counts': {'":': 5, '<': 6, '<?xml version=': 0, '>': 10, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '1', 'dir': 'jupyter-structured-clean-dedup', 'id': '525430', 'license': 'MIT', 'n_tokens_mistral': 4364, 'n_tokens_neox': 4395, 'n_words': 553, 'path': 'part03-neural-network/time-series/UCI_HAR_torch_cnn.ipynb', 'repo_name': 'edgardeng/machine-learning-pytorch', 'size': '10965'}