（1）对抗域自适应与风力发电机跨域特征对齐：

针对不同风机或不同风电场之间数据分布差异导致的模型泛化能力下降问题，提出了一种基于对抗域自适应的跨域故障识别方法。该方法的核心在于构建一个特征提取器与域判别器相互博弈的对抗网络。特征提取器致力于从源域和目标域数据中提取出域不变的特征表示，使得两个域的特征分布尽可能相似；而域判别器则试图区分特征来源于哪个域。通过这种对抗训练，特征提取器最终能够生成域不变的特征，从而有效减小源域（有标签）与目标域（无标签或少量标签）之间的分布差异。在此基础上，进一步引入梯度反转层简化训练过程，使得整个网络可以端到端地优化。该方法无需对目标域数据进行人工标注，即可实现跨风机叶片结冰和行星齿轮箱故障的准确识别，显著提升了模型在未知风机上的适应能力。

（2）时序卷积与自注意力融合的时空特征提取：

为了充分挖掘风机运行数据中蕴含的深层次时空关联信息，设计了一种融合时序卷积网络与自注意力机制的特征提取模块。时序卷积网络通过因果卷积和膨胀卷积，能够有效捕捉长时间跨度的序列依赖关系，同时保持计算的高效性。该模块首先利用多层膨胀卷积层对原始振动信号和功率信号进行局部特征建模，提取出不同时间尺度的特征图。随后，将特征图输入到多头自注意力层中，自注意力机制能够动态地为不同时间步和不同特征通道分配不同的权重，从而聚焦于对故障最敏感的时刻和频段。通过这种时空特征的互补结合，模型不仅能够感知信号中的局部突变模式，还能理解全局的周期性变化规律，这对于识别风力发电机叶片结冰这种渐变式故障尤为重要。实验表明，该融合模块在少量训练样本下仍能保持较高的识别率。

（3）基于原型校正的小样本跨域故障分类：

针对实际工程中目标域故障样本极度稀缺甚至为零的情况，提出了一种基于原型校正的小样本学习方法。首先，利用源域的大量标注数据训练一个类别原型生成器，为每种故障类型计算一个初始原型向量（即该类样本特征的中心）。然而，由于域偏移，源域原型不能直接适用于目标域。为此，设计了一个原型自适应校正模块，该模块通过少量无标签目标域数据的统计特性（如特征分布的一阶矩和二阶矩）来微调源域原型，使其向目标域的特征空间靠拢。校正后的原型作为目标域的类别参考中心。对于待诊断的目标域样本，只需计算其特征与各个校正后原型的距离，最近的类别即为预测结果。该方法避免了复杂的对抗训练，计算开销小，特别适合风电场边缘计算设备。在真实风机数据集上的实验结果表明，即使目标域只有几十个无标签样本，该方法也能达到90%以上的跨域识别准确率。

import torch
import torch.nn as nn
import torch.optim as optim
from torch.autograd import Function

# 梯度反转层（用于域对抗）
class GradientReversal(Function):
    @staticmethod
    def forward(ctx, x, alpha):
        ctx.alpha = alpha
        return x.view_as(x)
    @staticmethod
    def backward(ctx, grad_output):
        return grad_output.neg() * ctx.alpha, None

# 时序卷积块
class TemporalConvBlock(nn.Module):
    def __init__(self, in_ch, out_ch, kernel_size, dilation):
        super().__init__()
        self.conv = nn.Conv1d(in_ch, out_ch, kernel_size, padding=dilation*(kernel_size-1)//2, dilation=dilation)
        self.bn = nn.BatchNorm1d(out_ch)
        self.relu = nn.ReLU()
    def forward(self, x):
        return self.relu(self.bn(self.conv(x)))

# 完整模型：特征提取器 + 域判别器 + 分类器
class WindTurbineDomainAdaptNet(nn.Module):
    def __init__(self, input_dim=12, seq_len=512, num_classes=4):
        super().__init__()
        # 时序卷积特征提取
        self.tcn = nn.Sequential(
            TemporalConvBlock(input_dim, 32, 3, 1),
            TemporalConvBlock(32, 64, 3, 2),
            TemporalConvBlock(64, 128, 3, 4),
            nn.AdaptiveAvgPool1d(1)
        )
        # 自注意力
        self.self_attn = nn.MultiheadAttention(embed_dim=128, num_heads=4, batch_first=True)
        self.classifier = nn.Linear(128, num_classes)
        self.domain_classifier = nn.Sequential(nn.Linear(128, 64), nn.ReLU(), nn.Linear(64, 2))
        
    def forward(self, x, alpha=1.0):
        # x: (batch, channels, seq_len)
        feat = self.tcn(x).squeeze(-1)  # (batch, 128)
        feat_attn, _ = self.self_attn(feat.unsqueeze(1), feat.unsqueeze(1), feat.unsqueeze(1))
        feat = feat_attn.squeeze(1)
        cls_out = self.classifier(feat)
        # 域对抗部分
        reversed_feat = GradientReversal.apply(feat, alpha)
        domain_out = self.domain_classifier(reversed_feat)
        return cls_out, domain_out

# 原型校正伪代码（示意）
def prototype_correction(source_protos, target_features):
    # source_protos: dict {class: tensor}
    # target_features: list of unlabeled target features
    target_mean = torch.stack(target_features).mean(dim=0)
    target_std = torch.stack(target_features).std(dim=0)
    corrected_protos = {}
    for cls, proto in source_protos.items():
        # 仿射变换校正
        corrected = (proto - proto.mean()) / (proto.std() + 1e-6) * target_std + target_mean
        corrected_protos[cls] = corrected
    return corrected_protos

# 训练循环示例
def train_adaptation(model, source_loader, target_loader, epochs):
    optimizer = optim.Adam(model.parameters(), lr=0.001)
    criterion_cls = nn.CrossEntropyLoss()
    criterion_domain = nn.CrossEntropyLoss()
    for epoch in range(epochs):
        p = float(epoch) / epochs
        alpha = 2. / (1. + np.exp(-10 * p)) - 1   # 逐步增加对抗强度
        for (src_x, src_y), (tgt_x, _) in zip(source_loader, target_loader):
            cls_out, domain_out = model(src_x, alpha)
            loss_cls = criterion_cls(cls_out, src_y)
            # 源域域标签为0，目标域为1
            domain_label_src = torch.zeros(src_x.size(0), dtype=torch.long)
            domain_label_tgt = torch.ones(tgt_x.size(0), dtype=torch.long)
            _, domain_out_tgt = model(tgt_x, alpha)
            loss_domain = criterion_domain(domain_out, domain_label_src) + criterion_domain(domain_out_tgt, domain_label_tgt)
            total_loss = loss_cls + 0.1 * loss_domain
            optimizer.zero_grad(); total_loss.backward(); optimizer.step()
    return model
",

如有问题，可以直接沟通

👇👇👇👇👇👇👇👇👇👇👇👇👇👇👇👇👇👇👇👇👇👇