强噪声下的故障诊断：深度残差收缩网络（DRSN）算法解析与Python复现

匆忙游

805人浏览 · 2026-02-27 15:25:59

匆忙游 · 2026-02-27 15:25:59 发布

本文探讨了论文《Deep Residual Shrinkage Networks for Fault Diagnosis》中所提出的深度残差收缩网络（DRSN）。在工业旋转机械的故障诊断中，振动信号往往夹杂着较强的背景噪声。为了提升模型在低信噪比SNR环境下的效果，DRSN通过引入可学习的阈值机制，将信号降噪与特征学习进行了集成。

一、自适应软阈值化的非线性变换

深度残差收缩网络的核心创新点在于将“软阈值化（Soft Thresholding）”作为非线性变换层嵌入到深度网络架构中。

如上图所示，软阈值化并非简单地像ReLU那样将负值归零，而是将绝对值低于某一阈值τ的特征置为零，并将超过阈值的特征向零收缩。这种操作在信号处理领域被广泛用于小波去噪（Wavelet Denoising）。在深度残差收缩网络中，这个阈值τ并非由专家经验手动设定，而是通过一个专门设计的子网络自动学习得到，实现了针对不同输入样本的自适应降噪。

二、通道级残差收缩块（RSBU-CW）

为了实现自动化特征过滤，论文设计了带有通道级阈值的残差收缩块（Residual Shrinkage Building Unit with Channel-Wise thresholds, RSBU-CW）。

首先对输入特征图进行绝对值处理，通过全局平均池化（Global Average Pooling, GAP）将其压缩为一个特征向量。该向量经过两层全连接层进行非线性映射，通过Sigmoid函数输出一个介于0到1之间的比例因子。这个因子与原始特征的均值相乘，便得到了每个通道专属的缩放阈值。这种通道级的灵活处理，使得模型能够针对性地抑制不同通道中的噪声干扰。

三、基于西储大学数据集的端到端复现

在实验部分，采用西储大学（CWRU）轴承数据集进行性能验证。实验涵盖了正常状态、内圈故障、滚珠故障和外圈故障四类状态，并细分为不同的损伤尺寸。

以下是基于TensorFlow/Keras框架复现的深度残差收缩网络通道级版本（DRSN-CW）完整代码。代码实现了从数据加载、滑动窗口切片到模型构建与训练的全流程：

# =============================================================================
# 
# 本程序旨在复现 Zhao 等人于 2020 年发表在 IEEE Transactions on Industrial
# Informatics 上的关于深度残差收缩网络（DRSN）的原创工作。代码使用
# TensorFlow/Keras 框架，针对 CWRU 轴承振动数据集，实现了一维深度残差收缩
# 网络通道级阈值版本（DRSN-CW）的端到端故障诊断流程。
#
# 参考文献（Reference）：
# Zhao, M., Zhong, S., Fu, X., Tang, B., & Pecht, M. (2020).
# Deep residual shrinkage networks for fault diagnosis.
# IEEE Transactions on Industrial Informatics, 16(7), 4681–4690.
# https://doi.org/10.1109/TII.2019.2943898
#
# 核心功能模块涵盖：
# 1. 环境配置：自动检测并优化 TensorFlow 的计算资源分配策略。
# 2. 信号处理：实现 MATLAB 原始信号的加载、驱动端信号提取、滑动窗口切片，
#    以及基于训练集均方差的特征集标准化。
# 3. 鲁棒性验证：支持高斯白噪声注入（可控 SNR）和动态在线数据增强机制。
# 4. 网络构建：核心构建通道级自适应软阈值收缩单元（RSBU-CW），通过学习
#    通道特定阈值来抑制噪声干扰。
# 5. 模型验证：完成模型的编译、训练、验证，并在模拟的强噪声工况下进行
#    最终性能评估。
#
# 网络设计理念聚焦于将软阈值操作作为可学习的非线性变换层嵌入到残差结构中，
# 以期在处理高背景噪声的振动信号时，提升特征的判别性和模型的泛化能力。
# =============================================================================

