CWRU轴承振动信号导入包:从.mat文件到分类器的高效实现
2026/9/23 7:35:38 网站建设 项目流程

简介:这份资源是面向机械健康监测与故障诊断方向的科研人员和学生准备的CWRU轴承振动信号导入包,用于快速加载并处理凯斯西储大学公开的轴承数据集,为故障检测与预测实验提供标准化的数据入口。压缩包共6个文件,约6KB,以Python脚本为主,辅以打包配置、许可协议与说明文档,其中核心模块提供多变量振动信号的组织与读取能力,方便直接接入后续分析流程。已有260人学习下载,适合刚接触CWRU数据集、希望省去数据整理环节的入门者,也适合需要统一数据接口的算法验证场景。使用者可借助该包完成信号预处理、特征提取与异常识别等环节的快速搭建,把精力集中在诊断模型的设计与对比上,从而更高效地复现和验证轴承故障诊断方法。

1. 从一堆 .mat 文件到能跑的分类器:CWRU轴承振动信号导入包到底解决什么

如果你做过旋转机械故障诊断,大概率绕不开凯斯西储大学(CWRU)轴承数据集。它几乎是入门轴承振动信号分析的第一站:文件不大、标签清晰、工况齐全,网上讲解一抓一大把。但真正动手时,很多人卡在第一步——下载下来是一堆 .mat 文件,文件名像97.mat105.mat,打开后变量名又是X097_DE_timeX105_DE_time这种,到底哪个文件对应哪种故障、哪个通道是驱动端、采样率多少、怎么切成训练集,全靠翻文档和猜。所谓 CWRU轴承振动信号导入包,本质上就是把这层“脏活”封装掉:给定数据目录,自动扫描文件、解析文件名与标签映射、抽取指定通道、按窗口切分、归一化,最后吐出可以直接喂给 sklearn 或 PyTorch 的数组。它解决的不是模型精度问题,而是让一个新手在半小时内从零跑通“加载—切分—训练—评估”的闭环,而不是花两天在文件解析上翻车。适合谁?适合刚接触 CWRU 数据集的学生、需要快速搭 baseline 的算法工程师,以及想把振动信号分类做成教学案例的人。下面我按自己实际搭这套导入流程的顺序,把选型、实现、参数和坑一次讲清。

2. 先搞懂 CWRU 文件命名与通道结构,再谈导入

2.1 文件名数字与故障标签的对应关系

CWRU 数据集的 .mat 文件命名不是随机的,每个数字对应一种工况和故障直径。常见做法是维护一张映射表,把文件编号区间映射到标签。以驱动端(DE)数据为例,正常基线文件编号在 97 到 100 附近,内圈故障、外圈故障、滚动体故障各有自己的编号段。不同直径(0.007、0.014、0.021 英寸)又分属不同编号。很多人第一次导入时直接把所有文件混在一起,结果标签全错,训练出来的模型准确率虚高,因为同一类故障的不同负载工况被当成了不同类。

我一般会先写一个解析函数,从文件名提取编号,再查表得到标签。表可以硬编码,也可以放在单独的 JSON 里方便修改。下面是一个最小可用的映射片段,覆盖最常用的 12 类(正常 + 3 种故障 × 3 种直径 + 少量工况):

# label_map.py # 键为文件名中的数字编号,值为标签字符串 # 仅列出最常用的一部分,实际使用时可扩展 CWRU_LABEL_MAP = { 97: "normal", 98: "normal", 99: "normal", 100: "normal", 105: "inner_race_007", 106: "inner_race_007", 107: "inner_race_007", 108: "inner_race_007", 118: "outer_race_007", 119: "outer_race_007", 120: "outer_race_007", 121: "outer_race_007", 130: "ball_007", 131: "ball_007", 132: "ball_007", 133: "ball_007", # 0.014 和 0.021 英寸的编号段可继续补充 }

逻辑说明:编号到标签的映射必须和官方文档核对,不能凭记忆。参数说明:CWRU_LABEL_MAP的键是整数,值是字符串标签;如果后续要做多分类,标签字符串可以直接作为类别名。注意,同一故障直径下不同负载的文件编号不同,但标签应该相同,所以映射表里会出现多个编号指向同一个标签的情况。

2.2 驱动端、风扇端与基座通道的取舍

每个 .mat 文件里通常包含多个变量,比如X097_DE_time是驱动端加速度计数据,X097_FE_time是风扇端,X097_BA_time是基座。做轴承故障分类,绝大多数论文和工程实践只用驱动端(DE),因为信号信噪比最高、故障特征最明显。风扇端信号衰减大,基座信号更弱。导入包应该允许指定通道后缀,默认DE

读取 .mat 文件用scipy.io.loadmat即可,但要注意变量名是动态的,不能写死。常见做法是遍历文件里的键,找到以_DE_time结尾的那个。下面是一个通道抽取函数:

