datawhalechina PyTorch tutorial

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

from sklearn.metrics import confusion_matrix


def compute_confusion_matrix(true_labels_list, pred_labels_list, labels):
    true_labels_list = np.asarray(true_labels_list)
    pred_labels_list = np.asarray(pred_labels_list)
    matrix = confusion_matrix(true_labels_list,
                              pred_labels_list,
                              labels=labels)
    return matrix

1
2
3
4
5
6

def compute_oa(matrix):
    """
    计算总体准确率
    OA=(TP+TN)/(TP+TN+FP+FN)
    """
    return np.trace(matrix) / np.sum(matrix)

1
2
3
4
5
6
7

def compute_aa(matrix):
    """
    计算平均精度
    AA=(TP/(TP+FN)+TN/(FP+TN))/2
    """
    # 将混淆矩阵的对角线元素取出，并且对于混淆矩阵进行列求和，用对角线元素除以求和后的结果，最后对结果计算求出平均值
    return np.mean(np.diag(matrix) / np.sum(matrix, axis=1))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

def compute_kappa(matrix):
    """
    计算kappa系数
    """
    po = compute_oa(matrix)
    pe = 0
    for i in range(len(matrix)):
        pe += np.sum(matrix[i]) * np.sum(matrix[:, i])
    pe = pe / np.sum(matrix) ** 2
    return (po - pe) / (1 - pe)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# 反向传播
x = torch.ones(2, 2, requires_grad=True)
out2 = x.sum()
out2.backward()
print(x.grad)

# 注意若不将梯度清零，grad 是累加的
out3 = x.sum()
# x.grad.data.zero_()
out3.backward()
print(x.grad)

1
2
3
4
5
6

import os
import numpy as np
import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader
import torch.optim as optimizer

1
2
3
4
5
6
7
8

# 批次的大小
batch_size = 16

# 优化器的学习率
lr = 1e-4

# 模型迭代次数
max_epochs = 100

1
2

device = torch.device("cuda:1" if torch.cuda.is_available() else "cpu")
# 指明调用的GPU为1号

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

# 自定义 Dataset
import os
import pandas as pd
import matplotlib.pyplot as plt
from torch.utils.data import Dataset, DataLoader
from torchvision.io import read_image


class MyDataset(Dataset):
    def __init__(self, annotations_file, img_dir, transform=None, target_transform=None):
        """
        Args:
            annotations_file (string):带注释的csv文件的路径。
            img_dir (string):包含所有图像的目录。
            transform(可调用，可选):应用于样本的转换。
            target_transform(可调用，可选):应用于标签的转换。
        """
        self.img_labels = pd.read_csv(annotations_file)
        self.img_dir = img_dir
        self.transform = transform
        self.target_transform = target_transform

    def __len__(self):
        return len(self.img_labels)

    def __getitem__(self, idx):
        img_path = os.path.join(self.img_dir, self.img_labels.iloc[idx, 0])

        # 读取图片
        image = read_image(img_path)
        label = self.img_labels.iloc[idx, 1]

        # 根据传入的参数对图像和标签进行转换
        if self.transform:
            image = self.transform(image)
        if self.target_transform:
            label = self.target_transform(label)
        return image, label

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

# 批量读入数据
train_loader = DataLoader(
    train_data,
    batch_size=batch_size,
    num_workers=4,    # 用于读取数据的进程数
    shuffle=True,
    drop_last=True    # 忽略掉剩余达不到批次数的样本
)
val_loader = DataLoader(
    val_data,
    batch_size=batch_size,
    num_workers=4,
    shuffle=False    # 验证集的数据无需打乱
)

# 利用迭代器获得一个样本和对应标签
images, labels = next(iter(val_loader))
print(images.shape)
# 可视化图片，transpose调整颜色通道顺序
plt.imshow(images[0].transpose(1,2,0))
plt.show()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

import torch
from torch import nn


class MLP(nn.Module):
    def __init__(self, **kwargs):
        super(MLP, self).__init__(**kwargs)
        self.hidden = nn.Linear(784, 256)
        self.act = nn.ReLU()
        self.output = nn.Linear(256,10)

    # 定义模型的正向传播（前向计算），即如何根据输入 x 计算返回所需要的模型输出
    # 这里先经过一层隐藏层，然后再使用激活函数，最后经过线性层
    def forward(self, x):
        o = self.act(self.hidden(x))
        return self.output(o)

