【域适应之遥感滑坡检测】学习笔记

    import torch
    import torch.nn as nn
    import torch.optim as optim
    from torch.utils.data import DataLoader
    
    # 1. 定义Gradient Reversal Layer (GRL)
    class GradientReversal(torch.autograd.Function):
    @staticmethod
    def forward(ctx, x, alpha):
        ctx.alpha = alpha
        return x.view_as(x)
    
    @staticmethod
    def backward(ctx, grad_output):
        output = grad_output.neg() * ctx.alpha
        return output, None
    
    # 2. 定义滑坡检测的域适应模型
    class DomainAdaptationModel(nn.Module):
    def __init__(self):
        super(DomainAdaptationModel, self).__init__()
        
        # 特征提取器
        self.feature_extractor = nn.Sequential(
            nn.Conv2d(1, 32, kernel_size=5),
            nn.ReLU(),
            nn.MaxPool2d(2),
            nn.Conv2d(32, 64, kernel_size=5),
            nn.ReLU(),
            nn.MaxPool2d(2)
        )
        
        # 滑坡分类器
        self.class_classifier = nn.Sequential(
            nn.Linear(64 * 4 * 4, 100),
            nn.ReLU(),
            nn.Linear(100, 2),  # 二分类：滑坡区域和非滑坡区域
            nn.LogSoftmax(dim=1)
        )
        
        # 域判别器
        self.domain_classifier = nn.Sequential(
            nn.Linear(64 * 4 * 4, 100),
            nn.ReLU(),
            nn.Linear(100, 2),  # 二分类：源域和目标域
            nn.LogSoftmax(dim=1)
        )
    
    def forward(self, x, alpha=1.0):
        features = self.feature_extractor(x)
        features = features.view(-1, 64 * 4 * 4)
        
        class_output = self.class_classifier(features)
        reverse_feature = GradientReversal.apply(features, alpha)
        domain_output = self.domain_classifier(reverse_feature)
        
        return class_output, domain_output
    
    # 3. 数据加载
    def get_dataloader():
    # 假设已准备好A区域和B区域的数据
    # A区域为源领域（有标签数据），B区域为目标领域（无标签数据）
    pass
    
    source_loader, target_loader = get_dataloader()
    
    # 4. 定义训练函数
    def train(model, source_loader, target_loader, optimizer, epoch, alpha):
    model.train()
    for batch_idx, (source_data, source_label, target_data) in enumerate(zip(source_loader, target_loader)):
        source_data, source_label = source_data.cuda(), source_label.cuda()
        target_data = target_data.cuda()
        
        optimizer.zero_grad()
        
        # 源领域滑坡分类
        class_output, domain_output_source = model(source_data, alpha)
        class_loss = nn.CrossEntropyLoss()(class_output, source_label)
        
        # 源领域域分类
        domain_label_source = torch.zeros(len(source_data)).long().cuda()  # 0表示源领域
        domain_loss_source = nn.CrossEntropyLoss()(domain_output_source, domain_label_source)
        
        # 目标领域域分类
        _, domain_output_target = model(target_data, alpha)
        domain_label_target = torch.ones(len(target_data)).long().cuda()  # 1表示目标领域
        domain_loss_target = nn.CrossEntropyLoss()(domain_output_target, domain_label_target)
        
        # 总损失
        loss = class_loss + domain_loss_source + domain_loss_target
        loss.backward()
        optimizer.step()
        
        if batch_idx % 100 == 0:
            print(f'Epoch {epoch} [{batch_idx}/{len(source_loader)}]: Loss={loss.item()}')
    
    # 5. 初始化模型与优化器
    model = DomainAdaptationModel().cuda()
    optimizer = optim.Adam(model.parameters(), lr=0.001)
    
    # 6. 训练模型
    for epoch in range(1, 11):
    train(model, source_loader, target_loader, optimizer, epoch, alpha=0.1)