聊聊基於pytorch實現Resnet對本地資料集的訓練問題

from dataset import data_dataloader    #電腦本地寫的讀取資料的函數
from torch import nn                   #匯入pytorch的nn模組
from torch import optim                #匯入pytorch的optim模組
from network import Res_net            #電腦本地寫的網路框架的函數
from train import train                #電腦本地寫的訓練函數

def main():
    # 以下是通過Data_dataloader函數輸入為：資料的路徑，資料模式，資料大小，batch的大小，有幾線並用 （把dataset和Dataloader功能合在了一起）
    train_loader = data_dataloader(data_path='./data', mode='train', size=64, batch_size=24, num_workers=4)
    val_loader = data_dataloader(data_path='./data', mode='val', size=64, batch_size=24, num_workers=2)
    test_loader = data_dataloader(data_path='./data', mode='test', size=64, batch_size=24, num_workers=2)
    # 以下是超引數的定義
    lr = 1e-4           #學習率
    epochs = 10         #訓練輪次
    model = Res_net(2)  # resnet網路
    optimizer = optim.Adam(model.parameters(), lr=lr)  # 優化器
    loss_function = nn.CrossEntropyLoss()  # 損失函數
    # 訓練以及驗證測試函數
    train(model=model, optimizer=optimizer, loss_function=loss_function, train_data=train_loader, val_data=val_loader,test_data= test_loader, epochs=epochs)
if __name__ == '__main__':
    main()

import torch
import os,glob
import random
import csv
from torch.utils.data import Dataset
from PIL import Image
from torchvision import transforms
from torch.utils.data import DataLoader

# 第一部分：通過三個步驟得到輸出的tensor型別的資料
class Dataset_self(Dataset):                    #如果是nn.moduel 則是編寫網路模型框架，這裡需要繼承的是dataset的資料，所以括號中的是Dataset
    #第一步：初始化
    def __init__(self,root,mode,resize,):       #root是檔案根目錄，mode是選擇什麼樣的資料集，resize是影象重新調整大小
        super(Dataset_self, self).__init__()
        self.resize = resize
        self.root = root
        self.name_label = {}       #建立一個字典來儲存每個檔案的標籤
        #首先得到標籤相對於的字典（標籤和名稱一一對應）
        for name in sorted(os.listdir(os.path.join(root))):     #排序並且用列表的形式開啟資料夾
            if not os.path.isdir(os.path.join(root,name)):      #不是資料夾就不需要讀取
                continue
            self.name_label[name] = len(self.name_label.keys())  #每個檔案的名字為name_Label字典中有多少對鍵值對的個數
        #print(self.name_label)
        self.image,self.label = self.make_csv('images.csv')       #編寫一共函數來讀取圖片和標籤的路徑
        #在得到image和label的基礎上對圖片資料進行一共劃分  （注意：如果需要交叉驗證就不需要驗證集，只劃分為訓練集和測試集）
        if mode == 'train':
            self.image ,self.label= self.image[:int(0.6*len(self.image))],self.label[:int(0.6*len(self.label))]
        if mode == 'val':
            self.image ,self.label= self.image[int(0.6*len(self.image)):int(0.8*len(self.image))],self.label[int(0.6*len(self.label)):int(0.8*len(self.label))]
        if mode == 'test':
            self.image ,self.label= self.image[int(0.8*len(self.image)):],self.label[int(0.8*len(self.label)):]
    # 獲得圖片和標籤的函數
    def make_csv(self,filename):
        if not os.path.exists(os.path.join(self.root,filename)):  #如果不存在彙總的目錄就新建一個
            images = []
            for image in self.name_label.keys():                            # 讓image到name_label中的每個檔案中去讀取圖片
                images += glob.glob(os.path.join(self.root,image,'*jpg'))   #加* 貪婪搜尋關於jpg的所有檔案
            #print('長度為：{}，第二張圖片為：{}'.format(len(images),images[1]))
            random.shuffle(images)                                         #把images列表中的資料洗牌
            # images[0]: ./dataants382971067_0bfd33afe0.jpg
            with open(os.path.join(self.root,filename),mode='w',newline='') as f :  #建立檔案
                writer = csv.writer(f)
                for image in images:
                    name = image.split(os.sep)[-2]  #得到與圖片相對應的標籤
                    label = self.name_label[name]
                    writer.writerow([image,label])  #寫入檔案  第一行：./dataants382971067_0bfd33afe0.jpg,0
        images,labels = [],[]
        with open(os.path.join(self.root,filename)) as f:   #讀取檔案
            reader = csv.reader(f)
            for row in reader:
                image, label = row
                label = int(label)
                images.append(image)
                labels.append(label)
        assert len(images) == len(labels)   #類似if語句，只有兩者長度一致才繼續執行，否則報錯
        return images,labels                #返回所有！！是所有的圖片和標籤（此處的圖片不是圖片資料本身，而是它的檔案目錄）
    #第二步：得到圖片資料的長度（標籤資料長度與圖片一致）
    def __len__(self):
        return len(self.image)
    #第三步：讀取圖片和標籤，並輸出
    def __getitem__(self, item):   # 單張返回張量的影象與標籤
        image,label = self.image[item],self.label[item]      #得到單張圖片和相應的標籤（此處都是image都是檔案目錄）
        image = Image.open(image).convert('RGB')             #得到圖片資料
        #使用transform對圖片進行處理以及變成tensor型別資料
        transf = transforms.Compose([transforms.Resize((int(self.resize),int(self.resize))),
                                     transforms.RandomRotation(15),
                                     transforms.CenterCrop(self.resize),
                                     transforms.ToTensor(),  #先變成tensor型別資料，然後在進行下面的標準化
                                     ])
        image = transf(image)
        label = torch.tensor(label)   #把圖片標籤也變成tensor型別
        return image,label