import os
import sys
import logging
import numpy as np
import scipy.io as io
import tensorflow as tf
from tensorflow.keras import layers, models, regularizers
from sklearn.model_selection import train_test_split

# =============================================================================
# 第一部分：运行时计算环境优化与资源配置
# =============================================================================

# 全局日志记录器配置
logging.basicConfig(level=logging.INFO, format='[%(asctime)s] %(levelname)s: %(message)s')

class DeviceConfig:
    """
    硬件资源管理类：
    负责探测底层异构计算架构（CPU/GPU）并配置 TensorFlow 的显存增量分配策略。
    """
    @staticmethod
    def init_device():
        """
        初始化计算环境。
        激活动态显存增长策略（Memory Growth），以实现 GPU 显存的高效按需分配。
        """
        try:
            import sklearn
            import scipy
        except ImportError as dependency_error:
            logging.error("核心依赖库缺失: {0}".format(dependency_error))
            sys.exit(1)

        os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2'
        physical_gpus = tf.config.list_physical_devices('GPU')
        
        if physical_gpus:
            try:
                for device in physical_gpus:
                    tf.config.experimental.set_memory_growth(device, True)
                logging.info("成功初始化 GPU 环境，已启用 {0} 个设备的显存自适应模式。".format(len(physical_gpus)))
            except RuntimeError as runtime_err:
                logging.error("GPU 运行时策略设置失败: {0}".format(runtime_err))
        else:
            logging.info("未检测到可用 GPU，计算流水线将切换至 CPU 模式。")

# 初始化运行环境
DeviceConfig.init_device()

# =============================================================================
# 第二部分：数据集构建逻辑
# =============================================================================

class CWRULoader:
    """
    时序数据驱动引擎：
    封装了针对 CWRU 原始 MAT 信号的解析逻辑、驱动端分量提取及滑动窗口切片技术。
    """
    def __init__(self, directory, window_size=1024):
        """
        初始化数据引擎。
        :param directory: 原始数据集存放的物理路径。
        :param window_size: 序列样本的时间跨度。
        """
        self.base_dir = os.path.abspath(directory)
        self.seq_len = window_size
        self.step_size = window_size  # 默认步长等于窗口长度，即无重叠采样

    def _load_mat(self, file_path):
        """
        私有方法：解析 .mat 文件并定位驱动端振动分量。
        """
        try:
            raw_content = io.loadmat(file_path)
            for identifier in raw_content.keys():
                if 'DE_time' in identifier:
                    return raw_content[identifier].ravel()
        except Exception:
            return None
        return None

    def load_data(self, FAULT_CONFIG_MAPPING):
        """
        构建带标签的特征矩阵。
        :param FAULT_CONFIG_MAPPING: 映射关系字典，格式为 {标签编号: [对应文件名列表]}。
        :return: (特征张量, 标签向量) 的 Numpy 数组。
        """
        feature_pool = []
        label_pool = []
        valid_file_counter = 0

        for class_id, resource_list in FAULT_CONFIG_MAPPING.items():
            for resource_name in resource_list:
                full_path = os.path.join(self.base_dir, "{0}.mat".format(resource_name))
                if not os.path.exists(full_path):
                    continue
                
                time_series = self._load_mat(full_path)
                if time_series is None:
                    continue
                
                valid_file_counter += 1
                # 滑动窗口切片处理
                num_segments = (len(time_series) - self.seq_len) // self.step_size + 1
                for i in range(num_segments):
                    offset = i * self.step_size
                    segment = time_series[offset : offset + self.seq_len]
                    feature_pool.append(segment)
                    label_pool.append(class_id)
        
        if valid_file_counter == 0:
            raise IOError("在指定目录 {0} 下未检索到符合条件的 .mat 信号文件。".format(self.base_dir))
            
        return np.array(feature_pool, dtype='float32'), np.array(label_pool, dtype='int32')

