PyTorch中图像多分类的实现

from torchvision import datasets
import torchvision.transforms as transforms
import os

path2data="./data"
# 如果数据路径不存在，则创建
if not os.path.exists(path2data):
    os.mkdir(path2data)
    
# 定义数据转换
data_transformer = transforms.Compose([transforms.ToTensor()])
    
# 从datasets库中导入STL10数据集，并指定数据集的路径、分割方式、是否下载以及数据转换器
train_ds=datasets.STL10(path2data, split='train',download=True,transform=data_transformer)

# 打印数据形状
print(train_ds.data.shape)

import collections

# 获取标签
y_train=[y for _,y in train_ds]

# 统计标签
counter_train=collections.Counter(y_train)
print(counter_train)

# 加载数据
test0_ds=datasets.STL10(path2data, split='test', download=True,transform=data_transformer)
# 打印数据形状
print(test0_ds.data.shape)

# 导入StratifiedShuffleSplit模块
from sklearn.model_selection import StratifiedShuffleSplit

# 创建StratifiedShuffleSplit对象，设置分割次数为1，测试集大小为0.2，随机种子为0
sss = StratifiedShuffleSplit(n_splits=1, test_size=0.2, random_state=0)

# 获取test0_ds的索引
indices=list(range(len(test0_ds)))

# 获取test0_ds的标签
y_test0=[y for _,y in test0_ds]

# 对索引和标签进行分割
for test_index, val_index in sss.split(indices, y_test0):
    # 打印测试集和验证集的索引
    print("test:", test_index, "val:", val_index)
    # 打印测试集和验证集的大小
    print(len(val_index),len(test_index))

# 从torch.utils.data中导入Subset类
from torch.utils.data import Subset

# 从test0_ds中选取val_index索引的子集，赋值给val_ds
val_ds=Subset(test0_ds,val_index)
# 从test0_ds中选取test_index索引的子集，赋值给test_ds
test_ds=Subset(test0_ds,test_index)

import collections
import numpy as np

# 获取标签
y_test=[y for _,y in test_ds]
y_val=[y for _,y in val_ds]

# 统计测试集和验证集的标签数量
counter_test=collections.Counter(y_test)
counter_val=collections.Counter(y_val)

# 打印测试集和验证集的标签数量
print(counter_test)
print(counter_val)

from torchvision import utils
import matplotlib.pyplot as plt
import numpy as np
%matplotlib inline

# 设置随机种子为0
np.random.seed(0)

# 定义一个函数，用于显示图像
def show(img,y=None,color=True):
    # 将图像转换为numpy数组
    npimg = img.numpy()
    # 将图像的维度从（C,H,W）转换为（H,W,C）
    npimg_tr=np.transpose(npimg, (1,2,0))
    # 显示图像
    plt.imshow(npimg_tr)
    # 如果有标签，则显示标签
    if y is not None:
        plt.title("label: "+str(y))
        
# 定义网格大小
grid_size=4
# 随机生成4个索引
rnd_inds=np.random.randint(0,len(train_ds),grid_size)
print("image indices:",rnd_inds)

# 从训练集中获取这4个索引对应的图像和标签
x_grid=[train_ds[i][0] for i in rnd_inds]
y_grid=[train_ds[i][1] for i in rnd_inds]

# 将这4个图像拼接成一个网格
x_grid=utils.make_grid(x_grid, nrow=4, padding=2)
print(x_grid.shape)

# 调用helper函数显示网格
plt.figure(figsize=(10,10))
show(x_grid,y_grid)

# 设置随机种子为0 
np.random.seed(0)

# 设置网格大小
grid_size=4
# 从验证数据集中随机选择grid_size个索引
rnd_inds=np.random.randint(0,len(val_ds),grid_size)
print("image indices:",rnd_inds)

# 从验证数据集中选择对应的图像
x_grid=[val_ds[i][0] for i in rnd_inds]
# 从验证数据集中选择对应的标签
y_grid=[val_ds[i][1] for i in rnd_inds]

# 将图像排列成网格
x_grid=utils.make_grid(x_grid, nrow=4, padding=2)
print(x_grid.shape)

# 调用辅助函数
plt.figure(figsize=(10,10))
# 显示网格图像和标签
show(x_grid,y_grid)

import numpy as np

