动手学机器学习——网络模型基础

E:\MiniConda	# MiniConda安装路径
E:\MiniConda\Scripts
E:\MiniConda\Library\bin

E:\CUDA\Computing\bin
E:\CUDA\Computing\lib\x64
E:\CUDA\Computing\extras\CUPTI\lib64
E:\CUDA\Computing\include

import numpy as np
import matplotlib.pyplot as plt
from matplotlib.pyplot import MultipleLocator
import math

if __name__ == '__main__':
    # 生成等差数列
    X = np.linspace(0.01, 1, 256, endpoint=True)
    print(X)

    # 计算信息量
    aoi = [-math.log(x) for x in X]
    print(aoi)

    # 实例化横坐间隔
    x_major_locator = MultipleLocator(0.5)

    ax = plt.gca()
    # 设置横坐标间隔
    ax.xaxis.set_major_locator(x_major_locator)
    # 设置纵坐标长度
    plt.ylim(0, 5.0)
    plt.plot(X, aoi)

    plt.show()

import torch
from d2l import torch as d2l


x = torch.arange(-8.0, 8.0, 0.1, requires_grad=True)
y = torch.relu(x)
d2l.plot(x.detach(), y.detach(), 'x', 'relu(x)', figsize=(5, 2.5))
d2l.plt.show()

import torch
from d2l import torch as d2l


x = torch.arange(-8.0, 8.0, 0.1, requires_grad=True)
y = torch.relu(x)
y.backward(torch.ones_like(x), retain_graph=True)
d2l.plot(x.detach(), x.grad, 'x', 'grad of relu', figsize=(5, 2.5))
d2l.plt.show()

import torch
from d2l import torch as d2l


x = torch.arange(-8.0, 8.0, 0.1, requires_grad=True)

y = torch.sigmoid(x)
d2l.plot(x.detach(), y.detach(), 'x', 'sigmoid(x)', figsize=(5, 2.5))
d2l.plt.show()

import torch
from d2l import torch as d2l


x = torch.arange(-8.0, 8.0, 0.1, requires_grad=True)

y = torch.sigmoid(x)
y.backward(torch.ones_like(x), retain_graph=True)
d2l.plot(x.detach(), x.grad, 'x', 'grad of sigmoid', figsize=(5, 2.5))
d2l.plt.show()

import torch
from d2l import torch as d2l


x = torch.arange(-8.0, 8.0, 0.1, requires_grad=True)
y = torch.tanh(x)
d2l.plot(x.detach(), y.detach(), 'x', 'relu(x)', figsize=(5, 2.5))
d2l.plt.show()

import torch
from d2l import torch as d2l

x = torch.arange(-8.0, 8.0, 0.1, requires_grad=True)
y = torch.tanh(x)
y.backward(torch.ones_like(x), retain_graph=True)
d2l.plot(x.detach(), x.grad, 'x', 'grad of relu', figsize=(5, 2.5))
d2l.plt.show()

import torch
from torch import nn
from d2l import torch as d2l

"""获取数据
"""
batch_size = 256
train_iter, test_iter = d2l.load_data_fashion_mnist(batch_size)

"""准备参数
"""

# 输出为784维的特征，输出为10项的类别，隐藏层中神经元的个数设置为256
num_inputs, num_outputs, num_hiddens = 784, 10, 256


# W1为隐藏层权重，784 * 256维
W1 = nn.Parameter(torch.randn(
    num_inputs, num_hiddens, requires_grad=True) * 0.01)

# b1为隐藏层偏置值，为隐藏层神经元个数256
b1 = nn.Parameter(torch.zeros(num_hiddens, requires_grad=True))

# W2为输出层权重，256 * 10维
W2 = nn.Parameter(torch.randn(
    num_hiddens, num_outputs, requires_grad=True) * 0.01)

# b2为输出层偏置值，为类别个数10
b2 = nn.Parameter(torch.zeros(num_outputs, requires_grad=True))

params = [W1, b1, W2, b2]

"""激活函数
"""

def relu(x):
    a = torch.zeros_like(x)
    return torch.max(x, a)