def add_noise(signal_batch, snr_ratio):
    """
    加性高斯白噪声 (AWGN) 注入算子：模拟实际工况下的信噪比环境。
    基于信号功率密度动态计算并叠加随机噪声。
    """
    signal_batch = np.array(signal_batch)
    random_state = np.random.default_rng()
    
    # 确定当前的信噪比数值
    target_snr = snr_ratio if isinstance(snr_ratio, (int, float)) else random_state.uniform(snr_ratio[0], snr_ratio[1])
    
    # 计算信号平均功率及对应的噪声功率阈值
    avg_signal_power = np.mean(np.square(signal_batch), axis=1, keepdims=True)
    avg_noise_power = avg_signal_power / (10 ** (target_snr / 10.0))
    
    noise_component = random_state.normal(0, np.sqrt(avg_noise_power), signal_batch.shape)
    return (signal_batch + noise_component).astype('float32')

# =============================================================================
# 第三部分：深度残差收缩网络 (DRSN) 架构实现
# =============================================================================

class SoftThresholding(layers.Layer):
    """
    非线性软阈值激活层：实现 DRSN 的核心收缩算子。
    数学模型：y = sgn(x) * max(|x| - τ, 0)，旨在滤除冗余噪声特征。
    """
    def __init__(self, **kwargs):
        super(SoftThresholding, self).__init__(**kwargs)

    def call(self, inputs):
        """
        前向传递逻辑。
        :param inputs: 包含 [特征张量, 阈值向量] 的列表。
        """
        feature_map, tau = inputs
        # 维度对齐：将阈值向量扩展至与特征图相同的空间维度
        tau_expanded = tf.expand_dims(tau, axis=1)
        
        magnitude = tf.abs(feature_map)
        shrinkage_term = tf.maximum(magnitude - tau_expanded, 0.0)
        return tf.sign(feature_map) * shrinkage_term

class RSBU_CW(layers.Layer):
    """
    通道级残差收缩块 (RSBU-CW)：DRSN 的核心架构组件。
    集成子网络分支以学习通道特定阈值，实现自适应特征降噪。
    """
    def __init__(self, filters, kernel_size, strides=1, **kwargs):
        super(RSBU_CW, self).__init__(**kwargs)
        self.out_channels = filters
        self.stride = strides
        self.k_size = kernel_size
        self.weight_decay = regularizers.l2(1e-4)

        self.identity_mapping = None
        
        # 卷积变换流水线
        self.batch_norm_a = layers.BatchNormalization()
        self.relu_a = layers.Activation('relu')
        self.conv_a = layers.Conv1D(filters, kernel_size, strides=strides, padding='same', 
                                   kernel_initializer='he_normal', kernel_regularizer=self.weight_decay)
        
        self.batch_norm_b = layers.BatchNormalization()
        self.relu_b = layers.Activation('relu')
        self.conv_b = layers.Conv1D(filters, kernel_size, strides=1, padding='same', 
                                   kernel_initializer='he_normal', kernel_regularizer=self.weight_decay)
        
        # 阈值学习子网络
        self.pooling_global = layers.GlobalAveragePooling1D()
        self.dense_fc1 = layers.Dense(filters, kernel_initializer='he_normal')
        self.bn_fc = layers.BatchNormalization()
        self.relu_fc = layers.Activation('relu')
        self.dense_fc2 = layers.Dense(filters, activation='sigmoid')
        self.shrinkage_layer = SoftThresholding()

    def build(self, input_shape):
        """
        参数化残差快捷路径：
        当特征图维度（步长或通道数）不匹配时，自动引入 1x1 卷积执行线性空间投影。
        """
        if self.stride != 1 or input_shape[-1] != self.out_channels:
            self.identity_mapping = models.Sequential([
                layers.Conv1D(self.out_channels, 1, strides=self.stride, padding='same'),
            ])
        super(RSBU_CW, self).build(input_shape)

    def call(self, inputs):
        """
        逻辑流程：输入 -> 两次卷积变换 -> 计算通道权重 -> 派生阈值 -> 软阈值降噪 -> 残差融合。
        """
        shortcut_path = inputs
        if self.identity_mapping:
            shortcut_path = self.identity_mapping(inputs)

        # 特征映射分支
        feat = self.batch_norm_a(inputs)
        feat = self.relu_a(feat)
        feat = self.conv_a(feat)
        feat = self.batch_norm_b(feat)
        feat = self.relu_b(feat)
        feat = self.conv_b(feat)

        # 阈值计算分支
        signal_abs = tf.abs(feat)
        abs_mean_vec = self.pooling_global(signal_abs)
        
        z = self.dense_fc1(abs_mean_vec)
        z = self.bn_fc(z)
        z = self.relu_fc(z)
        alpha = self.dense_fc2(z)
        
        # 阈值 = 绝对值均值 * 学习到的比例因子
        tau = tf.multiply(alpha, abs_mean_vec)
        
        # 执行非线性收缩与相加
        denoised_feat = self.shrinkage_layer([feat, tau])
        return layers.Add()([denoised_feat, shortcut_path])