# 计算训练集中每个样本的RGB均值
meanRGB=[np.mean(x.numpy(),axis=(1,2)) for x,_ in train_ds]  
# 计算训练集中每个样本的RGB标准差
stdRGB=[np.std(x.numpy(),axis=(1,2)) for x,_ in train_ds]  

meanR=np.mean([m[0] for m in meanRGB])  # 计算所有样本的R通道均值的平均值
meanG=np.mean([m[1] for m in meanRGB])  
meanB=np.mean([m[2] for m in meanRGB])  

stdR=np.mean([s[0] for s in stdRGB])  # 计算所有样本的R通道标准差的平均值
stdG=np.mean([s[1] for s in stdRGB])  
stdB=np.mean([s[2] for s in stdRGB])  

print(meanR,meanG,meanB)  # 打印R、G、B通道的均值
print(stdR,stdG,stdB)  # 打印R、G、B通道的标准差

# 定义训练数据的转换器
train_transformer = transforms.Compose([
    # 随机水平翻转，翻转概率为0.5
    transforms.RandomHorizontalFlip(p=0.5),  
    # 随机垂直翻转，翻转概率为0.5
    transforms.RandomVerticalFlip(p=0.5),  
    # 将图像转换为张量
    transforms.ToTensor(),
    # 对图像进行归一化，均值和标准差分别为meanR, meanG, meanB和stdR, stdG, stdB
    transforms.Normalize([meanR, meanG, meanB], [stdR, stdG, stdB])])
                 

# 定义测试数据的转换器
test0_transformer = transforms.Compose([
    # 将图像转换为张量
    transforms.ToTensor(),
    # 对图像进行归一化，均值和标准差分别为meanR, meanG, meanB和stdR, stdG, stdB
    transforms.Normalize([meanR, meanG, meanB], [stdR, stdG, stdB]),
    ])   

# 将训练数据集的转换器赋值给训练数据集的transform属性
train_ds.transform=train_transformer
# 将测试数据集的转换器赋值给测试数据集的transform属性
test0_ds.transform=test0_transformer

import torch
import numpy as np
import matplotlib.pyplot as plt

# 设置随机种子
np.random.seed(0)
torch.manual_seed(0)

# 定义网格大小
grid_size=4

# 从训练数据集中随机选择grid_size个样本的索引
rnd_inds=np.random.randint(0,len(train_ds),grid_size)
print("image indices:",rnd_inds)

# 根据索引从训练数据集中获取对应的样本
x_grid=[train_ds[i][0] for i in rnd_inds]
y_grid=[train_ds[i][1] for i in rnd_inds]

# 将样本转换为网格形式
x_grid=utils.make_grid(x_grid, nrow=4, padding=2)
print(x_grid.shape)

# 创建一个10x10的图像
plt.figure(figsize=(10,10))
# 显示网格和对应的标签
show(x_grid,y_grid)

from torch.utils.data import DataLoader

# 创建训练数据集的DataLoader，batch_size为32，shuffle为True，表示每次迭代时都会打乱数据集
train_dl = DataLoader(train_ds, batch_size=32, shuffle=True)
# 创建验证数据集的DataLoader，batch_size为64，shuffle为False，表示每次迭代时不会打乱数据集
val_dl = DataLoader(val_ds, batch_size=64, shuffle=False)  

# 遍历训练数据集
for x, y in train_dl:
    # 打印x的形状
    print(x.shape)
    # 打印y的形状
    print(y.shape)
    # 跳出循环
    break

# 遍历val_dl中的每个元素，x和y分别表示输入和标签
for x, y in val_dl:
    # 打印输入的形状
    print(x.shape)
    # 打印标签的形状
    print(y.shape)
    # 退出循环
    break

# 从datasets库中导入FashionMNIST数据集，并将其设置为训练集
fashion_train=datasets.FashionMNIST(path2data, train=True, download=True)

from torchvision import models
import torch

# 创建一个resnet18模型，pretrained参数设置为False，表示不使用预训练的权重
model_resnet18 = models.resnet18(pretrained=False)
# 打印模型ResNet18
print(model_resnet18)

from torch import nn
# 定义类别数量
num_classes=10
# 获取模型ResNet18的全连接层输入特征数量
num_ftrs = model_resnet18.fc.in_features 
# 将全连接层替换为新的全连接层，输出特征数量为类别数量
model_resnet18.fc = nn.Linear(num_ftrs, num_classes)

