核心内容摘要
黑土腿法：舌尖上的匠心传承，风味中的乡土情深
✨ 本团队擅长数据搜集与处理、建模仿真、程序设计、仿真代码、论文写作与指导毕业论文、期刊论文经验交流。
✅ 专业定制毕设、代码✅ 成品或定制查看文章底部微信二维码(
基于逐次变分模态分解与注意力残差网络的轴承故障诊断方法旋转机械设备在工业生产中广泛应用其运行状态的实时监测和故障及时诊断对于保障生产安全和避免经济损失至关重要。
滚动轴承作为旋转机械的关键支撑部件其振动信号中蕴含着丰富的设备健康状态信息。
然而在实际工业环境中轴承振动信号不可避免地受到强背景噪声的干扰包括其他机械部件的振动传递、电磁干扰和测量系统本身的噪声等导致有用的故障特征信息被噪声掩盖而难以有效提取。
传统的信号处理方法如傅里叶变换和小波分析在处理强噪声干扰下的非平稳信号时存在固有局限性。
本研究提出一种结合逐次变分模态分解与注意力残差网络的轴承故障诊断方法。
逐次变分模态分解通过迭代方式逐个提取信号中的本征模态分量每次提取都通过变分优化确定当前模态的最优中心频率和带宽避免了传统变分模态分解需要预先指定模态数的困难。
针对分解得到的多个模态分量设计了基于能量分布和峭度特征的融合评价指标自动识别并剔除主要包含噪声成分的冗余模态保留包含故障冲击信息的有效模态进行信号重构。
重构后的去噪信号转换为短时傅里叶变换时频图像作为深度学习模型的输入。
在网络结构设计上采用残差网络作为基础架构以解决深层网络的梯度消失问题同时嵌入卷积注意力模块实现对时频图像中关键故障特征区域的自适应聚焦。
卷积注意力模块包含通道注意力和空间注意力两个子模块分别学习不同特征通道和不同空间位置的重要性权重增强与故障判别相关的特征响应抑制噪声和无关成分的干扰。
(
基于时间卷积与Transformer融合的多通道信号端到端诊断模型工业设备的振动监测系统通常部署多个传感器从不同方向和位置采集振动信号形成多通道的监测数据。
单一通道的诊断模型难以充分利用多通道信息之间的互补性和关联性而简单的通道拼接又忽略了不同通道数据的时序特性差异。
此外传统的特征工程方法依赖领域专家的先验知识来设计和选择故障特征在面对新型设备或未知故障模式时缺乏适应性。
针对上述问题本研究提出一种端到端的时间卷积网络与Transformer融合模型能够直接从原始多通道振动信号学习故障判别特征而无需人工特征提取。
在模型的前端采用时间卷积网络结构对输入信号序列进行局部模式提取时间卷积通过因果卷积和膨胀卷积的组合实现对长时序依赖关系的高效建模同时保持时序因果性防止信息泄露。
针对多通道信号的融合问题设计了单向补丁序列标记方法将多通道信号沿时间维度分割为若干补丁每个补丁包含所有通道在该时间段内的采样值通过线性投影转换为统一维度的标记向量。
这种表示方式自然地保留了通道间的同步关系便于后续模型学习跨通道的相关特性。
在模型的后端引入Transformer架构捕捉全局范围内的特征依赖关系。
为了增强模型对位置信息和通道信息的感知能力在标准自注意力机制的基础上设计了混合注意力模块同时计算位置维度和通道维度的注意力权重实现更加全面的特征交互。
(
面向嵌入式部署的Transformer模型剪枝与轻量化方法深度神经网络模型在故障诊断任务中展现出优异的性能但其庞大的参数量和计算复杂度给工业现场的实际部署带来了严峻挑战。
工业监测系统往往需要在资源受限的嵌入式设备上运行这些设备的计算能力、内存容量和功耗预算都十分有限难以支撑复杂深度模型的实时推理。
模型压缩技术是解决这一矛盾的有效途径其中结构化剪枝通过移除网络中冗余的权重、神经元或层来减少模型规模同时尽可能保持原有的诊断精度。
本研究提出一种基于重要性评分的Transformer模型剪枝方法针对Transformer架构的特点设计了适用于注意力模块和前馈网络模块的剪枝策略。
重要性评分通过计算反向传播过程中各网络模块的梯度与激活值的乘积来评估该乘积反映了对应模块对最终输出的贡献程度贡献较小的模块被认为是冗余的可以进行剪枝移除。
对于多头注意力层独立评估每个注意力头的重要性移除贡献最小的若干注意力头以减少计算量。
对于前馈网络层沿隐藏维度进行神经元级别的剪枝保留重要性评分最高的神经元子集。
剪枝后的模型需要经过微调训练以恢复因结构改变导致的精度损失。
import numpy as np import torch import torch.nn as nn import torch.nn.functional as F from scipy.signal import stft from sklearn.preprocessing import StandardScaler class SuccessiveVMD: def __init__(self, max_modes8, alpha
: self.max_modes max_modes self.alpha alpha self.modes [] def extract_single_mode(self, signal, omega_init, tau0, tol1e-7, max_iter
: N len(signal) freqs np.fft.fftfreq(N) f_hat np.fft.fft(signal) u_hat np.zeros(N, dtypecomplex) omega omega_init for _ in range(max_iter): u_hat_old u_hat.copy() numerator f_hat denominator 1 2 * self.alpha * (freqs - omega)**2 u_hat numerator / denominator if np.sum(np.abs(u_hat)**
0: omega np.abs(np.sum(freqs * np.abs(u_hat)**
/ np.sum(np.abs(u_hat)**
) if np.sum(np.abs(u_hat - u_hat_old)**
/ (np.sum(np.abs(u_hat_old)**
1e-
tol: break return np.real(np.fft.ifft(u_hat)), omega def decompose(self, signal): self.modes [] residual signal.copy() for _ in range(self.max_modes): spectrum np.abs(np.fft.fft(residual)) omega_init np.argmax(spectrum[:len(spectrum)//2]) / len(spectrum) mode, _ self.extract_single_mode(residual, omega_init) self.modes.append(mode) residual residual - mode if np.std(residual)
01 * np.std(signal): break return np.array(self.modes) def compute_fusion_index(self, mode): energy np.sum(mode**
kurtosis np.mean(mode**
/ (np.mean(mode**
**2 1e-
return energy * kurtosis def select_and_reconstruct(self, threshold_ratio
0.
: indices [self.compute_fusion_index(m) for m in self.modes] max_index max(indices) selected [m for m, idx in zip(self.modes, indices) if idx threshold_ratio * max_index] return np.sum(selected, axis
class CBAMBlock(nn.Module): def __init__(self, channels, reduction
: super(CBAMBlock, self).__init__() self.channel_attention nn.Sequential( nn.AdaptiveAvgPool2d(
, nn.Flatten(), nn.Linear(channels, channels // reduction), nn.ReLU(), nn.Linear(channels // reduction, channels), ) self.spatial_attention nn.Sequential( nn.Conv2d(2, 1, kernel_size7, padding
, nn.Sigmoid() ) def forward(self, x): avg_out self.channel_attention(x) max_out self.channel_attention(F.adaptive_max_pool2d(x,
.view(x.size(
, -
) channel_att torch.sigmoid(avg_out max_out).unsqueeze(-
.unsqueeze(-
x x * channel_att avg_spatial torch.mean(x, dim1, keepdimTrue) max_spatial, _ torch.max(x, dim1, keepdimTrue) spatial_att self.spatial_attention(torch.cat([avg_spatial, max_spatial], dim
) x x * spatial_att return x class ResidualBlockWithCBAM(nn.Module): def __init__(self, in_channels, out_channels, stride
: super(ResidualBlockWithCBAM, self).__init__() self.conv1 nn.Conv2d(in_channels, out_channels, 3, stride,
self.bn1 nn.BatchNorm2d(out_channels) self.conv2 nn.Conv2d(out_channels, out_channels, 3, 1,
self.bn2 nn.BatchNorm2d(out_channels) self.cbam CBAMBlock(out_channels) self.shortcut nn.Sequential() if stride ! 1 or in_channels ! out_channels: self.shortcut nn.Sequential( nn.Conv2d(in_channels, out_channels, 1, stride), nn.BatchNorm2d(out_channels) ) def forward(self, x): out F.relu(self.bn1(self.conv1(x))) out self.bn2(self.conv2(out)) out self.cbam(out) out self.shortcut(x) return F.relu(out) class CBAMResNet(nn.Module): def __init__(self, num_classes): super(CBAMResNet, self).__init__() self.conv1 nn.Conv2d(1, 64, 7, 2,
self.bn1 nn.BatchNorm2d(
self.pool nn.MaxPool2d(3, 2,
self.layer1 self._make_layer(64, 64,
self.layer2 self._make_layer(64, 128, 2, stride
self.layer3 self._make_layer(128, 256, 2, stride
self.avgpool nn.AdaptiveAvgPool2d((1,
) self.fc nn.Linear(256, num_classes) def _make_layer(self, in_channels, out_channels, num_blocks, stride
: layers [ResidualBlockWithCBAM(in_channels, out_channels, stride)] for _ in range(1, num_blocks): layers.append(ResidualBlockWithCBAM(out_channels, out_channels)) return nn.Sequential(*layers) def forward(self, x): x F.relu(self.bn1(self.conv1(x))) x self.pool(x) x self.layer1(x) x self.layer2(x) x self.layer3(x) x self.avgpool(x) x x.view(x.size(
, -
return self.fc(x) class TemporalBlock(nn.Module): def __init__(self, in_channels, out_channels, kernel_size, dilation): super(TemporalBlock, self).__init__() padding (kernel_size -
* dilation // 2 self.conv1 nn.Conv1d(in_channels, out_channels, kernel_size, paddingpadding, dilationdilation) self.bn1 nn.BatchNorm1d(out_channels) self.conv2 nn.Conv1d(out_channels, out_channels, kernel_size, paddingpadding, dilationdilation) self.bn2 nn.BatchNorm1d(out_channels) self.downsample nn.Conv1d(in_channels, out_channels,
if in_channels ! out_channels else None def forward(self, x): out F.relu(self.bn1(self.conv1(x))) out self.bn2(self.conv2(out)) res x if self.downsample is None else self.downsample(x) return F.relu(out res) class TCN(nn.Module): def __init__(self, input_channels, num_channels, kernel_size
: super(TCN, self).__init__() layers [] num_levels len(num_channels) for i in range(num_levels): dilation 2 ** i in_ch input_channels if i 0 else num_channels[i-1] out_ch num_channels[i] layers.append(TemporalBlock(in_ch, out_ch, kernel_size, dilation)) self.network nn.Sequential(*layers) def forward(self, x): return self.network(x) class MixedAttention(nn.Module): def __init__(self, d_model, num_heads): super(MixedAttention, self).__init__() self.d_model d_model self.num_heads num_heads self.d_k d_model // num_heads self.W_q nn.Linear(d_model, d_model) self.W_k nn.Linear(d_model, d_model) self.W_v nn.Linear(d_model, d_model) self.W_o nn.Linear(d_model, d_model) self.channel_attention nn.Sequential( nn.Linear(d_model, d_model //
, nn.ReLU(), nn.Linear(d_model // 4, d_model), nn.Sigmoid() ) def forward(self, x): batch_size, seq_len, _ x.size() Q self.W_q(x).view(batch_size, seq_len, self.num_heads, self.d_k).transpose(1,
K self.W_k(x).view(batch_size, seq_len, self.num_heads, self.d_k).transpose(1,
V self.W_v(x).view(batch_size, seq_len, self.num_heads, self.d_k).transpose(1,
scores torch.matmul(Q, K.transpose(-2, -
) / np.sqrt(self.d_k) attention F.softmax(scores, dim-
context torch.matmul(attention, V) context context.transpose(1,
.contiguous().view(batch_size, seq_len, self.d_model) channel_weights self.channel_attention(context.mean(dim
).unsqueeze(
context context * channel_weights return self.W_o(context) class TCNTransformer(nn.Module): def __init__(self, input_channels, seq_length, num_classes, d_model128, num_heads
: super(TCNTransformer, self).__init__() self.tcn TCN(input_channels, [32, 64, d_model]) self.patch_size 16 self.num_patches seq_length // self.patch_size self.patch_embedding nn.Linear(self.patch_size * d_model, d_model) self.pos_embedding nn.Parameter(torch.randn(1, self.num_patches, d_model)) self.attention_layers nn.ModuleList([ MixedAttention(d_model, num_heads) for _ in range(
]) self.layer_norms nn.ModuleList([nn.LayerNorm(d_model) for _ in range(
]) self.classifier nn.Linear(d_model, num_classes) def forward(self, x): tcn_out self.tcn(x) batch_size, channels, seq_len tcn_out.size() patches tcn_out.view(batch_size, channels, self.num_patches, self.patch_size) patches patches.permute(0, 2, 1,
.contiguous() patches patches.view(batch_size, self.num_patches, -
x self.patch_embedding(patches) self.pos_embedding for attn, norm in zip(self.attention_layers, self.layer_norms): x norm(x attn(x)) x x.mean(dim
return self.classifier(x) class TransformerPruner: def __init__(self, model, target_sparsity
0.
: self.model model self.target_sparsity target_sparsity self.importance_scores {} def compute_importance_scores(self, data_loader, criterion): self.model.train() gradients {name: [] for name, _ in self.model.named_parameters()} activations {name: [] for name, _ in self.model.named_modules()} for batch in data_loader: x, y batch self.model.zero_grad() output self.model(x) loss criterion(output, y) loss.backward() for name, param in self.model.named_parameters(): if param.grad is not None: gradients[name].append(param.grad.abs().mean().item()) for name in gradients: if gradients[name]: self.importance_scores[name] np.mean(gradients[name]) def prune_attention_heads(self, layer, num_heads_to_prune): with torch.no_grad(): W_q layer.W_q.weight.view(layer.num_heads, layer.d_k, -
head_importance torch.norm(W_q, dim(1,
) _, indices_to_prune torch.topk(head_importance, num_heads_to_prune, largestFalse) mask torch.ones(layer.num_heads, dtypetorch.bool) mask[indices_to_prune] False new_num_heads layer.num_heads - num_heads_to_prune new_d_model new_num_heads * layer.d_k return mask, new_num_heads, new_d_model def prune_feedforward(self, linear_layer, num_neurons_to_prune): with torch.no_grad(): neuron_importance torch.norm(linear_layer.weight, dim
_, indices_to_keep torch.topk(neuron_importance, linear_layer.out_features - num_neurons_to_prune) new_weight linear_layer.weight[indices_to_keep] new_bias linear_layer.bias[indices_to_keep] if linear_layer.bias is not None else None new_layer nn.Linear(linear_layer.in_features, len(indices_to_keep)) new_layer.weight.data new_weight if new_bias is not None: new_layer.bias.data new_bias return new_layer def iterative_pruning(self, data_loader, criterion, num_iterations
: sparsity_per_iteration self.target_sparsity / num_iterations for iteration in range(num_iterations): self.compute_importance_scores(data_loader, criterion) for name, module in self.model.named_modules(): if isinstance(module, MixedAttention): num_to_prune int(module.num_heads * sparsity_per_iteration) if num_to_prune 0: mask, new_heads, new_dim self.prune_attention_heads(module, num_to_prune) return self.model class LightweightFaultDiagnoser: def __init__(self, input_channels1, num_classes
: self.svmd SuccessiveVMD() self.model TCNTransformer(input_channels, seq_length1024, num_classesnum_classes) self.pruner TransformerPruner(self.model) self.scaler StandardScaler() def preprocess(self, signal): modes self.svmd.decompose(signal) reconstructed self.svmd.select_and_reconstruct() f, t, Zxx stft(reconstructed, nperseg
return np.abs(Zxx) def train(self, train_signals, train_labels, epochs
: processed [self.preprocess(s) for s in train_signals] X torch.FloatTensor(np.array(processed)).unsqueeze(
y torch.LongTensor(train_labels) dataset torch.utils.data.TensorDataset(X, y) loader torch.utils.data.DataLoader(dataset, batch_size32, shuffleTrue) optimizer torch.optim.Adam(self.model.parameters(), lr
0.