《数据驱动的可重复性研究课堂作业》——手搓神经网络模型

# 创建一个新的 Conda 环境，命名为 pytorch_env，使用 Python 3.10 版本
# conda create -n pytorch_env python=3.10

# 激活该环境
# conda activate pytorch_env

# 使用 conda 安装 PyTorch 及其相关工具包
# conda install pytorch torchvision torchaudio -c pytorch

# 使用 conda 安装 PyTorch 及其相关工具包
# conda install pytorch torchvision torchaudio -c pytorch

# 导入必要的库
import torch
import torch.nn as nn
import torch.optim as optim
from torchvision import datasets, transforms
import matplotlib.pyplot as plt
import torch.nn.functional as F

# 定义图像预处理流程
transform = transforms.Compose([
    transforms.ToTensor(),
    transforms.Normalize((0.1307,), (0.3081,))
])

# 下载并加载数据集（利用 torchvision 自动下载 MNIST 数据集。这个数据集包含手写数字图像，是机器学习领域的经典数据集。）
train_dataset = datasets.MNIST(
    root='./data',
    train=True,
    download=True,
    transform=transform
)

test_dataset = datasets.MNIST(
    root='./data',
    train=False,
    download=True,
    transform=transform
)

# 绘制训练集图像
plt.figure(figsize=(10, 6))
plt.title("MNIST Dataset Examples")
for i in range(21): # 绘制 21 张训练集图像
    plt.subplot(4, 7, i+1) # 绘制第 i+1 张图像
    plt.axis("off") # 不显示坐标轴
    img = train_dataset[i][0].squeeze() # 获取第 i 张图像
    label = train_dataset[i][1] # 获取第 i 张图像的标签
    plt.imshow(img, cmap="gray") # 绘制第 i 张图像
    plt.text(18, 26, f"{label}", fontsize=10, color="red") # 在图像右下角标出红色标签
# 绘制测试集图像
for i in range(7): # 绘制 7 张测试集图像
    plt.subplot(4, 7, i+22) # 绘制第 i+22 张图像
    plt.axis("off") # 不显示坐标轴
    img = test_dataset[i][0].squeeze() # 获取第 i 张图像
    label = test_dataset[i][1] # 获取第 i 张图像的标签
    plt.imshow(img, cmap="gray") # 绘制第 i 张图像
    plt.text(20, 25, f"{label}", fontsize=10, color="green") # 在图像右下角标出绿色标签
plt.tight_layout()
plt.show()

# 导入必要的库
import torch
import torch.nn as nn
import torch.optim as optim
from torchvision import datasets, transforms
from torch.utils.data import DataLoader
import matplotlib.pyplot as plt
import torch.nn.functional as F
from sklearn.metrics import confusion_matrix
import numpy as np
import matplotlib.pyplot as plt
from scipy.signal import convolve2d
from torch.nn import MaxPool2d
import math

# 定义 LeNet 神经网络模型类，继承自 nn.Module
class LeNet(nn.Module):
    def __init__(self):
        # 初始化父类 nn.Module
        super(LeNet, self).__init__()
        # 第一个卷积层：
        # 输入通道：1（灰度图像），输出通道：6，卷积核大小：5x5
        self.conv1 = nn.Conv2d(in_channels=1, out_channels=6, kernel_size=5)
        
        # 定义池化层：
        # 使用 2x2 的最大池化，能够减小特征图的尺寸
        self.pool = nn.MaxPool2d(kernel_size=2, stride=2)
        
        # 第二个卷积层：
        # 输入通道：6，输出通道：16，卷积核大小：5x5
        self.conv2 = nn.Conv2d(in_channels=6, out_channels=16, kernel_size=5)
        
        # 第一个全连接层：
        # 输入特征数为 16*4*4（经过两次卷积和池化后的特征图尺寸），输出特征数为 120
        self.fc1 = nn.Linear(in_features=16*4*4, out_features=120)
        
        # 第二个全连接层：将 120 个特征映射到 84 个特征
        self.fc2 = nn.Linear(in_features=120, out_features=84)
        
        # 第三个全连接层：输出 10 个类别，对应 MNIST 中 10 个数字
        self.fc3 = nn.Linear(in_features=84, out_features=10)

    def forward(self, x):
        # 将输入通过第一个卷积层，并使用 ReLU 激活函数增加非线性
        x = torch.relu(self.conv1(x))
        # 应用池化层，减小特征图尺寸
        x = self.pool(x)
        # 第二个卷积层 + ReLU 激活
        x = torch.relu(self.conv2(x))
        # 再次池化
        x = self.pool(x)
        # 将多维特征图展平为一维向量，为全连接层做准备
        x = x.view(-1, 16*4*4)
        # 第一个全连接层 + ReLU 激活
        x = torch.relu(self.fc1(x))
        # 第二个全连接层 + ReLU 激活
        x = torch.relu(self.fc2(x))
        # 第三个全连接层得到最终输出（未经过激活，后续会结合损失函数使用）
        x = self.fc3(x)
        return x

