机械设备早期故障的冲击特征极易隐匿于非平稳的随机环境扰动之中，如何在复杂的背景噪声下实现故障信息的鲁棒提取，已成为制约智能监测技术工程化应用的关键难题。2020年，《IEEE Transactions on Industrial Informatics》发表了题为“Deep Residual Shrinkage Networks for Fault Diagnosis”的论文，提出了深度残差收缩网络（DRSN）。该架构通过在残差网络中集成可训练的软阈值收缩单元，实现了在特征学习过程中对噪声成分的自适应抑制，有助于提升模型在低信噪比环境下的诊断鲁棒性。

本研究完成了对DRSN框架的复现，并开展了相关实验。在测试阶段，通过引入-8 dB的高斯白噪声（噪声功率远高于信号功率）来评估模型的抗干扰能力。在凯斯西储大学（Case Western Reserve University，CWRU）轴承数据集上的实验结果表明，该模型在强噪声干扰下仍能保持较高的分类准确率。

深度残差收缩网络的核心创新在于将传统信号处理中的软阈值化（Soft Thresholding）算子嵌入深度残差网络。与ReLU等仅对负半轴进行截断的激活函数不同，软阈值函数对正负特征进行对称压缩，其数学表达为：y = sgn(x) * max(|x| - τ, 0)，其中τ为正阈值参数。

该机制对特征学习的贡献主要体现在以下两个方面：

（1）基于“噪声成分幅值通常较小”的假设，当特征绝对值低于阈值τ时，该分量被识别为噪声成分并被置为零，从而在特征层面实现非线性去噪。（2）对于幅值超过阈值的有效信息，函数在保留符号的同时，使特征值向零平滑压缩，在一定程度上减轻了硬截断导致的信号不连续问题，有助于提取更具判别性的故障冲击特征。

DRSN舍弃了人工预设固定阈值的做法，转而在每个残差收缩单元中嵌入轻量化子网络（Sub-network），利用注意力机制实现阈值的端到端自适应调节。

（1）全局统计特性提取：利用绝对值化后的全局平均池化（GAP），将输入特征图映射为能够反映各通道响应强度的向量。

（2）缩放因子映射：通过两层全连接层（FC）学习非线性权重系数α。该系数经 Sigmoid函数归一化至(0, 1)区间。

（3）自适应阈值生成：根据公式τ=α*average(|x|)计算最终阈值，确保阈值水平能够随输入特征分布的变化动态更新。

在此基础上，DRSN-CW（Channel-Wise）变体将该机制扩展至通道维度，为每个卷积通道配置独立的收缩阈值，实现了较为精细化的特征调控能力。

模型在结构上采用残差收缩模块（RSBU）逐层堆叠，卷积通道数由32逐步提升至128。随着层级加深，模型从底层的时域波形特征逐步抽象出具备高阶判别性的故障模式表示。

（1）正则化约束：引入全局L2权重衰减（系数为1e-4），抑制模型在训练过程中对随机噪声模式的过拟合。

（2）动态数据增强：复现代码集成了在线数据增强策略，包括随机循环平移（Cyclic Shift）以减轻相位影响，以及-8 dB至8 dB的动态信噪比混合训练，提升模型在极端环境下提取关键故障特征的能力。

本研究基于凯斯西储大学（CWRU）轴承公开数据集进行验证。将样本划分为10类故障状态（涵盖正常运行及三种不同损伤直径的内圈、外圈、滚动体故障），并按70%、15%、15%的比例构建训练集、验证集与测试集。

实验采用的故障分类如下表所示：

本文实现的DRSN代码遵循模块化设计的原则。实验结果显示，当测试阶段信噪比低至-8 dB时，模型的分类准确率仍稳定在90%以上。通过将软阈值先验算子与深度学习架构融合，该模型能够有效识别强背景干扰下的微弱故障类别，为强噪声背景下的机械故障诊断提供了一种有效的技术手段。

# =============================================================================
# 本程序基于 Zhao 等人在 IEEE Transactions on Industrial Informatics 上
# 提出的深度残差收缩网络（DRSN）方法，实现其在 CWRU 轴承数据集上的
# 一维振动信号故障诊断流程复现，包括数据预处理、软阈值残差单元构建、
# 网络训练及强噪声环境下的鲁棒性评估。
#
# 参考文献：
# 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.
# =============================================================================

import os
import sys
import logging
import numpy as np
import scipy.io as matlab_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 RuntimeConfigurator:
    """
    负责执行底层硬件加速策略及环境依赖性的预检
    """
    @staticmethod
    def configure_computation_backend():
        """
        配置深度学习计算后端：
        1. 屏蔽 TensorFlow 冗余的底层调试信息。
        2. 验证核心科学计算库的版本兼容性。
        3. 激活 GPU 显存的动态按需分配策略，规避预分配导致的资源锁死。
        """
        os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2'
        
        try:
            import sklearn
            import scipy
        except ImportError as exc:
            logging.critical("运行环境缺失核心科学计算库支持：{}".format(exc))
            sys.exit(1)

        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 显存动态增长。")
            except RuntimeError as err:
                logging.error("计算设备配置失败：{}".format(err))
        else:
            logging.info("未检测到硬件加速设备，系统降级至 CPU 计算模式。")

