11-卷积神经网络(CNN): 原理和示例

# 输入矩阵和卷积核定义
input_matrix = [
    [1, 2, 3],
    [4, 5, 6],
    [7, 8, 9]
]
kernel = [
    [1, 2, 3],
    [4, 5, 6],
    [7, 8, 9]
]

# 超参数设置
pad = 1
stride = 1

# 对输入矩阵进行零填充
n_rows = len(input_matrix)
n_cols = len(input_matrix[0]) if n_rows > 0 else 0
padded_input = []
# 添加上方的填充行
for _ in range(pad):
    padded_input.append([0] * (n_cols + 2 * pad))
# 添加中间行（左右填充）
for row in input_matrix:
    padded_row = [0] * pad + row + [0] * pad
    padded_input.append(padded_row)
# 添加下方的填充行
for _ in range(pad):
    padded_input.append([0] * (n_cols + 2 * pad))

# 计算输出矩阵尺寸
kernel_rows = len(kernel)
kernel_cols = len(kernel[0]) if kernel_rows > 0 else 0
output_rows = (n_rows + 2 * pad - kernel_rows) // stride + 1
output_cols = (n_cols + 2 * pad - kernel_cols) // stride + 1

# 初始化输出矩阵
output = [[0 for _ in range(output_cols)] for _ in range(output_rows)]

# 执行卷积（互相关）运算
for i in range(output_rows):
    for j in range(output_cols):
        # 计算窗口起始位置
        row_start = i * stride
        row_end = row_start + kernel_rows
        col_start = j * stride
        col_end = col_start + kernel_cols

        # 提取当前窗口
        window = []
        for r in range(row_start, row_end):
            window.append(padded_input[r][col_start:col_end])

        # 计算窗口与卷积核的点乘和
        sum_val = 0
        for x in range(kernel_rows):
            for y in range(kernel_cols):
                sum_val += window[x][y] * kernel[x][y]
        output[i][j] = sum_val

# 打印结果
print("填充后的输入矩阵：")
for row in padded_input:
    print(row)
print("\n卷积结果矩阵：")
for row in output:
    print(row)
    
    
##
# 填充后的输入矩阵：
# [0, 0, 0, 0, 0]
# [0, 1, 2, 3, 0]
# [0, 4, 5, 6, 0]
# [0, 7, 8, 9, 0]
# [0, 0, 0, 0, 0]
# 
# 卷积结果矩阵：
# [94, 154, 106]
# [186, 285, 186]
# [106, 154, 94]
# #

import tensorflow as tf
import keras

# other platform
from tensorflow.keras.datasets import mnist
# macOS may
# from keras._tf_keras.keras.datasets import mnist


# 加载数据集
(X_train, y_train), (X_test, y_test) = mnist.load_data()

# 预处理：适配 CNN 输入。(还可以加上归一化)
X_train = X_train[..., tf.newaxis]
X_test = X_test[..., tf.newaxis]

# 模型
def cnn_net():
    return keras.Sequential([
        # 输入
        keras.layers.Input(shape=(28, 28, 1)),
        # 卷积
        keras.layers.Conv2D(filters=32, kernel_size=(3, 3), padding='same', strides=1, activation='relu'),
        keras.layers.Conv2D(filters=64, kernel_size=(2, 2), padding='same', strides=1, activation='relu'),
        # 最大池化
        keras.layers.MaxPool2D(pool_size=(2, 2)),

        # 卷积
        keras.layers.Conv2D(filters=32, kernel_size=(3, 3), padding='same', strides=1, activation='relu'),
        keras.layers.Conv2D(filters=64, kernel_size=(2, 2), padding='same', strides=1, activation='relu'),
        # 最大池化
        keras.layers.MaxPool2D(pool_size=(2, 2)),

        # 全连接
        keras.layers.Flatten(),
        # 输出
        keras.layers.Dense(10, activation='softmax')
    ])


model = cnn_net()

# 损失函数
loss = keras.losses.SparseCategoricalCrossentropy() 

# 优化器
optimizer = keras.optimizers.Adam() 

# 训练批次大小
batch_size = 32

# 训练批次
epoch_num = 5

# 编译模型
model.compile(
    optimizer=optimizer,
    loss=loss,
    metrics=['accuracy']
)