class DRSN_CW(models.Model):
    """
    深度残差收缩网络主体架构：
    包含输入层、多级 RSBU 模块堆叠、全局池化以及 Softmax 分类器。
    """
    def __init__(self, num_classes):
        super(DRSN_CW, self).__init__(name="DRSN_Bearing_Expert")
        
        # 输入层
        self.stem_conv = layers.Conv1D(32, 15, strides=2, padding='same', kernel_initializer='he_normal')
        self.stem_bn = layers.BatchNormalization()
        self.stem_act = layers.Activation('relu')
        
        # 核心特征提取骨架：三级收缩块序列
        self.stage_1 = [RSBU_CW(32, 5, strides=2), RSBU_CW(32, 5, strides=1)]
        self.stage_2 = [RSBU_CW(64, 5, strides=2), RSBU_CW(64, 5, strides=1)]
        self.stage_3 = [RSBU_CW(128, 5, strides=2), RSBU_CW(128, 5, strides=1)]
        
        # 输出头
        self.head_bn = layers.BatchNormalization()
        self.head_act = layers.Activation('relu')
        self.gap_layer = layers.GlobalAveragePooling1D()
        self.softmax_layer = layers.Dense(num_classes, activation='softmax')

    def call(self, inputs):
        """
        端到端推理路径。
        """
        x = self.stem_conv(inputs)
        x = self.stem_act(self.stem_bn(x))
        
        for block in self.stage_1 + self.stage_2 + self.stage_3:
            x = block(x)
            
        x = self.head_act(self.head_bn(x))
        x = self.gap_layer(x)
        return self.softmax_layer(x)

# =============================================================================
# 第四部分：自动化诊断流水线与评估逻辑
# =============================================================================

