第五十一天打卡

DAY 51 复习日
作业：day43的时候我们安排大家对自己找的数据集用简单cnn训练，现在可以尝试下借助这几天的知识来实现精度的进一步提高

kaggl的一个图像数据集；数据集地址：Lung Nodule Malignancy 肺结核良恶性判断 

三层卷积CNN做到的精度63%，现在需要实现提高。

import os
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
from torch.utils.data import Dataset, DataLoader
from PIL import Image
import torch
from torchvision import transforms
import torch.nn as nn
import torch.optim as optim
import matplotlib.pyplot as plt# 1. 读标签并映射 0/1
df = pd.read_csv('archive/malignancy.csv')# 2. 按 patch_id 划 train/val
ids    = df['NoduleID'].values
labels = df['malignancy'].values
train_ids, val_ids = train_test_split(ids, test_size=0.2, random_state=42, stratify=labels
)
train_df = df[df['NoduleID'].isin(train_ids)].reset_index(drop=True)
val_df   = df[df['NoduleID'].isin(val_ids)].reset_index(drop=True)# 3. Dataset：多页 TIFF 按页读取
class LungTBDataset(Dataset):def __init__(self, tif_path, df, transform=None):self.tif_path = tif_pathself.df = dfself.transform = transformdef __len__(self):return len(self.df)def __getitem__(self, idx):row = self.df.iloc[idx]pid = int(row['NoduleID'])label = int(row['malignancy'])try:with Image.open(self.tif_path) as img:# 检查 pid 是否超出实际帧数total_pages = sum(1 for _ in ImageSequence.Iterator(img))if pid >= total_pages:pid = total_pages - 1  # 取最后一帧img.seek(pid)img = img.convert('RGB')except Exception as e:# 返回黑色占位图img = Image.new('RGB', (224, 224), (0, 0, 0))if self.transform:img = self.transform(img)return img, label# 4. 变换 & DataLoader
transform = transforms.Compose([transforms.Resize((224,224)),transforms.ToTensor(),transforms.Normalize(mean=[0.485,0.456,0.406],std =[0.229,0.224,0.225])
])
train_ds = LungTBDataset('archive/ct_tiles.tif', train_df, transform)
val_ds   = LungTBDataset('archive/ct_tiles.tif',   val_df, transform)
train_loader = DataLoader(train_ds, batch_size=16, shuffle=True,  num_workers=0, pin_memory=True)
val_loader   = DataLoader(val_ds,   batch_size=16, shuffle=False, num_workers=0, pin_memory=True)import torch
import torch.nn as nn
import torch.optim as optim
from torchvision import models
from torch.optim.lr_scheduler import ReduceLROnPlateau# ==================== 1. 定义VGG16-CBAM模型 ====================
class ChannelAttention(nn.Module):def __init__(self, in_planes, ratio=16):super().__init__()self.avg_pool = nn.AdaptiveAvgPool2d(1)self.max_pool = nn.AdaptiveMaxPool2d(1)self.fc = nn.Sequential(nn.Conv2d(in_planes, in_planes//ratio, 1, bias=False),nn.ReLU(),nn.Conv2d(in_planes//ratio, in_planes, 1, bias=False))self.sigmoid = nn.Sigmoid()def forward(self, x):avg_out = self.fc(self.avg_pool(x))max_out = self.fc(self.max_pool(x))out = avg_out + max_outreturn self.sigmoid(out)class SpatialAttention(nn.Module):def __init__(self, kernel_size=7):super().__init__()self.conv = nn.Conv2d(2, 1, kernel_size, padding=kernel_size//2, bias=False)self.sigmoid = nn.Sigmoid()def forward(self, x):avg_out = torch.mean(x, dim=1, keepdim=True)max_out, _ = torch.max(x, dim=1, keepdim=True)x = torch.cat([avg_out, max_out], dim=1)x = self.conv(x)return self.sigmoid(x)class CBAM(nn.Module):def __init__(self, in_planes, ratio=16, kernel_size=7):super().__init__()self.ca = ChannelAttention(in_planes, ratio)self.sa = SpatialAttention(kernel_size)def forward(self, x):x = x * self.ca(x)x = x * self.sa(x)return x# ==== 定义VGG16-CBAM模型 ====
class VGG16_CBAM(nn.Module):def __init__(self, num_classes=2, pretrained=True):super().__init__()original_vgg = models.vgg16(pretrained=pretrained)# 特征提取部分self.features = original_vgg.features# 在block4和block5后插入CBAMself.cbam_block4 = CBAM(512)  # 对应block4输出self.cbam_block5 = CBAM(512)  # 对应block5输出# 分类器部分self.avgpool = nn.AdaptiveAvgPool2d((7, 7))self.classifier = nn.Sequential(nn.Linear(512 * 7 * 7, 4096),nn.ReLU(True),nn.Dropout(),nn.Linear(4096, 4096),nn.ReLU(True),nn.Dropout(),nn.Linear(4096, num_classes),)def forward(self, x):# 前向传播过程x = self.features(x)x = self.cbam_block4(x)  # 在block4后应用CBAMx = self.cbam_block5(x)  # 在block5后应用CBAMx = self.avgpool(x)x = torch.flatten(x, 1)x = self.classifier(x)return x
# ==================== 2. 训练策略配置 ====================
def set_trainable_layers(model, trainable_layers):"""阶段式解冻层"""for name, param in model.named_parameters():param.requires_grad = any(layer in name for layer in trainable_layers)def get_optimizer(model, lr_dict):"""差异化学习率优化器"""params = []for name, param in model.named_parameters():if param.requires_grad:# 不同层组设置不同学习率lr = lr_dict['features'] if 'features' in name else lr_dict['classifier']params.append({'params': param, 'lr': lr})return optim.Adam(params)# ==================== 3. 训练流程 ====================
def train_model(model, train_loader, val_loader, num_epochs=10):device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')model = model.to(device)# 阶段式训练配置training_phases = [{'name': 'Phase1-Classifier', 'train_layers': ['classifier'], 'epochs': 2, 'lr': {'features': 1e-5, 'classifier': 1e-4}},{'name': 'Phase2-Conv5+CBAM', 'train_layers': ['features.24', 'features.25', 'features.26', 'features.27', 'features.28', 'classifier'], 'epochs': 3, 'lr': {'features': 5e-5, 'classifier': 1e-4}},{'name': 'Phase3-Conv4+CBAM', 'train_layers': ['features.16', 'features.17', 'features.18', 'features.19', 'features.20', 'features.21', 'features.22', 'features.23', 'features.24', 'features.25', 'features.26', 'features.27', 'features.28', 'classifier'], 'epochs': 3, 'lr': {'features': 1e-4, 'classifier': 1e-4}},{'name': 'Phase4-FullModel', 'train_layers': ['features', 'classifier'], 'epochs': 2, 'lr': {'features': 2e-4, 'classifier': 1e-4}}]criterion = nn.CrossEntropyLoss()best_acc = 0.0for phase in training_phases:print(f"\n=== {phase['name']} ===")set_trainable_layers(model, phase['train_layers'])optimizer = get_optimizer(model, phase['lr'])scheduler = ReduceLROnPlateau(optimizer, mode='max', patience=1, factor=0.5)for epoch in range(phase['epochs']):# 训练阶段model.train()running_loss = 0.0for inputs, labels in train_loader:inputs, labels = inputs.to(device), labels.to(device)optimizer.zero_grad()outputs = model(inputs)loss = criterion(outputs, labels)loss.backward()optimizer.step()running_loss += loss.item() * inputs.size(0)epoch_loss = running_loss / len(train_loader.dataset)# 验证阶段model.eval()correct = 0with torch.no_grad():for inputs, labels in val_loader:inputs, labels = inputs.to(device), labels.to(device)outputs = model(inputs)_, preds = torch.max(outputs, 1)correct += (preds == labels).sum().item()epoch_acc = correct / len(val_loader.dataset)print(f'Epoch {epoch+1}/{phase["epochs"]} - Loss: {epoch_loss:.4f} Acc: {epoch_acc:.4f}')scheduler.step(epoch_acc)# 保存最佳模型if epoch_acc > best_acc:best_acc = epoch_acc# torch.save(model.state_dict(), 'best_vgg16_cbam.pth')print(f'Best Validation Accuracy: {best_acc:.4f}')# ==================== 4. 初始化并训练模型 ====================
model = VGG16_CBAM(num_classes=2, pretrained=True)
train_model(model, train_loader, val_loader, num_epochs=10)