简介:这份资源是面向深度学习与工业智能方向初学者及研究者的故障诊断入门示例,围绕数据预处理、模型搭建与模型训练三个核心环节展开,帮助读者理解如何用神经网络识别和预测复杂设备故障模式。压缩包共57个文件,约46.13MB,以40个mat数据文件、10张png结果图、3个py脚本、2个h5模型权重和2个md说明文档为主,覆盖数据、代码、模型与文档四类内容,便于直接复现与二次修改。资源中涉及数据清洗、标准化、特征工程与数据集划分,并对比CNN、RNN/LSTM、自编码器、预训练模型等结构在故障诊断中的适用场景,同时给出初始化、学习率调整、正则化与模型融合等训练策略,以及准确率、精确率、召回率、F1分数等评估指标。目前已有908人学习,适合希望快速跑通完整流程、积累排错与调参思路的读者参考。
1. 从一段轴承振动信号说起:这个项目到底能干什么
手里拿到一份工业设备的振动信号数据,采样频率 12kHz,几万条时序点,标签是“正常/内圈故障/外圈故障/滚动体故障”四类。想用深度学习做故障诊断,但打开论文一看全是公式,打开开源仓库一看全是几百行的工程封装,根本不知道从哪下手。这个项目就是解决这个问题的——它把基于深度学习的故障诊断拆成了一条最小可运行链路:数据预处理、模型搭建、模型训练三段,代码量控制在一个入门开发者能逐行读完的范围内。适合刚接触深度学习故障诊断的学生、需要快速验证算法可行性的工程师,以及想拿一个干净模板做二次开发的人。它不追求 SOTA 精度,追求的是“你能看懂每一行在干什么,并且能跑通”。
2. 项目整体架构:从原始信号到分类结果的完整链路
2.1 三段式流水线的设计思路
这个项目没有用复杂的框架封装,而是把整个故障诊断流程拆成三个独立可测试的阶段。第一阶段是数据预处理,负责把原始振动信号切成模型能吃的样本;第二阶段是模型搭建,定义网络结构;第三阶段是训练循环,把数据喂进模型并保存权重。三个阶段之间通过文件系统解耦——预处理阶段把处理好的数据存成.npy或.pt文件,训练阶段直接读这些文件。这样做的好处是:调模型的时候不用反复跑预处理,调预处理的时候不用动模型代码。
常见做法是把这三个阶段写成三个 Python 脚本:preprocess.py、model.py、train.py,再加一个config.py统一管理超参数。我一般会再加一个dataset.py专门封装 Dataset 类,这样训练脚本里就不会出现大段的数据加载逻辑。
2.2 目录结构长什么样
一个干净的项目目录大概是这样组织的:
fault_diagnosis_demo/ ├── config.py # 超参数、路径配置 ├── preprocess.py # 数据预处理脚本 ├── dataset.py # Dataset 封装 ├── model.py # 网络结构定义 ├── train.py # 训练主循环 ├── evaluate.py # 测试集评估 ├── data/ │ ├── raw/ # 原始信号文件 │ └── processed/ # 预处理后的样本 ├── checkpoints/ # 模型权重保存目录 └── requirements.txt # 依赖清单这个结构的好处是职责清晰。config.py里放所有路径和超参数,比如SAMPLE_RATE = 12000、WINDOW_SIZE = 1024、BATCH_SIZE = 64、EPOCHS = 50。改参数不用翻遍所有文件,只动一个地方。data/raw/放原始数据,data/processed/放切好的样本,checkpoints/放训练好的权重。这种目录划分在工业故障诊断项目里是通用做法,换成自己的数据只需要改config.py里的路径。
2.3 技术栈选型:为什么是 PyTorch 而不是 TensorFlow
项目用 PyTorch 作为深度学习框架。原因很直接:故障诊断领域的学术论文和开源实现绝大多数用 PyTorch,遇到问题搜到的解决方案更多。而且 PyTorch 的动态图机制在调试时更直观——你可以直接在 forward 函数里 print 张量形状,不用先建图再跑 session。
依赖清单大概包括:torch、numpy、scipy(用于信号处理)、scikit-learn(用于混淆矩阵和分类报告)、matplotlib(用于画损失曲线和频谱图)。如果要用公开数据集,可能还需要pandas读 CSV 或h5py读 HDF5 文件。这些库的版本不用卡太死,PyTorch 1.10 以上都能跑。
提示:如果用的是 GPU 训练,安装 PyTorch 时注意 CUDA 版本要和驱动匹配。CPU 也能跑这个项目,只是训练时间从几分钟变成十几分钟。
3. 数据预处理:把原始振动信号切成模型能吃的样本
3.1 信号切片与归一化
原始振动信号是一条长时序,比如 12kHz 采样 10 秒就是 120000 个点。模型不能直接吃这么长的序列,需要切成固定长度的窗口。常见做法是用滑动窗口,窗口大小 1024 或 2048,步长取窗口的一半或四分之一。这样既能保证每个样本包含足够的故障特征周期,又能通过重叠增加样本数量。
import numpy as np def sliding_window(signal, window_size=1024, step=512): """ 对一维振动信号做滑动窗口切片 signal: 一维 numpy 数组,原始振动信号 window_size: 每个样本的长度 step: 窗口滑动步长 返回: shape=(样本数, window_size) 的数组 """ samples = [] for start in range(0, len(signal) - window_size + 1, step): samples.append(signal[start:start + window_size]) return np.array(samples) def normalize(samples): """ 对每个样本做 Z-score 归一化 避免不同工况下幅值差异影响训练 """ mean = samples.mean(axis=1, keepdims=True) std = samples.std(axis=1, keepdims=True) + 1e-8 return (samples - mean) / stdsliding_window函数里window_size决定了每个样本覆盖的时间跨度。12kHz 采样下 1024 个点约 85 毫秒,对于轴承故障特征频率来说通常够用。step越小样本越多,但重叠太大会导致训练集和测试集泄漏——如果同一个窗口的切片同时出现在训练和测试里,精度会虚高。我一般会把原始信号按时间顺序先切分训练/测试,再分别做滑动窗口,避免这个问题。
normalize做的是 Z-score 归一化,每个样本减自己的均值除自己的标准差。这一步在故障诊断里很关键,因为不同负载、不同转速下振动幅值可能差一个数量级,不归一化的话模型会偏向幅值大的样本。
3.2 标签编码与数据集划分
故障诊断通常是多分类任务,标签是“正常”“内圈故障”“外圈故障”等字符串。模型需要整数标签,所以要做编码。同时要划分训练集、验证集、测试集,常见比例是 7:1.5:1.5 或 8:1:1。
from sklearn.model_selection import train_test_split from sklearn.preprocessing import LabelEncoder def prepare_data(X, y, test_size=0.2, val_size=0.1, random_state=42): """ X: 样本数组 shape=(N, window_size) y: 标签列表,字符串或整数 返回: 训练/验证/测试集 """ le = LabelEncoder() y_encoded = le.fit_transform(y) # 先分训练+验证 和 测试 X_temp, X_test, y_temp, y_test = train_test_split( X, y_encoded, test_size=test_size, random_state=random_state, stratify=y_encoded ) # 再从训练+验证里分验证 val_ratio = val_size / (1 - test_size) X_train, X_val, y_train, y_val = train_test_split( X_temp, y_temp, test_size=val_ratio, random_state=random_state, stratify=y_temp ) return X_train, X_val, X_test, y_train, y_val, y_test, le这里用了stratify参数保证每个类别在训练/验证/测试里的比例一致。故障诊断数据经常类别不平衡——正常样本远多于故障样本,不做分层抽样的话可能某个故障类别在测试集里一个都没有。LabelEncoder的classes_属性保存了类别到整数的映射,后面画混淆矩阵的时候要用到。
3.3 保存预处理结果
预处理完的数据存成.npz文件,训练脚本直接加载。这样调模型的时候不用每次重新切片。
def save_processed(X_train, X_val, X_test, y_train, y_val, y_test, save_path): np.savez_compressed( save_path, X_train=X_train, X_val=X_val, X_test=X_test, y_train=y_train, y_val=y_val, y_test=y_test )np.savez_compressed比np.savez省磁盘空间,加载速度也够快。如果数据量特别大,比如几十万样本,可以考虑用 HDF5 格式分块存储,但入门项目用 npz 足够了。
4. 模型搭建:一维卷积网络处理时序信号的实现细节
4.1 为什么用 1D-CNN 而不是 LSTM
故障诊断的振动信号是典型的一维时序,理论上 LSTM 也能做。但 1D-CNN 在这个场景下有几个实际优势:训练速度快、参数量少、对局部冲击特征敏感。轴承故障的振动信号里,故障特征往往表现为周期性的冲击脉冲,1D-CNN 的卷积核正好能捕捉这种局部模式。LSTM 虽然能建模长程依赖,但在这个任务上并没有明显优势,反而训练更慢、调参更麻烦。
常见做法是堆叠几层 1D 卷积 + 池化,最后接全连接分类头。卷积核大小一般取 3、5、7 这种奇数,层数 3 到 5 层足够。每层通道数从 16 或 32 开始,逐层翻倍。
4.2 网络结构定义
import torch import torch.nn as nn class FaultCNN(nn.Module): def __init__(self, num_classes=4, input_length=1024): super(FaultCNN, self).__init__() # 第一层卷积:1通道输入,32通道输出,卷积核大小7 self.conv1 = nn.Conv1d(1, 32, kernel_size=7, padding=3) self.bn1 = nn.BatchNorm1d(32) self.pool1 = nn.MaxPool1d(2) # 第二层:32->64通道 self.conv2 = nn.Conv1d(32, 64, kernel_size=5, padding=2) self.bn2 = nn.BatchNorm1d(64) self.pool2 = nn.MaxPool1d(2) # 第三层:64->128通道 self.conv3 = nn.Conv1d(64, 128, kernel_size=3, padding=1) self.bn3 = nn.BatchNorm1d(128) self.pool3 = nn.MaxPool1d(2) # 自适应池化,不管输入长度多少都输出固定大小 self.global_pool = nn.AdaptiveAvgPool1d(1) # 分类头 self.fc = nn.Sequential( nn.Linear(128, 64), nn.ReLU(), nn.Dropout(0.5), nn.Linear(64, num_classes) ) def forward(self, x): # x shape: (batch, 1, input_length) x = torch.relu(self.bn1(self.conv1(x))) x = self.pool1(x) x = torch.relu(self.bn2(self.conv2(x))) x = self.pool2(x) x = torch.relu(self.bn3(self.conv3(x))) x = self.pool3(x) # 全局池化后 shape: (batch, 128, 1) x = self.global_pool(x).squeeze(-1) return self.fc(x)这个网络结构有几个设计点值得说。第一,每层卷积后接 BatchNorm,这在故障诊断里很重要,因为不同批次的振动信号分布可能有差异,BN 能稳定训练。第二,用AdaptiveAvgPool1d(1)做全局池化,这样不管输入窗口长度是 1024 还是 2048,网络都能处理,不用改全连接层。第三,分类头里加了 Dropout 0.5,防止过拟合——故障诊断数据集通常不大,过拟合是常见问题。
input_length参数其实在 forward 里没直接用到,因为自适应池化会自动处理。但保留这个参数是为了文档化,让读代码的人知道预期输入长度。
4.3 参数量与感受野估算
这个网络的参数量大概在 10 万左右,具体计算:第一层卷积1*32*7+32=256,第二层32*64*5+64=10304,第三层64*128*3+128=24704,全连接128*64+64=8256和64*4+4=260。加起来约 4.4 万参数,加上 BN 的参数总共 5 万左右。这个量级在 CPU 上也能快速训练。
感受野方面,三层卷积核大小分别是 7、5、3,池化层每层缩小一半。第一层每个输出点看到 7 个原始点,第二层看到(7-1)+(5-1)*2+1=15个点,第三层看到(15-1)+(3-1)*4+1=23个点。23 个点在 12kHz 采样下约 2 毫秒,对于捕捉轴承故障的冲击特征来说偏小。如果故障特征周期较长,可以增大卷积核或增加层数。
注意:如果发现模型在训练集上精度很高但测试集很差,优先检查数据泄漏——滑动窗口的重叠部分是否同时出现在训练和测试里。
5. 训练与评估:损失曲线、混淆矩阵和常见翻车点
5.1 训练循环的骨架
训练循环是标准 PyTorch 流程:前向传播、算损失、反向传播、更新参数。但故障诊断场景下有几个细节要注意。
import torch.optim as optim from torch.utils.data import TensorDataset, DataLoader def train_model(model, X_train, y_train, X_val, y_val, config): device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') model = model.to(device) # 把 numpy 数组转成 Tensor train_ds = TensorDataset( torch.FloatTensor(X_train).unsqueeze(1), # 加通道维度 torch.LongTensor(y_train) ) val_ds = TensorDataset( torch.FloatTensor(X_val).unsqueeze(1), torch.LongTensor(y_val) ) train_loader = DataLoader(train_ds, batch_size=config['batch_size'], shuffle=True) val_loader = DataLoader(val_ds, batch_size=config['batch_size'], shuffle=False) criterion = nn.CrossEntropyLoss() optimizer = optim.Adam(model.parameters(), lr=config['lr']) scheduler = optim.lr_scheduler.StepLR(optimizer, step_size=20, gamma=0.5) best_val_acc = 0.0 for epoch in range(config['epochs']): # 训练阶段 model.train() train_loss = 0.0 for xb, yb in train_loader: xb, yb = xb.to(device), yb.to(device) optimizer.zero_grad() logits = model(xb) loss = criterion(logits, yb) loss.backward() optimizer.step() train_loss += loss.item() * xb.size(0) # 验证阶段 model.eval() correct, total = 0, 0 with torch.no_grad(): for xb, yb in val_loader: xb, yb = xb.to(device), yb.to(device) logits = model(xb) preds = logits.argmax(dim=1) correct += (preds == yb).sum().item() total += yb.size(0) val_acc = correct / total scheduler.step() # 保存最佳模型 if val_acc > best_val_acc: best_val_acc = val_acc torch.save(model.state_dict(), config['checkpoint_path']) print(f"Epoch {epoch+1}/{config['epochs']} | " f"Loss: {train_loss/len(train_ds):.4f} | Val Acc: {val_acc:.4f}") return best_val_acc几个关键点:TensorDataset要求输入是 Tensor,所以 numpy 数组要先转。unsqueeze(1)是给数据加一个通道维度,因为 Conv1d 要求输入 shape 是(batch, channels, length)。优化器用 Adam,学习率 1e-3 起步,配合 StepLR 每 20 个 epoch 降一半。保存最佳模型而不是最后一个 epoch 的模型,因为故障诊断数据少,后面容易过拟合。
5.2 评估指标:不只看准确率
故障诊断里准确率会骗人。如果正常样本占 90%,模型全预测正常也能拿 90% 准确率。所以要看混淆矩阵和每个类别的 F1。
from sklearn.metrics import classification_report, confusion_matrix import matplotlib.pyplot as plt import seaborn as sns def evaluate_model(model, X_test, y_test, label_encoder, config): device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') model.load_state_dict(torch.load(config['checkpoint_path'])) model = model.to(device) model.eval() test_ds = TensorDataset( torch.FloatTensor(X_test).unsqueeze(1), torch.LongTensor(y_test) ) test_loader = DataLoader(test_ds, batch_size=config['batch_size'], shuffle=False) all_preds, all_labels = [], [] with torch.no_grad(): for xb, yb in test_loader: xb = xb.to(device) preds = model(xb).argmax(dim=1).cpu().numpy() all_preds.extend(preds) all_labels.extend(yb.numpy()) # 分类报告 print(classification_report(all_labels, all_preds, target_names=label_encoder.classes_)) # 混淆矩阵 cm = confusion_matrix(all_labels, all_preds) plt.figure(figsize=(8, 6)) sns.heatmap(cm, annot=True, fmt='d', cmap='Blues', xticklabels=label_encoder.classes_, yticklabels=label_encoder.classes_) plt.xlabel('Predicted') plt.ylabel('True') plt.title('Confusion Matrix') plt.tight_layout() plt.savefig('confusion_matrix.png', dpi=150)classification_report会输出每个类别的 precision、recall、F1。如果某个故障类别的 recall 特别低,说明模型漏检了这类故障,在工业场景下漏检比误报更危险。混淆矩阵能看出模型把哪两类搞混了——比如内圈故障和外圈故障容易混,因为它们的冲击特征频率接近。
5.3 避坑与常见问题排查
现象一:训练 loss 不下降,一直在 2.0 附近。原因通常是学习率太大或数据没归一化。解决:先把学习率降到 1e-4 试试,再检查预处理阶段有没有做 Z-score。如果输入幅值在几千的量级,网络第一层就饱和了。
现象二:验证集准确率比训练集还高。这听起来是好事,但通常是验证集太小或数据泄漏。检查滑动窗口的 step 是不是太小导致训练集和验证集有重叠样本。解决:按时间顺序切分原始信号,先分训练/验证/测试,再各自做滑动窗口。
现象三:测试集上某一类故障全部预测错。原因大概率是类别不平衡,那一类样本太少。解决:在损失函数里加weight参数,给少数类更高权重。nn.CrossEntropyLoss(weight=torch.FloatTensor([1.0, 3.0, 3.0, 3.0]))这种。
现象四:GPU 显存不够。故障诊断的样本通常不大,但如果窗口长度 4096 以上、batch size 128,显存可能吃紧。解决:减小 batch size 或窗口长度,或者用梯度累积模拟大 batch。
现象五:训练完模型在测试集上精度 99%,换一组数据掉到 60%。这是过拟合的典型表现。故障诊断数据集如果来自同一台设备同一工况,训练集和测试集分布太接近,模型学到的是设备特定特征而不是故障特征。解决:用不同负载、不同转速的数据做交叉验证,或者加数据增强(加噪声、幅值缩放、时间偏移)。
6. 二次开发:换数据、换模型、换任务的改造思路
拿到这个项目之后,最常做的改造是换自己的数据。假设你手里有一组新的振动信号,采样频率 20kHz,四类故障,每类 500 个样本。第一步改config.py里的SAMPLE_RATE和WINDOW_SIZE。20kHz 下如果故障特征周期是 5ms,窗口至少覆盖 100 个点,但为了卷积核有足够感受野,建议窗口 2048 起步。第二步改preprocess.py里的数据加载逻辑——如果你的数据是 CSV 格式,用pandas.read_csv读进来转 numpy;如果是 MAT 文件,用scipy.io.loadmat。第三步改dataset.py里的类别数,num_classes从 4 改成你的实际类别数。
换模型结构是另一个常见需求。如果你想试 LSTM,在model.py里加一个FaultLSTM类,输入 shape 从(batch, 1, length)改成(batch, length, 1),因为 LSTM 要求通道在最后。如果你想试 Transformer,注意位置编码要加上,因为振动信号的时序顺序很重要。我一般会保留FaultCNN作为 baseline,新模型在验证集上超过 baseline 才继续调。
换任务的话,比如从故障诊断换成异常检测,标签只有正常和异常两类,num_classes改成 2,损失函数可以用BCEWithLogitsLoss。如果做的是无监督异常检测,那整个训练逻辑都要改——用自编码器重构正常样本,测试时看重构误差。这个改造幅度比较大,建议先把有监督版本跑通再动。
验证改造是否成功,我习惯用三个检查点:第一,训练 loss 是否稳定下降;第二,验证集 F1 是否在 0.85 以上;第三,混淆矩阵对角线是否明显。如果三个都过了,再去看测试集。测试集精度比验证集低 5 个点以内算正常,低 10 个点以上就要查数据分布。
从那以后我每次换新数据,都强制先跑一遍预处理可视化——把切片后的样本画成波形图,确认每个类别的波形有肉眼可辨的差异。如果波形图上看不出区别,模型大概率也学不出来。希望帮到你。
本文还有配套的精品资源,点击获取