✅博主简介：擅长数据搜集与处理、建模仿真、程序设计、仿真代码、论文写作与指导，毕业论文、期刊论文经验交流。
✅成品或者定制，扫描文章底部微信二维码。
(1) 基于时频域多模态信息的旋转机械特征表示方法
旋转机械在运行过程中产生的振动信号蕴含着丰富的设备健康状态信息，但单一时域信号的表达能力有限，难以充分揭示复杂故障模式的细微特征差异。时域信号能够直观反映振动幅值的瞬时变化和冲击响应特性，对于突发性故障如轴承滚动体剥落具有较好的敏感性，但对于渐进性退化故障如齿轮磨损的早期特征捕捉能力较弱。频域分析通过傅里叶变换将信号分解为不同频率成分的叠加，能够清晰地呈现与旋转部件特征频率相关的故障信息，但丢失了时间维度的动态演化规律。为了综合利用不同信号表示形式的互补优势，本研究提出将原始振动信号同时转换为时域序列和时频谱图两种模态作为诊断模型的输入。时频变换采用连续小波变换方法，通过选择与机械振动特性相匹配的母小波函数，将一维时间序列映射为二维的时频能量分布图像，该图像的横轴表示时间、纵轴表示频率、像素亮度表示对应时频点的能量强度。与短时傅里叶变换相比，小波变换具有多分辨率分析的特性，能够在低频段提供较高的频率分辨率以识别旋转频率及其谐波成分，在高频段提供较高的时间分辨率以捕捉瞬态冲击响应。这种多模态的数据表示策略使得后续的深度学习模型能够从不同的视角挖掘故障特征信息，时域通道关注信号的整体包络变化和统计特性，频域通道聚焦于特定频率成分的异常增强或消失，两者的协同作用有望实现比单一模态更加全面和准确的故障识别效果。
(2) 多尺度特征融合神经网络的架构设计与优化
针对多模态输入数据的特征提取和融合问题，本研究设计了一种由时域处理通道和频域处理通道并行构成的双分支神经网络架构。时域通道采用一维卷积神经网络结构处理原始振动序列，通过堆叠多层卷积和池化操作逐步提取从局部波形细节到全局统计特性的层次化特征表示。为了缓解深层网络训练过程中的梯度消失问题并保留浅层特征的有效信息，在卷积模块之间引入残差跳跃连接，使得网络能够学习输入与输出之间的残差映射而非直接映射。频域通道采用二维卷积神经网络结构处理时频谱图图像，由于时频图中不同区域的信息重要性存在差异，某些局部区域可能包含与故障高度相关的特征而其他区域则主要是背景噪声，为此引入卷积注意力机制模块自适应地调整不同空间位置和通道维度的特征权重。该注意力模块首先沿空间维度进行全局平均池化和最大池化生成通道注意力向量，然后沿通道维度进行池化生成空间注意力图，两者依次作用于原始特征图实现对关键故障特征的突出强调和对无关信息的抑制。在特征融合阶段，将时域通道和频域通道提取的高级语义特征向量进行拼接，通过全连接层进行跨模态的信息交互和综合推理，最终输出各故障类别的概率分布。网络的训练采用交叉熵损失函数和自适应学习率优化算法，通过在标准旋转机械故障数据集上的大量对比实验验证，所提出的多模态特征融合网络在诊断准确率、训练收敛速度和对噪声干扰的鲁棒性等方面均优于单模态基线模型。
(3) 面向标注数据稀缺场景的域适应迁移学习方法
深度学习模型的优异性能依赖于大规模标注数据集的支撑，但在实际工业应用中获取充足的故障标签样本往往面临诸多困难。一方面，旋转机械通常在正常状态下运行大部分时间，故障发生的概率相对较低且分布不均匀；另一方面，对故障样本进行准确标注需要经验丰富的领域专家参与，人工标注成本高昂且难以保证一致性。针对目标域标注数据稀缺的实际情况，本研究提出一种基于域适应技术的迁移学习策略，利用已有的源域标注数据训练诊断模型，并通过无监督域适应方法将模型迁移到目标域实现跨域故障诊断。考虑到源域和目标域数据之间可能存在的分布差异，包括由不同采集设备、工况条件和环境因素造成的数据偏移，直接应用源域训练的模型往往难以取得理想效果。为此在特征提取网络之后引入域对抗训练机制，通过一个辅助的域判别器区分特征来自源域还是目标域，同时训练特征提取器生成能够混淆域判别器的域不变特征表示。这种对抗博弈过程促使网络学习到与具体域无关但与故障类别相关的本质特征
import numpy as np import torch import torch.nn as nn import torch.optim as optim from torch.utils.data import DataLoader, TensorDataset import pywt from sklearn.model_selection import train_test_split class WaveletTransformer: def __init__(self, wavelet='morl', scales=64): self.wavelet = wavelet self.scales = np.arange(1, scales + 1) def transform(self, signal): coefficients, _ = pywt.cwt(signal, self.scales, self.wavelet) return np.abs(coefficients) def batch_transform(self, signals): return np.array([self.transform(s) for s in signals]) class ChannelAttention(nn.Module): def __init__(self, channels, reduction=16): super(ChannelAttention, self).__init__() self.avg_pool = nn.AdaptiveAvgPool2d(1) self.max_pool = nn.AdaptiveMaxPool2d(1) self.fc = nn.Sequential( nn.Linear(channels, channels // reduction, bias=False), nn.ReLU(), nn.Linear(channels // reduction, channels, bias=False) ) self.sigmoid = nn.Sigmoid() def forward(self, x): b, c, _, _ = x.size() avg_out = self.fc(self.avg_pool(x).view(b, c)) max_out = self.fc(self.max_pool(x).view(b, c)) attention = self.sigmoid(avg_out + max_out).view(b, c, 1, 1) return x * attention class SpatialAttention(nn.Module): def __init__(self, kernel_size=7): super(SpatialAttention, self).__init__() self.conv = nn.Conv2d(2, 1, kernel_size, padding=kernel_size // 2, bias=False) self.sigmoid = nn.Sigmoid() def forward(self, x): avg_out = torch.mean(x, dim=1, keepdim=True) max_out, _ = torch.max(x, dim=1, keepdim=True) concat = torch.cat([avg_out, max_out], dim=1) attention = self.sigmoid(self.conv(concat)) return x * attention class CBAMBlock(nn.Module): def __init__(self, channels): super(CBAMBlock, self).__init__() self.channel_attention = ChannelAttention(channels) self.spatial_attention = SpatialAttention() def forward(self, x): x = self.channel_attention(x) x = self.spatial_attention(x) return x class ResidualBlock1D(nn.Module): def __init__(self, in_channels, out_channels, stride=1): super(ResidualBlock1D, self).__init__() self.conv1 = nn.Conv1d(in_channels, out_channels, 3, stride, 1) self.bn1 = nn.BatchNorm1d(out_channels) self.conv2 = nn.Conv1d(out_channels, out_channels, 3, 1, 1) self.bn2 = nn.BatchNorm1d(out_channels) self.shortcut = nn.Sequential() if stride != 1 or in_channels != out_channels: self.shortcut = nn.Sequential( nn.Conv1d(in_channels, out_channels, 1, stride), nn.BatchNorm1d(out_channels) ) self.relu = nn.ReLU() def forward(self, x): out = self.relu(self.bn1(self.conv1(x))) out = self.bn2(self.conv2(out)) out += self.shortcut(x) return self.relu(out) class TimeDomainEncoder(nn.Module): def __init__(self, input_length, embed_dim=128): super(TimeDomainEncoder, self).__init__() self.conv1 = nn.Conv1d(1, 32, 7, 2, 3) self.bn1 = nn.BatchNorm1d(32) self.relu = nn.ReLU() self.res1 = ResidualBlock1D(32, 64, 2) self.res2 = ResidualBlock1D(64, 128, 2) self.pool = nn.AdaptiveAvgPool1d(1) self.fc = nn.Linear(128, embed_dim) def forward(self, x): if x.dim() == 2: x = x.unsqueeze(1) x = self.relu(self.bn1(self.conv1(x))) x = self.res1(x) x = self.res2(x) x = self.pool(x).squeeze(-1) return self.fc(x) class FrequencyDomainEncoder(nn.Module): def __init__(self, embed_dim=128): super(FrequencyDomainEncoder, self).__init__() self.conv1 = nn.Conv2d(1, 32, 3, 1, 1) self.bn1 = nn.BatchNorm2d(32) self.conv2 = nn.Conv2d(32, 64, 3, 2, 1) self.bn2 = nn.BatchNorm2d(64) self.cbam = CBAMBlock(64) self.conv3 = nn.Conv2d(64, 128, 3, 2, 1) self.bn3 = nn.BatchNorm2d(128) self.pool = nn.AdaptiveAvgPool2d(1) self.fc = nn.Linear(128, embed_dim) self.relu = nn.ReLU() def forward(self, x): if x.dim() == 3: x = x.unsqueeze(1) x = self.relu(self.bn1(self.conv1(x))) x = self.relu(self.bn2(self.conv2(x))) x = self.cbam(x) x = self.relu(self.bn3(self.conv3(x))) x = self.pool(x).squeeze(-1).squeeze(-1) return self.fc(x) class MultimodalFusionNetwork(nn.Module): def __init__(self, input_length, num_classes, embed_dim=128): super(MultimodalFusionNetwork, self).__init__() self.time_encoder = TimeDomainEncoder(input_length, embed_dim) self.freq_encoder = FrequencyDomainEncoder(embed_dim) self.fusion = nn.Sequential( nn.Linear(embed_dim * 2, 256), nn.ReLU(), nn.Dropout(0.3), nn.Linear(256, 128), nn.ReLU(), nn.Dropout(0.3), nn.Linear(128, num_classes) ) def forward(self, time_signal, freq_spectrogram): time_feat = self.time_encoder(time_signal) freq_feat = self.freq_encoder(freq_spectrogram) fused = torch.cat([time_feat, freq_feat], dim=1) return self.fusion(fused) class DomainDiscriminator(nn.Module): def __init__(self, input_dim): super(DomainDiscriminator, self).__init__() self.classifier = nn.Sequential( nn.Linear(input_dim, 128), nn.ReLU(), nn.Dropout(0.3), nn.Linear(128, 64), nn.ReLU(), nn.Linear(64, 1), nn.Sigmoid() ) def forward(self, x): return self.classifier(x) class GradientReversalLayer(torch.autograd.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 DomainAdaptationNetwork(nn.Module): def __init__(self, input_length, num_classes, embed_dim=128): super(DomainAdaptationNetwork, self).__init__() self.feature_extractor = MultimodalFusionNetwork(input_length, num_classes, embed_dim) self.domain_classifier = DomainDiscriminator(embed_dim * 2) def forward(self, time_signal, freq_spectrogram, alpha=1.0): time_feat = self.feature_extractor.time_encoder(time_signal) freq_feat = self.feature_extractor.freq_encoder(freq_spectrogram) features = torch.cat([time_feat, freq_feat], dim=1) class_output = self.feature_extractor.fusion(features) reversed_features = GradientReversalLayer.apply(features, alpha) domain_output = self.domain_classifier(reversed_features) return class_output, domain_output, features class RotatingMachineryDiagnosisSystem: def __init__(self, signal_length=2048, num_classes=10): self.wavelet = WaveletTransformer() self.model = MultimodalFusionNetwork(signal_length, num_classes) self.optimizer = optim.Adam(self.model.parameters(), lr=0.001) self.criterion = nn.CrossEntropyLoss() def preprocess(self, signals): spectrograms = self.wavelet.batch_transform(signals) return torch.FloatTensor(signals), torch.FloatTensor(spectrograms) def train(self, train_signals, train_labels, epochs=100, batch_size=32): time_data, freq_data = self.preprocess(train_signals) labels = torch.LongTensor(train_labels) dataset = TensorDataset(time_data, freq_data, labels) loader = DataLoader(dataset, batch_size=batch_size, shuffle=True) for epoch in range(epochs): total_loss = 0 for time_batch, freq_batch, label_batch in loader: self.optimizer.zero_grad() outputs = self.model(time_batch, freq_batch) loss = self.criterion(outputs, label_batch) loss.backward() self.optimizer.step() total_loss += loss.item() def diagnose(self, signal): self.model.eval() time_data, freq_data = self.preprocess(signal.reshape(1, -1)) with torch.no_grad(): output = self.model(time_data, freq_data) pred = torch.argmax(output, dim=1).item() fault_names = ['normal', 'inner_race', 'outer_race', 'ball', 'cage', 'gear_wear', 'gear_crack', 'shaft_misalignment', 'unbalance', 'looseness'] return fault_names[pred]如有问题，可以直接沟通
👇👇👇👇👇👇👇👇👇👇👇👇👇👇👇👇👇👇👇👇👇👇
)
