核心内容摘要
四川亲子乱对白：那些年，我们一起“坑”过的娃
✅博主简介擅长数据搜集与处理、建模仿真、程序设计、仿真代码、论文写作与指导毕业论文、期刊论文经验交流。
✅成品或者定制扫描文章底部微信二维码。
(
无监督时频对比预训练的特征表示学习框架针对工业现场轴承故障标注样本稀缺而无标签运行数据充足的现状,本研究构建基于对比学习的自监督预训练策略充分挖掘未标注数据的特征表达价值。
时频对比学习通过构造正负样本对引导模型学习判别性特征表示,将原始振动时序信号同时转换为时域波形和频域频谱作为双视图输入,对同一信号源的时频表示视为正样本对,不同信号源的表示视为负样本对。
编码器网络采用孪生结构分别处理时域和频域输入,通过共享权重的卷积层提取局部特征,全局平均池化层生成紧凑的嵌入向量。
对比损失函数基于InfoNCE准则设计,最大化正样本对在特征空间的相似度同时推开负样本对,促使模型学习与故障模式相关的不变特征而忽略噪声和工况变化带来的干扰。
为增强样本多样性,设计针对振动信号的数据增强策略,包括时域随机裁剪、幅值缩放、高斯噪声注入和频域随机掩码,通过对同一信号应用不同增强操作生成多个视图构成正样本对。
预训练阶段在大规模无标签数据集上进行,编码器通过对比学习目标优化参数,学习到的特征表示能够捕获轴承运行状态的本质特征。
微调阶段将预训练编码器作为特征提取器,仅在少量标注样本上训练轻量级分类头,通过迁移预训练知识快速适应特定故障诊断任务,显著降低对标注数据的依赖。
(
时频信息深度融合的Transformer原型网络架构考虑到时域和频域信息分别反映轴承故障的不同侧面,本研究设计多模态融合网络实现时频特征的互补增强。
时域分支采用倒置残差卷积结构处理原始振动信号,通过深度可分离卷积降低参数量,扩张卷积层增大感受野捕获长距离时序依赖,批归一化和ReLU激活函数增强非线性表达能力。
频域分支对短时傅里叶变换得到的时频谱进行处理,利用二维卷积提取频率分布模式和时频局部纹理特征。
提取的时频特征分别输入Transformer编码器进行全局上下文建模,通过多头自注意力机制捕获特征序列的长距离依赖关系,前馈网络进一步提升特征的抽象层次。
为实现时频特征的深度交互,设计交叉注意力融合模块,将时域特征作为查询向量,频域特征作为键值向量进行注意力计算,使时域特征能够选择性地关注频域中的判别性成分,反之亦然。
融合后的多模态特征进入原型分类网络,通过计算查询样本与各类别原型中心的距离进行分类。
原型中心定义为支持集中同类样本嵌入向量的均值,分类决策基于欧氏距离或余弦相似度进行最近邻匹配。
针对小样本场景下原型估计不稳定的问题,引入原型分布平衡正则化,通过最小化类间原型分布的KL散度促使各类别原型在特征空间均匀分布,避免某些类别的原型过度聚集导致分类偏差。
训练采用episode采样策略,每个episode从训练集中随机抽取N个类别,每类K个支持样本和Q个查询样本,模拟真实小样本测试场景。
(
跨工况泛化的元学习与领域适应联合优化针对轴承在不同转速和负载工况下振动特性差异导致的诊断性能下降问题,本研究结合元学习和领域适应技术提升模型的跨工况泛化能力。
元学习通过在多个源工况任务上进行二阶优化,学习对新工况快速适应的初始化参数,使模型能够仅用少量目标工况样本实现高精度诊断。
任务级训练将不同工况下的故障诊断视为不同任务,每个任务包含支持集和查询集,模型在支持集上进行内循环更新,在查询集上计算元损失进行外循环优化,通过任务间梯度聚合学习任务无关的通用表示。
为进一步缩小源域和目标域的分布差异,引入领域适应模块对齐特征分布,采用最大均值差异准则度量两个分布的距离,通过最小化MMD损失促使特征提取器生成领域不变表示。
设计自适应权重调整机制动态平衡分类损失和领域对齐损失,在训练初期侧重特征提取能力的提升,后期逐步增强领域对齐约束。
import numpy as np import torch import torch.nn as nn import torch.nn.functional as F from scipy.fftpack import fft from sklearn.preprocessing import StandardScaler class TimeFrequencyEncoder(nn.Module): def __init__(self, signal_length1024, embedding_dim
: super(TimeFrequencyEncoder, self).__init__() self.time_encoder nn.Sequential( nn.Conv1d(1, 32, kernel_size64, stride
, nn.BatchNorm1d(
, nn.ReLU(), nn.Conv1d(32, 64, kernel_size16, stride
, nn.BatchNorm1d(
, nn.ReLU(), nn.AdaptiveAvgPool1d(
) self.freq_encoder nn.Sequential( nn.Conv1d(1, 32, kernel_size64, stride
, nn.BatchNorm1d(
, nn.ReLU(), nn.Conv1d(32, 64, kernel_size16, stride
, nn.BatchNorm1d(
, nn.ReLU(), nn.AdaptiveAvgPool1d(
) self.projection nn.Sequential( nn.Linear(128, embedding_dim), nn.ReLU(), nn.Linear(embedding_dim, embedding_dim) ) def forward(self, time_signal, freq_signal): time_feat self.time_encoder(time_signal).squeeze(-
freq_feat self.freq_encoder(freq_signal).squeeze(-
combined torch.cat([time_feat, freq_feat], dim
embedding self.projection(combined) return F.normalize(embedding, dim
class ContrastiveLoss(nn.Module): def __init__(self, temperature
0.
: super(ContrastiveLoss, self).__init__() self.temperature temperature def forward(self, embeddings, labels): batch_size embeddings.size(
similarity_matrix torch.matmul(embeddings, embeddings.T) / self.temperature mask torch.eye(batch_size, deviceembeddings.device).bool() similarity_matrix.masked_fill_(mask, -1e
labels labels.unsqueeze(
label_mask (labels labels.T).float() label_mask.masked_fill_(mask,
positives similarity_matrix * label_mask negatives similarity_matrix * (1 - label_mask) pos_exp torch.exp(positives).sum(dim
all_exp torch.exp(similarity_matrix).sum(dim
loss -torch.log(pos_exp / (all_exp 1e-
1e-
return loss.mean() class InvertedResidualBlock(nn.Module): def __init__(self, in_channels, out_channels, expand_ratio
: super(InvertedResidualBlock, self).__init__() hidden_dim in_channels * expand_ratio self.use_residual in_channels out_channels self.conv nn.Sequential( nn.Conv1d(in_channels, hidden_dim, 1, biasFalse), nn.BatchNorm1d(hidden_dim), nn.ReLU6(inplaceTrue), nn.Conv1d(hidden_dim, hidden_dim, 3, padding1, groupshidden_dim, biasFalse), nn.BatchNorm1d(hidden_dim), nn.ReLU6(inplaceTrue), nn.Conv1d(hidden_dim, out_channels, 1, biasFalse), nn.BatchNorm1d(out_channels) ) def forward(self, x): if self.use_residual: return x self.conv(x) return self.conv(x) class CrossAttentionFusion(nn.Module): def __init__(self, dim, num_heads
: super(CrossAttentionFusion, self).__init__() self.num_heads num_heads self.scale (dim // num_heads) ** -
5 self.q_proj nn.Linear(dim, dim) self.k_proj nn.Linear(dim, dim) self.v_proj nn.Linear(dim, dim) self.out_proj nn.Linear(dim, dim) def forward(self, x1, x
: batch_size, seq_len, dim x
size() q self.q_proj(x
.view(batch_size, seq_len, self.num_heads, -
.transpose(1,
k self.k_proj(x
.view(batch_size, -1, self.num_heads, dim // self.num_heads).transpose(1,
v self.v_proj(x
.view(batch_size, -1, self.num_heads, dim // self.num_heads).transpose(1,
attn torch.matmul(q, k.transpose(-2, -
) * self.scale attn F.softmax(attn, dim-
out torch.matmul(attn, v) out out.transpose(1,
.contiguous().view(batch_size, seq_len, dim) out self.out_proj(out) return out class PrototypicalNetwork(nn.Module): def __init__(self, input_dim128, hidden_dim
: super(PrototypicalNetwork, self).__init__() self.time_branch nn.Sequential( InvertedResidualBlock(1,
, InvertedResidualBlock(32,
, nn.AdaptiveAvgPool1d(
) encoder_layer nn.TransformerEncoderLayer(d_model64, nhead4, dim_feedforward256, batch_firstTrue) self.transformer nn.TransformerEncoder(encoder_layer, num_layers
self.cross_attn CrossAttentionFusion(64, num_heads
self.fc nn.Sequential( nn.Linear(64 * 16, hidden_dim), nn.ReLU(), nn.Dropout(
0.
, nn.Linear(hidden_dim, input_dim) ) def forward(self, x): batch_size x.size(
x self.time_branch(x) x x.transpose(1,
x self.transformer(x) x_fusion self.cross_attn(x, x) x (x x_fusion).view(batch_size, -
x self.fc(x) return x def compute_prototypes(self, support_embeddings, support_labels, num_classes): prototypes [] for c in range(num_classes): mask support_labels c class_embeddings support_embeddings[mask] prototype class_embeddings.mean(dim
prototypes.append(prototype) return torch.stack(prototypes) def classify(self, query_embeddings, prototypes): distances torch.cdist(query_embeddings, prototypes) return -distances def euclidean_distance(x, y): return torch.cdist(x, y) def prototype_regularization(prototypes): num_prototypes prototypes.size(
pairwise_dist euclidean_distance(prototypes, prototypes) mask torch.eye(num_prototypes, deviceprototypes.device).bool() pairwise_dist.masked_fill_(mask, 1e
min_dist pairwise_dist.min(dim
[0] reg_loss -torch.log(min_dist 1e-
.mean() return reg_loss def generate_bearing_data(num_samples1000, signal_length
: X_time [] X_freq [] y [] for i in range(num_samples): fault_type np.random.randint(0,
t np.linspace(0, 1, signal_length) if fault_type 0: signal np.sin(2 * np.pi * 50 * t)
2 * np.random.randn(signal_length) elif fault_type 1: signal np.sin(2 * np.pi * 80 * t)
5 * np.sin(2 * np.pi * 200 * t) impulse_locs np.random.randint(0, signal_length,
signal[impulse_locs] np.random.randn(
*
0 elif fault_type 2: signal np.sin(2 * np.pi * 120 * t)
3 * np.sin(2 * np.pi * 300 * t) else: freq 50 fault_type * 30 signal np.sin(2 * np.pi * freq * t)
3 * np.random.randn(signal_length) freq_signal np.abs(fft(signal))[:signal_length//2] freq_signal np.pad(freq_signal, (0, signal_length - len(freq_signal))) X_time.append(signal) X_freq.append(freq_signal) y.append(fault_type) X_time np.array(X_time)[:, np.newaxis, :] X_freq np.array(X_freq)[:, np.newaxis, :] y np.array(y) return X_time.astype(np.float
, X_freq.astype(np.float
, y X_time, X_freq, y generate_bearing_data(
encoder TimeFrequencyEncoder(signal_length1024, embedding_dim
contrastive_loss ContrastiveLoss(temperature
0.
time_tensor torch.FloatTensor(X_time) freq_tensor torch.FloatTensor(X_freq) label_tensor torch.LongTensor(y) optimizer torch.optim.Adam(encoder.parameters(), lr
0.
encoder.train() for epoch in range(
: optimizer.zero_grad() embeddings encoder(time_tensor, freq_tensor) loss contrastive_loss(embeddings, label_tensor) loss.backward() optimizer.step() if (epoch
% 10 0: print(fEpoch {epoch1}: Contrastive Loss{loss.item():.4f}) proto_net PrototypicalNetwork(input_dim128, hidden_dim
optimizer_proto torch.optim.Adam(proto_net.parameters(), lr
0.