1
2
3
4

X = torch.rand(2, 784)    # 生成随机的输入张量
net = MLP()    # 实例化模型
print(net)    # 打印模型结构
net(X)    # 正向传播

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

import torch
from torch import nn


class MyLayer(nn.Module):
    def __init__(self, **kwargs):
        super(MyLayer, self).__init__(**kwargs)

    # 在这里定义层的计算
    def forward(self, x):
        # 该层所做计算为输入减去均值
        return x - x.mean()


tmp_in = torch.tensor([1, 2, 3, 4, 5], dtype=torch.float)
# 实例化该层，然后做前向计算
layer = MyLayer()
layer(tmp_in)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

import torch
from torch import nn


class MyListDense(nn.Module):
    def __init__(self):
        super(MyListDense, self).__init__()
        # 模型参数列表，初始化3个服从标准正态分布的4x4随机张量
        self.params = nn.ParameterList(
            [nn.Parameter(torch.randn(4, 4)) for i in range(3)]
        )
        # 额外添加一个服从标准正态分布的4x1随机张量
        self.params.append(nn.Parameter(torch.randn(4, 1)))

    def forward(self, x):
        for i in range(len(self.params)):
            # 将输入与参数一一进行矩阵乘法
            x = torch.mm(x, self.params[i])
        return x

net = MyListDense()
print(net)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

import torch
from torch import nn


# 卷积运算（二维互相关）
def corr2d(X, K):
    # 输入 X 和卷积核 K
    h, w = K.shape
    X, K = X.float(), K.float()
    # 初始化卷积运算后的返回张量
    Y = torch.zeros((X.shape[0] - h + 1, X.shape[1] - w + 1))
    for i in range(Y.shape[0]):
        for j in range(Y.shape[1]):
            # 与卷积核对应位置元素相乘，最后求和
            Y[i, j] = (X[i: i + h, j: j + w] * K).sum()
    return Y

# 二维卷积层
class Conv2D(nn.Module):
    def __init__(self, kernel_size):
        super(Conv2D, self).__init__()
        # 随机初始化卷积核参数和标准偏差
        self.weight = nn.Parameter(torch.randn(kernel_size))
        self.bias = nn.Parameter(torch.randn(1))

    def forward(self, x):
        return corr2d(x, self.weight) + self.bias

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

import torch
from torch import nn


# 默认使用最大池化
def pool2d(X, pool_size, mode='max'):
    p_h, p_w = pool_size
    Y = torch.zeros((X.shape[0] - p_h + 1, X.shape[1] - p_w + 1))
    for i in range(Y.shape[0]):
        for j in range(Y.shape[1]):
            if mode == 'max':
                Y[i, j] = X[i: i + p_h, j: j + p_w].max()
            elif mode == 'avg':
                Y[i, j] = X[i: i + p_h, j: j + p_w].mean()
    return Y

X = torch.tensor([[0, 1, 2], [3, 4, 5], [6, 7, 8]], dtype=torch.float)
pool2d(X, (2, 2))
pool2d(X, (2, 2), 'avg')

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

import torch
import torch.nn as nn
import torch.nn.functional as F


class Net(nn.Module):

    def __init__(self):
        super(Net, self).__init__()
        # 输入图像的 channel 为 1
        # 输出 channel 为 6
        # 卷积核大小为 5x5
        self.conv1 = nn.Conv2d(1, 6, 5)
        self.conv2 = nn.Conv2d(6, 16, 5)
        # an affine operation: y = Wx + b
        self.fc1 = nn.Linear(16 * 5 * 5, 120)
        self.fc2 = nn.Linear(120, 84)
        self.fc3 = nn.Linear(84, 10)

    def forward(self, x):
        # 2x2 Max pooling
        x = F.max_pool2d(F.relu(self.conv1(x)), (2, 2))
        x = F.max_pool2d(F.relu(self.conv2(x)), 2)
        x = x.view(-1, self.num_flat_features(x))
        x = F.relu(self.fc1(x))
        x = F.relu(self.fc2(x))
        x = self.fc3(x)
        return x

    def num_flat_features(self, x):
        size = x.size()[1:]  # 忽略批处理维度
        num_features = 1
        for s in size:
            num_features *= s
        return num_features