import scipy.io as sio import numpy as np def load_signal(mat_path, channel="DE"): """ 从 .mat 文件中读取指定通道的振动信号。 mat_path: .mat 文件路径 channel: "DE"、"FE" 或 "BA" 返回一维 numpy 数组 """ mat = sio.loadmat(mat_path) suffix = f"_{channel}_time" for key in mat: if key.endswith(suffix): return mat[key].ravel().astype(np.float32) raise KeyError(f"未找到通道 {channel} 对应的变量,文件:{mat_path}")

逻辑说明:ravel()把列向量压平成一维,astype(np.float32)减少内存占用。参数说明:channel默认DE,如果要做多通道融合可以改成列表循环读取。注意,有些文件可能没有_BA_time,所以异常处理要保留。

2.3 采样率与窗口切分的参数怎么定

CWRU 驱动端采样率常见为 12 kHz 和 48 kHz 两种。12 kHz 数据用于故障直径 0.007 到 0.021 英寸的多数文件,48 kHz 用于部分 0.014 和 0.021 英寸文件。导入包必须记录每个文件的采样率,否则后续做频谱分析时频率轴全错。窗口切分通常用固定长度滑动窗,比如窗长 1024 点、步长 512 点。窗长太短,频率分辨率不够;窗长太长,样本数太少。我一般先用 1024 点试,如果分类效果差再调到 2048。

def segment_signal(signal, window_size=1024, step=512): """ 对一维信号做滑动窗切分。 返回形状为 (n_windows, window_size) 的数组。 """ n = len(signal) windows = [] for start in range(0, n - window_size + 1, step): windows.append(signal[start:start + window_size]) return np.stack(windows)

逻辑说明:range的终点是n - window_size + 1,保证最后一个窗完整。参数说明:window_size默认 1024,step默认 512,即 50% 重叠。如果样本不够,可以减小 step 到 256 增加样本量,但要注意相邻窗相关性变高。

3. 把导入包写成可复用的 Dataset:从目录扫描到批量化

3.1 目录扫描与标签自动匹配

导入包的核心入口应该是一个函数,输入数据根目录,输出特征矩阵和标签向量。目录结构可以扁平,也可以按故障类型分文件夹。我倾向于扁平结构,因为 CWRU 原始下载就是一堆 .mat 文件。扫描时用glob找所有 .mat,然后从文件名提取数字,查映射表。如果某个文件不在映射表里,跳过并打印警告,不要静默丢弃。

import os import glob import re from label_map import CWRU_LABEL_MAP def scan_dataset(root_dir, channel="DE", window_size=1024, step=512): """ 扫描目录下所有 .mat 文件,返回 (X, y, file_names)。 X: (n_samples, window_size) y: (n_samples,) 整数标签 file_names: 每个样本对应的源文件名,便于排查 """ mat_files = glob.glob(os.path.join(root_dir, "*.mat")) X_list, y_list, name_list = [], [], [] label_to_id = {} for mat_path in sorted(mat_files): base = os.path.basename(mat_path) match = re.match(r"(\d+)", base) if not match: print(f"跳过无法解析的文件:{base}") continue file_id = int(match.group(1)) if file_id not in CWRU_LABEL_MAP: print(f"跳过未映射的文件:{base}") continue label_str = CWRU_LABEL_MAP[file_id] if label_str not in label_to_id: label_to_id[label_str] = len(label_to_id) signal = load_signal(mat_path, channel=channel) windows = segment_signal(signal, window_size, step) X_list.append(windows) y_list.append(np.full(len(windows), label_to_id[label_str])) name_list.extend([base] * len(windows)) X = np.concatenate(X_list, axis=0) y = np.concatenate(y_list, axis=0) return X, y, name_list, label_to_id

逻辑说明:label_to_id动态生成,保证标签从 0 开始连续。name_list记录每个窗口来自哪个文件,方便后续做按文件划分训练测试集,避免同一文件窗口同时出现在训练和测试中导致数据泄漏。参数说明:root_dir是 .mat 文件所在目录;channel默认DEwindow_sizestep按上一节建议设置。

3.2 按文件划分训练集与测试集,避免数据泄漏

这是最容易被忽略的一步。如果随机划分窗口,同一个 .mat 文件切出的窗口可能一部分在训练集、一部分在测试集,模型相当于见过测试数据的“兄弟窗口”,准确率会虚高十几个百分点。正确做法是按源文件划分:先把文件列表分成训练文件和测试文件,再分别取窗口。我一般用 70% 文件做训练,30% 做测试,且保证每类都有文件进入测试集。

from sklearn.model_selection import train_test_split def split_by_file(X, y, file_names, test_size=0.3, random_state=42): """ 按源文件划分训练集和测试集。 """ unique_files = np.unique(file_names) train_files, test_files = train_test_split( unique_files, test_size=test_size, random_state=random_state ) train_mask = np.isin(file_names, train_files) test_mask = np.isin(file_names, test_files) return (X[train_mask], y[train_mask], X[test_mask], y[test_mask])

逻辑说明:np.isin生成布尔掩码,避免循环。参数说明:test_size默认 0.3,random_state固定保证可复现。注意,如果某类文件数很少,train_test_split可能把某类全部划到一边,需要加stratify参数按标签分层,但分层对象应该是文件对应的标签,不是窗口标签。