# 打印模型结构
model = LeNet()
print(model)

transform = transforms.Compose([
    transforms.ToTensor(),
    transforms.Normalize((0.1307,), (0.3081,))
])

test_dataset = datasets.MNIST(
    root='./data',
    train=False,
    download=True,
    transform=transform
)

train_dataset = datasets.MNIST(root='./data', train=True, transform=transform, download=True)
train_loader = DataLoader(train_dataset, batch_size=64, shuffle=True)

# 创建数据加载器
train_loader = torch.utils.data.DataLoader(
    dataset=train_dataset,
    batch_size=64,
    shuffle=True
)

test_loader = torch.utils.data.DataLoader(
    dataset=test_dataset,
    batch_size=1000,
    shuffle=False
)

# 定义训练函数，用于在训练集上训练模型
def train(model, device, train_loader, optimizer, criterion, epoch):
    model.train()
    train_loss = 0
    correct = 0
    for batch_idx, (data, target) in enumerate(train_loader):
        data, target = data.to(device), target.to(device)
        optimizer.zero_grad()
        output = model(data)
        loss = criterion(output, target)
        loss.backward()
        optimizer.step()
        
        # 累计损失和正确预测数
        train_loss += loss.item() * data.size(0)
        pred = output.argmax(dim=1, keepdim=True)
        correct += pred.eq(target.view_as(pred)).sum().item()
        
        if batch_idx % 5000 == 0:
            print(f"Train Epoch: {epoch} [{batch_idx * len(data)}/{len(train_loader.dataset)}]\tLoss: {loss.item():.6f}")
    
    # 计算平均损失和准确率
    train_loss /= len(train_loader.dataset)
    accuracy = 100. * correct / len(train_loader.dataset)
    return train_loss, accuracy

# 定义测试函数，用于评估模型在测试集上的表现
def test(model, device, test_loader, criterion):
    model.eval()  # 将模型设置为评估模式，关闭 dropout 等训练特性
    test_loss = 0  # 初始化测试损失
    correct = 0    # 初始化预测正确的样本计数
    all_preds = []  # 用于存储所有预测结果
    all_targets = []  # 用于存储所有真实标签
    
    # 在测试阶段不计算梯度，节省内存和加快计算速度
    with torch.no_grad():
        for data, target in test_loader:
            data, target = data.to(device), target.to(device)
            output = model(data)
            test_loss += criterion(output, target).item() * data.size(0)
            pred = output.argmax(dim=1, keepdim=True)
            correct += pred.eq(target.view_as(pred)).sum().item()
            
            # 收集预测结果和真实标签
            all_preds.extend(pred.cpu().numpy().flatten())
            all_targets.extend(target.cpu().numpy())
    
    test_loss /= len(test_loader.dataset)  # 计算平均损失
    accuracy = 100. * correct / len(test_loader.dataset)  # 计算准确率
    
    # 计算混淆矩阵

    cm = confusion_matrix(all_targets, all_preds)
    
    return test_loss, accuracy, cm

# 检查是否有 GPU 可用，否则使用 CPU
device = torch.device("cuda" if torch.cuda.is_available() else "mps" if torch.mps.is_available() else "cpu")

# 实例化 LeNet 模型，并移动到指定设备上
model = LeNet().to(device)

# 定义优化器：使用随机梯度下降（SGD），学习率为 0.01，动量为 0.9
optimizer = optim.SGD(model.parameters(), lr=0.01, momentum=0.9)

# 定义损失函数：交叉熵损失函数常用于分类问题
criterion = nn.CrossEntropyLoss()

