遥感图像应用：在低分辨率图像上实现洪水损害检测（迁移学习）-Toy模板网

这篇具有很好参考价值的文章主要介绍了遥感图像应用：在低分辨率图像上实现洪水损害检测（迁移学习）。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

本文是上一篇关于“在低分辨率图像上实现洪水损害检测”的博客的延申。

代码来源：https://github.com/weining20000/Flooding-Damage-Detection-from-Post-Hurricane-Satellite-Imagery-Based-on-CNN/tree/master

数据储存地址：https://github.com/JeffereyWu/FloodDamageDetection/tree/main

目标：利用迁移学习训练两个预训练的CNN模型（VGG和Resnet），自动化识别一个区域是否存在洪水损害。

运行环境：Google Colab

1. 导入库

# Pytoch
import torch
from torchvision import datasets, models
from torch.utils.data import Dataset, DataLoader
import torchvision.transforms as transforms
import torch.nn as nn
from torch_lr_finder import LRFinder

# Data science tools
import numpy as np
import pandas as pd
import os
from sklearn.metrics import accuracy_score
from sklearn.metrics import confusion_matrix

from PIL import Image

# Visualizations
import matplotlib.pyplot as plt
import seaborn as sns

2. 迁移学习知识点

对于卷积神经网络（CNN）等模型，通常包括一些卷积层和池化层，这些层的权重用于提取图像的特征。当这些层的参数被冻结时，这些权重将保持不变，不会在训练过程中进行更新。这意味着模型会继续使用预训练模型的特征提取能力。
如果模型还包含其他的预训练层，例如预训练的全连接层，这些层的权重也将被冻结，不会更新。
通常，当使用预训练模型进行微调时，会替换模型的最后一层或几层，以适应新的任务。新添加的自定义分类器层的权重将被训练和更新，以适应特定的分类任务。

3. 加载和配置预训练的深度学习模型

#Load pre-trained model
def get_pretrained_model(model_name):
  """
  获取预训练模型的函数。

  参数：
  model_name: 要加载的预训练模型的名称（例如，'vgg16' 或 'resnet50'）

  返回：
  MODEL: 加载并配置好的预训练模型
  """

  if model_name == 'vgg16':
      model = models.vgg16(pretrained=True)

      # 将模型的参数（权重）冻结，不进行微调。这意味着这些参数在训练过程中不会更新
      for param in model.parameters():
          param.requires_grad = False
      n_inputs = model.classifier[6].in_features # 获取模型分类器最后一层的输入特征数
      n_classes = 2

      # 替换模型的分类器部分，添加自定义的分类器
      model.classifier[6] = nn.Sequential(
          nn.Linear(n_inputs, 256), nn.ReLU(), nn.Dropout(0.2),
          nn.Linear(256, n_classes))

  elif model_name == 'resnet50':
      model = models.resnet50(pretrained=True)

      for param in model.parameters():
          param.requires_grad = False

	  # 获取模型最后一层全连接层的输入特征数
      n_inputs = model.fc.in_features
      n_classes = 2
      model.fc = nn.Sequential(
          nn.Linear(n_inputs, 256), nn.ReLU(), nn.Dropout(0.2),
          nn.Linear(256, n_classes))

  # Move to GPU
  MODEL = model.to(device)

  return MODEL # 返回加载和配置好的预训练模型

注意，这里vgg16的classifier结构原本为：
Sequential(
(0): Linear(in_features=25088, out_features=4096, bias=True)
(1): ReLU(inplace=True)
(2): Dropout(p=0.5, inplace=False)
(3): Linear(in_features=4096, out_features=4096, bias=True)
(4): ReLU(inplace=True)
(5): Dropout(p=0.5, inplace=False)
(6): Linear(in_features=4096, out_features=1000, bias=True)
)
以上代码替换了最后一层的classifier，改为：
Sequential(
(0): Linear(in_features=25088, out_features=4096, bias=True)
(1): ReLU(inplace=True)
(2): Dropout(p=0.5, inplace=False)
(3): Linear(in_features=4096, out_features=4096, bias=True)
(4): ReLU(inplace=True)
(5): Dropout(p=0.5, inplace=False)
(6): Sequential(
(0): Linear(in_features=4096, out_features=256, bias=True)
(1): ReLU()
(2): Dropout(p=0.2, inplace=False)
(3): Linear(in_features=256, out_features=2, bias=True)
)
)