3.3 归一化与数据加载器对接

振动信号幅值范围因工况而异,归一化能加速收敛。常见做法是每个窗口减均值除标准差,或者全局 min-max。我倾向于按窗口做 z-score,因为不同负载下幅值差异大,按窗口归一化更鲁棒。

def normalize_windows(X): """ 对每个窗口做 z-score 归一化。 """ mean = X.mean(axis=1, keepdims=True) std = X.std(axis=1, keepdims=True) + 1e-8 return (X - mean) / std

逻辑说明:keepdims=True保持二维形状,便于广播。1e-8防止除零。参数说明:如果要做全局归一化,把axis=1改成axis=None,但我不推荐,因为会丢失工况差异信息。

对接 PyTorch 时,把X转成float32张量,y转成long张量,用TensorDatasetDataLoader包装即可。批大小一般设 64 或 128,太小训练慢,太大内存吃紧。

4. 导入环节的避坑与排查:那些让我重跑一晚上的细节

4.1 现象:读取 .mat 报错“Unknown mat file type”

原因:CWRU 部分文件是 v7.3 格式的 HDF5,scipy.io.loadmat不支持。解决:用h5py读取,或者用mat73库。判断方法:用sio.loadmat报错时捕获异常,改用h5py.File遍历键。注意 h5py 读出来是转置的,需要.T

4.2 现象:标签全变成 0,模型准确率 100%

原因:label_to_id在循环里动态生成,但y_list里存的是当时的值,如果后面又遇到新标签,前面的映射没更新。更常见的是映射表里所有文件都映射到同一个标签。解决:先遍历所有文件收集唯一标签,建立固定映射,再第二次遍历读取信号。或者直接用sklearn.preprocessing.LabelEncoder统一编码。

4.3 现象:训练集准确率高,测试集准确率低得离谱

原因:按窗口随机划分导致数据泄漏。解决:按源文件划分,如 3.2 节所述。检查方法:打印训练集和测试集里出现的文件名,看是否有交集。

4.4 现象:内存爆掉,读几个文件就卡死

原因:48 kHz 文件很长,一个文件几十万点,全部读入内存再切窗,内存占用是原始信号的数倍。解决:边读边切,切完立即释放原始信号;或者用生成器逐文件 yield 窗口。如果数据量特别大,可以先把切好的窗口存成.npy内存映射文件。

4.5 现象:同一类故障不同负载的样本被当成不同类

原因:映射表把不同编号映射到了不同标签,但实际它们属于同一故障类型。解决:映射表的键是文件编号,值是故障类型字符串,确保同一故障不同负载的编号指向同一个字符串。检查方法:打印每个标签对应的文件编号列表,人工核对。

5. 进阶:用导入包快速验证一个 1D-CNN baseline

导入包搭好后,验证它是否可靠的最好方式是用一个简单模型跑一遍。我一般用 1D-CNN,因为振动信号是一维时序,CNN 能自动提取局部特征,比手工特征省事。下面是一个最小 PyTorch 模型,输入窗长 1024,输出类别数由label_to_id长度决定。

import torch import torch.nn as nn class CNN1D(nn.Module): def __init__(self, n_classes, in_channels=1): super().__init__() self.features = nn.Sequential( nn.Conv1d(in_channels, 16, kernel_size=15, stride=2, padding=7), nn.ReLU(), nn.MaxPool1d(2), nn.Conv1d(16, 32, kernel_size=7, stride=2, padding=3), nn.ReLU(), nn.MaxPool1d(2), nn.Conv1d(32, 64, kernel_size=3, stride=2, padding=1), nn.ReLU(), nn.AdaptiveAvgPool1d(1) ) self.classifier = nn.Linear(64, n_classes) def forward(self, x): # x: (batch, 1, window_size) x = self.features(x) x = x.squeeze(-1) return self.classifier(x)

逻辑说明:三层卷积逐步增大通道数,AdaptiveAvgPool1d(1)把时间维压成 1,再接全连接分类。参数说明:kernel_size第一层用 15 是为了覆盖足够长的局部波形;stride=2降采样减少计算量。训练时用CrossEntropyLoss,优化器选 Adam,学习率 1e-3,跑 20 个 epoch 通常能看到测试准确率到 95% 以上。如果准确率上不去,先检查导入环节的标签和划分,而不是调模型。

验证导入包是否正确,还有一个笨办法但很有效:随机抽一个测试窗口,画出波形,再打印它的标签和源文件名,人工看一眼波形是否像对应故障。内圈故障会有周期性冲击,外圈故障冲击间隔更稳定,滚动体故障调制现象明显。这一步花五分钟,能省掉后面几小时的玄学调参。

我自己的习惯是,每次换数据集或改导入逻辑,先跑一个“小样本过拟合”测试:取 10 个窗口,让模型过拟合到 100% 准确率。如果做不到,说明导入的标签或数据有问题,而不是模型不行。这个习惯帮我省了很多后悔药。希望帮到你。

本文还有配套的精品资源,点击获取

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询