# 定义设备为GPU
device = torch.device("cuda:0")
# 将模型移动到GPU上
model_resnet18.to(device)

from torchsummary import summary

# 打印模型结构，输入大小为(3, 224, 224)，即3个通道，224x224大小的图像
summary(model_resnet18, input_size=(3, 224, 224))

# 遍历模型ResNet18的参数
for w in model_resnet18.parameters():
    # 将参数转换为CPU数据
    w=w.data.cpu()
    # 打印参数的形状
    print(w.shape)
    break

# 计算参数的最小值
min_w=torch.min(w)
# 计算w1，其中w1 = (-1/(2*min_w))*w + 0.5 
w1 = (-1/(2*min_w))*w + 0.5 
# 打印w1的最小值和最大值
print(torch.min(w1).item(),torch.max(w1).item())

# 计算网格大小
grid_size=len(w1)
# 生成网格
x_grid=[w1[i] for i in range(grid_size)]
x_grid=utils.make_grid(x_grid, nrow=8, padding=1)
print(x_grid.shape)

# 创建一个5x5的图像
plt.figure(figsize=(5,5))
show(x_grid)

from torchvision import models
import torch

# 加载预训练的resnet18模型
resnet18_pretrained = models.resnet18(pretrained=True)

# 定义分类的类别数
num_classes=10
# 获取resnet18模型的最后一层全连接层的输入特征数
num_ftrs = resnet18_pretrained.fc.in_features
# 将最后一层全连接层替换为新的全连接层，新的全连接层的输出特征数为num_classes
resnet18_pretrained.fc = nn.Linear(num_ftrs, num_classes)

# 定义设备为cuda:0
device = torch.device("cuda:0")
# 将模型移动到cuda:0设备上
resnet18_pretrained.to(device) 

# 遍历resnet18_pretrained的参数
for w in resnet18_pretrained.parameters():
    # 将参数转换为cpu格式
    w=w.data.cpu()
    print(w.shape)
    break

# 计算w的最小值
min_w=torch.min(w)
# 计算w1，其中w1=(-1/(2*min_w))*w + 0.5
w1 = (-1/(2*min_w))*w + 0.5 
# 打印w1的最小值和最大值
print(torch.min(w1).item(),torch.max(w1).item())

# 计算w1的网格大小
grid_size=len(w1)
# 将w1转换为网格形式
x_grid=[w1[i] for i in range(grid_size)]
x_grid=utils.make_grid(x_grid, nrow=8, padding=1)
print(x_grid.shape)

# 创建一个5x5的图像
plt.figure(figsize=(5,5))
show(x_grid)

torch.manual_seed(0)

# 定义输入数据的维度
n,c=4,5
# 生成随机输入数据，并设置requires_grad=True，表示需要计算梯度
y = torch.randn(n, c, requires_grad=True)
# 打印输入数据的形状
print(y.shape)

# 定义交叉熵损失函数，reduction参数设置为"sum"，表示将所有样本的损失相加
loss_func = nn.CrossEntropyLoss(reduction="sum")
# 生成随机目标数据，表示每个样本的类别
target = torch.randint(c,size=(n,))
# 打印目标数据的形状
print(target.shape)

# 计算损失
loss = loss_func(y, target)
# 打印损失值
print(loss.item())

# 反向传播，计算梯度
loss.backward()
# 打印输出y的值
print (y.data)

from torch import optim
# 定义优化器，使用Adam优化算法，优化model_resnet18的参数，学习率为1e-4
opt = optim.Adam(model_resnet18.parameters(), lr=1e-4)
# 定义一个函数，用于获取优化器的学习率
def get_lr(opt):
    # 遍历优化器的参数组
    for param_group in opt.param_groups:
        # 返回学习率
        return param_group['lr']

# 调用函数，获取当前学习率
current_lr=get_lr(opt)
# 打印当前学习率
print('current lr={}'.format(current_lr))

from torch.optim.lr_scheduler import CosineAnnealingLR