def execute_diagnostic_workflow(data_root, input_length=1024):
    """
    主控程序：数据加载、模型训练、强噪声测试的全流程。
    """
    # CWRU 故障类型
    FAULT_CONFIG_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']
    }
    
    # 实例化处理引擎
    loader = CWRULoader(directory=data_root, window_size=input_length)
    
    try:
        x_raw, y_raw = loader.load_data(FAULT_CONFIG_MAPPING)
    except Exception as fatal_err:
        logging.error("数据准备阶段发生严重错误: %s", fatal_err)
        return

    # 数据集切分 (7:1.5:1.5)
    train_x, tmp_x, train_y, tmp_y = train_test_split(x_raw, y_raw, test_size=0.3, random_state=42)
    val_x, test_x, val_y, test_y = train_test_split(tmp_x, tmp_y, test_size=0.5, random_state=42)
    
    # 基于训练统计量的 Z-Score 标准化
    train_mean = np.mean(train_x)
    train_std = np.std(train_x)
    
    def norm_and_reshape(data):
        return ((data - train_mean) / train_std).reshape(-1, input_length, 1)

    x_train_final = norm_and_reshape(train_x)
    x_val_final = norm_and_reshape(val_x)
    x_test_final = norm_and_reshape(test_x)
    
    num_classes = len(FAULT_CONFIG_MAPPING)
    y_train_encoded = tf.keras.utils.to_categorical(train_y, num_classes)
    y_val_encoded = tf.keras.utils.to_categorical(val_y, num_classes)
    y_test_encoded = tf.keras.utils.to_categorical(test_y, num_classes)

    # 稳健性验证配置：在测试数据中注入 -8dB 极强噪声，以评估模型在恶劣工业噪声下的特征判别能力。
    x_val_noisy = add_noise(x_val_final, snr_ratio=-8)
    x_test_noisy = add_noise(x_test_final, snr_ratio=-8)

    def augment_batch(features, labels):
        """
        实时在线数据增强策略：融合了时域循环移位、瞬态异常脉冲仿真以及宽动态范围的信噪比扰动机制。
        """
        prng = np.random.default_rng()
        augmented_x = features.copy()
        batch_n, step_n, _ = augmented_x.shape

        # 随机时域循环平移
        for idx in range(batch_n):
            pixel_shift = prng.integers(0, step_n)
            augmented_x[idx, :, 0] = np.roll(augmented_x[idx, :, 0], pixel_shift)

        # 瞬态异常脉冲注入
        if prng.random() > 0.9: 
            for idx in range(batch_n):
                if prng.random() > 0.5: 
                    spike_count = prng.integers(1, 3) 
                    spike_pos = prng.integers(0, step_n, spike_count)
                    pulse_amp = np.std(augmented_x[idx]) * prng.uniform(1.5, 2.5) 
                    augmented_x[idx, spike_pos, 0] += pulse_amp * prng.choice([-1, 1], size=spike_count)

        # 宽动态范围信噪比扰动
        if prng.random() > 0.5: 
            augmented_x = add_noise(augmented_x, snr_ratio=(-8, 8))

        return augmented_x.astype(np.float32), labels.astype(np.float32)

    def set_tensor_shape(f_tensor, l_tensor):
        f_tensor.set_shape([None, input_length, 1])
        l_tensor.set_shape([None, num_classes])
        return f_tensor, l_tensor

    # 构建高性能 I/O 流水线
    training_pipeline = tf.data.Dataset.from_tensor_slices((x_train_final.astype('float32'), y_train_encoded))
    training_pipeline = training_pipeline.shuffle(len(train_x)).batch(64)
    training_pipeline = training_pipeline.map(
        lambda x, y: tf.numpy_function(augment_batch, [x, y], [tf.float32, tf.float32]),
        num_parallel_calls=tf.data.AUTOTUNE
    ).map(set_tensor_shape).prefetch(tf.data.AUTOTUNE)

    # 模型实例化与编译
    drsn_model = DRSN_CW(num_classes=num_classes)
    drsn_model.compile(
        optimizer=tf.keras.optimizers.Adam(learning_rate=1e-3), 
        loss='categorical_crossentropy',
        metrics=['accuracy']
    )

    logging.info("基于 DRSN 的诊断系统已就绪。分类数: %d, 采样窗口: %d", num_classes, input_length)
    
    # 训练监控回调函数
    training_monitors = [
        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(
        training_pipeline,
        epochs=100,
        validation_data=(x_val_noisy, y_val_encoded),
        callbacks=training_monitors,
        verbose=2
    )

    # 极低信噪比环境下的最终泛化性能评估
    _, final_acc = drsn_model.evaluate(x_test_noisy, y_test_encoded, verbose=0)
    print("\n" + "="*50)
    print("性能报告：在 -8dB SNR 极强噪声环境下，系统识别准确率为: {0:.2f}%".format(final_acc * 100))
    print("="*50)

# =============================================================================
# 程序入口
# =============================================================================

if __name__ == "__main__":
    # 默认数据源搜索路径
    DEFAULT_DATA_DIR = os.path.join(os.getcwd(), 'data_path')
    
    if not os.path.exists(DEFAULT_DATA_DIR):
        logging.warning("缺省路径未发现数据集: {0}".format(DEFAULT_DATA_DIR))
        user_input_dir = input("请手动指定包含 CWRU .mat 文件的目录路径: ").strip()
        if user_input_dir:
            DEFAULT_DATA_DIR = user_input_dir
        else:
            logging.error("未输入有效路径，程序终止运行。")
            sys.exit(0)

    # 执行主程序逻辑
    execute_diagnostic_workflow(DEFAULT_DATA_DIR, input_length=1024)

四、极强噪声环境下的鲁棒性验证

为了模拟真实的工业恶劣环境，在测试阶段向原始振动信号中人为注入了高斯白噪声，将信噪比降低至-8dB。