# 启动环境初始化
RuntimeConfigurator.configure_computation_backend()

# =============================================================================
# 信号处理与特征工程
# =============================================================================

class CWRULoader:
    """
    信号数据存储库：实现对 CWRU 原始加速度信号的读取、解析与样本重构
    """
    def __init__(self, dataset_root, input_length=1024):
        """
        :param dataset_root: .mat 原始文件的根目录
        :param input_length: 单次采样的时间步长
        """
        self.base_dir = os.path.abspath(dataset_root)
        self.input_length = input_length
        self.hop_length = input_length  # 默认步长与采样长度一致（无重叠）

    def _load_vibration_signal(self, file_path):
        """
        解析 MATLAB 格式数据并定位驱动端传感器的时序信号
        """
        try:
            data_dict = matlab_io.loadmat(file_path)
            for identifier in data_dict.keys():
                if 'DE_time' in identifier:
                    return data_dict[identifier].ravel()
        except Exception:
            return None
        return None

    def segment_signal(self, label_mapping):
        """
        构建训练/测试特征库。
        :param label_mapping: 映射字典，关联类别标签与文件名称。
        :return: (特征矩阵, 类别向量)
        """
        features, labels = [], []
        is_accessible = False
        
        for class_idx, name_list in label_mapping.items():
            for filename in name_list:
                full_path = os.path.join(self.base_dir, "{}.mat".format(filename))
                if not os.path.exists(full_path):
                    continue
                
                signal = self._load_vibration_signal(full_path)
                if signal is None:
                    continue
                
                is_accessible = True
                # 基于滑动窗口的特征平铺逻辑
                num_samples = (signal.size - self.input_length) // self.hop_length + 1
                for i in range(num_samples):
                    offset = i * self.hop_length
                    frame = signal[offset : offset + self.input_length]
                    features.append(frame)
                    labels.append(class_idx)
        
        if not is_accessible:
            raise IOError("在指定路径下未检索到符合规范的数据源：{}".format(self.base_dir))
            
        return np.array(features, dtype='float32'), np.array(labels, dtype='int32')

def add_gaussian_noise(x_batch, snr):
    """
    针对输入信号批次施加加性高斯白噪声 (AWGN)，用于评估模型的鲁棒性。
    计算公式：P_noise = P_signal / 10^(SNR/10)
    """
    x_batch = np.array(x_batch)
    random_engine = np.random.default_rng()
    
    # 支持固定 SNR 或 SNR 区间采样
    snr_val = snr if isinstance(snr, (int, float)) else \
                 random_engine.uniform(snr[0], snr[1])
    
    # 统计功率并生成对应的噪声张量
    mean_power = np.mean(np.square(x_batch), axis=1, keepdims=True)
    noise_variance = mean_power / (10 ** (snr_val / 10.0))
    noise = random_engine.normal(0, np.sqrt(noise_variance), x_batch.shape)
    
    return (x_batch + noise).astype('float32')

# =============================================================================
# 深度残差收缩网络 (DRSN) 架构实现
# =============================================================================

class SoftThresholding(layers.Layer):
    """
    非线性缩进层：实现深度残差收缩网络中的核心软阈值去噪功能。
    计算逻辑：y = sign(x) * ReLU(|x| - threshold)
    """
    def __init__(self, **kwargs):
        super(SoftThresholding, self).__init__(**kwargs)

    def call(self, inputs):
        """
        :param inputs: 包含 [原始特征, 自适应阈值向量] 的列表
        """
        x, threshold = inputs
        # 将阈值向量调整为可广播的维度 (Batch, 1, Channels)
        thres_expanded = tf.expand_dims(threshold, axis=1)
        return tf.math.sign(x) * tf.math.maximum(tf.math.abs(x) - thres_expanded, 0.0)