"""模型
"""
def net(x):
    x = x.reshape((-1, num_inputs))
    H = relu(x@W1 + b1)  # @代表矩阵乘法
    return (H@W2 + b2)

"""损失函数
"""
loss = nn.CrossEntropyLoss(reduction='none')

"""训练
"""

num_epochs, lr = 10, 0.1
updater = torch.optim.SGD(params, lr=lr)

d2l.train_ch3(net, train_iter, test_iter, loss, num_epochs, updater)
d2l.predict_ch3(net, test_iter)
d2l.plt.show()

# TODO:定义模型
net = nn.Sequential(
    nn.Flatten(),
    nn.Linear(28*28, 256),
    nn.ReLU(),
    nn.Linear(256, 10),
)

# TODO:初始化权重


def init_Weights(m):
    if type(m) == nn.Linear:
        nn.init.normal_(m.weight, std=0.01)

        
        

net.apply(init_Weights)

# TODO:获取数据
batch_size, lr, num_epochs = 256, 0.1, 10
train_iter, test_iter = d2l.load_data_fashion_mnist(batch_size)
loss = nn.CrossEntropyLoss(reduction='none')
trainer = torch.optim.SGD(net.parameters(), lr=lr)

d2l.train_ch3(net, train_iter, test_iter, loss, num_epochs, trainer)
d2l.predict_ch3(net, test_iter)

import math
import numpy as np
import torch
from torch import nn
from d2l import torch as d2l


def dropout_layer(X, dropout):
    """dropout layer

    Args:
        X (Tensor): 某一层的输出
        dropout (float): dropout的概率p

    Returns:
        Tensor: dropout后的输出，丢弃的为0，未丢弃的进行了缩放即除以(1-p)
    """
    assert 0 <= dropout <= 1

    if dropout == 1:
        return torch.zeros_like(X)
    if dropout == 0:
        return X
    # 制造一个滤波随机生成一个值为0，1间均匀随机分布的张量
    # 大于dropout的取1，小于取0
    mask = (torch.rand(X.shape) > dropout).float()
    # 不使用Numpy中X[mask] = 0的方式来选择是因为对GPU而言直接乘法比选择快
    return mask * X / (1.0 - dropout)


# 定义模型
num_input, num_output, num_hidden1, num_hidden2 = 784, 10, 256, 256
dropout1, dropout2 = 0.2, 0.5


class Net(nn.Module):
    def __init__(self, num_input, num_output, num_hidden1, num_hidden2, is_training=True):
        super(Net, self).__init__()
        self.is_training = is_training
        self.num_input = num_input
        self.lin1 = nn.Linear(num_input, num_hidden1)
        self.lin2 = nn.Linear(num_hidden1, num_hidden2)
        self.lin3 = nn.Linear(num_hidden2, num_output)
        self.relu = nn.ReLU()

    def forward(self, X):
        H1 = self.relu(self.lin1(X.reshape((-1, self.num_input))))
        # 训练时使用dropout
        if self.is_training:
            H1 = dropout_layer(H1, dropout1)
        H2 = self.relu(self.lin2(H1))
        if self.is_training:
            H2 = dropout_layer(H2, dropout2)
        out = self.lin3(H2)
        return out


net = Net(num_input, num_output, num_hidden1, num_hidden2)


# 训练
num_epochs, lr, batch_size = 10, 0.5, 256
loss = nn.CrossEntropyLoss(reduction='none')
train_iter, test_iter = d2l.load_data_fashion_mnist(batch_size)
trainer = torch.optim.SGD(net.parameters(), lr=lr)
d2l.train_ch3(net, train_iter, test_iter, loss, num_epochs, trainer)
d2l.plt.show()

# 简洁实现
# 模型定义
net = nn.Sequential(
    nn.Flatten(),
    nn.Linear(num_input, num_hidden1),
    nn.ReLU(),
    nn.Dropout(dropout1),
    nn.Linear(num_hidden1, num_hidden1),
    nn.ReLU(),
    nn.Dropout(dropout2),
    nn.Linear(num_hidden2, num_output)
)


def init_weights(m):
    if type(m) == nn.Linear:
        nn.init.normal_(m.weight, std=0.01)


