✅博主简介擅长数据搜集与处理、建模仿真、程序设计、仿真代码、论文写作与指导毕业论文、期刊论文经验交流。✅成品或者定制扫描文章底部微信二维码。1基于先验知识的分层图构建风电机组各子系统如传动链、偏航系统、变桨系统内部传感器变量间存在强相关性而不同子系统间的相关性相对较弱。基于这一先验知识我们提出一种分层图构建方法。首先根据风电机组的物理结构将整个系统划分为多个局部子系统例如叶片子系统、齿轮箱子系统、发电机子系统等。在每个子系统中将传感器变量作为图的节点并根据机理关系如机械连接、热力学耦合确定节点间的初始边连接形成局部图。随后为了捕捉跨子系统的潜在关联构建全局图将各局部图的节点通过全连接或基于相关性阈值连接形成一个包含所有节点的大型图。这样分层图结构既保留了子系统内部的紧密连接又允许信息在不同子系统间流动为后续的时空建模提供了符合物理背景的拓扑结构。2多层时空图神经网络与故障溯源在构建的分层图基础上我们设计了一种多层时空图神经网络来提取时空特征。该网络由多个时空卷积模块堆叠而成每个模块包含图卷积层和时间卷积层。图卷积层用于聚合邻居节点的空间信息更新节点特征时间卷积层则沿时间维度进行一维卷积捕捉时间依赖。通过堆叠多个模块模型能够逐步学习到不同层次的空间关联和长期时间依赖。为了进行故障检测我们引入PCA重构残差分析。具体而言利用正常工况下的数据训练PCA模型得到每个变量的重构值。在监测阶段将时空图网络提取的特征作为输入通过PCA重构并计算每个变量的重构残差。当某个变量发生故障时其重构残差会显著增大从而定位故障变量。进一步结合分层图结构可以追溯故障的传播路径通过分析残差在图中相邻节点间的传递顺序推断故障的起始部位和扩散过程。在真实风电机组数据集上的验证表明该方法能够有效实现早期故障预警并准确定位故障源。3基于注意力机制的多机组故障诊断同一风电场内的多台风电机组通常处于相似的环境条件下其运行数据存在时空相关性。利用多机组信息可以增强单机组故障诊断的鲁棒性减少工况波动带来的误报。本文提出一种基于注意力机制的多机组故障诊断方法。首先对每台机组的传感器数据分别构建分层图并通过共享权重的多层时空图神经网络提取各自的时空特征。然后引入注意力机制计算不同机组特征之间的相似度权重将关联紧密的机组特征进行加权融合。例如对于目标机组通过注意力汇聚其他机组的特征形成增强的特征表示。融合后的特征再输入故障分类器或残差检测模块。这种多机组注意力融合能够有效利用风场整体信息抑制单机组上的随机噪声和工况波动提升诊断模型的稳定性和准确率。实验结果表明该方法在真实SCADA数据上相比单机组模型具有更低的误报率和更高的故障检出率。import torch import torch.nn as nn import torch.nn.functional as F import numpy as np from torch_geometric.nn import GCNConv from torch_geometric.data import Data, Batch # 1. 分层图构建简化版 def build_hierarchical_graph(num_nodes_per_subsystem, edges_within, edges_between): 构建分层图 :param num_nodes_per_subsystem: 每个子系统的节点数列表 :param edges_within: 每个子系统内部的边列表 (列表的列表) :param edges_between: 子系统之间的边列表 :return: edge_index 整体图的边索引 offset 0 all_edges [] for i, n_nodes in enumerate(num_nodes_per_subsystem): # 子系统内部边 for edge in edges_within[i]: all_edges.append([edge[0] offset, edge[1] offset]) offset n_nodes # 子系统之间的边 for edge in edges_between: all_edges.append([edge[0], edge[1]]) return torch.tensor(all_edges, dtypetorch.long).t().contiguous() # 2. 时空图卷积模块 class SpatialTemporalConv(nn.Module): def __init__(self, in_channels, out_channels, kernel_size3): super(SpatialTemporalConv, self).__init__() self.gcn GCNConv(in_channels, out_channels) self.temporal_conv nn.Conv1d(out_channels, out_channels, kernel_size, paddingkernel_size//2) def forward(self, x, edge_index): # x: [num_nodes, seq_len, features] (batch处理需合并) num_nodes, seq_len, feat x.shape x x.view(num_nodes * seq_len, feat) # 合并时间和节点 x self.gcn(x, edge_index) x x.view(num_nodes, seq_len, -1) x x.permute(1, 2, 0) # [seq_len, features, num_nodes] x self.temporal_conv(x) # [seq_len, features, num_nodes] x x.permute(2, 0, 1) # [num_nodes, seq_len, features] return x class HierarchicalSTGNN(nn.Module): def __init__(self, node_features, hidden_dim, num_classes): super(HierarchicalSTGNN, self).__init__() self.stconv1 SpatialTemporalConv(node_features, hidden_dim) self.stconv2 SpatialTemporalConv(hidden_dim, hidden_dim) self.fc nn.Linear(hidden_dim, num_classes) def forward(self, data): x, edge_index data.x, data.edge_index # x: [num_nodes, seq_len, features] x self.stconv1(x, edge_index) x F.relu(x) x self.stconv2(x, edge_index) # 全局平均池化 x x.mean(dim0) # [seq_len, hidden_dim] x x.mean(dim0) # [hidden_dim] return self.fc(x) # 3. 基于PCA的重构残差故障检测 class PCABasedDetector: def __init__(self, n_components0.95): from sklearn.decomposition import PCA self.pca PCA(n_componentsn_components) def fit(self, normal_data): self.pca.fit(normal_data) def reconstruct_error(self, data): transformed self.pca.transform(data) reconstructed self.pca.inverse_transform(transformed) errors np.mean((data - reconstructed) ** 2, axis1) return errors # 4. 多机组注意力融合模块 class MultiTurbineAttention(nn.Module): def __init__(self, feature_dim): super(MultiTurbineAttention, self).__init__() self.query nn.Linear(feature_dim, feature_dim) self.key nn.Linear(feature_dim, feature_dim) self.value nn.Linear(feature_dim, feature_dim) self.scale feature_dim ** 0.5 def forward(self, x): # x: [num_turbines, feature_dim] Q self.query(x) K self.key(x) V self.value(x) attn_weights torch.softmax(Q K.T / self.scale, dim-1) out attn_weights V return out, attn_weights class MultiTurbineFusionModel(nn.Module): def __init__(self, base_stgnn, feature_dim, num_classes): super(MultiTurbineFusionModel, self).__init__() self.base_stgnn base_stgnn # 共享的STGNN self.attention MultiTurbineAttention(feature_dim) self.classifier nn.Linear(feature_dim, num_classes) def forward(self, batch_list): # batch_list: 每个机组的Data对象列表 features [] for data in batch_list: feat self.base_stgnn(data) # 假设返回特征向量 features.append(feat) features torch.stack(features) # [num_turbines, feature_dim] fused_feat, attn self.attention(features) # 使用目标机组假设第一个的融合特征 out self.classifier(fused_feat[0]) return out, attn # 5. 示例训练 if __name__ __main__: # 模拟数据: 2个机组每个机组3个节点时间步长10特征维度5 num_nodes 3 seq_len 10 feat_dim 5 edge_index torch.tensor([[0,1,2,0,1,2], [1,2,0,2,0,1]], dtypetorch.long) # 简单图 # 构建数据 data_list [] for i in range(2): # 两个机组 x torch.randn(num_nodes, seq_len, feat_dim) data Data(xx, edge_indexedge_index) data_list.append(data) # 初始化模型 base_model HierarchicalSTGNN(node_featuresfeat_dim, hidden_dim16, num_classes3) # 3种故障 fusion_model MultiTurbineFusionModel(base_model, feature_dim16, num_classes3) # 前向 out, attn fusion_model(data_list) print(预测输出:, out.shape) print(注意力权重:, attn) # 训练步骤 optimizer torch.optim.Adam(fusion_model.parameters(), lr0.001) labels torch.randint(0, 3, (1,)) # 假设目标机组标签 loss F.cross_entropy(out.unsqueeze(0), labels) optimizer.zero_grad() loss.backward() optimizer.step() print(损失:, loss.item())如有问题可以直接沟通