# 训练
history = model.fit(
    x=X_train,
    y=y_train,
    batch_size=batch_size,
    epochs=epoch_num,
    validation_data=(X_test, y_test)
)

#保存CNN模型
model.save("mnist_cnn.keras")

##
# Epoch 1/5
# 1875/1875 ━━━━━━━━━━━━━━━━━━━━ 24s 13ms/step - accuracy: 0.9114 - loss: 0.4192 - val_accuracy: 0.9839 - val_loss: 0.0490
# Epoch 2/5
# 1875/1875 ━━━━━━━━━━━━━━━━━━━━ 24s 13ms/step - accuracy: 0.9850 - loss: 0.0468 - val_accuracy: 0.9865 - val_loss: 0.0412
# Epoch 3/5
# 1875/1875 ━━━━━━━━━━━━━━━━━━━━ 26s 14ms/step - accuracy: 0.9892 - loss: 0.0351 - val_accuracy: 0.9890 - val_loss: 0.0305
# Epoch 4/5
# 1875/1875 ━━━━━━━━━━━━━━━━━━━━ 25s 13ms/step - accuracy: 0.9914 - loss: 0.0256 - val_accuracy: 0.9824 - val_loss: 0.0534
# Epoch 5/5
# 1875/1875 ━━━━━━━━━━━━━━━━━━━━ 25s 13ms/step - accuracy: 0.9917 - loss: 0.0247 - val_accuracy: 0.9886 - val_loss: 0.0390
# #

import torch
import torchvision
from torch.utils.data import DataLoader
from tqdm import tqdm

# 定义设备（GPU/CPU）
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

# 下载训练数据
train_data = torchvision.datasets.MNIST(
    root='./data',
    train=True,
    download=True,
    transform=torchvision.transforms.ToTensor(),
)

# 加载测试数据
test_data = torchvision.datasets.MNIST(
    root='./data',
    train=False,
    transform=torchvision.transforms.ToTensor(),
)

# 批次大小
batch_size = 64

# 训练数据加载器
train_loader = DataLoader(
    dataset=train_data,
    batch_size=batch_size,
    shuffle=True,
    num_workers=2
)

# 测试数据加载器
test_loader = DataLoader(
    dataset=test_data,
    batch_size=batch_size,
    shuffle=False
)


# 模型
class CnnNet(torch.nn.Module):
    def __init__(self):
        super(CnnNet, self).__init__()

        # (卷积+激活)*2 + 最大池化
        self.conv1 = torch.nn.Sequential(
            torch.nn.Conv2d(in_channels=1, out_channels=32, kernel_size=(3, 3), stride=1, padding=1),
            torch.nn.ReLU(),
            torch.nn.Conv2d(in_channels=32, out_channels=64, kernel_size=(3, 3), stride=1, padding=1),
            torch.nn.ReLU(),
            torch.nn.MaxPool2d(kernel_size=2, stride=2),
        )

        self.conv2 = torch.nn.Sequential(
            torch.nn.Conv2d(in_channels=64, out_channels=128, kernel_size=(3, 3), stride=1, padding=1),
            torch.nn.ReLU(),
            torch.nn.Conv2d(in_channels=128, out_channels=64, kernel_size=(3, 3), stride=1, padding=1),
            torch.nn.ReLU(),
            torch.nn.MaxPool2d(kernel_size=2, stride=2),
        )

        self.classifier = torch.nn.Sequential(
            # 展平后输出
            torch.nn.Flatten(),
            torch.nn.Linear(in_features=64*7*7, out_features=256),
            torch.nn.ReLU(),
            torch.nn.Linear(in_features=256, out_features=128),
            torch.nn.ReLU(),
            torch.nn.Linear(in_features=128, out_features=10),
        )

    def forward(self, x):
        x = self.conv1(x)
        x = self.conv2(x)
        output = self.classifier(x)
        return output