net.apply(init_weights)


# 训练
num_epochs, lr, batch_size = 10, 0.5, 256
loss = nn.CrossEntropyLoss(reduction='none')
train_iter, test_iter = d2l.load_data_fashion_mnist(batch_size)
trainer = torch.optim.SGD(net.parameters(), lr=lr)
d2l.train_ch3(net, train_iter, test_iter, loss, num_epochs, trainer)
d2l.plt.show()

import torch
from d2l import torch as d2l


x = torch.arange(-8.0, 8.0, 1.0, requires_grad=True)
y = torch.sigmoid(x)
y.backward(torch.ones_like(x))

d2l.plot(x.detach().numpy(), [y.detach().numpy(), x.grad.numpy()], legend=[
         'sigmoid', 'gradient'], figsize=(4.5, 2.5))
d2l.plt.show()

import pandas as pd
import numpy as np
import torch
from torch import nn
from d2l import torch as d2l

DATA_HUB = dict()
DATA_URL = 'http://d2l-data.s3-accelerate.amazonaws.com/'


DATA_HUB['kaggle_house_train'] = (
    DATA_URL + 'kaggle_house_pred_train.csv',
    '585e9cc93e70b39160e7921475f9bcd7d31219ce'
)

DATA_HUB['kaggle_house_test'] = (
    DATA_URL + 'kaggle_house_pred_test.csv',
    'fa19780a7b011d9b009e8bff8e99922a8ee2eb90'
)

# 下载数据集
train_data = pd.read_csv(d2l.download('kaggle_house_train'))
test_data = pd.read_csv(d2l.download('kaggle_house_test'))


print(train_data.shape)
print(test_data.shape)

print(train_data.iloc[0:4, [0, 1, 2, 3, -3, -2, -1]])

# 删掉第一列id
# 并且取出训练数据的最后一列
all_feature = pd.concat((train_data.iloc[:, 1:-1], test_data.iloc[:, 1:]))
print(all_feature.shape)
print(all_feature.iloc[0:4, :])

numeric_features = all_feature.dtypes[all_feature.dtypes != 'object'].index

# 将数值均值变成0，方差变成1
all_feature[numeric_features] = all_feature[numeric_features].apply(
    lambda x: (x - x.mean()) / (x.std())
)

# 将NA数据替换为均值
all_feature[numeric_features] = all_feature[numeric_features].fillna(0)

# 使用one-hot处理离散值
all_feature = pd.get_dummies(all_feature, dummy_na=True)
# 使用one-hot编码处理离散值会使特征变大
print(all_feature.shape)

# 从pandas格式种提取NumPy格式，并将其转换为Tensor
n_train = train_data.shape[0]
train_features = torch.tensor(
    all_feature[:n_train].values, dtype=torch.float32)
test_features = torch.tensor(
    all_feature[n_train:].values, dtype=torch.float32)
train_labels = torch.tensor(
    train_data.SalePrice.values.reshape(-1, 1), dtype=torch.float32)


# 训练

loss = nn.MSELoss()
in_features = train_features.shape[1]

# 网络


def get_net():
    net = nn.Sequential(
        nn.Linear(in_features, 1)
    )
    return net

# 误差
# 为了防止更大的值将会带来更大的误差，我们通常使用
# (y - y')/y
# 为了防止除法，对上述式子再取对数


def log_rmse(net, features, labels):
    # 对于输出为INF的值，将其替换为1
    clipped_preds = torch.clamp(net(features), 1, float('inf'))
    rmse = torch.sqrt(loss(torch.log(clipped_preds), torch.log(labels)))
    return rmse.item()

# 训练


def train(net, train_features, train_labels, test_features, test_labels,
          num_epochs, learning_rate, weight_decay, batch_size):
    train_ls, test_ls = [], []
    train_iter = d2l.load_array((train_features, train_labels), batch_size)
    # 这里使用的是Adam优化算法，其对初始学习率不那么敏感
    optimizer = torch.optim.Adam(net.parameters(),
                                 lr=learning_rate,
                                 weight_decay=weight_decay)
    for epoch in range(num_epochs):
        for X, y in train_iter:
            optimizer.zero_grad()
            l = loss(net(X), y)
            l.backward()
            optimizer.step()
        train_ls.append(log_rmse(net, train_features, train_labels))
        if test_labels is not None:
            test_ls.append(log_rmse(net, test_features, test_labels))
    return train_ls, test_ls