net = Net()
print(net)
# 打印模型参数
params = list(net.parameters())
print(len(params))
print(params[0].size())  # conv1 的权重

# 随机生成测试输入
input = torch.randn(1, 1, 32, 32)
out = net(input)
print(out)
# 清零所有参数的梯度缓存，然后进行随机梯度的反向传播
net.zero_grad()
out.backward(torch.randn(1, 10))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

import torch
import torch.nn as nn

conv = nn.Conv2d(1,3,3)
linear = nn.Linear(10,1)

# 对conv进行kaiming初始化
torch.nn.init.kaiming_normal_(conv.weight.data)
conv.weight.data

# 对linear进行常数初始化
torch.nn.init.constant_(linear.weight.data,0.3)
linear.weight.data

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

import os
import torch


# 随机初始化权重，服从正态分布
weight = torch.randn((2, 2), requires_grad=True)
>>> tensor([[-0.3077, -0.1808],
            [-0.7462, -1.5556]])

# 设置梯度为全1矩阵
weight.grad = torch.ones((2, 2))
>>> tensor([[1., 1.],
            [1., 1.]])

1
2
3
4
5
6
7
8

# 实例化优化器
optimizer = torch.optim.SGD([weight], lr=0.1, momentum=0.9)
optimizer.step()
# 执行一步优化后更新的权重与梯度
>>> tensor([[-0.4077, -0.2808],
            [-0.8462, -1.6556]])
>>> tensor([[1., 1.],
            [1., 1.]])

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

# 进行梯度清零，权重的梯度变为全 0
optimizer.zero_grad()

# 输出参数
optimizer.param_groups
>>> [{'params': [tensor([[-0.4077, -0.2808],
            [-0.8462, -1.6556]], requires_grad=True)], 'lr': 0.1, 'momentum': 0.9, 'dampening': 0, 'weight_decay': 0, 'nesterov': False}]

# 查看参数位置
id(optimizer.param_groups[0]['params'][0])
>>> 1841923407424

id(weight)
>>> 1841923407424
# 注意到optimizer中的权重和weight变量在内存中的地址是一样的

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# 添加参数：weight2
weight2 = torch.randn((3, 3), requires_grad=True)
optimizer.add_param_group({"params": weight2, 'lr': 0.0001, 'nesterov': True})
# 查看现有的参数
optimizer.param_groups
>>> [{'params': [tensor([[-0.4077, -0.2808],
        [-0.8462, -1.6556]], requires_grad=True)], 'lr': 0.1, 'momentum': 0.9, 'dampening': 0, 'weight_decay': 0, 'nesterov': False}, {'params': [tensor([[ 0.4539, -2.1901, -0.6662],
        [ 0.6630, -1.5178, -0.8708],
        [-2.0222,  1.4573,  0.8657]], requires_grad=True)], 'lr': 0.0001, 'nesterov': True, 'momentum': 0.9, 'dampening': 0, 'weight_decay': 0}]

# 查看当前状态信息
opt_state_dict = optimizer.state_dict()
>>> {'state': {0: {'momentum_buffer': tensor([[1., 1.],
        [1., 1.]])}}, 'param_groups': [{'lr': 0.1, 'momentum': 0.9, 'dampening': 0, 'weight_decay': 0, 'nesterov': False, 'params': [0]}, {'lr': 0.0001, 'nesterov': True, 'momentum': 0.9, 'dampening': 0, 'weight_decay': 0, 'params': [1]}]}

1
2
3
4
5
6

# 进行5次step操作
for _ in range(50):
    optimizer.step()
optimizer.state_dict()
>>> {'state': {0: {'momentum_buffer': tensor([[0.0052, 0.0052],
        [0.0052, 0.0052]])}}, 'param_groups': [{'lr': 0.1, 'momentum': 0.9, 'dampening': 0, 'weight_decay': 0, 'nesterov': False, 'params': [0]}, {'lr': 0.0001, 'nesterov': True, 'momentum': 0.9, 'dampening': 0, 'weight_decay': 0, 'params': [1]}]}

1
2
3
4
5
6
7
8
9