if __name__ == '__main__':

    model = CnnNet()
    print(model)
    model = model.to(device)  # 将模型移动到设备
    
    # 损失函数
    criterion = torch.nn.CrossEntropyLoss()
    # 优化器
    optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
    # 训练轮次
    num_epoch = 5
    # 最高正确率
    best_val_acc = 0

    for epoch in range(num_epoch):
        # 训练阶段
        model.train()  # 设置为训练模式
        train_loss = 0.0
        train_correct = 0
        train_total = 0

        for inputs, labels in tqdm(train_loader, desc=f"Epoch {epoch + 1}/{num_epoch} [Train]"):
            inputs, labels = inputs.to(device), labels.to(device)

            optimizer.zero_grad()  # 清空梯度
            outputs = model(inputs)  # 前向传播
            loss = criterion(outputs, labels)  # 计算损失
            loss.backward()  # 反向传播
            optimizer.step()  # 更新参数

            # 统计训练结果
            train_loss += loss.item() * inputs.size(0)
            _, predicted = torch.max(outputs.data, 1)
            train_total += labels.size(0)
            train_correct += (predicted == labels).sum().item()

        train_loss = train_loss / len(train_loader.dataset)  # 平均训练损失
        train_acc = 100 * train_correct / train_total  # 训练准确率

        # 验证/测试阶段
        model.eval()  # 设置为评估模式
        val_loss = 0.0
        val_correct = 0
        val_total = 0

        with torch.no_grad():  # 关闭梯度计算
            for inputs, labels in tqdm(test_loader, desc=f"Epoch {epoch + 1}/{num_epoch} [Val]"):
                inputs, labels = inputs.to(device), labels.to(device)
                outputs = model(inputs)
                loss = criterion(outputs, labels)

                # 统计验证结果
                val_loss += loss.item() * inputs.size(0)
                _, predicted = torch.max(outputs.data, 1)
                val_total += labels.size(0)
                val_correct += (predicted == labels).sum().item()

        val_loss = val_loss / len(test_loader.dataset)  # 平均验证损失
        val_acc = 100 * val_correct / val_total  # 验证准确率

        # 打印结果
        print(f"Epoch {epoch + 1}/{num_epoch}: "
              f"Train Loss: {train_loss:.4f}, Train Acc: {train_acc:.2f}%, "
              f"Val Loss: {val_loss:.4f}, Val Acc: {val_acc:.2f}%")

        # 保存最佳模型
        if val_acc > best_val_acc:
            best_val_acc = val_acc
            torch.save(model.state_dict(), "best_model.pth")
            print(f"Saved best model with val acc {val_acc:.2f}%")


##
# Epoch 1/5 [Train]: 100%|██████████| 938/938 [01:35<00:00,  9.85it/s]
# Epoch 1/5 [Val]: 100%|██████████| 157/157 [00:04<00:00, 35.55it/s]
# Epoch 1/5: Train Loss: 1.6747, Train Acc: 78.58%, Val Loss: 1.5900, Val Acc: 87.09%
# Saved best model with val acc 87.09%
# Epoch 2/5 [Train]: 100%|██████████| 938/938 [01:36<00:00,  9.77it/s]
# Epoch 2/5 [Val]: 100%|██████████| 157/157 [00:04<00:00, 35.30it/s]
# Epoch 2/5: Train Loss: 1.5747, Train Acc: 88.60%, Val Loss: 1.5032, Val Acc: 95.77%
# Saved best model with val acc 95.77%
# Epoch 3/5 [Train]: 100%|██████████| 938/938 [01:36<00:00,  9.77it/s]
# Epoch 3/5 [Val]: 100%|██████████| 157/157 [00:04<00:00, 35.61it/s]
# Epoch 3/5: Train Loss: 1.5215, Train Acc: 93.95%, Val Loss: 1.5097, Val Acc: 95.14%
# Epoch 4/5 [Train]: 100%|██████████| 938/938 [01:36<00:00,  9.76it/s]
# Epoch 4/5 [Val]: 100%|██████████| 157/157 [00:04<00:00, 35.67it/s]
# Epoch 4/5: Train Loss: 1.5186, Train Acc: 94.24%, Val Loss: 1.5147, Val Acc: 94.64%
# Epoch 5/5 [Train]: 100%|██████████| 938/938 [01:35<00:00,  9.80it/s]
# Epoch 5/5 [Val]: 100%|██████████| 157/157 [00:04<00:00, 34.81it/s]
# Epoch 5/5: Train Loss: 1.5314, Train Acc: 92.97%, Val Loss: 1.5799, Val Acc: 88.11%
# #