# K折交叉验证
def get_k_fold_data(k, i, x, y):
    assert k > 1
    fold_size = x.shape[0] // k
    x_train, y_train = None, None
    for j in range(k):
        idx = slice(j*fold_size, (j + 1) * fold_size)
        x_part, y_part = x[idx, :], y[idx]
        if j == i:
            x_valid, y_valid = x_part, y_part
        elif x_train is None:
            x_train, y_train = x_part, y_part
        else:
            x_train = torch.cat([x_train, x_train], 0)
            y_train = torch.cat([y_train, y_train], 0)
    return x_train, y_train, x_valid, y_valid


def k_fold(k, x_train, y_train, num_epochs, learning_rate, weight_decay, batch_size):
    train_l_sum, valid_l_sum = 0, 0
    for i in range(k):
        data = get_k_fold_data(k, i, x_train, y_train)
        net = get_net()
        train_ls, valid_ls = train(
            net, *data, num_epochs, learning_rate, weight_decay, batch_size)
        train_l_sum += train_ls[-1]
        valid_l_sum += valid_ls[-1]
        if i == 0:
            d2l.plot(list(range(1, num_epochs + 1)), [train_ls, valid_ls],
                     xlabel='epoch', ylabel='rmse', xlim=[1, num_epochs],
                     legend=['train', 'valid'], yscale='log')
        print(f'折{i + 1}，训练log rmse{float(train_ls[-1]):f}, '
              f'验证log rmse{float(valid_ls[-1]):f}')
    return train_l_sum / k, valid_l_sum / k


## k, num_epochs, lr, weight_decay, batch_size = 5, 100, 5, 0, 64
k, num_epochs, lr, weight_decay, batch_size = 5, 150, 5, 0.08, 64
train_l, valid_l = k_fold(k, train_features, train_labels, num_epochs, lr,
                          weight_decay, batch_size)
print(f'{k}-折验证: 平均训练log rmse: {float(train_l):f}, '
      f'平均验证log rmse: {float(valid_l):f}')

d2l.plt.show()

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


class MLP(nn.Module):

    # 初始化
    def __init__(self):
        super().__init__()
        self.hidden = nn.Linear(20, 256)    # 隐藏层
        self.out = nn.Linear(256, 10)   # 输出层

    # 前向传播
    def forward(self, x):
        return self.out(F.relu(self.hidden(x)))


x = torch.rand(2, 20)
net = MLP()
print(net(x))

# 顺序快


class MySequential(nn.Module):
    def __init__(self, *args):
        super().__init__()
        for idx, module in enumerate(args):
            # 此处使用了父类中的熟悉_modules，其类型为OrderedDict
            # 且第kay指定类型str，value指定类型为Module
            self._modules[str(idx)] = module

    def forward(self, x):
        # 使用OrderedDict保证了调用层级的顺序
        for block in self._modules.values():
            x = block(x)
        return x

# 前向传播时执行代码

class FixedHiddenMLP(nn.Module):
    def __init__(self):
        super.__init__()
        # 创建一个不会被更新的参数
        self.rand_weight = torch.rand((20, 20), requires_grad=False)
        self.linear = nn.Linear(20, 20)

    def forward(self, x):
        x = self.linear(x)
        # 进入ReLU之前先对线性层结果进行了一次固定参数的线性变换
        x = F.relu(torch.mm(x, self.rand_weight) + 1)
        x = self.linear(x)
        # 控制流
        while x.abs().sum > 1:
            x /= 2
        return x.sum()

# 混合搭配


class NestMLP(nn.Module):
    def __init__(self):
        super().__init__()
        self.net = nn.Sequential(nn.Linear(20, 64), nn.ReLU(),
                                 nn.Linear(64, 32), nn.ReLU())
        self.linear = nn.Linear(32, 16)

    def forward(self, X):
        return self.linear(self.net(X))