# 创建学习率调度器，T_max表示周期长度，eta_min表示最小学习率
lr_scheduler = CosineAnnealingLR(opt,T_max=2,eta_min=1e-5)
# 定义一个空列表lrs
lrs=[]
# 循环10次
for i in range(10):
    # 调用lr_scheduler.step()方法
    lr_scheduler.step()
    # 调用get_lr()方法获取当前学习率
    lr=get_lr(opt)
    # 打印当前epoch和对应的学习率
    print("epoch %s, lr: %.1e" %(i,lr))
    # 将当前学习率添加到列表lrs中
    lrs.append(lr)
# 绘制lrs列表中的数据
plt.plot(lrs)

# 定义一个函数metrics_batch，用于计算预测结果和目标之间的正确率
def metrics_batch(output, target):
    # 将输出结果的最大值所在的索引作为预测结果
    pred = output.argmax(dim=1, keepdim=True)
    
    # 计算预测结果和目标之间的正确率
    corrects=pred.eq(target.view_as(pred)).sum().item()
    # 返回正确率
    return corrects
def loss_batch(loss_func, output, target, opt=None):
    # 计算batch的损失
    loss = loss_func(output, target)
    # 计算batch的评估指标
    metric_b = metrics_batch(output,target)
    
    # 如果有优化器，则进行反向传播和参数更新
    if opt is not None:
        opt.zero_grad()
        loss.backward()
        opt.step()

    # 返回损失和评估指标
    return loss.item(), metric_b

device = torch.device("cuda")

# 定义一个函数loss_epoch，用于计算模型在数据集上的损失
def loss_epoch(model,loss_func,dataset_dl,sanity_check=False,opt=None):
    # 初始化运行损失和运行指标
    running_loss=0.0
    running_metric=0.0
    # 获取数据集的长度
    len_data=len(dataset_dl.dataset)

    # 遍历数据集
    for xb, yb in dataset_dl:
        # 将数据移动到GPU上
        xb=xb.to(device)
        yb=yb.to(device)
        
        # 获取模型输出
        output=model(xb)
        
        # 计算当前批次的损失和指标
        loss_b,metric_b=loss_batch(loss_func, output, yb, opt)
        
        # 累加损失和指标
        running_loss+=loss_b
        
        if metric_b is not None:
            running_metric+=metric_b

        # 如果是sanity_check模式，则只计算一个批次
        if sanity_check is True:
            break
    
    # 计算平均损失和指标
    loss=running_loss/float(len_data)
    
    metric=running_metric/float(len_data)
    
    # 返回平均损失和指标
    return loss, metric

def train_val(model, params):
    # 获取参数
    num_epochs=params["num_epochs"]
    loss_func=params["loss_func"]
    opt=params["optimizer"]
    train_dl=params["train_dl"]
    val_dl=params["val_dl"]
    sanity_check=params["sanity_check"]
    lr_scheduler=params["lr_scheduler"]
    path2weights=params["path2weights"]
    
    # 初始化损失和指标历史记录
    loss_history={
        "train": [],
        "val": [],
    }
    
    metric_history={
        "train": [],
        "val": [],
    }
    
    # 复制模型参数
    best_model_wts = copy.deepcopy(model.state_dict())
    
    # 初始化最佳损失
    best_loss=float('inf')
    
    # 遍历每个epoch
    for epoch in range(num_epochs):
        
        # 获取当前学习率
        current_lr=get_lr(opt)
        print('Epoch {}/{}, current lr={}'.format(epoch, num_epochs - 1, current_lr))
        
        # 训练模型
        model.train()
        train_loss, train_metric=loss_epoch(model,loss_func,train_dl,sanity_check,opt)

        # 记录训练损失和指标
        loss_history["train"].append(train_loss)
        metric_history["train"].append(train_metric)
        
        # 评估模型
        model.eval()
        with torch.no_grad():
            val_loss, val_metric=loss_epoch(model,loss_func,val_dl,sanity_check)
        
        # 如果验证损失小于最佳损失，则更新最佳损失和最佳模型参数
        if val_loss < best_loss:
            best_loss = val_loss
            best_model_wts = copy.deepcopy(model.state_dict())
            
            # 将最佳模型参数保存到本地文件
            torch.save(model.state_dict(), path2weights)
            print("Copied best model weights!")
        
        # 记录验证损失和指标
        loss_history["val"].append(val_loss)
        metric_history["val"].append(val_metric)
        
        # 更新学习率
        lr_scheduler.step()

        # 打印训练损失、验证损失和准确率
        print("train loss: %.6f, dev loss: %.6f, accuracy: %.2f" %(train_loss,val_loss,100*val_metric))
        print("-"*10) 

    # 加载最佳模型参数
    model.load_state_dict(best_model_wts)
        
    # 返回模型、损失历史和指标历史
    return model, loss_history, metric_history