# 用于记录训练过程的指标
train_losses = []
train_accs = []
test_losses = []
test_accs = []

epochs = 20  # 设定训练轮数为 20
# 循环训练和测试模型
for epoch in range(1, epochs + 1):
    # 训练并记录指标
    train_loss, train_acc = train(model, device, train_loader, optimizer, criterion, epoch)
    test_loss, test_acc, cm = test(model, device, test_loader, criterion)
    
    # 保存指标
    train_losses.append(train_loss)
    train_accs.append(train_acc)
    test_losses.append(test_loss)
    test_accs.append(test_acc)
    
    print(f"\nEpoch {epoch}:")
    print(f"Train - Loss: {train_loss:.4f}, Accuracy: {train_acc:.2f}%")
    print(f"Test  - Loss: {test_loss:.4f}, Accuracy: {test_acc:.2f}%\n")

# 绘制训练过程图表
epochs_range = range(1, epochs + 1)

plt.figure(figsize=(12, 5))

# 绘制损失曲线
plt.subplot(1, 2, 1)
plt.plot(epochs_range, train_losses, 'bo-', label='Training Loss')
plt.plot(epochs_range, test_losses, 'ro-', label='Test Loss')
plt.title('Model Loss')
plt.xlabel('Epoch')
plt.ylabel('Loss')
plt.legend()

# 绘制准确率曲线
plt.subplot(1, 2, 2)
plt.plot(epochs_range, train_accs, 'bo-', label='Training Accuracy')
plt.plot(epochs_range, test_accs, 'ro-', label='Test Accuracy')
plt.title('Model Accuracy')
plt.xlabel('Epoch')
plt.ylabel('Accuracy (%)')
plt.legend()

plt.tight_layout()
plt.show()

import numpy as np
import matplotlib.pyplot as plt
from scipy.signal import convolve2d

# 原始矩阵
matrix = test_dataset[0][0].squeeze()

# 3x3 卷积核
kernel = np.array([[-1, -1, -1],
                    [ 0,  0,  0],
                    [ 1,  1,  1]])

# 进行卷积运算
convolved = convolve2d(matrix, kernel, mode='valid')

# 计算子图尺寸比例
original_shape = matrix.shape
convolved_shape = convolved.shape

fig, axes = plt.subplots(1, 2, figsize=(10, 5))

# 计算比例因子，使卷积后的小图与原图比例协调
scale_factor = original_shape[0] / convolved_shape[0]

# 调整原始矩阵子图
axes[0].imshow(matrix, cmap='gray', interpolation='nearest', aspect=1)
axes[0].set_title("Original Matrix")
axes[0].axis("off")

# 调整卷积后矩阵子图，缩放至与原图比例协调
axes[1].imshow(convolved, cmap='gray', interpolation='nearest', aspect=1/scale_factor)
axes[1].set_title("Convolved Matrix")
axes[1].axis("off")

plt.show()

conv1_weights = model.conv1.weight.data
print(conv1_weights)


# 最大池化层
maxpool = MaxPool2d(kernel_size=2, stride=2)

num_kernels = len(conv1_weights)
num_cols = 6
num_rows = math.ceil(num_kernels * 3 / num_cols)

plt.figure(figsize=(num_cols * 2, num_rows * 2))

for kernel_idx, kernel in enumerate(conv1_weights):
    convolved_image = convolve2d(matrix, kernel.squeeze().cpu().detach().numpy(), mode='valid')
    pooled_image = maxpool(torch.tensor(convolved_image).unsqueeze(0)).squeeze(0).cpu().detach().numpy()    

    plt.subplot(num_rows, num_cols, kernel_idx * 3 + 1)
    plt.imshow(matrix, cmap='gray', interpolation='nearest')
    plt.title("Original")
    plt.axis("off")

    plt.subplot(num_rows, num_cols, kernel_idx * 3 + 2)
    plt.imshow(convolved_image, cmap='gray', interpolation='nearest')
    plt.title(f"Convolved ({kernel_idx + 1})")
    plt.axis("off")

    plt.subplot(num_rows, num_cols, kernel_idx * 3 + 3)
    plt.imshow(pooled_image, cmap='gray', interpolation='nearest')
    plt.title(f"Pooled ({kernel_idx + 1})")
    plt.axis("off")

plt.tight_layout()
plt.show()