class RSBU_CW(layers.Layer):
    """
    深度残差收缩网络基本单元 (Residual Shrinkage Building Unit)。
    该模块通过集成子网络自适应计算通道阈值，消除信号中的冗余分量。
    """
    def __init__(self, filters, kernel_size, strides=1, **kwargs):
        super(RSBU_CW, self).__init__(**kwargs)
        self.filters = filters
        self.strides = strides
        self.kernel_size = kernel_size
        self.weight_decay = regularizers.l2(1e-4)

        self.shortcut_path = None
        
        # 特征映射分支
        self.norm_1 = layers.BatchNormalization()
        self.relu_1 = layers.Activation('relu')
        self.conv_1 = layers.Conv1D(filters, kernel_size, strides=strides, padding='same', 
                                   kernel_initializer='he_normal', kernel_regularizer=self.weight_decay)
        
        self.norm_2 = layers.BatchNormalization()
        self.relu_2 = layers.Activation('relu')
        self.conv_2 = layers.Conv1D(filters, kernel_size, strides=1, padding='same', 
                                   kernel_initializer='he_normal', kernel_regularizer=self.weight_decay)
        
        # 阈值学习子网络
        self.global_pool = layers.GlobalAveragePooling1D()
        self.dense_1 = layers.Dense(filters, kernel_initializer='he_normal')
        self.norm_dense = layers.BatchNormalization()
        self.relu_dense = layers.Activation('relu')
        self.dense_2 = layers.Dense(filters, activation='sigmoid')
        self.shrinkage_op = SoftThresholding()

    def build(self, input_shape):
        """
        根据输入输出特征图尺寸自动调整 Identity Mapping
        """
        if self.strides != 1 or input_shape[-1] != self.filters:
            self.shortcut_path = models.Sequential([
                layers.Conv1D(self.filters, 1, strides=self.strides, padding='same'),
            ])
        super(RSBU_CW, self).build(input_shape)

    def call(self, inputs):
        """
        残差前向传播逻辑
        """
        identity = inputs
        if self.shortcut_path:
            identity = self.shortcut_path(inputs)

        # 非线性特征转换
        net = self.norm_1(inputs)
        net = self.relu_1(net)
        net = self.conv_1(net)
        net = self.norm_2(net)
        net = self.relu_2(net)
        net = self.conv_2(net)

        # 阈值自动提取流
        abs_x = tf.abs(net)
        gap_val = self.global_pool(abs_x)
        
        alpha = self.dense_1(gap_val)
        alpha = self.norm_dense(alpha)
        alpha = self.relu_dense(alpha)
        alpha = self.dense_2(alpha)
        
        thres = tf.multiply(alpha, gap_val)
        
        # 执行软阈值收缩与残差融合
        x_denoised = self.shrinkage_op([net, thres])
        return layers.Add()([x_denoised, identity])

class DRSN_CW_Model(models.Model):
    """
    深度残差收缩网络完整分类器。
    堆叠多个 RSBU_CW 模块，通过多层抽象提取高阶判别特征。
    """
    def __init__(self, num_classes):
        super(DRSN_CW_Model, self).__init__(name="DRSN_Classifier_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, strides=2),
            RSBU_CW(32, 5, strides=1),
            RSBU_CW(64, 5, strides=2),
            RSBU_CW(64, 5, strides=1),
            RSBU_CW(128, 5, strides=2),
            RSBU_CW(128, 5, strides=1)
        ]
        
        # 分类
        self.post_bn = layers.BatchNormalization()
        self.post_relu = layers.Activation('relu')
        self.avg_pool = layers.GlobalAveragePooling1D()
        self.logits = layers.Dense(num_classes, activation='softmax')

    def call(self, inputs):
        """
        端到端前向推理
        """
        x = self.stem_conv(inputs)
        x = self.stem_bn(x)
        x = self.stem_relu(x)
        
        for stage in self.backbone:
            x = stage(x)
            
        x = self.post_bn(x)
        x = self.post_relu(x)
        x = self.avg_pool(x)
        return self.logits(x)

# =============================================================================
# 训练引擎与评估逻辑
# =============================================================================