import copy

# 定义交叉熵损失函数，reduction参数设置为"sum"，表示将所有样本的损失相加
loss_func = nn.CrossEntropyLoss(reduction="sum")
# 定义Adam优化器，优化模型参数，学习率为1e-4
opt = optim.Adam(model_resnet18.parameters(), lr=1e-4)
# 定义余弦退火学习率调度器，T_max参数设置为5，eta_min参数设置为1e-6
lr_scheduler = CosineAnnealingLR(opt,T_max=5,eta_min=1e-6)

# 定义训练参数字典
params_train={
 "num_epochs": 3,  # 训练轮数
 "optimizer": opt,  # 优化器
 "loss_func": loss_func,  # 损失函数
 "train_dl": train_dl,  # 训练数据集
 "val_dl": val_dl,  # 验证数据集
 "sanity_check": False,  # 是否进行sanity check
 "lr_scheduler": lr_scheduler,  # 学习率调度器
 "path2weights": "./models/resnet18.pt",  # 模型权重保存路径
}

# 训练和验证模型
model_resnet18,loss_hist,metric_hist=train_val(model_resnet18,params_train)

# 获取训练参数中的训练轮数
num_epochs=params_train["num_epochs"]

# 绘制训练和验证损失曲线
plt.title("Train-Val Loss")
plt.plot(range(1,num_epochs+1),loss_hist["train"],label="train")
plt.plot(range(1,num_epochs+1),loss_hist["val"],label="val")
plt.ylabel("Loss")
plt.xlabel("Training Epochs")
plt.legend()
plt.show()

# 绘制训练和验证准确率曲线
plt.title("Train-Val Accuracy")
plt.plot(range(1,num_epochs+1),metric_hist["train"],label="train")
plt.plot(range(1,num_epochs+1),metric_hist["val"],label="val")
plt.ylabel("Accuracy")
plt.xlabel("Training Epochs")
plt.legend()
plt.show()

import copy

# 定义损失函数，使用交叉熵损失，并设置reduction为sum
loss_func = nn.CrossEntropyLoss(reduction="sum")
# 定义优化器，使用Adam优化器，并设置学习率为1e-4
opt = optim.Adam(resnet18_pretrained.parameters(), lr=1e-4)
# 定义学习率调度器，使用余弦退火调度器，设置最大周期为5，最小学习率为1e-6
lr_scheduler = CosineAnnealingLR(opt,T_max=5,eta_min=1e-6)

# 定义训练参数
params_train={
 "num_epochs": 3,  # 设置训练周期为3
 "optimizer": opt,  # 设置优化器
 "loss_func": loss_func,  # 设置损失函数
 "train_dl": train_dl,  # 设置训练数据集
 "val_dl": val_dl,  # 设置验证数据集
 "sanity_check": False,  # 设置是否进行sanity check
 "lr_scheduler": lr_scheduler,  # 设置学习率调度器
 "path2weights": "./models/resnet18_pretrained.pt",  # 设置权重保存路径
}

# 调用train_val函数进行训练和验证，并返回训练后的模型、损失历史和指标历史
resnet18_pretrained,loss_hist,metric_hist=train_val(resnet18_pretrained,params_train)

# 获取训练参数中的训练轮数
num_epochs=params_train["num_epochs"]

# 绘制训练和验证损失曲线
plt.title("Train-Val Loss")
plt.plot(range(1,num_epochs+1),loss_hist["train"],label="train")
plt.plot(range(1,num_epochs+1),loss_hist["val"],label="val")
plt.ylabel("Loss")
plt.xlabel("Training Epochs")
plt.legend()
plt.show()

# 绘制训练和验证准确率曲线
plt.title("Train-Val Accuracy")
plt.plot(range(1,num_epochs+1),metric_hist["train"],label="train")
plt.plot(range(1,num_epochs+1),metric_hist["val"],label="val")
plt.ylabel("Accuracy")
plt.xlabel("Training Epochs")
plt.legend()
plt.show()