#第二部分：使用pytorch自帶的DataLoader函數批次得到圖片資料
def data_dataloader(data_path,mode,size,batch_size,num_workers):   #用一個函數載入上訴的資料，data_path、mode和size分別是以上定義的Dataset_self(）中的引數，batch_size是一次性輸出多少張影象，num_worker是同時處理幾張影象
    dataset = Dataset_self(data_path,mode,size)
    dataloader = DataLoader(dataset,batch_size,num_workers)  #使用pytorch中的dataloader函數得到資料
    return dataloader
#測試
def main():
    test = Dataset_self('./data','train',64)
if __name__ == '__main__':
    main()

import torch
from torch import nn

# 先寫好resnet的block塊
class Res_block(nn.Module):
    def __init__(self,in_num,out_num,stride):
        super(Res_block, self).__init__()
        self.cov1 = nn.Conv2d(in_num,out_num,(3,3),stride=stride,padding=1)    #(3,3)  padding=1 則影象大小不變，stride為幾影象就縮小几倍，能極大減少引數
        self.bn1 = nn.BatchNorm2d(out_num)
        self.cov2 = nn.Conv2d(out_num,out_num,(3,3),padding=1)
        self.bn2 = nn.BatchNorm2d(out_num)
        self.extra = nn.Sequential(
                nn.Conv2d(in_num,out_num,(1,1),stride=stride),
                nn.BatchNorm2d(out_num)
            )   #使得輸入前後的影象資料大小是一致的
        self.relu = nn.ReLU()
    def forward(self,x):
        out = self.relu(self.bn1(self.cov1(x)))
        out = self.relu(self.bn2(self.cov2(out)))
        out = self.extra(x) + out
        return out
class Res_net(nn.Module):
    def __init__(self,num_class):
        super(Res_net, self).__init__()
        self.init = nn.Sequential(
            nn.Conv2d(3,16,(3,3)),
            nn.BatchNorm2d(16)
        )   #預處理
        self.bn1 = Res_block(16,32,2)
        self.bn2 = Res_block(32,64,2)
        self.bn3 = Res_block(64,128,2)
        self.bn4 = Res_block(128,256,2)
        self.fl = nn.Flatten()
        self.linear1 = nn.Linear(8192,10)
        self.linear2 = nn.Linear(10,num_class)
        out = self.relu(self.init(x))
        #print('inint:',out.shape)
        out = self.bn1(out)
        #print('bn1:', out.shape)
        out = self.bn2(out)
        #print('bn2:', out.shape)
        out = self.bn3(out)
        #print('bn3:', out.shape)
        out = self.fl(out)
        #print('flatten:', out.shape)
        out = self.relu(self.linear1(out))
        #print('linear1:', out.shape)
        out = self.relu(self.linear2(out))
        #print('linear2:', out.shape)
#測試
def main():
    x = torch.randn(2,3,64,64)
    net = Res_net(2)
    out = net(x)
    print(out.shape)