# 保存参数信息
torch.save(
    optimizer.state_dict(),
    os.path.join(r"D:\python_project\pytorch_learning", "optimizer_state_dict.pkl")
)

# 加载参数信息
state_dict = torch.load(r"D:\python_project\pytorch_learning\optimizer_state_dict.pkl")
optimizer.load_state_dict(state_dict)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# 给网络不同的层赋予不同的优化器参数
from torchvision.models import resnet18


net = resnet18()
optimizer = optim.SGD(
    [
        {'params':net.fc.parameters()},
        {'params':net.layer4[0].conv1.parameters(), 'lr':1e-2}
    ],
    lr=1e-5
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

from collections import OrderedDict


class MySequential(nn.Module):
    def __init__(self, *args):
        super(MySequential, self).__init__()
        # 如果传入的是一个 OrderedDict
        if len(args) == 1 and isinstance(args[0], OrderedDict):
            for key, module in args[0].items():
                self.add_module(key, module)
        # 传入的是列表形式的 Module
        else:
            for idx, module in enumerate(args):
                self.add_module(str(idx), module)

    def forward(self, input):
        # self._modules 返回一个 OrderedDict，保证会按照成员添加时的顺序遍历
        for module in self._modules.values():
            input = module(input)  # 简单串联
        return input

1
2
3
4
5
6
7
8
9

import torch.nn as nn


net = nn.Sequential(
    nn.Linear(784, 256),
    nn.ReLU(),
    nn.Linear(256, 10),
)
print(net)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

import torch.nn as nn


class model(nn.Module):
    def __init__(self, ...):
        super().__init__()
        self.modulelist = nn.ModuleList([nn.Linear(784, 256), nn.ReLU()])
        self.modulelist.append(nn.Linear(256, 10))    # 类似List的append操作
        ...

    def forward(self, x):
        for layer in self.modulelist:
            x = layer(x)
        return x

1
2
3
4
5

nn.ModuleDict({
    'linear': nn.Linear(784, 256),
    'act': nn.ReLU(),
})
net['output'] = nn.Linear(256, 10)    # 类似字典的添加操作

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

# 两次卷积
import torch
import torch.nn as nn
import torch.nn.functional as F


class DoubleConv(nn.Module):
    """(convolution => [BN] => ReLU) * 2"""

    def __init__(self, in_channels, out_channels, mid_channels=None):
        super().__init__()
        if not mid_channels:
            mid_channels = out_channels
        self.double_conv = nn.Sequential(
            nn.Conv2d(in_channels, mid_channels, kernel_size=3, padding=1, bias=False),
            nn.BatchNorm2d(mid_channels),
            nn.ReLU(inplace=True),
            nn.Conv2d(mid_channels, out_channels, kernel_size=3, padding=1, bias=False),
            nn.BatchNorm2d(out_channels),
            nn.ReLU(inplace=True)
        )

    def forward(self, x):
        return self.double_conv(x)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

# 下采样
class Down(nn.Module):
    """Downscaling with maxpool then double conv"""

    def __init__(self, in_channels, out_channels):
        super().__init__()
        self.maxpool_conv = nn.Sequential(
            nn.MaxPool2d(2),
            DoubleConv(in_channels, out_channels)
        )

    def forward(self, x):
        return self.maxpool_conv(x)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

# 上采样
class Up(nn.Module):
    """Upscaling then double conv"""

    def __init__(self, in_channels, out_channels, bilinear=False):
        super().__init__()

        # if bilinear, use the normal convolutions to reduce the number of channels
        if bilinear:
            self.up = nn.Upsample(scale_factor=2, mode='bilinear', align_corners=True)
            self.conv = DoubleConv(in_channels, out_channels, in_channels // 2)
        else:
            self.up = nn.ConvTranspose2d(in_channels, in_channels // 2, kernel_size=2, stride=2)
            self.conv = DoubleConv(in_channels, out_channels)

    def forward(self, x1, x2):
        x1 = self.up(x1)
        # input is CHW
        diffY = x2.size()[2] - x1.size()[2]
        diffX = x2.size()[3] - x1.size()[3]

        x1 = F.pad(x1, [diffX // 2, diffX - diffX // 2,
                        diffY // 2, diffY - diffY // 2])
        # if you have padding issues, see
        # https://github.com/HaiyongJiang/U-Net-Pytorch-Unstructured-Buggy/commit/0e854509c2cea854e247a9c615f175f76fbb2e3a
        # https://github.com/xiaopeng-liao/Pytorch-UNet/commit/8ebac70e633bac59fc22bb5195e513d5832fb3bd
        x = torch.cat([x2, x1], dim=1)
        return self.conv(x)

1
2
3
4
5
6
7

class OutConv(nn.Module):
    def __init__(self, in_channels, out_channels):
        super(OutConv, self).__init__()
        self.conv = nn.Conv2d(in_channels, out_channels, kernel_size=1)

    def forward(self, x):
        return self.conv(x)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

class DiceLoss(nn.Module):
    def __init__(self,weight=None,size_average=True):
        super(DiceLoss,self).__init__()

    def forward(self,inputs,targets,smooth=1):
        inputs = F.sigmoid(inputs)
        inputs = inputs.view(-1)
        targets = targets.view(-1)
        intersection = (inputs * targets).sum()
        dice = (2.*intersection + smooth)/(inputs.sum() + targets.sum() + smooth)
        return 1 - dice


criterion = DiceLoss()
loss = criterion(input,targets)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# 自定义 scheduler
def adjust_learning_rate(optimizer, epoch):
    # 每训练 30 轮后，学习率下降为原来的 10%
    lr = args.lr * (0.1 ** (epoch // 30))
    for param_group in optimizer.param_groups:
        param_group['lr'] = lr


optimizer = torch.optim.SGD(model.parameters(),lr = args.lr,momentum = 0.9)
for epoch in range(10):
    train(...)
    validate(...)
    # 每一轮训练结束后，调整学习率
    adjust_learning_rate(optimizer,epoch)

1
2

self.model = models.resnet50(pretrained=False)
self.model.load_state_dict(torch.load('./model/resnet50-19c8e357.pth'))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

# demo.py
import argparse


# 创建 ArgumentParser() 对象
parser = argparse.ArgumentParser()

# 添加参数
parser.add_argument(
    '-o', '--output',
    action='store_true',
    help="shows output"
)
parser.add_argument(
    '--lr',
    type=float,
    default=3e-5,
    help='select the learning rate, default=1e-3'
)
parser.add_argument(
    '--batch_size',
    type=int,
    required=True,
    help='input batch size'
)
# action=store_true 会将 output 参数记录为 True
# type 规定了参数的格式
# default 规定了默认值
# required 规定可选或者必需

# 使用 parse_args() 解析函数
args = parser.parse_args()


if args.output:
    print("This is some output")
    print(f"learning rate:{args.lr} ")

1
2

This is some output
learning rate: 3e-4

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

# config.py
import argparse


def get_options(parser=argparse.ArgumentParser()):
    parser.add_argument('--batch_size', type=int, default=4, help='input batch size, default=64')

    parser.add_argument('--niters', type=int, default=10, help='number of epochs to train for, default=10')

    parser.add_argument('--lr', type=float, default=3e-5, help='select the learning rate, default=1e-3')

    parser.add_argument('--seed', type=int, default=118, help="random seed")

    parser.add_argument('--cuda', action='store_true', default=True, help='enables cuda')

    parser.add_argument('--checkpoint_path',type=str,default='',
    help='Path to load a previous trained model if not empty (default empty)')

    opt = parser.parse_args()

    if opt.output:
        print(f'batch_size: {opt.batch_size}')
        print(f'learning rate : {opt.lr}')
        print(f'manual_seed: {opt.seed}')
        print(f'cuda enable: {opt.cuda}')
        print(f'checkpoint_path: {opt.checkpoint_path}')
    return opt

if __name__ == '__main__':
    opt = get_options()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

# train.py
import config


opt = config.get_options()

batch_size = opt.batch_size
lr = opt.lr
manual_seed = opt.seed
niters = opt.niters
checkpoint_path = opt.checkpoint_path

# 随机数的设置，保证复现结果
def set_seed(seed):
    torch.manual_seed(seed)
    torch.cuda.manual_seed_all(seed)
    random.seed(seed)
    np.random.seed(seed)
    torch.backends.cudnn.benchmark = False
    torch.backends.cudnn.deterministic = True


if __name__ == '__main__':
    set_seed(manual_seed)
    for epoch in range(niters):
        ......