在工业实际场景中，振动信号往往夹杂着强烈的背景噪声，导致传统的卷积神经网络（CNN）有可能过度拟合噪声特征。本文将阐述2020年发表于IEEE TII的论文《Deep Residual Shrinkage Networks for Fault Diagnosis》，深入解析其核心原理，并分享基于TensorFlow的完整复现代码。
一、软阈值化与深度学习的结合
深度残差收缩网络（DRSN）的核心创新点在于将自适应软阈值化（Soft Thresholding）引入残差网络（ResNet）。在信号处理领域，软阈值化是小波去噪等经典算法的核心，其数学表达式为：y = sign(x) * max(|x| - τ, 0)。
DRSN通过神经网络自动学习阈值τ。相比于普通的ResNet，DRSN不仅能通过恒等路径缓解梯度消失，还能利用软阈值化非线性层剔除不重要的特征（即噪声），保留对故障诊断有益的判别性特征。
二、逐通道缩放的残差收缩模块
本复现采用了逐通道缩放的残差收缩模块（RSBU-CW，Channel-Wise架构）。如图所示，该模块在普通残差模块的基础上增加了一个子网络：
（1）首先对特征图取绝对值，通过全局平均池化得到一维向量。
（2）经过两层全连接层和Sigmoid激活，输出每个通道的缩放参数α。
（3）最终阈值 τ = α * average(|x|)，实现对不同通道噪声的差异化抑制。
三、完整复现代码
以下代码实现了自适应软阈值层、RSBU-CW模块以及完整的DRSN-CW模型。
"""
旋转机械故障识别程序 (基于深度残差收缩网络)
技术架构：
本程序采用深度残差收缩网络（Deep Residual Shrinkage Network, DRSN）作为核心诊断引擎。
通过在传统残差学习框架中植入自适应软阈值化组件，模型能够从高噪声背景的振动序列中
捕捉故障敏感特征，并实现非线性噪声抑制。
参考文献：
Zhao M, Zhong S, Fu X, Tang B, Pecht M.
Deep residual shrinkage networks for fault diagnosis.
IEEE Transactions on Industrial Informatics, 2020, 16(7): 4681–4690.
"""
import os
import sys
import logging
import numpy as np
import scipy.io as sio
import tensorflow as tf
from tensorflow.keras import layers, models, regularizers
from sklearn.model_selection import train_test_split
# -----------------------------------------------------------------------------
# 1. 运算环境优化配置
# -----------------------------------------------------------------------------
def init_gpu_config():
"""
配置底层的计算环境与硬件资源。
1. 屏蔽非必要的底层框架日志。
2. 针对 NVIDIA GPU 启用显存动态分配策略，优化多任务并发。
"""
os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2'
# 检索物理计算单元
gpus = tf.config.list_physical_devices('GPU')
if gpus:
try:
for gpu in gpus:
tf.config.experimental.set_memory_growth(gpu, True)
logging.info("GPU 加速引擎已就绪，显存自适应模式开启。个数: {}".format(len(gpus)))
except (RuntimeError, Exception) as error:
logging.error("计算加速单元配置冲突: %s", error)
else:
logging.info("未发现可用 GPU，系统切换至 CPU 仿真运行。")
# 初始化配置
logging.basicConfig(level=logging.INFO, format='[%(asctime)s] %(levelname)s: %(message)s')
init_gpu_config()
# -----------------------------------------------------------------------------
# 2. 数据加载管道
# -----------------------------------------------------------------------------
class CWRUDataLoader:
"""
数据加载类。
负责对 CWRU 原始加速度信号进行解析、基于时间窗口的重采样。
"""
def __init__(self, data_root, window_size=1024):
self.base_dir = os.path.abspath(data_root)
self.seq_len = window_size
self.stride_step = window_size # 采用无重叠切片策略
def _load_cwru_signal(self, file_path):
"""
解析 MATLAB 数据，提取驱动端振动分量。
"""
try:
raw_data = sio.loadmat(file_path)
for key in raw_data.keys():
if 'DE_time' in key:
return raw_data[key].ravel()
except Exception:
return None
return None
def load_dataset(self, label_mapping):
"""
构建结构化诊断数据集。
"""
x_samples = []
y_labels = []
is_accessible = False
for class_id, file_list in label_mapping.items():
for mat_file in file_list:
full_path = os.path.join(self.base_dir, "{}.mat".format(mat_file))
if not os.path.exists(full_path):
continue
vibration_signal = self._load_cwru_signal(full_path)
if vibration_signal is None:
continue
is_accessible = True
# 执行窗口化分割
total_len = len(vibration_signal)
for start in range(0, total_len - self.seq_len + 1, self.stride_step):
sample = vibration_signal[start : start + self.seq_len]
x_samples.append(sample)
y_labels.append(class_id)
if not is_accessible:
raise FileNotFoundError("指定的路径 '{}' 下未检测到有效的 .mat 信号文件。".format(self.base_dir))
return np.array(x_samples, dtype='float32'), np.array(y_labels, dtype='int32')
def add_awgn_noise(input_signal, target_snr):
"""
向原始信号注入加性高斯白噪声。
通过信噪比公式：P_noise = P_signal / (10^(SNR/10)) 计算噪声强度。
"""
signal_batch = np.array(input_signal)
rng = np.random.default_rng()
# 动态确定 SNR 参数
actual_snr = target_snr if not isinstance(target_snr, tuple) \
else rng.uniform(target_snr[0], target_snr[1])
# 计算信号功率
signal_power = np.mean(signal_batch**2, axis=1, keepdims=True)
noise_var = signal_power / (10 ** (actual_snr / 10.0))
white_noise = rng.normal(0, np.sqrt(noise_var), signal_batch.shape)
return (signal_batch + white_noise).astype('float32')
# -----------------------------------------------------------------------------
# 3. 深度残差收缩网络 (DRSN) 架构实现
# -----------------------------------------------------------------------------
class SoftThresholding(layers.Layer):
"""
深度残差收缩网络特有的非线性激活层。
实现数学公式：y = sign(x) * max(|x| - τ, 0)，其中 τ 为学习到的阈值。
"""
def __init__(self, **kwargs):
super(SoftThresholding, self).__init__(**kwargs)
def call(self, inputs):
feature_map, tau = inputs
# 广播阈值维度以适配 (Batch, Length, Channels)
tau_expanded = tf.expand_dims(tau, axis=1)
abs_x = tf.abs(feature_map)
return tf.sign(feature_map) * tf.maximum(abs_x - tau_expanded, 0.0)
class RSBU_CW(layers.Layer):
"""
深度残差收缩网络的基本构建块 (RSBU-CW)。
集成了多通道注意力机制与自适应软阈值化。
"""
def __init__(self, out_channels, kernel_size, stride=1, **kwargs):
super(RSBU_CW, self).__init__(**kwargs)
self.out_channels = out_channels
self.stride = stride
self.kernel_size = kernel_size
self.l2_reg = regularizers.l2(1e-4)
# 恒等变换分支（Shortcut Connection）
self.shortcut_layer = None
# 卷积变换
self.bn_a = layers.BatchNormalization()
self.relu_a = layers.Activation('relu')
self.conv_a = layers.Conv1D(self.out_channels, self.kernel_size, strides=self.stride, padding='same',
kernel_initializer='he_normal', kernel_regularizer=self.l2_reg)
self.bn_b = layers.BatchNormalization()
self.relu_b = layers.Activation('relu')
self.conv_b = layers.Conv1D(self.out_channels, self.kernel_size, strides=1, padding='same',
kernel_initializer='he_normal', kernel_regularizer=self.l2_reg)
# 软阈值生成子网
self.gap = layers.GlobalAveragePooling1D()
self.fc1 = layers.Dense(self.out_channels, kernel_initializer='he_normal')
self.bn_fc1 = layers.BatchNormalization()
self.relu_fc1 = layers.Activation('relu')
self.fc2 = layers.Dense(self.out_channels, activation='sigmoid')
self.thresholding_op = SoftThresholding()
def build(self, input_shape):
# 若步长非1或通道数不匹配，则启用 1x1 卷积对齐维度
if self.stride != 1 or input_shape[-1] != self.out_channels:
self.shortcut_layer = layers.Conv1D(self.out_channels, 1, strides=self.stride, padding='same')
super(RSBU_CW, self).build(input_shape)
def call(self, inputs):
shortcut = inputs
if self.shortcut_layer:
shortcut = self.shortcut_layer(inputs)
# 顺序特征提取
x = self.bn_a(inputs)
x = self.relu_a(x)
x = self.conv_a(x)
x = self.bn_b(x)
x = self.relu_b(x)
x = self.conv_b(x)
# 自适应阈值预测流程
abs_x = tf.abs(x)
avg_abs_x = self.gap(abs_x)
# 注意力加权
alpha = self.fc1(avg_abs_x)
alpha = self.bn_fc1(alpha)
alpha = self.relu_fc1(alpha)
alpha = self.fc2(alpha)
# 计算逐通道收缩阈值 τ = alpha * average(|x|)
tau = tf.multiply(alpha, avg_abs_x)
# 软阈值降噪与残差合并
denoised_x = self.thresholding_op([x, tau])
return layers.Add()([denoised_x, shortcut])
class DRSN_CW(models.Model):
"""
深度残差收缩网络完整模型架构 (Channel-Wise variant)。
"""
def __init__(self, num_classes):
super(DRSN_CW, self).__init__(name="DRSN_Core_Diagnostic_Model")
# 初始特征提取
self.stem_conv = layers.Conv1D(32, 15, strides=2, padding='same', kernel_initializer='he_normal')
self.stem_bn = layers.BatchNormalization()
self.stem_relu = layers.Activation('relu')
# 深度残差收缩模块的堆叠
self.backbone = [
RSBU_CW(32, 5, stride=2),
RSBU_CW(32, 5, stride=1),
RSBU_CW(64, 5, stride=2),
RSBU_CW(64, 5, stride=1),
RSBU_CW(128, 5, stride=2),
RSBU_CW(128, 5, stride=1)
]
# 输出决策层
self.head_bn = layers.BatchNormalization()
self.head_relu = layers.Activation('relu')
self.global_pool = layers.GlobalAveragePooling1D()
self.classifier = layers.Dense(num_classes, activation='softmax')
def call(self, inputs):
z = self.stem_conv(inputs)
z = self.stem_bn(z)
z = self.stem_relu(z)
for stage in self.backbone:
z = stage(z)
z = self.head_bn(z)
z = self.head_relu(z)
z = self.global_pool(z)
return self.classifier(z)
# -----------------------------------------------------------------------------
# 4. 训练协议与自动化流水线
# -----------------------------------------------------------------------------
def train_and_test_pipeline(data_path, window_size=1024):
"""
执行端到端的训练与诊断评估流程。
"""
# 标签映射
label_mapping = {
0: ['Normal_0', 'Normal_1', 'Normal_2', 'Normal_3'],
1: ['IR007_0', 'IR007_1', 'IR007_2', 'IR007_3'],
2: ['IR014_0', 'IR014_1', 'IR014_2', 'IR014_3'],
3: ['IR021_0', 'IR021_1', 'IR021_2', 'IR021_3'],
4: ['B007_0', 'B007_1', 'B007_2', 'B007_3'],
5: ['B014_0', 'B014_1', 'B014_2', 'B014_3'],
6: ['B021_0', 'B021_1', 'B021_2', 'B021_3'],
7: ['OR007@6_0', 'OR007@6_1', 'OR007@6_2', 'OR007@6_3'],
8: ['OR014@6_0', 'OR014@6_1', 'OR014@6_2', 'OR014@6_3'],
9: ['OR021@6_0', 'OR021@6_1', 'OR021@6_2', 'OR021@6_3']
}
data_loader = CWRUDataLoader(data_root=data_path, window_size=window_size)
try:
x_raw, y_raw = data_loader.load_dataset(label_mapping)
except Exception as exc:
logging.error("数据预处理阶段异常中止: %s", exc)
return
# 随机化数据集切分
x_train_unnorm, x_holdout, y_train_unnorm, y_holdout = train_test_split(
x_raw, y_raw, test_size=0.3, random_state=42
)
x_val_unnorm, x_test_unnorm, y_val_unnorm, y_test_unnorm = train_test_split(
x_holdout, y_holdout, test_size=0.5, random_state=42
)
# 执行标准化处理 (Standardization)
mu = np.mean(x_train_unnorm)
sigma = np.std(x_train_unnorm)
def normalize_samples(data):
return ((data - mu) / sigma).reshape(-1, window_size, 1)
x_train = normalize_samples(x_train_unnorm)
x_val = normalize_samples(x_val_unnorm)
x_test = normalize_samples(x_test_unnorm)
num_classes = len(label_mapping)
y_train_cat = tf.keras.utils.to_categorical(y_train_unnorm, num_classes)
y_val_cat = tf.keras.utils.to_categorical(y_val_unnorm, num_classes)
y_test_cat = tf.keras.utils.to_categorical(y_test_unnorm, num_classes)
# 模拟极端噪声环境 (-8dB SNR)
x_val_noisy = add_awgn_noise(x_val, target_snr=-8)
x_test_noisy = add_awgn_noise(x_test, target_snr=-8)
def data_augmentation_pipeline(features, labels):
"""
在线数据增强：通过循环移位、随机脉冲及混合噪声提升深度残差收缩网络的泛化能力。
"""
gen = np.random.default_rng()
aug_x = features.copy()
m_batch, m_length, _ = aug_x.shape
# A. 信号循环平移
for i in range(m_batch):
shift = gen.integers(0, m_length)
aug_x[i, :, 0] = np.roll(aug_x[i, :, 0], shift)
# B. 稀疏脉冲干扰
if gen.random() > 0.9:
for i in range(m_batch):
if gen.random() > 0.5:
spike_num = gen.integers(1, 3)
spike_idx = gen.integers(0, m_length, spike_num)
amp = np.std(aug_x[i]) * gen.uniform(1.5, 2.5)
aug_x[i, spike_idx, 0] += amp * gen.choice([-1, 1], size=spike_num)
# C. 随机信噪比扰动
if gen.random() > 0.5:
aug_x = add_awgn_noise(aug_x, target_snr=(-8, 8))
return aug_x.astype(np.float32), labels.astype(np.float32)
def _fix_tensor_shape(feat, lab):
feat.set_shape([None, window_size, 1])
lab.set_shape([None, num_classes])
return feat, lab
# 构建高性能数据加载流
train_ds = tf.data.Dataset.from_tensor_slices((x_train.astype('float32'), y_train_cat.astype('float32')))
train_ds = train_ds.shuffle(2000).batch(64)
train_ds = train_ds.map(
lambda x, y: tf.numpy_function(data_augmentation_pipeline, [x, y], [tf.float32, tf.float32]),
num_parallel_calls=tf.data.AUTOTUNE
).map(_fix_tensor_shape).prefetch(tf.data.AUTOTUNE)
# 编译深度残差收缩模型
drsn_model = DRSN_CW(num_classes=num_classes)
drsn_model.compile(
optimizer=tf.keras.optimizers.Adam(learning_rate=0.001),
loss='categorical_crossentropy',
metrics=['accuracy']
)
logging.info("诊断模型已封装。目标类别: %d, 特征长度: %d", num_classes, window_size)
# 训练监控回调
training_callbacks = [
tf.keras.callbacks.ReduceLROnPlateau(monitor='val_loss', factor=0.5, patience=7, min_lr=1e-6, verbose=1),
tf.keras.callbacks.EarlyStopping(monitor='val_loss', patience=20, restore_best_weights=True)
]
# 启动梯度下降训练
drsn_model.fit(
train_ds,
epochs=100,
validation_data=(x_val_noisy, y_val_cat),
callbacks=training_callbacks,
verbose=2
)
# 最终准确率测试
_, final_acc = drsn_model.evaluate(x_test_noisy, y_test_cat, verbose=0)
print("\n" + "="*50)
print("在强环境噪声 (-8dB SNR) 下，深度残差收缩网络测试准确率为: {:.2f}%".format(final_acc * 100))
print("="*50)
# -----------------------------------------------------------------------------
# 5. 主程序入口
# -----------------------------------------------------------------------------
if __name__ == "__main__":
# 默认检索目录
TARGET_DATA_ROOT = os.path.join(os.getcwd(), 'bearing_data')
if not os.path.exists(TARGET_DATA_ROOT):
logging.warning("默认数据路径未找到: %s", TARGET_DATA_ROOT)
interactive_path = input(">>> 请手动输入 CWRU 数据集 (.mat) 文件夹路径: ").strip()
if interactive_path:
TARGET_DATA_ROOT = interactive_path
else:
logging.error("未获取到有效数据源，系统退出。")
sys.exit(0)
# 开启诊断序列
train_and_test_pipeline(TARGET_DATA_ROOT, window_size=1024)
四、噪声环境下的实测表现
实验数据采用了加噪的CWRU轴承数据集。样本长度为1024个采样点，类别数为10个（正常状态及不同损伤程度的故障），验证集和测试集的数据信噪比（SNR）为-8 dB（较强噪声）。
从控制台输出可以看到，在-8 dB的测试环境下，DRSN-CW的测试准确率稳定在90%以上，初步展示了自适应软阈值模块在特征提取过程中的“去伪存真”作用。
论文标题: Deep residual shrinkage networks for fault diagnosis
出版期刊: IEEE Transactions on Industrial Informatics. 2020, 16(7): 4681-4690.
来自 “ ITPUB博客 ” ，链接：https://blog.itpub.net/70049566/viewspace-3111202/ ，如需转载，请注明出处，否则将追究法律责任。
上一篇：
没有了~
下一篇：
没有了~
请登录后发表评论
登录
全部评论