注意，这里resnet50的fc结构原本为：
Linear(in_features=2048, out_features=1000, bias=True)
以上代码替换了最后一层fc，改为：
Sequential(
(0): Linear(in_features=2048, out_features=256, bias=True)
(1): ReLU()
(2): Dropout(p=0.2, inplace=False)
(3): Linear(in_features=256, out_features=2, bias=True)
)

4. 建立模型

# VGG 16
model_vgg = get_pretrained_model('vgg16') # 包含加载和配置好的 VGG16 模型
criterion_vgg = nn.CrossEntropyLoss()
optimizer_vgg = torch.optim.Adam(model_vgg.parameters(), lr=0.00002)

# ResNet 50
model_resnet50 = get_pretrained_model('resnet50') # 包含加载和配置好的 ResNet50 模型
criterion_resnet50 = nn.CrossEntropyLoss() 
optimizer_resnet50 = torch.optim.Adam(model_resnet50.parameters(), lr=0.001)

5. 定义计算准确率的函数

def acc_vgg(x, y, return_labels=False):

  with torch.no_grad(): # 禁止梯度计算，因为在准确率计算中不需要梯度信息
      logits = model_vgg(x)
      pred_labels = np.argmax(logits.cpu().numpy(), axis=1)
  if return_labels:
      return pred_labels
  else:
      return 100*accuracy_score(y.cpu().numpy(), pred_labels)

def acc_resnet50(x, y, return_labels=False):
  
  with torch.no_grad():
      logits = model_resnet50(x)
      pred_labels = np.argmax(logits.cpu().numpy(), axis=1)
  if return_labels:
      return pred_labels
  else:
      return 100*accuracy_score(y.cpu().numpy(), pred_labels)

6. 定义一个用于训练深度学习模型的函数

def train(model, criterion, optimizer, acc, xtrain, ytrain, xval, yval, save_file_name, n_epochs, BATCH_SIZE):
    """
    训练深度学习模型的函数。

    参数：
    model: 要训练的深度学习模型
    criterion: 损失函数
    optimizer: 优化器
    acc: 准确率计算函数
    xtrain: 训练数据
    ytrain: 训练标签
    xval: 验证数据
    yval: 验证标签
    save_file_name: 保存训练后模型权重的文件名
    n_epochs: 训练的总轮数（epochs）
    BATCH_SIZE: 每个批次的样本数量

    返回：
    训练完成的模型和训练历史记录
    """

    history1 = []

    # Number of epochs already trained (if using loaded in model weights)
    try:
        print(f'Model has been trained for: {model.epochs} epochs.\n')
    except:
        model.epochs = 0
        print(f'Starting Training from Scratch.\n')

    # Main loop
    for epoch in range(n_epochs):

        # keep track of training and validation loss each epoch
        train_loss = 0.0
        val_loss = 0.0

        train_acc = 0
        val_acc = 0

        # Set to training
        model.train()

        #Training loop
        for batch in range(len(xtrain)//BATCH_SIZE):
            idx = slice(batch * BATCH_SIZE, (batch+1)*BATCH_SIZE)

            # Clear gradients
            optimizer.zero_grad()
            # Predicted outputs
            output = model(xtrain[idx])
            # Loss and BP of gradients
            loss = criterion(output, ytrain[idx])
            loss.backward()
            # Update the parameters
            optimizer.step()
            # Track train loss
            train_loss += loss.item()
            train_acc = acc(xtrain, ytrain)

        # After training loops ends, start validation
        # set to evaluation mode
        model.eval()
        # Don't need to keep track of gradients
        with torch.no_grad():
            # Evaluation loop
            # F.P.
            y_val_pred = model(xval)
            # Validation loss
            loss = criterion(y_val_pred, yval)
            val_loss = loss.item()
            val_acc = acc(xval, yval)

            history1.append([train_loss / BATCH_SIZE, val_loss, train_acc, val_acc])
            torch.save(model.state_dict(), save_file_name) # 保存模型权重
            torch.cuda.empty_cache()

            # Print training and validation results
        print("Epoch {} | Train Loss: {:.5f} | Train Acc: {:.2f} | Valid Loss: {:.5f} | Valid Acc: {:.2f} |".format(
            epoch, train_loss / BATCH_SIZE, acc(xtrain, ytrain), val_loss, acc(xval, yval)))
        # Format history
        history = pd.DataFrame(history1, columns=['train_loss', 'val_loss', 'train_acc', 'val_acc'])
    return model, history