三一国际网站设计,获取网站js,重庆品牌设计公司,深圳学网站开发1. 从零开始#xff1a;为什么CWRU轴承数据集是故障诊断的“必修课”#xff1f; 如果你刚接触工业设备的故障诊断#xff0c;或者想找一个经典、干净、又足够有挑战的数据集来练手#xff0c;那凯斯西储大学的这个轴承数据集#xff0c;绝对是你的不二之选。我刚开始做这…1. 从零开始为什么CWRU轴承数据集是故障诊断的“必修课”如果你刚接触工业设备的故障诊断或者想找一个经典、干净、又足够有挑战的数据集来练手那凯斯西储大学的这个轴承数据集绝对是你的不二之选。我刚开始做这个方向的时候也试过不少公开数据但要么数据量太小要么工况太单一要么就是数据格式乱七八糟预处理就得花掉一大半时间。CWRU轴承数据集不一样它就像教科书里的标准例题结构清晰背景明确几乎成了这个领域论文和项目的“基准线”。你去看任何一篇关于旋转机械故障诊断的顶会文章十有八九都会用它来做性能对比。所以搞定它你不仅是在完成一个项目更是在掌握一套行业通用的“语言”和“标尺”。这个数据集到底好在哪首先它非常“真实”。数据是在一个真实的实验台架上采集的模拟了电机驱动端和风扇端的轴承在不同负载0到3马力下运行的状态。故障类型也覆盖得很全从最轻微的单点损伤比如轴承内圈、外圈、滚动体上有一个小坑到更严重的复合故障都有。传感器采集的是振动信号这是工业现场最常用、也最有效的监测手段之一。其次它非常“干净”。数据已经按不同的故障位置、故障尺寸从0.007英寸到0.021英寸、电机负载和采样频率12kHz和48kHz分门别类地整理好了。你不需要自己去猜测哪个文件对应什么状态省去了大量数据清洗和标注的麻烦。这对于我们快速搭建原型、验证算法想法来说简直是天大的福音。那么这个实战项目要做什么我们的目标很明确用Python构建一个端到端的轴承故障诊断系统。从把那一堆.mat数据文件读进内存开始到把它们变成模型能“吃”的格式再到设计并训练一个能准确识别“健康”、“内圈故障”、“外圈故障”、“滚动体故障”的智能模型最后把这个模型打包保存甚至做成一个简单的Web应用来演示。整个过程我会把我踩过的坑、试过的好方法、以及那些让代码跑得更稳的小技巧都毫无保留地分享给你。不管你是数据科学新手还是想在这个领域深耕的算法工程师跟着这个流程走一遍你都能获得一个可以直接复用到自己项目里的“工业级”模板。2. 数据加载与预处理把“原始振动”变成“规整表格”拿到数据的第一步千万别急着往模型里塞。很多新手项目效果不好问题往往就出在数据准备的草率上。CWRU的数据文件是MATLAB的.mat格式这在科研领域很常见。我们用Python的scipy.io库就能轻松读取。import numpy as np import pandas as pd from scipy.io import loadmat import os # 假设你的数据放在 ‘CWRU/‘ 目录下结构是驱动端/12k驱动端轴承数据/... data_path ‘CWRU/12k驱动端轴承数据/‘ file_list os.listdir(data_path) # 我们先加载一个文件看看结构 sample_file ‘97.mat‘ # 例如97对应的是驱动端内圈故障0.007英寸损伤1马力负载 sample_data loadmat(os.path.join(data_path, sample_file)) # 查看这个.mat文件里有什么键 print(sample_data.keys())你会发现文件里通常包含一个类似‘X097_DE_time‘的键其对应的值就是一个长长的振动信号序列。这里DE代表驱动端加速度计数据。我们需要把这个一维时间序列信号以及它的标签健康还是哪种故障提取出来。但这里有个关键点我们不能直接用整个长时间序列去训练。一个信号文件可能有几十万甚至上百万个点直接扔进模型计算量巨大而且序列太长模型也很难学到有效的局部特征。标准的做法是进行滑动窗口分割。想象一下你用一把固定长度的尺子在长长的信号带上滑动每次截取一小段这一小段就是一个样本。这样一个长文件就能生成成百上千个训练样本极大地扩充了数据量。def segment_signal(data, window_size, step_size): 将一维信号按滑动窗口分割成多个样本 segments [] num_samples len(data) for start in range(0, num_samples - window_size 1, step_size): segment data[start:start window_size] segments.append(segment) return np.array(segments) # 假设我们读取了振动信号 ‘signal‘ 和对应的标签 ‘label‘ window_size 1024 # 每个样本的长度也是后续模型输入的长度。这个值需要根据信号频率和故障特征周期来调整。 step_size 512 # 滑动步长步长越小生成的样本越多但样本间相关性也越大。 segments segment_signal(signal, window_size, step_size) # 此时 segments 的形状是 (n_samples, window_size) # 为每一个segment都赋予相同的标签 label labels np.full(len(segments), label)接下来是数据标准化。振动信号的幅值可能因为负载不同而有很大差异为了不让模型被这些量纲差异所误导我们需要对每个特征这里是信号的每个时间点进行标准化通常使用Z-score标准化即减去均值再除以标准差使得数据分布接近均值为0标准差为1。from sklearn.preprocessing import StandardScaler # 假设我们已将所有样本堆叠成一个大矩阵 all_segments形状为 (total_samples, window_size) scaler StandardScaler() # 注意拟合时应该只使用训练集数据避免数据泄露 scaler.fit(train_segments) train_segments_scaled scaler.transform(train_segments) test_segments_scaled scaler.transform(test_segments)最后别忘了划分训练集、验证集和测试集。一个严谨的划分能帮你客观评估模型性能。我习惯用sklearn的train_test_split先分出训练验证集和测试集再从训练集中分出一部分作为验证集。from sklearn.model_selection import train_test_split # 第一次分割分出训练验证集 和 测试集 X_train_val, X_test, y_train_val, y_test train_test_split(all_segments_scaled, all_labels, test_size0.15, random_state42, stratifyall_labels) # 第二次分割从训练验证集中再分出验证集 X_train, X_val, y_train, y_val train_test_split(X_train_val, y_train_val, test_size0.176, random_state42, stratifyy_train_val) # 0.176 使得最终比例为 70:15:15做完这些你的数据就从一堆散乱的.mat文件变成了规规矩矩的(样本数, 特征长度)的数组和对应的标签向量可以放心地喂给模型了。这个过程虽然有点繁琐但就像做饭前洗菜切菜一样必不可少而且很大程度上决定了最后“菜品”的质量。3. 特征工程与模型选择是直接“喂”原始信号还是先“加工”一下数据准备好了下一个问题就是我们用什么模型又该给模型输入什么在故障诊断领域尤其是处理振动信号时主要有两条技术路线。第一条路线是“端到端”学习直接把预处理后的原始一维振动信号切片输入给深度学习模型比如1D-CNN一维卷积神经网络、LSTM或者Transformer。模型的卷积层或注意力层会自动学习信号中的层次化特征从浅层的边缘可能是冲击的起始点到深层的复杂模式可能是某种故障的周期性表现。这种方法省去了人工设计特征的步骤非常方便。import torch import torch.nn as nn class Simple1DCNN(nn.Module): def __init__(self, input_length, num_classes): super(Simple1DCNN, self).__init__() self.conv1 nn.Conv1d(in_channels1, out_channels64, kernel_size3, padding1) self.pool1 nn.MaxPool1d(kernel_size2) self.conv2 nn.Conv1d(64, 128, kernel_size3, padding1) self.pool2 nn.MaxPool1d(2) # 计算经过卷积池化后的特征长度 self.flattened_size self._get_flattened_size(input_length) self.fc1 nn.Linear(self.flattened_size, 256) self.fc2 nn.Linear(256, num_classes) self.relu nn.ReLU() self.dropout nn.Dropout(0.5) def _get_flattened_size(self, length): # 模拟前向传播计算尺寸 x torch.randn(1, 1, length) x self.pool1(self.relu(self.conv1(x))) x self.pool2(self.relu(self.conv2(x))) return x.numel() def forward(self, x): x x.unsqueeze(1) # 增加通道维度: (batch, 1, length) x self.pool1(self.relu(self.conv1(x))) x self.pool2(self.relu(self.conv2(x))) x x.view(x.size(0), -1) # 展平 x self.relu(self.fc1(x)) x self.dropout(x) x self.fc2(x) return x第二条路线是“时频分析图像分类”。振动信号在时域上可能就是一串杂乱无章的波形但很多故障特征在频率域或者时频联合域上会表现得更加明显。我们可以先把一维信号转换成二维的时频图像比如短时傅里叶变换谱图、连续小波变换尺度图然后把生成的图像当成图片用成熟的2D-CNN如ResNet、Swin Transformer去做分类。这条路线的优势在于我们可以利用在ImageNet等大型图像数据集上预训练好的模型权重进行迁移学习这在数据量不是特别大的情况下往往能取得更好的效果。import librosa import librosa.display import matplotlib.pyplot as plt def compute_spectrogram(signal, sr12000, n_fft256, hop_length64): 计算并绘制STFT谱图返回谱图矩阵 D librosa.stft(signal, n_fftn_fft, hop_lengthhop_length) S_db librosa.amplitude_to_db(np.abs(D), refnp.max) # 这里可以保存S_db为图像文件或者直接作为特征输入 return S_db # 对一个样本信号计算谱图 spec compute_spectrogram(segments[0]) plt.figure(figsize(10, 4)) librosa.display.specshow(spec, sr12000, hop_length64, x_axis‘time‘, y_axis‘hz‘) plt.colorbar(format‘%2.0f dB‘) plt.title(‘STFT Spectrogram‘) plt.tight_layout() plt.show()那么到底选哪条路我的经验是如果你的目标是快速出一个高精度的基准结果或者数据量有限优先尝试时频图像2D CNN迁移学习。像Swin Transformer这种视觉模型对图像特征的捕捉能力极强在CWRU这种数据集上达到99%以上的准确率并不难。如果你更关注模型的实时性、轻量化或者想深入研究信号本身的特征那么端到端的1D CNN或Transformer是更好的选择。在实际项目中我通常会两条路都走一遍对比一下效果和推理速度再根据实际部署环境做决定。下面我们就以目前大热的Transformer架构为例看看如何构建一个端到端的故障诊断模型。4. 模型构建与训练用Transformer给振动信号做“深度体检”Transformer最初是为自然语言处理设计的但它的核心——自注意力机制能够捕捉序列中任意两个位置之间的依赖关系这个特性对于振动信号分析同样有用。故障冲击可能和前后很长一段时间的信号都有关系自注意力机制正好擅长建模这种长程依赖。但是原始Transformer处理的是离散的单词序列我们的振动信号是连续的数值序列。所以我们需要做一些适配。首先要把输入的一维信号通过一个线性投影或者一个浅层的1D CNN映射成一个序列的“特征向量”。然后为这个序列加上位置编码因为Transformer本身没有内置的顺序信息而振动信号的时间顺序至关重要。import torch.nn as nn import torch.nn.functional as F import math class PositionalEncoding(nn.Module): 标准的位置编码适用于序列建模 def __init__(self, d_model, max_len5000): super(PositionalEncoding, self).__init__() pe torch.zeros(max_len, d_model) position torch.arange(0, max_len, dtypetorch.float).unsqueeze(1) div_term torch.exp(torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model)) pe[:, 0::2] torch.sin(position * div_term) pe[:, 1::2] torch.cos(position * div_term) pe pe.unsqueeze(0).transpose(0, 1) # 形状: (max_len, 1, d_model) self.register_buffer(‘pe‘, pe) def forward(self, x): # x: (seq_len, batch_size, d_model) x x self.pe[:x.size(0), :] return x class SignalTransformer(nn.Module): def __init__(self, input_dim, d_model, nhead, num_encoder_layers, dim_feedforward, num_classes, max_seq_len1024): super(SignalTransformer, self).__init__() # 信号嵌入层将每个时间点的标量值映射到高维空间 self.embedding nn.Linear(1, d_model) # 假设输入是(batch, seq_len)我们将其视为(batch, seq_len, 1) self.pos_encoder PositionalEncoding(d_model, max_seq_len) encoder_layer nn.TransformerEncoderLayer(d_modeld_model, nheadnhead, dim_feedforwarddim_feedforward, batch_firstTrue) self.transformer_encoder nn.TransformerEncoder(encoder_layer, num_layersnum_encoder_layers) # 全局平均池化将序列维度聚合 self.global_pool nn.AdaptiveAvgPool1d(1) self.fc_out nn.Linear(d_model, num_classes) def forward(self, x): # x: (batch_size, seq_len) x x.unsqueeze(-1) # (batch_size, seq_len, 1) x self.embedding(x) # (batch_size, seq_len, d_model) x self.pos_encoder(x) # 加入位置信息 # Transformer期望的输入是(seq_len, batch, d_model)或(batch, seq_len, d_model) for batch_firstTrue x self.transformer_encoder(x) # (batch_size, seq_len, d_model) # 将序列维度通过全局平均池化压缩 x x.transpose(1, 2) # (batch_size, d_model, seq_len) x self.global_pool(x).squeeze(-1) # (batch_size, d_model) output self.fc_out(x) # (batch_size, num_classes) return output模型定义好了训练环节同样重要。对于多分类问题损失函数通常用交叉熵损失。优化器我习惯用AdamW它比Adam加了权重衰减能更好地防止过拟合。学习率调度器比如余弦退火能让训练过程更平滑有助于找到更优的解。import torch.optim as optim from torch.optim.lr_scheduler import CosineAnnealingLR device torch.device(‘cuda‘ if torch.cuda.is_available() else ‘cpu‘) model SignalTransformer(input_dim1, d_model128, nhead8, num_encoder_layers4, dim_feedforward512, num_classes4).to(device) criterion nn.CrossEntropyLoss() optimizer optim.AdamW(model.parameters(), lr1e-4, weight_decay1e-4) scheduler CosineAnnealingLR(optimizer, T_max50) # 假设训练50个epoch # 训练循环示例 for epoch in range(num_epochs): model.train() for batch_x, batch_y in train_loader: batch_x, batch_y batch_x.to(device), batch_y.to(device) optimizer.zero_grad() outputs model(batch_x) loss criterion(outputs, batch_y) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm1.0) # 梯度裁剪防止爆炸 optimizer.step() scheduler.step() # 每个epoch后在验证集上评估...训练时一定要监控训练集和验证集上的损失和准确率。如果训练集损失持续下降但验证集损失开始上升那就是过拟合了需要加大Dropout比率、增加数据增强比如对信号加轻微的高斯噪声、随机缩放或者使用更早的停止策略。我通常会保存验证集上性能最好的那个模型权重而不是最后一个epoch的。5. 模型评估与可解释性别只看准确率看看模型到底“学”会了什么模型训练完成在测试集上跑出了99%的准确率是不是就大功告成了远远没有。高准确率可能掩盖很多问题比如模型可能只是记住了数据中某些无关紧要的噪声或者对某一类故障判断得很好但对另一类却一塌糊涂。我们需要更细致的评估工具。混淆矩阵是分类任务中最直观的评估工具之一。它能清晰展示模型在每个类别上的分类情况对角线上的数字代表预测正确的样本数其他位置则代表混淆的情况。通过它你能一眼看出模型最容易把哪两类故障搞混。比如是不是经常把“内圈故障”预测成“滚动体故障”from sklearn.metrics import confusion_matrix, classification_report import seaborn as sns model.eval() all_preds [] all_labels [] with torch.no_grad(): for batch_x, batch_y in test_loader: batch_x batch_x.to(device) outputs model(batch_x) _, preds torch.max(outputs, 1) all_preds.extend(preds.cpu().numpy()) all_labels.extend(batch_y.numpy()) cm confusion_matrix(all_labels, all_preds) plt.figure(figsize(8,6)) sns.heatmap(cm, annotTrue, fmt‘d‘, cmap‘Blues‘, xticklabelsclass_names, yticklabelsclass_names) plt.ylabel(‘True Label‘) plt.xlabel(‘Predicted Label‘) plt.title(‘Confusion Matrix‘) plt.show() print(classification_report(all_labels, all_preds, target_namesclass_names))除了整体指标我们还需要知道模型是根据什么做出判断的。这就是模型可解释性要解决的问题。SHAP是一个强大的工具它可以为每个输入特征对我们来说就是振动信号的每个时间点分配一个重要性分数告诉你这个点对最终预测结果的贡献是正面的还是负面的。想象一下你有一个“健康”状态的信号和一个“外圈故障”的信号用SHAP分析模型对这两个信号的预测。你可能会发现在故障信号中模型特别关注那些周期性出现的冲击峰值所在的时间点而这些点在健康信号中重要性就很低。这不仅能验证模型是否学到了我们期望的故障特征还能帮助我们发现一些意想不到的、但模型认为很重要的信号模式甚至可能启发新的特征工程思路。import shap # 注意SHAP计算量可能较大建议在测试集的一个子集上运行 background X_train_scaled[:100].to(device) # 选择一个背景数据集 test_samples X_test_scaled[:10].to(device) # 创建一个函数包装模型预测 def model_predict(x): model.eval() with torch.no_grad(): output model(x) return output.cpu().numpy() # 使用DeepExplainer针对深度学习模型 explainer shap.DeepExplainer(model, background) shap_values explainer.shap_values(test_samples) # 可视化某个样本的SHAP值 shap.initjs() shap.force_plot(explainer.expected_value[0], shap_values[0][0], feature_names[f‘t{i}‘ for i in range(window_size)])通过混淆矩阵和SHAP分析我们就不再是模型的“黑盒用户”了。我们能清晰地知道模型的强项和弱点知道它决策的依据这对于将模型投入实际应用至关重要。毕竟在工业现场我们不仅要模型“准”还要它“稳”和“可信”。6. 模型部署与保存让训练好的模型“活”起来模型在笔记本上跑得再好如果不能应用到实际系统中也只是个玩具。部署的第一步是保存训练好的模型。在PyTorch中我们通常保存模型的state_dict状态字典它只包含模型的可学习参数比较轻量并且加载灵活。# 保存模型 torch.save({ ‘epoch‘: best_epoch, ‘model_state_dict‘: model.state_dict(), ‘optimizer_state_dict‘: optimizer.state_dict(), ‘loss‘: best_loss, ‘scaler_state‘: scaler, # 别忘了保存数据标准化器 }, ‘best_bearing_transformer_model.pth‘) # 加载模型 checkpoint torch.load(‘best_bearing_transformer_model.pth‘, map_locationdevice) model.load_state_dict(checkpoint[‘model_state_dict‘]) scaler checkpoint[‘scaler_state‘]保存时强烈建议把数据预处理用的StandardScaler也一并保存。因为在线预测时新的原始数据必须用和训练时完全相同的均值和方差进行标准化否则模型性能会严重下降。模型保存后怎么用起来呢最简单的方式是写一个预测脚本。但更工程化的做法是将其封装成一个API服务比如使用FastAPI。这样其他系统比如设备监控平台就可以通过发送HTTP请求来获取故障诊断结果。from fastapi import FastAPI, File, UploadFile import numpy as np import torch import joblib # 用于加载scaler app FastAPI() model SignalTransformer(...).to(device) model.load_state_dict(torch.load(‘best_model.pth‘, map_locationdevice)) model.eval() scaler joblib.load(‘standard_scaler.pkl‘) # 假设scaler用joblib保存 def preprocess_signal(raw_signal: np.ndarray): 模拟离线时的预处理流程分段、标准化 segments segment_signal(raw_signal, window_size1024, step_size512) segments_scaled scaler.transform(segments) # 取最后一个segment或者对所有segment预测后投票 return torch.FloatTensor(segments_scaled[-1:]) # 返回一个样本 app.post(/predict/) async def predict_bearing_condition(file: UploadFile File(...)): # 假设上传的文件是一个包含振动信号.npy文件或.csv文件 contents await file.read() # 这里需要根据实际文件格式解析出原始信号数组 raw_signal # raw_signal np.load(io.BytesIO(contents)) # 例如.npy格式 raw_signal np.genfromtxt(io.BytesIO(contents), delimiter‘,‘) # 例如.csv格式 processed_tensor preprocess_signal(raw_signal).to(device) with torch.no_grad(): output model(processed_tensor) prediction torch.argmax(output, dim1).item() confidence torch.nn.functional.softmax(output, dim1).max().item() class_map {0: ‘健康‘, 1: ‘内圈故障‘, 2: ‘外圈故障‘, 3: ‘滚动体故障‘} return { “status“: “success“, “prediction“: class_map[prediction], “confidence“: f“{confidence:.2%}“, “raw_logits“: output.cpu().numpy().tolist() }把这个脚本用uvicorn跑起来你就有了一个本地诊断服务。更进一步你可以用Docker把模型、API和环境打包成一个镜像这样就能在任何支持Docker的服务器上一致地运行。对于资源受限的边缘设备比如工控机你可能还需要考虑模型量化、剪枝等技术来压缩模型大小提升推理速度。PyTorch提供了torch.quantization模块来帮助实现动态或静态量化这能让模型在几乎不损失精度的情况下显著减少内存占用和计算延迟。从数据加载到API部署这一整套流程走下来你已经拥有了一个完整的、可复用的轴承故障诊断项目框架。下次当你遇到新的振动数据集或者不同的故障诊断场景时只需要替换数据加载部分微调一下模型结构或超参数就能快速搭建起一个新的原型系统。这才是实战的意义——不仅解决一个问题更是掌握一套方法论。