chimera = nn.Sequential(NestMLP(), nn.Linear(16, 20), FixedHiddenMLP())
print(chimera(x))

import torch
from torch import nn

net = nn.Sequential(nn.Linear(4, 8), nn.ReLU(), nn.Linear(8, 1))

x = torch.rand(size=(2, 4))
print(net(x))

# 提取第二个神经网络中的参数
# 下标为Sequential中输入参数的位置
print(net[2].state_dict())

# TODO:访问指定参数
print(type(net[2].bias))
print(net[2].bias)
print(net[2].bias.data)
print(net.state_dict()['2.bias'].data)

# 访问参数梯度
print(net[2].weight.grad)

# 访问全部参数
print(*[(name, params.shape) for name, params in net[0].named_parameters()])
print(*[(name, params.shape) for name, params in net.named_parameters()])

# TODO:访问嵌套块的参数


def block1():
    return nn.Sequential(nn.Linear(4, 8), nn.ReLU(),
                         nn.Linear(8, 4), nn.ReLU())


def block2():
    net = nn.Sequential()
    for i in range(4):
        # 在这里嵌套
        net.add_module(f'block {i}', block1())
    return net


rgnet = nn.Sequential(block2(), nn.Linear(4, 1))
rgnet(X)
print(rgnet)

import torch
from torch import nn

net = nn.Sequential(nn.Linear(4, 8), nn.ReLU(), nn.Linear(8, 1))

x = torch.rand(size=(2, 4))
print(net(x))

# TODO:内置初始化器初始化参数
def init_normal(m):
    if type(m) == nn.Linear:
        nn.init.normal_(m.weight, mean=0, std=0.01)
        nn.init.zeros_(m.bias)


net.apply(init_normal)
net[0].weight.data[0], net[0].bias.data[0]

# TODO:使用常数初始化参数


def init_constant(m):
    if type(m) == nn.Linear:
        nn.init.constant_(m.weight, 1)
        nn.init.zeros_(m.bias)


net.apply(init_constant)
net[0].weight.data[0], net[0].bias.data[0]


# TODO:不同层使用不同的初始化方案
def init_xavier(m):
    if type(m) == nn.Linear:
        nn.init.xavier_uniform_(m.weight)


def init_42(m):
    if type(m) == nn.Linear:
        nn.init.constant_(m.weight, 42)


net[0].apply(init_xavier)
net[2].apply(init_42)
print(net[0].weight.data[0])
print(net[2].weight.data)

# TODO:自定义初始化


def my_init(m):
    if type(m) == nn.Linear:
        print("Init", *[(name, param.shape)
                        for name, param in m.named_parameters()][0])
        nn.init.uniform_(m.weight, -10, 10)
        m.weight.data *= m.weight.data.abs() >= 5


net.apply(my_init)
print(net[0].weight[:2])

# TODO:共享参数的层

# 我们需要给共享层一个名称，以便可以引用它的参数
shared = nn.Linear(8, 8)
net = nn.Sequential(nn.Linear(4, 8), nn.ReLU(),
                    shared, nn.ReLU(),
                    shared, nn.ReLU(),
                    nn.Linear(8, 1))
net(X)
# 检查参数是否相同
print(net[2].weight.data[0] == net[4].weight.data[0])
net[2].weight.data[0, 0] = 100
# 确保它们实际上是同一个对象，而不只是有相同的值
print(net[2].weight.data[0] == net[4].weight.data[0])

import tensorflow as tf

# TODO: 定义网络
net = tf.keras.models.Sequential([
    tf.keras.layers.Dense(256, activation=tf.nn.relu),
    tf.keras.layers.Dense(10),
])

# TODO: 访问网络参数
print([net.layers[i].get_weights() for i in range(len(net.layers))])

X = tf.random.uniform((2, 20))
net(X)
print([w.shape for w in net.get_weights()])

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


class CenteredLayer(nn.Module):
    def __init__(self):
        super().__init__()
       
    def forward(self, X):
        return x - x.mean()

layer = CenteredLayer()
print(layer(torch.FloatTensor([1, 2, 3, 4, 5])))