def run_diagnosis_workflow(dataset_path, input_length=1024):
    """
    主控工作流：执行从原始信号加载到深度残差收缩网络训练的全过程。
    """
    
    # 定义 10 类故障诊断体系
    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']
    }
    
    loader = CWRULoader(dataset_root=dataset_path, input_length=input_length)
    
    try:
        x_raw, y_raw = loader.segment_signal(label_mapping)
    except Exception as failure:
        logging.error("数据流水线加载失败：{}".format(failure))
        return

    # 随机化数据集划分
    train_x, temp_x, train_y, temp_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(temp_x, temp_y, test_size=0.5, random_state=42)
    
    # 零均值标准化 (Z-score Normalization)
    data_mu, data_std = np.mean(train_x), np.std(train_x)
    
    def normalize_tensor(data):
        return ((data - data_mu) / data_std).reshape(-1, input_length, 1)

    train_x = normalize_tensor(train_x)
    val_x = normalize_tensor(val_x)
    test_x = normalize_tensor(test_x)
    
    num_classes = len(label_mapping)
    train_y_onehot = tf.keras.utils.to_categorical(train_y, num_classes).astype('float32')
    val_y_onehot = tf.keras.utils.to_categorical(val_y, num_classes).astype('float32')
    test_y_onehot = tf.keras.utils.to_categorical(test_y, num_classes).astype('float32')

    # 为验证与测试集引入 -8dB 的固定强背景噪声
    val_x_noisy = add_gaussian_noise(val_x, snr=-8)
    test_x_noisy = add_gaussian_noise(test_x, snr=-8)

    def dynamic_augmentation_strategy(features, labels):
        """
        在线数据增强：提升深度残差收缩网络在极端环境下的泛化能力。
        """
        rng = np.random.default_rng()
        aug_x = features.copy()
        batch_size, seq_len, _ = aug_x.shape

        # 变换 1：随机时域循环位移
        for i in range(batch_size):
            step = rng.integers(0, seq_len)
            aug_x[i, :, 0] = np.roll(aug_x[i, :, 0], step)

        # 变换 2：瞬态冲击噪声模拟
        if rng.random() > 0.9: 
            for i in range(batch_size):
                if rng.random() > 0.5: 
                    spike_num = rng.integers(1, 3) 
                    indices = rng.integers(0, seq_len, spike_num)
                    impact_mag = np.std(aug_x[i]) * rng.uniform(1.5, 2.5) 
                    aug_x[i, indices, 0] += impact_mag * rng.choice([-1, 1], size=spike_num)

        # 变换 3：混合动态信噪比噪声
        if rng.random() > 0.5: 
            aug_x = add_gaussian_noise(aug_x, snr=(-8, 8))

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

    def _tensor_signature_fix(feat, lab):
        feat.set_shape([None, input_length, 1])
        lab.set_shape([None, num_classes])
        return feat, lab

    # 构建高性能 tf.data 数据流
    train_ds = tf.data.Dataset.from_tensor_slices((train_x.astype('float32'), train_y_onehot))
    train_ds = train_ds.shuffle(train_x.shape[0]).batch(64)
    train_ds = train_ds.map(
        lambda x, y: tf.numpy_function(dynamic_augmentation_strategy, [x, y], [tf.float32, tf.float32]),
        num_parallel_calls=tf.data.AUTOTUNE
    ).map(_tensor_signature_fix).prefetch(tf.data.AUTOTUNE)

    # 编译深度残差收缩模型
    model_instance = DRSN_CW_Model(num_classes=num_classes)
    
    model_instance.compile(
        optimizer=tf.keras.optimizers.Adam(learning_rate=1e-3), 
        loss='categorical_crossentropy',
        metrics=['accuracy']
    )

    logging.info("诊断模型准备就绪。目标分类数：{}，窗口跨度：{}".format(num_classes, input_length))
    
    # 训练监控回调
    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)
    ]

    # 启动网络优化过程
    model_instance.fit(
        train_ds,
        epochs=100,
        validation_data=(val_x_noisy, val_y_onehot),
        callbacks=training_callbacks,
        verbose=2
    )

    # 最终性能验证（强噪声验证环境）
    eval_loss, eval_acc = model_instance.evaluate(test_x_noisy, test_y_onehot, verbose=0)
    print("\n" + "="*50)
    print("模型最终性能指标 (SNR = -8dB):")
    print("测试准确率: {:.2f}%".format(eval_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("缺省路径不可用：{}".format(DEFAULT_DATA_DIR))
        input_path = input("请手动指定 CWRU 数据集所在的目录路径: ").strip()
        if input_path:
            DEFAULT_DATA_DIR = input_path
        else:
            logging.error("路径为空，系统强制退出。")
            sys.exit(0)

    # 执行主程序
    run_diagnosis_workflow(DEFAULT_DATA_DIR, input_length=1024)

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.

https://ieeexplore.ieee.org/document/8850096

深度残差收缩网络的抗噪性能验证与Python工程实现

人工智能

算法

后端

深度残差收缩网络(DRSN)创新性地将软阈值化算子嵌入深度残差网络，通过自适应阈值学习机制实现强噪声环境下的机械故障诊断。该模型在CWRU轴承数据集测试中，即使面对-8dB的高斯白噪声干扰，仍保持90%以上的分类准确率，为工业设备智能监测提供了可靠解决方案。

数据预处理

数据存储

深度学习

数据流

4核4G3M云服务器 新用户低至38元/年！

对象存储COS新用户低至1元！


文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云AI代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

功能1上新10个字符

功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符。

功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符。

功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符

功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符

功能4上新

文章&问答评论现已支持表情

全新交互，全新视觉，新增快捷键、悬浮工具栏、高亮块等功能并同时优化现有功能，全面提升创作效率和体验

社区富文本编辑器全新改版！诚邀体验～ 

精选全网热门MCP server，让你的AI更好用 🚀

💥开发者 MCP广场重磅上线！

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

深度残差收缩网络的抗噪性能验证与Python工程实现

深度残差收缩网络的抗噪性能验证与Python工程实现

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