传统传感器数据直接采集，程序加入动态滤波算法，剔除随机干扰，测数比传统准30%
智能传感器动态滤波与抗干扰采集系统
一、实际应用场景描述
在智能仪器课程设计的高精度环境监测站项目中，我们需要采集温度、压力、振动等关键参数。该系统部署在工业现场，环境极其复杂：电机启停产生的电磁干扰、空气压缩机造成的气压波动、变频设备产生的高频噪声等。
典型采集场景：
- 热电偶测温：受变频器电磁干扰，读数跳动±5℃
- 压力变送器：液压泵脉冲导致压力值周期性抖动
- 加速度传感器：机械振动叠加电气噪声
- 数据采集卡：工频50Hz及其谐波干扰
二、引入痛点
1. 传统采集的致命缺陷
传统方式：直接读取ADC值
↓
原始数据：[23.1, 28.7, 22.3, 31.2, 24.8, 29.9, 23.5]
↓
直接使用：平均值 = 26.5°C
↓
实际真实值：25.0°C ± 0.2°C
↓
误差高达：6% (完全不可用!)
2. 现有方案的不足
方案 问题 后果
硬件RC滤波 截止频率固定，无法适应动态变化 响应慢/滤波不充分
算术平均 对所有异常值同等对待 异常值严重拉偏结果
移动平均 窗口大小难确定 要么滞后要么噪声大
卡尔曼滤波 参数整定困难，计算量大 嵌入式难以部署
3. 核心需求
- 🎯 精度提升30%：从±5%提升到±0.5%
- ⚡ 实时性：单样本处理时间<1ms
- 🔧 自适应：自动识别干扰类型并调整
- 📊 可信度评估：输出数据质量指标
三、核心逻辑讲解
动态滤波算法架构
┌─────────────────────────────────────────────────────────────┐
│ 动态滤波算法流程图 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 原始ADC采样 ──► 预处理层 ──► 多级滤波引擎 ──► 后处理层 │
│ │ │ │ │ │
│ ▼ ▼ ▼ ▼ │
│ [0-4095] 异常值检测 自适应加权 置信度评估 │
│ │ 移动平均 数据质量评分 │
│ ▼ 卡尔曼融合 趋势预测 │
│ 中值替代 Hampel滤波 输出限幅 │
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 滤波引擎内部决策逻辑 │ │
│ ├─────────────────────────────────────────────────────┤ │
│ │ │ │
│ │ IF 当前噪声水平 > 阈值: │ │
│ │ 启用 Hampel滤波 + 扩窗移动平均 │ │
│ │ ELSE IF 检测到阶跃变化: │ │
│ │ 减小窗口 + 卡尔曼增益增大 │ │
│ │ ELSE IF 稳态测量: │ │
│ │ 标准加权移动平均 + 轻微阻尼 │ │
│ │ ELSE: │ │
│ │ 自适应参数调整 │ │
│ │ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘
核心算法：自适应加权滤波
算法思想：不是所有数据点都平等！
传统平均： y = (x₁ + x₂ + x₃ + x₄ + x₅) / 5
权重均为 0.2
动态加权： wᵢ = exp(-|xᵢ - median|² / σ²)
权重根据偏离程度动态调整
优势：
✓ 远离群体的异常值获得极低权重
✓ 靠近群体的正常值获得高权重
✓ 自动适应噪声水平变化
四、代码模块化实现
项目结构
smart_sensor_filter/
├── main.py # 主程序入口
├── config.py # 系统配置
├── data_acquisition.py # 数据采集模块
├── dynamic_filters.py # 动态滤波算法
├── noise_detector.py # 噪声检测器
├── quality_assessor.py # 数据质量评估
├── statistics.py # 统计工具函数
└── README.md # 使用说明
1. config.py - 系统配置
"""
智能传感器动态滤波系统 - 配置文件
针对智能仪器课程的工业现场数据采集场景优化
"""
from dataclasses import dataclass, field
from typing import List, Dict
import numpy as np
@dataclass
class FilterConfig:
"""滤波器配置参数"""
# ===== 基础采集参数 =====
SAMPLE_RATE: int = 100 # 采样率 Hz
BUFFER_SIZE: int = 50 # 数据缓冲区大小
ADC_RESOLUTION: int = 12 # ADC位数
ADC_VREF: float = 3.3 # 参考电压
# ===== 移动平均滤波 =====
MOVING_AVG_WINDOW: int = 10 # 基础窗口大小
MIN_WINDOW: int = 3 # 最小窗口（快速响应）
MAX_WINDOW: int = 30 # 最大窗口（强滤波）
# ===== Hampel滤波参数 =====
HAMPEL_WINDOW: int = 7 # Hampel窗口半宽
HAMPEL_THRESHOLD: float = 3.0 # 异常值判定阈值(σ倍数)
# ===== 卡尔曼滤波参数 =====
KALMAN_Q: float = 0.01 # 过程噪声协方差
KALMAN_R: float = 0.5 # 测量噪声协方差
KALMAN_P: float = 1.0 # 估计误差协方差初值
# ===== 自适应加权参数 =====
WEIGHT_SIGMA: float = 1.0 # 权重计算高斯核宽度
OUTLIER_THRESHOLD: float = 0.2 # 异常值权重阈值
# ===== 噪声检测参数 =====
NOISE_BASELINE: float = 0.1 # 基准噪声水平
NOISE_JUMP_FACTOR: float = 3.0 # 噪声突增判定因子
# ===== 质量评估参数 =====
QUALITY_EXCELLENT: float = 0.9 # 优秀质量阈值
QUALITY_GOOD: float = 0.7 # 良好质量阈值
QUALITY_ACCEPTABLE: float = 0.5 # 可接受阈值
# ===== 传感器校准参数 =====
CALIBRATION_SLOPE: float = 1.0 # 斜率校准
CALIBRATION_OFFSET: float = 0.0 # 偏移校准
# ===== 模拟器参数 =====
SIMULATE_NOISE: bool = True # 启用噪声模拟
NOISE_LEVEL: float = 0.5 # 噪声强度(0-1)
GLITCH_PROBABILITY: float = 0.05 # 尖峰干扰概率
# 全局配置实例
CONFIG = FilterConfig()
# 传感器类型配置
SENSOR_CONFIGS = {
"temperature": {
"unit": "°C",
"range": (-40, 125),
"noise_profile": "gaussian",
"typical_drift": 0.1 # °C/s
},
"pressure": {
"unit": "kPa",
"range": (0, 1000),
"noise_profile": "periodic",
"typical_drift": 1.0
},
"vibration": {
"unit": "g",
"range": (-10, 10),
"noise_profile": "impulsive",
"typical_drift": 0.5
},
"current": {
"unit": "A",
"range": (0, 20),
"noise_profile": "electromagnetic",
"typical_drift": 0.2
}
}
2. statistics.py - 统计工具函数
"""
统计工具模块
提供滤波算法所需的各种统计计算函数
"""
import numpy as np
from typing import List, Tuple, Optional
from collections import deque
def moving_median(data: List[float], window: int) -> float:
"""
计算移动中位数
中位数对异常值不敏感，适合作为稳健的中心趋势估计
Args:
data: 数据序列
window: 窗口大小
Returns:
当前窗口的中位数
"""
if len(data) == 0:
return 0.0
# 取最近的window个数据
recent_data = list(data[-window:])
return float(np.median(recent_data))
def moving_std(data: List[float], window: int) -> float:
"""
计算移动标准差（噪声水平估计）
Args:
data: 数据序列
window: 窗口大小
Returns:
当前窗口的标准差
"""
if len(data) < 2:
return 0.0
recent_data = np.array(data[-window:])
return float(np.std(recent_data, ddof=1)) # 样本标准差
def weighted_average(values: List[float], weights: List[float]) -> float:
"""
计算加权平均值
Args:
values: 数值列表
weights: 对应权重列表
Returns:
加权平均值
"""
if not values or not weights:
return 0.0
# 确保长度一致
n = min(len(values), len(weights))
values = values[-n:]
weights = weights[-n:]
# 归一化权重
weight_sum = sum(weights)
if weight_sum == 0:
return sum(values) / len(values)
normalized_weights = [w / weight_sum for w in weights]
# 计算加权平均
return sum(v * w for v, w in zip(values, normalized_weights))
def adaptive_weights(values: List[float], sigma: float) -> List[float]:
"""
计算自适应权重（基于与中位数的距离）
权重计算公式：
w_i = exp(-(x_i - median)² / (2 * sigma²))
Args:
values: 数值列表
sigma: 高斯核宽度参数
Returns:
权重列表
"""
if not values:
return []
median_val = np.median(values)
if sigma <= 0:
sigma = 1.0
# 计算权重
weights = []
for v in values:
distance = abs(v - median_val)
weight = np.exp(-(distance ** 2) / (2 * sigma ** 2))
weights.append(float(weight))
return weights
def hampel_filter(values: List[float], window: int, threshold: float) -> Tuple[float, bool]:
"""
Hampel滤波器 - 检测和剔除异常值
原理：
1. 计算窗口中位数和中位数绝对偏差(MAD)
2. 若 |x - median| > threshold * MAD，则判定为异常值
3. 异常值用中位数替代
Args:
values: 数据序列
window: 窗口半宽（总窗口=2*window+1）
threshold: 异常值阈值（MAD倍数）
Returns:
(滤波后的值, 是否被修正)
"""
if len(values) < window * 2 + 1:
return values[-1], False
# 取当前窗口数据
idx = -1 # 当前索引（最后一个）
window_data = values[-(window * 2 + 1):]
# 计算中位数和MAD
median_val = np.median(window_data)
mad = np.median(np.abs(window_data - median_val))
# 防止MAD为0
if mad < 1e-6:
mad = 1e-6
# 当前值
current_val = values[idx]
# 检测异常值
deviation = abs(current_val - median_val)
is_outlier = deviation > threshold * mad
if is_outlier:
return median_val, True
else:
return current_val, False
def kalman_predict(x_est: float, p_est: float, q: float) -> Tuple[float, float]:
"""
卡尔曼滤波预测步骤
Args:
x_est: 上一时刻状态估计
p_est: 上一时刻估计误差协方差
q: 过程噪声协方差
Returns:
(预测状态, 预测误差协方差)
"""
# 状态预测（假设常值模型）
x_pred = x_est
p_pred = p_est + q
return x_pred, p_pred
def kalman_update(x_pred: float, p_pred: float, z: float, r: float) -> Tuple[float, float, float]:
"""
卡尔曼滤波更新步骤
Args:
x_pred: 预测状态
p_pred: 预测误差协方差
z: 测量值
r: 测量噪声协方差
Returns:
(更新后状态, 更新后误差协方差, 卡尔曼增益)
"""
# 计算卡尔曼增益
k = p_pred / (p_pred + r)
# 更新状态估计
x_est = x_pred + k * (z - x_pred)
# 更新误差协方差
p_est = (1 - k) * p_pred
return x_est, p_est, k
class CircularBuffer:
"""
环形缓冲区
高效存储最近N个数据点
"""
def __init__(self, capacity: int):
self.capacity = capacity
self.buffer = deque(maxlen=capacity)
def append(self, value: float) -> None:
"""添加新值"""
self.buffer.append(value)
def extend(self, values: List[float]) -> None:
"""批量添加"""
self.buffer.extend(values)
def get_all(self) -> List[float]:
"""获取所有数据"""
return list(self.buffer)
def get_last(self, n: int) -> List[float]:
"""获取最后n个数据"""
return list(self.buffer)[-n:]
def __len__(self) -> int:
return len(self.buffer)
def clear(self) -> None:
"""清空缓冲区"""
self.buffer.clear()
@property
def is_full(self) -> bool:
"""缓冲区是否已满"""
return len(self.buffer) >= self.capacity
3. noise_detector.py - 噪声检测器
"""
噪声检测器模块
实时监测噪声水平并分类干扰类型
"""
import numpy as np
from typing import Tuple, Dict
from collections import deque
from config import CONFIG
from statistics import moving_std, moving_median
class NoiseLevel(Enum):
"""噪声等级枚举"""
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
EXTREME = "extreme"
class NoiseDetector:
"""
噪声检测器
功能：
1. 实时估计噪声水平
2. 识别干扰类型（高斯噪声、脉冲干扰、周期性干扰）
3. 为滤波器参数自适应调整提供依据
"""
def __init__(self, window: int = 20):
self.window = window
self.noise_history = deque(maxlen=100)
self.current_level = NoiseLevel.LOW
self.interference_type = "unknown"
# 噪声统计
self.baseline_noise = CONFIG.NOISE_BASELINE
self.noise_trend = 0.0
def analyze(self, data: deque) -> Dict[str, any]:
"""
分析当前噪声特性
Args:
data: 数据缓冲区
Returns:
噪声分析结果字典
"""
if len(data) < self.window:
return self._default_result()
# 1. 计算当前噪声水平（标准差）
recent_data = list(data)[-self.window:]
current_noise = moving_std(recent_data, self.window)
# 更新历史
self.noise_history.append(current_noise)
# 2. 计算噪声趋势
self.noise_trend = self._calculate_trend()
# 3. 确定噪声等级
self.current_level = self._determine_level(current_noise)
# 4. 识别干扰类型
self.interference_type = self._identify_interference(recent_data)
return {
"noise_level": current_noise,
"noise_class": self.current_level.value,
"interference_type": self.interference_type,
"trend": self.noise_trend,
"baseline_ratio": current_noise / self.baseline_noise,
"is_spike": self._detect_spike(recent_data)
}
def _calculate_trend(self) -> float:
"""计算噪声变化趋势"""
if len(self.noise_history) < 5:
return 0.0
history = list(self.noise_history)[-10:]
# 简单线性回归斜率
x = np.arange(len(history))
slope = np.polyfit(x, history, 1)[0]
return slope
def _determine_level(self, noise: float) -> NoiseLevel:
"""确定噪声等级"""
ratio = noise / self.baseline_noise
if ratio < 1.5:
return NoiseLevel.LOW
elif ratio < 3.0:
return NoiseLevel.MEDIUM
elif ratio < 6.0:
return NoiseLevel.HIGH
else:
return NoiseLevel.EXTREME
def _identify_interference(self, data: List[float]) -> str:
"""
识别干扰类型
特征分析：
- 高斯噪声：正态分布，无显著尖峰
- 脉冲干扰：偶发大幅跳变
- 周期性干扰：FFT有明显峰值
- 阶跃干扰：均值突然偏移
"""
data_array = np.array(data)
# 1. 检查尖峰（脉冲干扰）
median = np.median(data_array)
mad = np.median(np.abs(data_array - median))
spike_count = np.sum(np.abs(data_array - median) > 5 * mad)
spike_ratio = spike_count / len(data_array)
if spike_ratio > 0.1:
return "impulsive"
# 2. 检查周期性（简化版）
if len(data_array) >= 16:
# 计算差分
diff = np.diff(data_array)
# 检查周期性强相关
autocorr = np.correlate(diff, diff, mode='full')
autocorr = autocorr[len(autocorr)//2:]
if autocorr[1] > 0.5 * autocorr[0]:
return "periodic"
# 3. 检查阶跃
first_half = np.mean(data_array[:len(data_array)//2])
second_half = np.mean(data_array[len(data_array)//2:])
step_size = abs(second_half - first_half) / np.std(data_array)
if step_size > 3:
return "step"
return "gaussian"
def _detect_spike(self, data: List[float]) -> bool:
"""检测是否存在尖峰干扰"""
if len(data) < 3:
return False
# 比较最新值与历史
latest = data[-1]
historical = data[-5:-1]
mean_hist = np.mean(historical)
std_hist = np.std(historical)
if std_hist < 1e-6:
return False
z_score = abs(latest - mean_hist) / std_hist
return z_score > 4.0
def _default_result(self) -> Dict[str, any]:
"""默认返回结果"""
return {
"noise_level": 0.0,
"noise_class": NoiseLevel.LOW.value,
"interference_type": "unknown",
"trend": 0.0,
"baseline_ratio": 1.0,
"is_spike": False
}
def get_optimal_window(self) -> int:
"""
根据噪声水平推荐最优滤波窗口大小
噪声越大，窗口越大（更强的平滑）
"""
window_map = {
NoiseLevel.LOW: CONFIG.MIN_WINDOW,
NoiseLevel.MEDIUM: CONFIG.MOVING_AVG_WINDOW,
NoiseLevel.HIGH: CONFIG.MAX_WINDOW,
NoiseLevel.EXTREME: CONFIG.MAX_WINDOW + 10
}
return window_map[self.current_level]
def get_kalman_r(self) -> float:
"""
根据噪声水平调整卡尔曼测量噪声R
噪声越大，R越大，信任模型更多
"""
r_map = {
NoiseLevel.LOW: 0.1,
NoiseLevel.MEDIUM: 0.5,
NoiseLevel.HIGH: 2.0,
NoiseLevel.EXTREME: 5.0
}
return r_map[self.current_level]
4. dynamic_filters.py - 动态滤波算法核心
"""
动态滤波算法模块
实现多种自适应滤波器的组合与切换
"""
import numpy as np
from typing import Tuple, Optional, Dict
from enum import Enum, auto
from config import CONFIG
from statistics import (
moving_median,
weighted_average,
adaptive_weights,
hampel_filter,
kalman_predict,
kalman_update
)
from noise_detector import NoiseDetector
class FilterMode(Enum):
"""滤波模式枚举"""
FAST_RESPONSE = auto() # 快速响应（阶跃变化）
NORMAL = auto() # 正常模式
HEAVY_FILTER = auto() # 强滤波（高噪声）
TRANSIENT = auto() # 瞬态处理
class DynamicFilterBank:
"""
动态滤波器组
核心创新：根据实时噪声分析，动态组合多种滤波算法
算法组合策略：
┌─────────────────────────────────────────────────────┐
│ 噪声等级 │ 主导滤波器 │ 辅助处理 │ 窗口 │
├─────────────────────────────────────────────────────┤
│ 低噪声 │ 卡尔曼滤波 │ 轻微加权 │ 小 │
│ 中等噪声 │ 自适应加权 │ Hampel │ 中 │
│ 高噪声 │ 移动中值 │ 扩窗平均 │ 大 │
│ 瞬态尖峰 │ Hampel │ 冻结更新 │ 自适应 │
└─────────────────────────────────────────────────────┘
"""
def __init__(self, sensor_type: str = "generic"):
self.sensor_type = sensor_type
self.noise_detector = NoiseDetector()
# 卡尔曼滤波器状态
self.kalman_x = 0.0 # 状态估计
self.kalman_p = CONFIG.KALMAN_P # 误差协方差
# 滤波历史
self.filtered_history = []
self.raw_history = []
# 当前模式
self.current_mode = FilterMode.NORMAL
self.mode_duration = 0
# 性能指标
self.stats = {
"total_samples": 0,
"outliers_rejected": 0,
"mode_switches": 0,
"processing_time_ms": 0.0
}
print(f"[DynamicFilter] 初始化完成，传感器类型: {sensor_type}")
def process(self, raw_value: float, timestamp: float = None) -> Tuple[float, Dict]:
"""
处理单个采样点
这是核心处理函数，所有滤波逻辑在此汇聚
Args:
raw_value: 原始采样值
timestamp: 时间戳（可选）
Returns:
(滤波后的值, 处理元数据)
"""
import time
start_time = time.time()
# 1. 记录原始数据
self.raw_history.append(raw_value)
if len(self.raw_history) > CONFIG.BUFFER_SIZE:
self.raw_history.pop(0)
# 2. 噪声分析
noise_info = self.noise_detector.analyze(deque(self.raw_history))
# 3. 确定滤波模式
self._update_filter_mode(noise_info)
# 4. 执行滤波流水线
filtered_value, metadata = self._filter_pipeline(raw_value, noise_info)
# 5. 更新卡尔曼状态
self._update_kalman(filtered_value, noise_info)
# 6. 记录结果
self.filtered_history.append(filtered_value)
if len(self.filtered_history) > CONFIG.BUFFER_SIZE:
self.filtered_history.pop(0)
# 7. 更新统计
processing_time = (time.time() - start_time) * 1000
self.stats["total_samples"] += 1
self.stats["processing_time_ms"] += processing_time
# 构建返回元数据
result_metadata = {
"noise_info": noise_info,
"filter_mode": self.current_mode.name,
"processing_time_ms": processing_time,
"kalman_gain": metadata.get("kalman_gain", 0),
"outlier_corrected": metadata.get("outlier_corrected", False)
}
return filtered_value, result_metadata
def _update_filter_mode(self, noise_info: Dict) -> None:
"""根据噪声信息更新滤波模式"""
old_mode = self.current_mode
# 瞬态尖峰优先处理
if noise_info["is_spike"]:
self.current_mode = FilterMode.TRANSIENT
self.mode_duration = 0
# 高噪声环境
elif noise_info["noise_class"] == "extreme":
self.current_mode = FilterMode.HEAVY_FILTER
self.mode_duration += 1
# 中等噪声
elif noise_info["noise_class"] == "high":
self.current_mode = FilterMode.HEAVY_FILTER
# 低噪声但可能有阶跃
elif noise_info["noise_class"] == "low" and self._detect_step():
self.current_mode = FilterMode.FAST_RESPONSE
self.mode_duration = 0
# 正常模式
else:
self.current_mode = FilterMode.NORMAL
self.mode_duration += 1
# 记录模式切换
if old_mode != self.current_mode:
self.stats["mode_switches"] += 1
print(f"[DynamicFilter] 模式切换: {old_mode.name} -> {self.current_mode.name}")
def _detect_step(self) -> bool:
"""检测是否存在阶跃变化"""
if len(self.filtered_history) < 10:
return False
recent = self.filtered_history[-10:]
first_half = np.mean(recent[:5])
second_half = np.mean(recent[5:])
step_siz
利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！