net = nn.Sequential(nn.Linear(8, 128), CenteredLayer())

Y = net(torch.rand(4,8))
print(Y.mean())

class MyLinear(nn.Module):
    def __init__(self, in_units, units):
        super().__init__()
        self.weight = nn.Parameter(torch.randn(in_units, units))
        self.bias = nn.Parameter(torch.randn(units,))
    def forward(self, X):
        linear = torch.matmul(X, self.weight.data) + self.bias.data
        return F.relu(linear)

linear = MyLinear(5, 3)
print(linear.weight)

print(linear(torch.rand(2, 5)))

net = nn.Sequential(MyLinear(64, 8), MyLinear(8, 1))
print(net(torch.rand(2, 64)))

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

x = torch.arange(4)
torch.save(x, 'x-file')

x2 = torch.load('x-file')
print(x2)

y = torch.zeros(4)
torch.save([x, y],'x-files')
x2, y2 = torch.load('x-files')
(x2, y2)

mydict = {'x': x, 'y': y}
npx.save('mydict', mydict)
mydict2 = npx.load('mydict')
mydict2

class MLP(nn.Module):
    def __init__(self):
        super().__init__()
        self.hidden = nn.Linear(20, 256)
        self.output = nn.Linear(256, 10)

    def forward(self, x):
        return self.output(F.relu(self.hidden(x)))

net = MLP()
X = torch.randn(size=(2, 20))
Y = net(X)

torch.save(net.state_dict(), 'mlp.params')

clone = MLP()
clone.load_state_dict(torch.load('mlp.params'))
print(clone.eval())

Y_clone = clone(X)
Y_clone == Y

import torch
from torch import nn

gpu_num = torch.cuda.device_count()
print(gpu_num)
for i in range(gpu_num):
    print(torch.device('cuda' + ':' + str(i)))
print(torch.device('cpu'))

def try_gpu(i=0):  #@save
    """如果存在，则返回gpu(i)，否则返回cpu()"""
    if torch.cuda.device_count() >= i + 1:
        return torch.device(f'cuda:{i}')
    return torch.device('cpu')

def try_all_gpus():  #@save
    """返回所有可用的GPU，如果没有GPU，则返回[cpu(),]"""
    devices = [torch.device(f'cuda:{i}')
             for i in range(torch.cuda.device_count())]
    return devices if devices else [torch.device('cpu')]

try_gpu(), try_gpu(10), try_all_gpus()

X = torch([1,2,3])
print(X.device)

X = torch.ones(2, 3, device=try_gpu())
print(X)

Z = X.cuda(1)
print(X)
print(Z)

X.cuda(0) is Z

net = nn.Sequential(nn.Linear(3,1))
net = net.to(device=try_gpu())

print(net(X))

print(net[0].weight.data.device)

import torch
from torch import nn
from d2l import torch as d2l


# TODO: 二维互相关运算
def corr2d(X, K):
    """二维互相关运算

    Args:
        X (Tensor(w,h)): 原始图片
        K (Tensor(x,y)): 卷积核

    Returns:
        Tensor(w-x+ 1, h-y+1): 卷积后的图片
    """

    h, w = K.shape
    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


X = torch.tensor([[0.0, 1.0, 2.0], [3.0, 4.0, 5.0], [6.0, 7.0, 8.0]])
K = torch.tensor([[0.0, 1.0], [2.0, 3.0]])
print(corr2d(X, K))

class Conv2D(nn.Module):
    def __init__(self, kernel_size):
        super().__init__()
        self.weight = nn.Parameter(torch.rand(kernel_size))
        self.bias = nn.Parameter(torch.zeros(1))

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

# TODO: 模拟一个边缘检测任务
X = torch.ones((6, 8))
X[:, 2:6] = 0
K = torch.tensor([[1.0, -1.0]])
Y = corr2d(X, K)

# TODO: 训练二维卷积层
# 具有1个输出通道和形状为(1,2)的卷积核
conv2d = nn.Conv2d(1, 1, kernel_size=(1, 2), bias=False)

