简介面向机械故障诊断与Python深度学习开发者的完整源码包以多层感知机与卷积神经网络混合模型处理滚动轴承、齿轮箱振动频谱数据实现故障类型智能识别适用于毕业设计、课程实践及工业智能运维研发。资源共40个文件含12个py脚本、12个xml配置、4个pyc编译文件、6个zbak工程备份以及docx技术文档和txt说明压缩包整体235KB轻量易用便于完整下载与本地复现。工程按RNN、CNN、SAE、MLP等模块划分覆盖数据预处理、特征提取、模型训练与评估全流程并附深度学习在故障诊断中的应用研究文档和备份代码方便对照阅读与二次开发。已有66人学习下载代码通过单元测试与集成验证关键模块设有异常处理能够为机械设备预测性维护和故障根因分析提供直接参考。1. 机械设备故障诊断系统为什么“听声音”值得用 Python 深度学习重做一遍机械设备故障诊断系统这几年在工厂里越来越常见核心做法就是把电机、轴承、齿轮箱的振动信号采集下来用 Python 做预处理再用深度学习模型自动判断设备是正常还是已经出现磨损、断齿、不对中。很多人以为难点在模型结构其实我见过更多团队在数据切分和标签对齐上翻车训练集准确率 98%换一台设备直接掉到 70%。这套流程适合要做预测性维护的运维工程师、刚转工业算法方向的开发以及想从源码层把整个数据流吃透的同学。下面按“数据准备 → 模型 → 源码解析 → 避坑 → 现场验证”的顺序把一套能落地的最小方案讲清楚。2. 把振动信号变成模型能吃的样本滑动窗口、FFT 与切分雷区机械设备故障诊断系统的输入通常是一维振动波形。加速度传感器以固定采样率采集信号比如 25600 Hz意味着每秒记录 25600 个点。原始波形很长不可能整段塞进网络第一步是把连续信号切成固定长度的“样本”这个过程叫滑动窗口。切得好不好直接决定后面模型能不能收敛。网上 python 教程里很少讲这一层因为图像任务不需要自己切窗口而振动数据恰恰卡在最前面。2.1 滑动窗口切分window_size、step_size 与重叠比例怎么定先看一段最基础的切分代码把一维信号变成 (N, window_size) 的样本数组。import numpy as np def sliding_window(signal, window_size, step_size): 把一维振动信号切成多个固定长度样本。 Args: signal: 一维 numpy 数组原始振动波形 window_size: 每个样本包含的采样点数例如 10240 step_size: 每次滑动的采样点数例如 5120 Returns: samples: 形状为 (N, window_size) 的二维数组 samples [] for start in range(0, len(signal) - window_size 1, step_size): samples.append(signal[start:start window_size]) return np.array(samples)这段代码把一段连续信号变成 N 个等长片段。range 的终点用len(signal) - window_size 1保证最后一个窗口不越界。step_size 小于 window_size 时相邻窗口有重叠样本量变大重叠太多会让样本高度相关我一般先设 50% 重叠也就是 step_size 取 window_size 的一半作为起点。window_size 怎么定要看设备转速和采样率。常见做法是让一个窗口至少包含 510 个旋转周期。例如电机转速 1500 rpm转一圈是 40 ms采样率 25600 Hz 时一个周期约 1024 点10 个周期就是 10240 点。如果窗口只取 1024 点相当于只看一圈故障特征可能还没完整出现模型容易把单圈噪声当成故障模式。参数常见取值影响window_size102410240太小装不下完整周期太大引入无关噪声step_sizewindow_size 的一半50% 重叠样本量与独立性折中重叠比例0%75%越高样本越多过高会让相邻样本近乎重复2.2 频谱特征提取采样率、FFT 分辨率和汉宁窗的作用滑动窗口切出的是时域波形。很多故障在时域上不明显比如轴承内圈早期点蚀时域上只是几个小冲击转到频域却能清楚看到故障特征频率及其边带。所以传统诊断流程里FFT 几乎是必经步骤。深度学习模型可以直接吃原始波形但保留一个频谱提取函数对后面做特征对比、现场排查都很有用。import numpy as np def extract_fft_spectrum(signal, sampling_rate): 对单个窗口做 FFT返回幅值谱与频率轴。 Args: signal: 一维时域信号长度 N sampling_rate: 采样率单位 Hz Returns: freqs: 频率轴单位 Hz magnitudes: 幅值谱 n len(signal) window np.hanning(n) spectrum np.fft.rfft(signal * window) magnitudes np.abs(spectrum) freqs np.fft.rfftfreq(n, d1.0 / sampling_rate) return freqs, magnitudes先乘汉宁窗再做 FFT是为了减少频谱泄漏。直接截断一段信号等价于给信号乘了一个矩形窗频谱旁瓣会拖得很长汉宁窗把两端压到接近零主瓣变宽一点但旁瓣压下去很多故障特征频率更容易辨认。rfft只计算实信号的正频率部分输出点数约为 n/2 1rfftfreq返回对应的频率刻度两者一一对应。频率分辨率等于采样率除以窗口长度。采样率 25600 Hz、窗口 10240 点时分辨率约 2.5 Hz要看几十赫兹的轴承特征频率足够了如果窗口只有 1024 点分辨率变成 25 Hz两条靠得近的边带直接糊在一起。所以现场做频谱分析时宁可用长窗口降低分辨率也别贪快。2.3 切分训练/验证集按设备切而不是按样本切样本切好之后下一个关键动作是划分训练集和验证集。很多人直接从切好的样本里随机抽 70% 训练、30% 验证这是整套系统里最隐蔽的雷区。同一台设备连续采集的信号相邻窗口高度相似随机切分等于把几乎重复的数据同时放进训练和验证验证指标会虚高到不真实。import numpy as np def split_by_machine(samples, labels, machine_ids, train_ratio0.7): 按设备编号划分训练/验证集避免同一台设备的数据同时出现在两边。 Args: samples: (N, L) 样本数组 labels: (N, ) 标签 machine_ids: (N, ) 每个样本所属设备编号 train_ratio: 划分到的设备中训练集设备占比 Returns: train_samples, train_labels, val_samples, val_labels unique_ids np.unique(machine_ids) n_train_machines int(len(unique_ids) * train_ratio) train_ids set(unique_ids[:n_train_machines]) train_mask np.array([m in train_ids for m in machine_ids]) return (samples[train_mask], labels[train_mask], samples[~train_mask], labels[~train_mask])这段代码先取设备编号去重再按设备粒度划分。注意unique_ids[:n_train_machines]取前几个编号前应该先np.random.shuffle(unique_ids)否则设备编号有序会让某些类别全落进训练集或验证集。验证集里只要出现没见过的设备里面样本再像模型也没抄过答案准确率才有参考价值。我之前踩过一次同一段轴承数据随机切分验证准确率 99%换到另一台同型号电机上直接掉到 76%。原因就是训练集和验证集里混着同一台设备的相邻窗口模型背下了那台设备的噪声底。按设备切分之后验证集准确率降到 92%但现场测试稳定很多。做这套系统数据切分这步省不得。3. 从波形到诊断结果用 PyTorch 搭一个能跑的 1D-CNN从深度学习入门阶段过来的人习惯把图像那套思路直接搬过来。但振动信号是典型的一维时序数据模型结构和输入组织都得改。很多人找深度学习实战项目案例时第一反应是做图像分类其实振动数据分类更贴近工业现场而且样本量通常比图像小得多模型不需要很深就能出效果。3.1 为什么 1D-CNN 是振动信号诊断的可靠基线振动信号里的故障模式比如轴承外圈剥落产生的冲击在时域上是局部波形在频域上是集中在某几个频段的能量。一维卷积核沿着时间轴滑动天然适合捕捉这种局部模式卷积核参数共享又让它比全连接网络抗过拟合。相比 LSTM、Transformer1D-CNN 参数少、训练快、部署容易对机械设备故障诊断这种样本量不一定很大的场景是性价比最高的起点。我一般把 1D-CNN 当基线模型先跑通再决定要不要上更复杂的结构。如果故障特征本身在频谱上分得很开CNN 几层就能学到如果样本量很大、故障形态复杂再考虑加注意力或者换成 2D 方式把时频图喂给 ResNet。基线模型的意义是给后续所有改动定一个参照点。3.2 搭一个 1D-CNN网络结构、参数与输入输出形状下面这个网络是我常用的最小结构输入是 (batch, 1, window_size)输出是类别得分。import torch.nn as nn class FaultCNN1D(nn.Module): 输入形状 (batch, 1, window_size)输出未归一化的类别得分。 def __init__(self, num_classes4, base_channels16): super().__init__() self.features nn.Sequential( nn.Conv1d(1, base_channels, kernel_size7, stride2, padding3), nn.BatchNorm1d(base_channels), nn.ReLU(inplaceTrue), nn.MaxPool1d(kernel_size2, stride2), nn.Conv1d(base_channels, base_channels * 2, kernel_size5, stride2, padding2), nn.BatchNorm1d(base_channels * 2), nn.ReLU(inplaceTrue), nn.MaxPool1d(kernel_size2, stride2), nn.Conv1d(base_channels * 2, base_channels * 4, kernel_size3, stride1, padding1), nn.BatchNorm1d(base_channels * 4), nn.ReLU(inplaceTrue), ) self.classifier nn.Sequential( nn.AdaptiveAvgPool1d(1), nn.Flatten(), nn.Linear(base_channels * 4, num_classes), ) def forward(self, x): return self.classifier(self.features(x))卷积核从 7 逐渐缩到 3前层看更宽的波形模式后层聚焦细节。stride2 在卷积层里直接降采样后面跟 MaxPool 再降一次感受野增长快计算量小。padding 取 kernel_size 整除 2配合 stride2 时每层输出长度近似减半不会越卷越短到负数。BatchNorm 放在卷积和激活之间对小批量训练很关键能压住梯度波动。最后一层用AdaptiveAvgPool1d(1)把特征压成 1 个点再接线性层。这样有个额外好处如果现场换了一种窗口长度只要特征图长度不为零模型结构不用改。Linear 输入维度固定为base_channels * 4与 window_size 无关。3.3 训练与评估loss 选择、混淆矩阵和查全率模型定义好之后训练循环的写法直接决定能不能复现。下面是最小可跑的 epoch 训练函数。import torch def train_one_epoch(model, dataloader, optimizer, criterion, device): 跑完一个 epoch返回平均训练损失。 model.train() total_loss 0.0 num_samples 0 for batch_x, batch_y in dataloader: batch_x batch_x.unsqueeze(1).to(device) # (B,1,L) batch_y batch_y.to(device) optimizer.zero_grad() outputs model(batch_x) loss criterion(outputs, batch_y) loss.backward() optimizer.step() total_loss loss.item() * batch_x.size(0) num_samples batch_x.size(0) return total_loss / num_samplesunsqueeze(1)给每个 batch 加一维 channel因为Conv1d期望输入是 (B, C, L)。损失函数用torch.nn.CrossEntropyLoss()它内部已经做了 softmax模型输出层不需要再接。优化器用 Adam学习率从 1e-3 起步如果训练损失震荡明显降到 3e-4 再跑。评估不能只看准确率尤其故障诊断这种正负样本可能严重不均衡的任务。用 sklearn 直接出分类报告和混淆矩阵比对着 loss 曲线猜要直观得多。from sklearn.metrics import classification_report, confusion_matrix def evaluate(model, dataloader, device): 返回预测标签与真实标签用于后续计算指标。 model.eval() preds, trues [], [] with torch.no_grad(): for batch_x, batch_y in dataloader: batch_x batch_x.unsqueeze(1).to(device) outputs model(batch_x) preds.extend(outputs.argmax(dim1).cpu().numpy()) trues.extend(batch_y.numpy()) return trues, preds拿到 trues 和 preds 之后先看classification_report里每个类别的查准率、查全率、F1再看confusion_matrix里哪些类别互相混淆。振动诊断里我一般最关心两类指标正常样本被误报成故障的比例以及故障样本被漏报的比例。前者决定现场会不会三天两头响假警报后者决定真正出问题时能不能兜住。4. 源码解析把故障诊断系统的模块边界与训练主流程拆开看源码解析这部分不打算逐行贴长文件而是按模块边界讲清楚每个文件负责什么、数据怎么流动再拆训练主流程里最容易写错的 checkpoint 保存和早停逻辑。一套能维护的故障诊断系统代码结构比模型结构更重要。临时脚本能跑通 demo但参数一改就崩现场根本留不住。4.1 项目模块划分数据加载、特征、模型、训练、评估与推理我一般会把项目拆成下面这些模块每个文件只干一件事。文件职责对外接口data_loader.py读取原始 npz/csv构造 Dataset 和 DataLoaderload_dataset(path)features.py滑动窗口、FFT、频段能量统计sliding_window(), extract_fft_spectrum()models/fault_cnn.py模型定义FaultCNN1Dtrain.py训练主流程、checkpoint、早停main()evaluate.py混淆矩阵、分类报告、单类别指标evaluate(), report()infer.py单样本/实时流推理predict_one(), online_inference()config.yaml所有超参数与路径无数据流是单向的原始信号 → features.py 切窗 → data_loader.py 读成样本 → train.py 训练 → evaluate.py 评估 → infer.py 部署。特征模块和训练模块不互相 importdata_loader 只负责把切好的样本喂给模型。这样后续换模型、换特征、换数据格式都不需要牵一发动全身。4.2 训练主流程源码checkpoint 保存与早停逻辑训练主流程里最容易写错的不是模型前向而是模型保存时机。常见错误是每个 epoch 都保存一次最后磁盘被 checkpoints 塞满或者只在最后一个 epoch 保存结果过拟合的模型覆盖了最好的参数。我习惯只保存验证集上表现最好的那个 checkpoint配合早停逻辑。class EarlyStopping: 验证指标连续不提升就提前停避免无效训练。 def __init__(self, patience15, min_delta0.001): self.patience patience self.min_delta min_delta self.counter 0 self.best_score None def should_stop(self, val_acc): if self.best_score is None: self.best_score val_acc return False if val_acc self.best_score self.min_delta: self.best_score val_acc self.counter 0 else: self.counter 1 return self.counter self.patiencemin_delta是容忍噪声的阈值。验证准确率提升不足 0.001 也当成没进步否则一点点波动就重置 counter早停形同虚设。patience设 15 表示连续 15 个 epoch 没有明显提升就停。对振动数据来说100 个 epoch 里前 20 个通常已经决定 80% 的效果后 50 个 epoch 基本在磨验证集上最后几个百分点。主流程把数据加载、训练、保存串起来import numpy as np import torch import torch.nn as nn from torch.utils.data import DataLoader, TensorDataset def main(): train_data np.load(data/processed/train.npz) val_data np.load(data/processed/val.npz) train_x, train_y train_data[x], train_data[y] val_x, val_y val_data[x], val_data[y] train_loader DataLoader( TensorDataset(torch.from_numpy(train_x).float(), torch.from_numpy(train_y).long()), batch_size64, shuffleTrue, num_workers4, ) val_loader DataLoader( TensorDataset(torch.from_numpy(val_x).float(), torch.from_numpy(val_y).long()), batch_size64, shuffleFalse, num_workers4, ) device torch.device(cuda if torch.cuda.is_available() else cpu) model FaultCNN1D(num_classes4).to(device) criterion nn.CrossEntropyLoss() optimizer torch.optim.Adam(model.parameters(), lr1e-3) early_stopping EarlyStopping(patience15) best_acc 0.0 for epoch in range(100): train_loss train_one_epoch(model, train_loader, optimizer, criterion, device) trues, preds evaluate(model, val_loader, device) val_acc (np.array(trues) np.array(preds)).mean() print(fepoch {epoch:02d} | loss {train_loss:.4f} | val_acc {val_acc:.4f}) if val_acc best_acc: best_acc val_acc torch.save(model.state_dict(), checkpoints/best.pt) if early_stopping.should_stop(val_acc): break注意torch.from_numpy(...).long()把标签转成 longCrossEntropyLoss 不接受 float 标签。num_workers4在有足够内存时能明显加速数据读取现场机器核数少就改成 1。保存用state_dict()而不是整个 model 对象这样加载时不需要依赖原来的类定义路径部署更干净。checkpoint 保存最好带上验证集准确率torch.save({ model: model.state_dict(), val_acc: val_acc, epoch: epoch, config: cfg, }, checkpoints/best.pt)保存 config 的作用是后悔药。现场三个月后翻出一份模型如果没有当时的窗口长度和采样率参数你根本不知道自己训的是什么数据。4.3 超参数配置化把学习率、窗口长度写进 yaml超参数写死在代码里是最常见的项目烂摊子。窗口长度、步长、学习率、batch size、早停 patience这些参数现场调试时几乎都会改。写进 yaml 后改参数不用翻代码也不会不小心改坏逻辑。# config.yaml data: train_path: data/processed/train.npz val_path: data/processed/val.npz sample_rate: 25600 window_size: 10240 step_size: 5120 model: num_classes: 4 base_channels: 16 train: batch_size: 64 learning_rate: 0.001 epochs: 100 early_stop_patience: 15 num_workers: 4 seed: 42 save: best_model: checkpoints/best.pttrain.py 开头加几行读取 yaml后面所有参数一律从cfg取代码里不出现裸的数字常量。seed 一定要固定否则每次运行数据集打乱顺序不同模型结果无法复现也没法判断改动是好是坏。数据路径、模型保存路径也放进 yaml换一台机器部署时只要改文件路径不用动代码。5. 避坑指南机械设备故障诊断系统最常见的 4 个坑与排查顺序这个章节写的是我自己反复踩过的坑也是帮别人看这类项目时最常见的共性问题。每一条按现象、原因、解决三段写方便你在现场照着排查。5.1 按样本随机切分验证集虚高现场准确率暴跌现象训练准确率 90%验证准确率却能到 98%换到另一台同型号设备测试准确率直接掉到 70% 以下。原因切完窗口后直接按样本随机划分训练集和验证集同一台设备连续采样切出的相邻窗口被同时分到两边。模型在训练时看过验证集里相似度极高的波形本质上是把验证集的答案背下来了。解决按设备编号切分确保验证集里出现的设备没有参与过训练。参考 2.3 的split_by_machine。如果现场只有一台设备就按时间分段切用前 70% 时间段的样本训练用后 30% 验证模拟“未来数据”。切分这步要在预处理阶段一次完成后面训练和评估都用同一份划分不要临时再切。5.2 标签错位模型学的不是故障是窗口序号现象训练损失下降缓慢甚至不降验证准确率长时间停在类别数分之一附近比如 4 分类就卡在 25% 左右。原因窗口切好后标签没有跟着索引对齐。常见于窗口长度和步长不一致的时候人工给窗口标号时写错了位置或者设备状态是按时段标记的切窗口时窗口边界跨过了状态切换点标签取到了错误状态。解决不要手动给每一段窗口标标签用一个可视化脚本打印窗口起点和对应时间戳检查起点所在的设备状态是否和标签一致。对齐逻辑简单验证一下# 打印前 5 个窗口起点和对应时刻的设备状态 for start in range(0, len(signal) - window_size 1, step_size): if start // step_size 5: break timestamp start / sampling_rate print(fwindow start{start}, time{timestamp:.2f}s, state{state_at(timestamp)})state_at(timestamp)是从设备运行记录里查到的状态比如 normal、inner_fault、outer_fault。如果窗口起点落在两个状态的边界上这个窗口要么删掉要么按起点的状态标千万别按窗口中心标。标签错位这个问题错一个样本影响不算大错一批样本模型就直接学废了。5.3 类别不均衡故障样本太少模型永远输出“正常”现象整体准确率看着有 90%但混淆矩阵里故障类别几乎全是 0 查全率模型把所有样本都判成正常。原因正常样本占 90% 以上交叉熵损失被多数类主导模型发现全判正常也能拿到很低的 loss就不再学故障特征了。解决先用WeightedRandomSampler做过采样原理是给故障类样本更高的采样权重让每个 batch 里正常和故障的比例不那么悬殊。from torch.utils.data import WeightedRandomSampler labels_np train_y.numpy() class_counts np.bincount(labels_np) weights 1.0 / class_counts[labels_np] sampler WeightedRandomSampler(weights, num_sampleslen(weights), replacementTrue) train_loader DataLoader( TensorDataset(torch.from_numpy(train_x).float(), torch.from_numpy(train_y).long()), batch_size64, samplersampler, )weights按每个样本所属类别的样本数倒数构造故障类样本少反而被多抽。replacementTrue允许同一个故障样本被重复采样缓解样本量不足。如果故障样本少到只有几十个过采样不够用先做数据增强振动信号可以加小幅高斯噪声、做时间平移、把幅值缩放 0.91.1 倍比在模型上折腾有效得多。5.4 部署后准确率骤降先查传感器再查数据分布现象实验室验证集准确率 95%装到现场只有 70%而且误报集中在一两个类别上。原因现场传感器安装位置和实验室不同比如从设备垂直方向换成了水平方向现场转速波动、负载变化、周围其他设备振动混入导致数据分布偏移。模型学到的是实验室数据的分布现场波形长得不像是很正常的。解决先做快速分布检查用 2.2 的extract_fft_spectrum对比现场和训练集的频谱看峰值频率是否整体偏移。如果传感器位置换了优先重新采集现场数据用少量现场样本微调模型微调时冻结前两层卷积只更新后面几层和分类层防止现场样本少导致过拟合。如果只是噪声水平不同先做带通滤波把训练时关注的频段留住再重新评估。6. 再往前一步用滑窗推理与置信度阈值做现场验证训练和评估都跑通之后模型还只是个静态评测报告。设备现场需要的是实时判断传感器一直传数据系统不能每隔几分钟才出一张报告。我一般的做法是把滑动窗口搬到推理阶段让模型对连续数据流做滚动预测而不是等一整段离线数据。下面这段代码接收最新一段缓冲数据输出多个窗口的平均概率和最终类别。def online_inference(model, buffer, window_size, step_size, device): 对连续信号缓冲做滑窗推理返回平均概率和最终类别。 model.eval() probs [] with torch.no_grad(): for start in range(0, len(buffer) - window_size 1, step_size): window buffer[start:start window_size] x torch.from_numpy(window).float().unsqueeze(0).unsqueeze(0).to(device) p torch.softmax(model(x), dim1).cpu().numpy()[0] probs.append(p) avg_prob np.mean(probs, axis0) cls int(avg_prob.argmax()) return cls, avg_prob单窗口预测很冲动稍微一点噪声扰动就会在类别之间跳变。对多个窗口的概率取平均等于做了一个时间上的平滑比单独看某一次预测稳很多。step_size决定推理间隔也决定系统多久输出一次诊断结果窗口越长看到的上下文越多但延迟也越高。现场部署时我给每个类别设一个置信度阈值通常 0.70.8。低于阈值就标成“不确定”不触发报警即使超过阈值也要连续三个窗口都判同一故障才报警。这套逻辑解决的是信任问题设备老师傅被假警报折腾几次后面真故障也没人信了。我现在的习惯是第一版就把置信度阈值和连续报警次数写进配置文件现场先用三到五天试跑宁可缓报也不要误报。跑过一轮之后再根据误报和漏报的比例回调阈值。这套故障诊断系统的价值其实不是模型准确率多少而是能不能让现场人员把警报当回事。希望帮到你。本文还有配套的精品资源点击获取