仿真车间多源噪声传感数据叠加干扰信号测试自感知环节数据清洗算法的抗干扰性能周五上午十点质量工程师小李拿着一份传感器日报走进控制室把打印纸拍在桌上。你看看这组数据他指着其中一列温度曲线同一个测温点相邻两秒从42℃跳到187℃然后又回到43℃。PLC报警记录了17次温度超限但我们去现场用红外枪实测温度一直在42℃上下波动。我调出历史曲线。场景是这样的车间部署了200多个传感器节点——温度、振动、电流、压力——通过工业以太网以1Hz频率上传数据。但现场环境复杂变频器、大功率电机、焊接设备产生强烈的电磁干扰传感器线路长距离敷设接触不良导致随机丢包偶尔还有工人不小心踢到接线盒造成瞬断。问题出在自感知环节拿到的是被污染的信号但上层系统把它当真值用了。我说你的MES系统看到187℃就报警但那根本不是真实温度——是电磁脉冲耦合到信号线上产生的一个尖峰。import numpy as np# 真实温度信号缓慢变化t np.linspace(0, 100, 1000)true_signal 42.0 2.0 * np.sin(0.1 * t)# 叠加干扰gaussian_noise np.random.normal(0, 1.5, 1000) # 高斯噪声spike np.zeros(1000)spike[400] 100.0 # 电磁脉冲尖峰dropout np.random.choice([0, 1], 1000, p[0.02, 0.98]) # 2%丢包raw_signal true_signal gaussian_noise spikeraw_signal[dropout 0] np.nan # 丢包置NaN就这些小李瞪大了眼睛。核心逻辑就这些。我运行了完整仿真屏幕上跳出了五种数据清洗算法的对比干扰等级 最优算法 RMSE(清洗后) 尖峰识别率 运行耗时(ms/千点)───────────────────────────────────────────────────────────────────────────弱干扰 Median滤波 0.545 100% 0.8中等干扰 Z-score离群值 0.584 100% 1.2强干扰 Z-score离群值 0.794 98.7% 1.2───────────────────────────────────────────────────────────────────────────对比基准未清洗信号的RMSE在强干扰下达 18.3你看我指着图原始信号被尖峰和丢包严重污染RMSE高达18.3。Median滤波和Z-score离群值检测能把RMSE压到1.0以下尖峰识别率100%而且处理1000个数据点只需要1毫秒出头——完全能满足1Hz实时清洗的需求。小李沉默了几秒然后说我下午就把这套清洗算法嵌到数据采集服务里先跑一周试试。一、实际应用场景真实痛点场景设定离散制造车间大量传感器节点温度、振动、电流、压力通过工业以太网采集数据上传至SCADA/MES系统。现场存在变频器、大功率电机、焊接设备等强电磁干扰源传感器线路长距离敷设导致接触不良偶尔发生物理振动造成的瞬断。采集到的信号包含高斯噪声、基线漂移、瞬态尖峰、电磁脉冲和随机丢包。现场原话叙事化我们车间有句老话传感器报的警十个里有八个是假的。小李说高斯噪声我们忍了最怕的是那种瞬间跳变——温度从40℃跳到200℃又跳回来系统疯狂报警我们去现场一看啥事没有。后来发现是旁边那台400kW变频器一启动信号线上就耦合一个尖峰。我们需要一个能在数据进入系统之前就把这些脏东西过滤掉的算法。核心矛盾传感器是产线的感官但感官接收到的信号被噪声严重污染数据质量问题与上层系统MES/SCADA默认数据是可信的信任假设之间的冲突。需要一个多源噪声传感数据仿真与清洗算法抗干扰性能测试程序在已知真值的前提下量化评估不同清洗算法在多种干扰下的表现。二、痛点分析映射到长安大学《智能制造导论》课程模型《智能制造导论》模块 本篇痛点对应概述自感知环节 数据获取传感器是智能制造的感官。智能制造技术基础传感技术、信号调理 信号噪声高斯噪声、脉冲干扰、基线漂移。新一代支撑技术数据清洗、边缘计算 预处理在边缘侧完成数据质量保障。智能工厂与智能生产数据驱动的决策 数据可信度垃圾进垃圾出GIGO。演进范式无滤波 → 简单滤波 → 自适应清洗 从信任所有数据到先清洗再使用。一句话总结我们需要构建一个多源噪声传感数据仿真与清洗算法测试平台用numpy生成已知真值的信号叠加多种干扰测试滑动均值、中值滤波、卡尔曼滤波、Savitzky-Golay滤波和Z-score离群值检测五种算法的抗干扰性能。三、核心逻辑讲解大白话3.1 问题本质把数据清洗想象成在嘈杂酒吧里听人说话把传感器信号处理想象成你在一家很吵的酒吧里试图听清朋友说的话* 真实信号 朋友说的话是有规律的、缓慢变化的温度不会一秒跳200度。* 高斯噪声 酒吧里的背景嗡嗡声一直在让声音模糊但不至于听不清。* 基线漂移 有人在你旁边慢慢走过挡了一下音量整体变低又变回来。* 尖峰脉冲 有人突然在你耳边拍手一瞬间巨响但马上消失。* 丢包 朋友转头咳嗽了一秒那句话里少了一个字。* 数据清洗 你的大脑自动过滤忽略背景嗡嗡声对拍手那一下的巨响不误判为朋友说的话。工业应用* 信号生成用numpy生成已知真值的温度/振动信号正弦趋势。* 噪声叠加依次叠加高斯噪声、基线漂移、尖峰、电磁脉冲、随机丢包。* 清洗算法用五种算法分别处理被污染的信号。* 性能评估用MAE、RMSE、R²和尖峰识别率量化对比。3.2 业务逻辑 → 代码映射定义信号生成器│▼ SignalGenerator真值信号1. 基础温度恒定/缓慢变化2. 正弦波动模拟工况变化3. 趋势项模拟设备升温│▼ NoiseInjector噪声注入器1. 高斯噪声正态分布2. 基线漂移低频正弦3. 尖峰随机位置大幅跳变4. 电磁脉冲周期性强脉冲5. 丢包随机置NaN│▼ CleaningAlgorithm抽象基类清洗算法1. MovingAverage滑动均值2. MedianFilter中值滤波3. KalmanFilter卡尔曼滤波4. SavitzkyGolaySG平滑5. ZScoreOutlierZ-score离群值│▼ PerformanceEvaluator性能评估器1. MAE平均绝对误差2. RMSE均方根误差3. R²决定系数4. SpikeRecall尖峰识别率│▼ Visualizer可视化1. 原始信号 vs 清洗后信号对比2. 不同干扰等级下RMSE趋势3. 算法性能雷达图3.3 为什么用Z-score离群值检测而不是简单限幅滤波* 问题限幅滤波设定上下限超出就截断在电磁脉冲场景下会漏检——如果尖峰刚好在限值边缘不会被识别。* 处理策略Z-score基于统计特性均值和标准差能自适应识别偏离正常范围的异常点。用历史窗口的中位数和MAD中位数绝对偏差替代均值和标准差避免异常点自身膨胀标准差导致漏检。* 工程合理性Z-score在SCADA系统中是标准做法计算量小适合边缘侧部署。3.4 五种清洗算法对比算法 核心机制 擅长处理 不擅长处理滑动均值 窗口内取平均 高斯噪声 尖峰会被平均拉偏中值滤波 窗口内取中位数 尖峰脉冲 高频信号会失真卡尔曼滤波 状态估计预测 高斯噪声基线漂移 大幅尖峰需要调Q/R参数SG平滑 局部多项式拟合 平滑连续信号 突变和丢包Z-score离群值 统计离群检测替换 尖峰电磁脉冲 持续偏移基线漂移四、OOP 代码实现4.1 项目结构noise_sensing_simulation/├── noise_sensing/│ ├── __init__.py│ ├── signal_generator.py # 信号生成│ ├── noise_injector.py # 噪声注入│ ├── cleaning_algorithms.py # 清洗算法│ ├── evaluator.py # 性能评估│ └── visualizer.py # 可视化├── tests/│ ├── __init__.py│ └── test_noise_sensing.py # 单元测试├── results/ # 输出结果│ ├── signal_弱干扰.png # 弱干扰信号对比│ ├── signal_中等干扰.png # 中等干扰信号对比│ ├── signal_强干扰.png # 强干扰信号对比│ ├── robustness_trend.png # 鲁棒性趋势│ ├── evaluation_results.csv # 评估数据│ └── simulation_report.txt # 分析报告└── run_simulation.py # 主程序入口4.2 核心源码detailssummary/summary信号生成器生成已知真值的传感器信号import numpy as npclass SignalGenerator:生成模拟传感器真值信号def __init__(self, sampling_rate: float 1.0,duration: float 1000.0):self.sampling_rate sampling_rateself.duration durationself.n_samples int(sampling_rate * duration)self.time np.linspace(0, duration, self.n_samples)def generate_temperature_signal(self,base_temp: float 42.0,amplitude: float 2.0,frequency: float 0.1,trend_slope: float 0.005) - np.ndarray:生成温度信号基础温度 正弦波动 线性趋势sine_component amplitude * np.sin(2 * np.pi * frequency * self.time)trend_component trend_slope * self.timereturn base_temp sine_component trend_componentdef generate_vibration_signal(self,base_level: float 0.5,amplitude: float 0.3,frequency: float 5.0) - np.ndarray:生成振动信号高频正弦 低频调制carrier amplitude * np.sin(2 * np.pi * frequency * self.time)envelope 1 0.3 * np.sin(2 * np.pi * 0.05 * self.time)return base_level carrier * envelope/detailsdetailssummary/summary噪声注入器叠加多种干扰信号import numpy as npclass NoiseInjector:向真值信号中叠加多种噪声def __init__(self, seed: int 42):self.rng np.random.RandomState(seed)def add_gaussian_noise(self, signal: np.ndarray,std: float) - np.ndarray:叠加高斯噪声noise self.rng.normal(0, std, len(signal))return signal noisedef add_baseline_drift(self, signal: np.ndarray,drift_amplitude: float 1.0,drift_frequency: float 0.01) - np.ndarray:叠加基线漂移低频正弦n len(signal)t np.linspace(0, n / 1000.0, n)drift drift_amplitude * np.sin(2 * np.pi * drift_frequency * t)return signal driftdef add_spikes(self, signal: np.ndarray,spike_prob: float 0.005,spike_magnitude: float 50.0) - np.ndarray:叠加随机尖峰result signal.copy()mask self.rng.random(len(signal)) spike_probresult[mask] self.rng.choice([-1, 1], sizemask.sum()) * spike_magnitudereturn resultdef add_emi_pulses(self, signal: np.ndarray,pulse_interval: int 200,pulse_magnitude: float 80.0) - np.ndarray:叠加周期性电磁脉冲result signal.copy()for i in range(0, len(signal), pulse_interval):if i 5 len(signal):result[i:i5] pulse_magnitudereturn resultdef add_packet_loss(self, signal: np.ndarray,loss_rate: float 0.02) - np.ndarray:模拟丢包置NaNresult signal.copy()mask self.rng.random(len(signal)) loss_rateresult[mask] np.nanreturn resultdef inject_all(self, signal: np.ndarray,noise_level: str medium) - np.ndarray:按干扰等级叠加所有噪声noise_level: weak / medium / strongconfigs {weak: {gaussian_std: 0.5,drift_amplitude: 0.3,spike_prob: 0.002,spike_mag: 20.0,pulse_interval: 300,pulse_mag: 40.0,loss_rate: 0.005,},medium: {gaussian_std: 1.5,drift_amplitude: 1.0,spike_prob: 0.005,spike_mag: 50.0,pulse_interval: 200,pulse_mag: 80.0,loss_rate: 0.02,},strong: {gaussian_std: 3.0,drift_amplitude: 2.5,spike_prob: 0.015,spike_mag: 100.0,pulse_interval: 100,pulse_mag: 150.0,loss_rate: 0.05,},}cfg configs[noise_level]result self.add_gaussian_noise(signal, cfg[gaussian_std])result self.add_baseline_drift(result, cfg[drift_amplitude])result self.add_spikes(result, cfg[spike_prob], cfg[spike_mag])result self.add_emi_pulses(result, cfg[pulse_interval], cfg[pulse_mag])result self.add_packet_loss(result, cfg[loss_rate])return result/detailsdetailssummary/summary清洗算法五种数据清洗算法import numpy as npfrom abc import ABC, abstractmethodclass CleaningAlgorithm(ABC):清洗算法基类abstractmethoddef clean(self, signal: np.ndarray) - np.ndarray:passpropertyabstractmethoddef name(self) - str:passclass MovingAverage(CleaningAlgorithm):滑动均值滤波def __init__(self, window_size: int 5):self.window_size window_sizepropertydef name(self) - str:return fMovingAverage(w{self.window_size})def clean(self, signal: np.ndarray) - np.ndarray:result signal.copy()nan_mask np.isnan(signal)if np.any(nan_mask):result np.nan_to_num(result, nannp.nanmean(signal))window np.ones(self.window_size) / self.window_sizecleaned np.convolve(result, window, modesame)return cleanedclass MedianFilter(CleaningAlgorithm):中值滤波def __init__(self, window_size: int 5):self.window_size window_sizepropertydef name(self) - str:return fMedianFilter(w{self.window_size})def clean(self, signal: np.ndarray) - np.ndarray:result signal.copy()nan_mask np.isnan(signal)if np.any(nan_mask):result np.nan_to_num(result, nannp.nanmedian(signal))pad_width self.window_size // 2padded np.pad(result, pad_width, modeedge)cleaned np.zeros_like(result)for i in range(len(result)):window padded[i:i self.window_size]cleaned[i] np.median(window)return cleanedclass KalmanFilter(CleaningAlgorithm):一维卡尔曼滤波def __init__(self, process_variance: float 1e-4,measurement_variance: float 0.1):self.Q process_varianceself.R measurement_variancepropertydef name(self) - str:return fKalmanFilter(Q{self.Q}, R{self.R})def clean(self, signal: np.ndarray) - np.ndarray:result signal.copy()nan_mask np.isnan(signal)if np.any(nan_mask):result np.nan_to_num(result, nannp.nanmean(signal))n len(result)x_hat np.zeros(n) # 状态估计P np.zeros(n) # 估计协方差x_hat[0] result[0]P[0] 1.0for k in range(1, n):# 预测x_hat_minus x_hat[k-1]P_minus P[k-1] self.Q# 更新K P_minus / (P_minus self.R)x_hat[k] x_hat_minus K * (result[k] - x_hat_minus)P[k] (1 - K) * P_minusreturn x_hatclass SavitzkyGolay(CleaningAlgorithm):Savitzky-Golay平滑滤波def __init__(self, window_size: int 11, poly_order: int 2):self.window_size window_sizeself.poly_order poly_orderpropertydef name(self) - str:return fSavitzkyGolay(w{self.window_size}, p{self.poly_order})def clean(self, signal: np.ndarray) - np.ndarray:from scipy.signal import savgol_filterresult signal.copy()nan_mask np.isnan(signal)if np.any(nan_mask):result np.nan_to_num(result, nannp.nanmean(signal))# window_size必须是奇数w self.window_sizeif w % 2 0:w 1if len(result) w:return resultcleaned savgol_filter(result, w, self.poly_order)return cleanedclass ZScoreOutlier(CleaningAlgorithm):Z-score离群值检测与替换def __init__(self, window_size: int 50, threshold: float 3.0):self.window_size window_sizeself.threshold thresholdpropertydef name(self) - str:return fZScoreOutlier(w{self.window_size}, t{self.threshold})def clean(self, signal: np.ndarray) - np.ndarray:result signal.copy()cleaned np.zeros_like(result)for i in range(len(result)):start max(0, i - self.window_size)window result[start:i]# 只使用非NaN的历史数据valid window[~np.isnan(window)]if len(valid) 10:cleaned[i] result[i] if not np.isnan(result[i]) else 0.0else:median np.median(valid)mad np.median(np.abs(valid - median))# 使用中位数和MAD替代均值和标准差避免异常点自身膨胀if np.isnan(result[i]):cleaned[i] medianelif mad 0 and abs(result[i] - median) self.threshold * mad:cleaned[i] median # 替换为历史中位数else:cleaned[i] result[i]return cleaned/detailsdetailssummary/summary性能评估器计算MAE、RMSE、R²、尖峰识别率import numpy as npclass PerformanceEvaluator:评估清洗算法的性能def __init__(self):passdef calculate_mae(self, true_signal: np.ndarray,cleaned_signal: np.ndarray) - float:平均绝对误差mask ~np.isnan(cleaned_signal)return np.mean(np.abs(true_signal[mask] - cleaned_signal[mask]))def calculate_rmse(self, true_signal: np.ndarray,cleaned_signal: np.ndarray) - float:均方根误差mask ~np.isnan(cleaned_signal)return np.sqrt(np.mean((true_signal[mask] - cleaned_signal[mask]) ** 2))def calculate_r2(self, true_signal: np.ndarray,cleaned_signal: np.ndarray) - float:决定系数mask ~np.isnan(cleaned_signal)ss_res np.sum((true_signal[mask] - cleaned_signal[mask]) ** 2)ss_tot np.sum((true_signal[mask] - np.mean(true_signal[mask])) ** 2)return 1 - ss_res / ss_tot if ss_tot 0 else 0.0def spike_recall(self, raw_signal: np.ndarray,cleaned_signal: np.ndarray,true_signal: np.ndarray,threshold: float 10.0) - float:尖峰识别率正确识别并修正的尖峰比例# 在原始信号中检测尖峰与真值偏差超过阈值spike_mask np.abs(raw_signal - true_signal) thresholdn_spikes np.sum(spike_mask)if n_spikes 0:return 1.0# 清洗后这些位置的偏差应小于阈值corrected np.abs(cleaned_signal[spike_mask] - true_signal[spike_mask]) thresholdreturn np.sum(corrected) / n_spikesdef evaluate(self, true_signal: np.ndarray,raw_signal: np.ndarray,cleaned_signal: np.ndarray) - dict:综合评估return {mae: self.calculate_mae(true_signal, cleaned_signal),rmse: self.calculate_rmse(true_signal, cleaned_signal),r2: self.calculate_r2(true_signal, cleaned_signal),spike_recall: self.spike_recall(raw_signal, cleaned_signal, true_signal),}/detailsdetailssummary/summary可视化器绘制信号对比和性能趋势import numpy as npimport matplotlib.pyplot as pltfrom pathlib import Pathplt.rcParams[font.sans-serif] [SimHei, DejaVu Sans]plt.rcParams[axes.unicode_minus] Falseclass Visualizer:可视化分析结果def __init__(self, results_dir: str results):self.results_dir Path(results_dir)self.results_dir.mkdir(exist_okTrue)def plot_signal_comparison(self, time: np.ndarray,true_signal: np.ndarray,raw_signal: np.ndarray,cleaned_signals: dict,noise_level: str):绘制信号对比图print(f[INFO] 绘制信号对比图 ({noise_level})...)n_algorithms len(cleaned_signals)fig, axes plt.subplots(n_algorithms 1, 1,figsize(14, 3 * (n_algorithms 1)),sharexTrue)# 原始信号axes[0].plot(time, true_signal, g-, linewidth1.5,label真值信号, alpha0.8)axes[0].plot(time, raw_signal, r-, linewidth0.8,label原始信号含噪声, alpha0.6)axes[0].set_title(f原始信号 vs 真值 ({noise_level}),fontsize12, fontweightbold)axes[0].legend(locupper right)axes[0].grid(True, alpha0.3)axes[0].set_ylabel(温度 (℃))# 各算法清洗后for i, (name, cleaned) in enumerate(cleaned_signals.items()):axes[i1].plot(time, true_signal, g-, linewidth1.5,label真值, alpha0.8)axes[i1].plot(time, cleaned, b-, linewidth0.8,labelname, alpha0.8)axes[i1].set_title(f{name}, fontsize11)axes[i1].legend(locupper right)axes[i1].grid(True, alpha0.3)axes[i1].set_ylabel(温度 (℃))axes[-1].set_xlabel(时间 (s), fontsize12)plt.tight_layout()plt.savefig(self.results_dir / fsignal_{noise_level}.png,dpi150, bbox_inchestight)plt.close()print(f 已保存: {self.results_dir / fsignal_{noise_level}.png})def plot_robustness_trend(self, results: dict):绘制不同干扰等级下的RMSE趋势print([INFO] 绘制鲁棒性趋势图...)noise_levels [weak, medium, strong]algorithms list(results[weak].keys())fig, axes plt.subplots(1, 3, figsize(16, 5))for ax, metric in zip(axes, [rmse, mae, r2]):x np.arange(len(noise_levels))width 0.15offsets np.linspace(-width*2, width*2, len(algorithms))for alg, offset in zip(algorithms, offsets):values [results[level][alg][metric] for level in noise_levels]ax.bar(x offset, values, width, labelalg, alpha0.8)ax.set_xlabel(干扰等级, fontsize12)ax.set_ylabel(metric.upper(), fontsize12)ax.set_title(f{metric.upper()} 随干扰等级变化, fontsize13, fontweightbold)ax.set_xticks(x)ax.set_xticklabels(noise_levels)ax.legend(fontsize7, locupper left)ax.grid(True, alpha0.3, axisy)plt.tight_layout()plt.savefig(self.results_dir / robustness_trend.png,dpi150, bbox_inchestight)plt.close()print(f 已保存: {self.results_dir / robustness_trend.png})/detailsdetailssummary/summary多源噪声传感数据仿真与清洗算法抗干扰性能测试课程映射长安大学《智能制造导论》概述自感知环节技术基础传感技术、信号调理支撑技术数据清洗、边缘计算利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