# 二维卷积层使用4维输入输出：(批大小, 通道, 高度, 宽度)
X = X.reshape((1, 1, 6, 8))
Y = Y.reshape((1, 1, 6, 7))
lr = 3e-2  # 学习率

for i in range(10):
    Y_hat = conv2d(X)
    # 使用平方误差作为损失
    l = (Y_hat - Y) ** 2
    conv2d.zero_grad()
    l.sum().backward()
    # 迭代卷积核
    conv2d.weight.data[:] -= lr * conv2d.weight.grad
    if (i + 1) % 2 == 0:
        print(f'epoch {i+1}, loss {l.sum():.3f}')

print(conv2d.weight.data.reshape((1, 2)))

import torch
from torch import nn

# TODO: 定义一个卷积运算


def comp_conv2d(conv2d, X):
    X = X.reshape((1, 1) + X.shape)
    Y = conv2d(X)
    return Y.reshape(Y.shape[2:])


# TODO: 构建卷积网络，设置填充为左右各1像素
conv2d = nn.Conv2d(1, 1, kernel_size=3, padding=1)
X = torch.rand(size=(8, 8))
print(comp_conv2d(conv2d, X).shape)

# TODO: 将步幅设置为2
conv2d = nn.Conv2d(1, 1, kernel_size=3, padding=1, stride=2)
print(comp_conv2d(conv2d, X).shape)

# TODO: 更复杂的情况
conv2d = nn.Conv2d(1, 1, kernel_size=(3, 5), padding=(0, 1), stride=(3, 4))
print(comp_conv2d(conv2d, X).shape)

import torch
from d2l import torch as d2l

# TODO: 用于处理多个通道的卷积
def corr2d_multi_in(X, K):
    return sum(d2l.corr2d(x, k) for x, z in zip(X, K))

# TODO: 对2通道的3*3的图片进行 2*2卷积核的卷积测试，最终应该得到一个2*2的特征图
X = torch.tensor([[[0.0, 1.0, 2.0], [3.0, 4.0, 5.0], [6.0, 7.0, 8.0]],
                  [[1.0, 2.0, 3.0], [4.0, 5.0, 6.0], [7.0, 8.0, 9.0]]])
K = torch.tensor([[[0.0, 1.0], [2.0, 3.0]], [[1.0, 2.0], [3.0, 4.0]]])

print(corr2d_multi_in(X, K))

# TODO: 对于多输出的情况


def corr2d_multi_in_out(X, K):
    """多通道多输出卷积

    Args:
        X (Tensor(c,h,w)): 多通道图片
        K (Tensor(o,c,h,w)): 多输出多通道卷积核

    Returns:
        Tensor(o,Xh - Kh + 1, Xw - Kw + 1): o个输出通道的二维特征图
    """
    # 迭代“K”的第0个维度，每次都对输入“X”执行互相关运算。
    # 最后将所有结果都叠加在一起
    return torch.stack([corr2d_multi_in(X, k) for k in K], 0)


# TODO: 通过将核张量K与K+1（K中每个元素加）和K+2连接起来，构造了一个具有个输出通道的卷积核。
K = torch.stack((K, K + 1, K + 2), 0)
print(K.shape)
print(corr2d_multi_in_out(X, K))

import torch
from torch import nn
from d2l import torch as d2l

# TODO: 单通道单输出池化层


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


# TODO: 池化层测试
X = torch.tensor([[0.0, 1.0, 2.0], [3.0, 4.0, 5.0], [6.0, 7.0, 8.0]])
print(pool2d(X, (2, 2)))
print(pool2d(X, (2, 2), 'avg'))

# TODO: Pytorch步幅填充调节
X = torch.arange(16, dtype=torch.float32).reshape((1, 1, 4, 4))
print(X)
# TODO: Pytorch中的最大池化层，该层默认步幅=核大小
pool2d = nn.MaxPool2d(3)
print(pool2d(X))

# TODO: 手动设置步长和填充
pool2d = nn.MaxPool2d(3, padding=1, stride=2)
print(pool2d(X))

# TODO: 多通道
X = torch.cat((X, X + 1), 1)
print(X)

pool2d = nn.MaxPool2d(3, padding=1, stride=2)
print(pool2d(X))