if __name__ == '__main__':
    main()

import torch
from torch import optim
from torch.utils.data import DataLoader
from dataset import Dataset_self
from network import Res_net
from torch import nn
from matplotlib import pyplot as plt
import numpy as np

def evaluate(model,loader):   #計算每次訓練後的準確率
    correct = 0
    total = len(loader.dataset)
    for x,y in loader:
        logits = model(x)
        pred = logits.argmax(dim=1)     #得到logits中分類值（要麼是[1,0]要麼是[0,1]表示分成兩個類別）
        correct += torch.eq(pred,y).sum().float().item()        #用logits和標籤label想比較得到分類正確的個數
    return correct/total
#把訓練的過程定義為一個函數
def train(model,optimizer,loss_function,train_data,val_data,test_data,epochs):  #輸入：網路架構，優化器，損失函數，訓練集，驗證集，測試集，輪次
    best_acc,best_epoch =0,0      #輸出驗證集中準確率最高的輪次和準確率
    train_list,val_List = [],[]   # 建立列表儲存每一次的acc，用來最後的畫圖
    for epoch in range(epochs):
            print('============第{}輪============'.format(epoch + 1))
            for steps,(x,y) in enumerate(train_data):   #  for x,y in train_data
                logits = model(x)                   #資料放入網路中
                loss = loss_function(logits,y)      #得到損失值
                optimizer.zero_grad()               #優化器先清零，不然會疊加上次的數值
                loss.backward()                     #後向傳播
                optimizer.step()
            train_acc =evaluate(model,train_data)
            train_list.append(train_acc)
            print('train_acc',train_acc)
            #if epoch % 1 == 2:   #這裡可以設定每兩次訓練驗證一次
            val_acc = evaluate(model,val_data)
            print('val_acc=',val_acc)
            val_List.append((val_acc))
            if val_acc > best_acc:  #判斷每次在驗證集上的準確率是否為最大
                best_epoch = epoch
                best_acc = val_acc
                torch.save(model.state_dict(),'best.mdl')   #儲存驗證集上最大的準確率
    print('===========================分割線===========================')
    print('best acc:',best_acc,'best_epoch:',best_epoch)
    #在測試集上檢測訓練好後模型的準確率
    model.load_state_dict((torch.load('best.mdl')))
    print('detect the test data!')
    test_acc = evaluate(model,test_data)
    print('test_acc:',test_acc)
    train_list_file = np.array(train_list)
    np.save('train_list.npy',train_list_file)
    val_list_file = np.array(val_List)
    np.save('val_list.npy',val_list_file)
    #畫圖
    x_label = range(1,len(val_List)+1)
    plt.plot(x_label,train_list,'bo',label='train acc')
    plt.plot(x_label,val_List,'b',label='validation acc')
    plt.title('train and validation accuracy')
    plt.xlabel('epochs')
    plt.legend()
    plt.show()
#測試
def main():
    train_dataset = Dataset_self('./data', 'train', 64)
    vali_dataset = Dataset_self('./data', 'val', 64)
    test_dataset = Dataset_self('./data', 'test', 64)
    train_loaber = DataLoader(train_dataset, 24, num_workers=4)
    val_loaber = DataLoader(vali_dataset, 24, num_workers=2)
    test_loaber = DataLoader(test_dataset, 24, num_workers=2)
    lr = 1e-4
    epochs = 5
    model = Res_net(2)
    optimizer = optim.Adam(model.parameters(), lr=lr)
    criteon = nn.CrossEntropyLoss()
    train(model,optimizer,criteon,train_loaber,val_loaber,test_loaber,epochs)
if __name__ == '__main__':
    main()

from network import Res_net
import torch
from PIL import Image
import torchvision

#匯入圖片
img = '1.jpg'
img =Image.open(img)
tf = torchvision.transforms.Compose([torchvision.transforms.Resize((64,64)),torchvision.transforms.ToTensor()])
img = tf(img)
image = torch.reshape(img,(1,3,64,64))
#載入模型
net = Res_net(2)
net.load_state_dict(torch.load('best.mdl'))
with torch.no_grad():
    out = net(image)
#確定分類
class_cl =out.argmax(dim=1)
class_num = class_cl.numpy()
if class_num == 0:
    print('這張照片是螞蟻')
else:
    print('這張照片是蜜蜂')