编写程序实现控制信号平滑滤波，消除变送器噪声造成调节阀频繁小幅动作。
调节阀“抖动”终结者 —— 基于信号平滑滤波的工业过程控制实践
“现场最怕的不是大故障，而是小抖动。阀门一天动作上万次，机械寿命就在一次次无意义的小幅震荡中耗尽。真正稳定的系统，不是反应最快的，而是最懂过滤噪声的。”
—— 哈尔滨工程大学《工业过程控制》课程核心思想延伸
一、实际应用场景描述
在化工精馏塔、锅炉汽包、反应釜温控等连续过程控制场景中，调节阀是执行机构的“手脚”，其稳定性直接决定产品质量与设备寿命：
┌──────────────────────────────────────────────┐
│ 典型过程控制回路（带噪声干扰） │
│ │
│ [被控对象] 精馏塔/锅炉/反应釜 │
│ • 温度、压力、流量、液位等热工参数 │
│ • 大滞后、大惯性、非线性特性 │
│ • 对控制平稳性要求极高 │
│ │ │
│ ▼ 过程响应（慢，几十秒~几分钟） │
│ ┌────────────────────────────┐ │
│ │ 测量变送环节 │ │
│ │ • 热电偶/热电阻/压力变送器 │ │
│ │ • 4~20mA 电流信号 │ │
│ │ • 电磁干扰（变频器/电机） │ ←── 噪声源 │
│ │ • 接线松动/接触不良 │ ←── 噪声源 │
│ │ • 模数转换量化误差 │ ←── 噪声源 │
│ └────────────┬───────────────┘ │
│ │ 含噪测量值 (PV_noisy) │
│ ▼ │
│ ┌────────────────────────────┐ │
│ │ PID控制器（常规） │ │
│ │ • 比例(P)：放大误差 │ │
│ │ • 积分(I)：消除余差 │ │
│ │ • 微分(D)：预判趋势 │ │
│ │ ⚠️ 噪声被放大，导致控制量抖动 │ │
│ └────────────┬───────────────┘ │
│ │ 抖动的控制输出 (MV_jitter) │
│ ▼ │
│ ┌────────────────────────────┐ │
│ │ 调节阀（执行机构） │ │
│ │ • 气动/电动调节阀 │ │
│ │ • 机械结构有死区/回差 │ │
│ │ • 频繁小幅动作 → 磨损 │ ←── 痛点 │
│ │ • 阀杆填料摩擦 → 卡涩 │ ←── 痛点 │
│ └────────────┬───────────────┘ │
│ │ 阀门开度指令 │
│ ▼ │
│ ┌────────────────────────────┐ │
│ │ 被控对象（再次反馈） │ │
│ │ • 过程响应滞后掩盖高频抖动 │ │
│ │ • 操作员误以为“控制正常” │ │
│ └───────────────────────────┘ │
│ │
│ 核心矛盾: 变送器噪声(快) vs 过程响应(慢) │
│ 解决方案: 在PID前加入信号平滑滤波环节 │
└──────────────────────────────────────────────┘
为什么噪声会导致阀门“发抖”？
这是一个典型的“快干扰、慢过程”失配问题：
1. 变送器噪声：电磁干扰、接线松动、电源纹波等产生的高频噪声（通常几Hz~几十Hz）
2. PID 放大作用：比例环节会将微小的测量波动放大 Kp 倍
3. 微分环节敏感：微分对快速变化的噪声极其敏感，产生剧烈脉冲
4. 执行机构响应：虽然过程本身响应慢，但阀门位置反馈灵敏，会跟随噪声抖动
5. 机械磨损加剧：阀门每天动作数千次，填料、阀杆、密封件快速老化
二、引入痛点
2.1 现场的真实困境
场景 现场发生了什么 根因
“阀门一天坏仨” “调节阀阀杆断裂，一个月换了三次” 噪声导致高频小幅振荡
“产品纯度波动” “精馏塔顶产品纯度忽高忽低” 阀门抖动破坏稳态流量
“电费异常升高” “压缩机频繁加载卸载，电耗飙升” 回流阀/进料阀频繁动作
“DCS报警不断” “PV偏差报警每分钟触发几十次” 噪声触发虚假偏差
“维护人员崩溃” “每天80%时间在调阀位，没空巡检” 噪声掩盖真实工艺问题
2.2 核心矛盾
过程控制的核心矛盾是“测量噪声”与“执行平稳”的冲突。变送器采集的是“真值+噪声”，而调节阀执行的是“指令+噪声”。如果不加处理，高频噪声会被PID放大，最终导致执行机构“疲于奔命”。
2.3 我们要解决什么
用一段精简的 Python 程序，构建一个 带信号平滑滤波的PID控制系统，实现：
1. 多种滤波算法 —— 滑动平均、一阶滞后、中值滤波
2. 自适应滤波 —— 根据噪声强度自动调整滤波强度
3. 死区+滞环 —— 防止微小波动触发阀门动作
4. 变化率限制 —— 限制阀门动作速度，保护机械
5. 性能量化 —— 对比滤波前后的阀门动作次数与磨损
三、核心逻辑讲解
3.1 理论基础：信号平滑与噪声抑制
本工具基于哈工程《工业过程控制》第三章“过程特性”和第四章“PID控制器”：
① 噪声来源与特性
噪声频谱分布:
┌─────────────────────────────────┐
│ 幅值 ▲ │
│ │ ●●●●●●●●●●●●●●● │
│ │ ● ● │
│ │● ● │
│ │ ●│
│ └────────────────────────→ 频率
│ 低频 中频 高频 │
│ (过程) (干扰) (噪声) │
└─────────────────────────────────┘
噪声类型:
• 白噪声: 均匀分布，各频率分量相等
• 脉冲噪声: 偶发尖峰，由电磁干扰引起
• 周期性噪声: 工频干扰(50Hz)、变频器谐波
• 量化噪声: A/D转换的舍入误差
② 常用滤波算法对比
滤波方式 原理 优点 缺点 适用场景
滑动平均 取N个采样的算术平均 算法简单，抑制随机噪声 滞后较大，对脉冲不敏感 平稳过程
一阶滞后 y(k)=αx(k)+(1-α)y(k-1) 连续平滑，无滞后积累 对突变响应慢 大多数场合
中值滤波 取N个采样的中值 有效抑制脉冲噪声 对连续变化响应慢 电磁干扰严重
卡尔曼滤波 最优估计算法 理论最优，自适应 计算复杂，需调参 高精度要求
死区+滞环 设置不灵敏区 彻底消除微小抖动 牺牲部分控制精度 阀门保护
③ 一阶滞后滤波（最常用）
y(k) = \alpha \cdot x(k) + (1-\alpha) \cdot y(k-1)
其中：
- y(k) ：当前滤波输出
- x(k) ：当前原始采样
- y(k-1) ：上一次滤波输出
- \alpha ：滤波系数（0~1），越小越平滑
截止频率：
f_c = \frac{\alpha}{2\pi T_s}
其中 T_s 为采样周期。
④ 死区与滞环（Deadband & Hysteresis）
死区+滞环特性:
输出 ▲
│
+δ │ ┌───────────────
│ │
0 │────┼───────────────→ 输入
│ │
-δ │ └───────────────
│
└─────────────────────
规则:
• 当 |输入-输出| < δ 时，输出保持不变
• 当 输入-输出 > +δ 时，输出 = 输入 - δ
• 当 输出-输入 > +δ 时，输出 = 输入 + δ
作用: 彻底消除微小波动引起的阀门动作
⑤ 变化率限制（Rate Limiting）
\Delta u_{max} = K_{rate} \cdot T_s
其中：
- \Delta u_{max} ：单步最大变化量
- K_{rate} ：变化率限制系数
- T_s ：采样周期
3.2 控制策略设计
┌──────────────────────────────────────────────┐
│ 带信号平滑的PID控制策略 │
│ │
│ 设定值(SP) │
│ │ │
│ ▼ │
│ ┌───────────────┐ │
│ │ 比较器 │←─────────────────────┐ │
│ │ e = SP - PV │ │ │
│ └───────┬───────┘ │ │
│ │ e(t) │ │
│ ▼ │ │
│ ┌─────────────────────────────────────┐ │ │
│ │ 信号平滑滤波模块 │ │ │
│ │ ┌───────────────────────────────┐ │ │ │
│ │ │ 1. 中值滤波（滤除脉冲） │ │ │ │
│ │ │ 2. 一阶滞后（平滑随机噪声） │ │ │ │
│ │ │ 3. 死区+滞环（消除微小波动） │ │ │ │
│ │ └───────────────────────────────┘ │ │ │
│ └───────┬───────────────────────────┘ │ │ │
│ │ e_filtered(t) │ │ │
│ ▼ │ │ │
│ ┌─────────────────────────────────────┐ │ │ │
│ │ PID控制器 │ │ │ │
│ │ • 比例(P): Kp · e_filtered │ │ │ │
│ │ • 积分(I): Ki ∫e_filtered dt │ │ │ │
│ │ • 微分(D): Kd · de_filtered/dt │ │ │ │
│ │ • 抗积分饱和 │ │ │ │
│ └───────┬───────────────────────────┘ │ │ │
│ │ MV_raw(t) │ │ │
│ ▼ │ │ │
│ ┌─────────────────────────────────────┐ │ │ │
│ │ 执行器信号处理模块 │ │ │ │
│ │ ┌───────────────────────────────┐ │ │ │
│ │ │ 1. 变化率限制（保护机械） │ │ │ │
│ │ │ 2. 输出限幅（硬件安全） │ │ │ │
│ │ │ 3. 死区处理（消除颤振） │ │ │ │
│ │ └───────────────────────────────┘ │ │ │ │
│ └───────┬───────────────────────────┘ │ │ │
│ │ MV_final(t) → 调节阀 │ │ │
│ ▼ │ │ │
│ ┌─────────────────────────────────────┐ │ │ │
│ │ 被控对象（大滞后） │ │ │ │
│ │ • 温度/压力/流量/液位 │ │ │ │
│ │ • 响应缓慢，过滤高频抖动 │ │ │ │
│ └───────┬───────────────────────────┘ │ │ │
│ │ PV_raw(t) │ │ │
│ ▼ │ │ │
│ ┌─────────────────────────────────────┐ │ │ │
│ │ 测量变送环节（含噪声） │ │ │ │
│ │ • 传感器噪声 │ │ │ │
│ │ • 电磁干扰 │ │ │ │
│ │ • 量化误差 │ │ │ │
│ └─────────────────────────────────────┘ │ │ │
│ │ │ │
│ 核心思想: 在PID前滤除噪声，在输出侧保 │ │ │
│ 护执行机构，中间过程保持线性 │ │ │
└───────────────────────────────────────────┴──┘
四、代码讲解（面向对象设计）
4.1 类结构总览
类名 职责 设计模式
"FilterConfig" 滤波器配置（dataclass） 值对象
"SignalFilter" 信号平滑滤波器（抽象基类） 策略模式
"MovingAverageFilter" 滑动平均滤波器 具体策略
"FirstOrderLagFilter" 一阶滞后滤波器 具体策略
"MedianFilter" 中值滤波器 具体策略
"AdaptiveFilter" 自适应滤波器 装饰器模式
"DeadbandHysteresis" 死区+滞环处理器 具体策略
"RateLimiter" 变化率限制器 具体策略
"FilteredPIDController" 带滤波的PID控制器 聚合根
"ValveSimulator" 调节阀仿真模型 领域模型
"ProcessSimulator" 过程对象仿真 领域模型
"NoiseGenerator" 噪声发生器 工厂模式
"PerformanceAnalyzer" 性能分析器 封装
"Visualizer" 可视化工具 封装
4.2 核心代码（精简版，CSDN友好）
完整源码约 280 行，包含 8 个类、多种滤波算法、死区滞环、变化率限制、性能分析。
以下为可直接运行的精简核心版。
"""
调节阀信号平滑滤波仿真 —— 消除变送器噪声引起的频繁动作
参考哈尔滨工程大学《工业过程控制》第三章"过程特性"与第四章"PID控制器"
"""
from dataclasses import dataclass, field
from typing import List, Tuple, Optional, Deque
from enum import Enum, auto
from abc import ABC, abstractmethod
import numpy as np
import matplotlib.pyplot as plt
from collections import deque
import math
from datetime import datetime
# ============================================================
# 1. 基础数据结构（值对象）
# ============================================================
@dataclass
class FilterConfig:
"""滤波器配置 —— 值对象"""
# 一阶滞后滤波
alpha: float = 0.1 # 滤波系数 (0~1)，越小越平滑
# 滑动平均滤波
window_size: int = 5 # 窗口大小
# 中值滤波
median_window: int = 3 # 中值窗口大小（奇数）
# 死区与滞环
deadband: float = 0.5 # 死区宽度 (%)
hysteresis: float = 0.2 # 滞环宽度 (%)
# 变化率限制
rate_limit: float = 2.0 # 最大变化率 (%/s)
# 自适应滤波
adaptive_enabled: bool = False
noise_threshold: float = 0.3 # 噪声检测阈值
# 通用参数
sample_time: float = 1.0 # 采样周期 (s)
max_sim_time: float = 1800.0 # 最大仿真时间 (s)
@dataclass
class PIDConfig:
"""PID控制器配置 —— 值对象"""
kp: float = 1.0 # 比例增益
ki: float = 0.05 # 积分增益
kd: float = 0.1 # 微分增益
output_min: float = 0.0 # 输出下限 (%)
output_max: float = 100.0 # 输出上限 (%)
anti_windup: bool = True # 抗积分饱和
@dataclass
class ValveConfig:
"""调节阀配置 —— 值对象"""
deadzone: float = 0.3 # 机械死区 (%)
hysteresis: float = 0.2 # 机械滞环 (%)
max_rate: float = 5.0 # 最大动作速度 (%/s)
response_time: float = 0.5 # 响应时间常数 (s)
stiction: float = 0.1 # 静摩擦力 (%)
@dataclass
class ProcessConfig:
"""过程对象配置 —— 值对象"""
gain: float = 1.0 # 过程增益
time_constant: float = 60.0 # 时间常数 (s)
dead_time: float = 10.0 # 纯滞后 (s)
noise_level: float = 0.2 # 噪声水平 (%)
# ============================================================
# 2. 滤波器接口与实现（策略模式）
# ============================================================
class SignalFilter(ABC):
"""信号滤波器抽象基类"""
@abstractmethod
def filter(self, value: float) -> float:
"""滤波处理"""
pass
@abstractmethod
def reset(self):
"""重置滤波器状态"""
pass
class FirstOrderLagFilter(SignalFilter):
"""一阶滞后滤波器 —— 最常用"""
def __init__(self, alpha: float = 0.1):
self.alpha = max(0.0, min(1.0, alpha))
self.prev_output: float = 0.0
self.initialized: bool = False
def filter(self, value: float) -> float:
if not self.initialized:
self.prev_output = value
self.initialized = True
return value
# y(k) = α·x(k) + (1-α)·y(k-1)
output = self.alpha * value + (1 - self.alpha) * self.prev_output
self.prev_output = output
return output
def reset(self):
self.prev_output = 0.0
self.initialized = False
class MovingAverageFilter(SignalFilter):
"""滑动平均滤波器"""
def __init__(self, window_size: int = 5):
self.window_size = max(1, window_size)
self.buffer: Deque[float] = deque(maxlen=self.window_size)
def filter(self, value: float) -> float:
self.buffer.append(value)
return sum(self.buffer) / len(self.buffer)
def reset(self):
self.buffer.clear()
class MedianFilter(SignalFilter):
"""中值滤波器 —— 抑制脉冲噪声"""
def __init__(self, window_size: int = 3):
self.window_size = max(1, window_size)
if self.window_size % 2 == 0:
self.window_size += 1 # 确保奇数
self.buffer: Deque[float] = deque(maxlen=self.window_size)
def filter(self, value: float) -> float:
self.buffer.append(value)
return np.median(list(self.buffer))
def reset(self):
self.buffer.clear()
class AdaptiveFilter(SignalFilter):
"""自适应滤波器 —— 根据噪声强度调整滤波强度"""
def __init__(self, base_filter: SignalFilter,
noise_threshold: float = 0.3, base_alpha: float = 0.1):
self.base_filter = base_filter
self.noise_threshold = noise_threshold
self.base_alpha = base_alpha
self.prev_value: float = 0.0
self.noise_estimate: float = 0.0
def filter(self, value: float) -> float:
# 估计噪声强度（当前值与上一值的差值）
if hasattr(self.base_filter, 'alpha'):
noise = abs(value - self.prev_value)
self.noise_estimate = 0.9 * self.noise_estimate + 0.1 * noise
# 根据噪声强度调整滤波系数
if self.noise_estimate > self.noise_threshold:
# 噪声大，增强滤波
self.base_filter.alpha = min(0.3, self.base_alpha * 2)
else:
# 噪声小，减弱滤波
self.base_filter.alpha = max(0.05, self.base_alpha)
self.prev_value = value
return self.base_filter.filter(value)
def reset(self):
self.base_filter.reset()
self.prev_value = 0.0
self.noise_estimate = 0.0
# ============================================================
# 3. 死区与滞环处理器
# ============================================================
class DeadbandHysteresis:
"""死区+滞环处理器 —— 消除微小波动"""
def __init__(self, deadband: float = 0.5, hysteresis: float = 0.2):
self.deadband = deadband
self.hysteresis = hysteresis
self.prev_output: float = 0.0
self.last_direction: int = 0 # -1:下降, 0:静止, 1:上升
def process(self, input_value: float, reference: float = 0.0) -> float:
"""
死区+滞环处理
input_value: 输入值
reference: 参考值（通常为设定值或上一输出）
"""
error = input_value - reference
# 死区判断
if abs(error) < self.deadband:
return self.prev_output
# 滞环判断
if error > self.deadband:
# 正向超过死区
if self.last_direction <= 0:
# 从静止或下降转为上升，需要额外越过滞环
if error > self.deadband + self.hysteresis:
output = reference + (error - self.hysteresis)
self.last_direction = 1
else:
output = self.prev_output
else:
# 已经在上升，正常跟随
output = reference + error
self.last_direction = 1
else:
# 负向超过死区
if self.last_direction >= 0:
# 从静止或上升转为下降，需要额外越过滞环
if abs(error) > self.deadband + self.hysteresis:
output = reference + (error + self.hysteresis)
self.last_direction = -1
else:
output = self.prev_output
else:
# 已经在下降，正常跟随
output = reference + error
self.last_direction = -1
self.prev_output = output
return output
def reset(self):
self.prev_output = 0.0
self.last_direction = 0
# ============================================================
# 4. 变化率限制器
# ============================================================
class RateLimiter:
"""变化率限制器 —— 保护执行机构"""
def __init__(self, max_rate: float = 2.0, sample_time: float = 1.0):
self.max_rate = max_rate # %/s
self.sample_time = sample_time
self.prev_output: float = 0.0
self.max_step = max_rate * sample_time # 单步最大变化
def limit(self, target_value: float) -> float:
"""限制变化率"""
delta = target_value - self.prev_output
if delta > self.max_step:
output = self.prev_output + self.max_step
elif delta < -self.max_step:
output = self.prev_output - self.max_step
else:
output = target_value
self.prev_output = output
return output
def reset(self):
self.prev_output = 0.0
# ============================================================
# 5. PID控制器（改进版）
# ============================================================
class FilteredPIDController:
"""带信号滤波的PID控制器 —— 聚合根"""
def __init__(self, pid_config: PIDConfig, filter_config: FilterConfig):
self.pid_config = pid_config
self.filter_config = filter_config
# PID状态
self.integral: float = 0.0
self.prev_error: float = 0.0
self.prev_output: float = 0.0
self.prev_pv: float = 0.0
# 滤波链
self.filters: List[SignalFilter] = []
# 1. 中值滤波（先滤除脉冲）
if filter_config.median_window > 1:
self.filters.append(MedianFilter(filter_config.median_window))
# 2. 一阶滞后（平滑随机噪声）
lag_filter = FirstOrderLagFilter(filter_config.alpha)
if filter_config.adaptive_enabled:
self.filters.append(AdaptiveFilter(lag_filter,
filter_config.noise_threshold,
filter_config.alpha))
else:
self.filters.append(lag_filter)
# 3. 滑动平均（进一步平滑）
if filter_config.window_size > 1:
self.filters.append(MovingAverageFilter(filter_config.window_size))
# 4. 死区+滞环
self.deadband_processor = DeadbandHysteresis(
filter_config.deadband, filter_config.hysteresis)
# 5. 变化率限制
self.rate_limiter = RateLimiter(
filter_config.rate_limit, filter_config.sample_time)
# 统计信息
self.valve_action_count: int = 0
self.total_valve_movement: float = 0.0
self.prev_filtered_error: float = 0.0
print("="*60)
print(" 调节阀信号平滑滤波仿真系统")
print(" 基于哈尔滨工程大学《工业过程控制》")
print("="*60)
self._print_config()
def _print_config(self):
"""打印配置信息"""
print(f"\n[CONFIG] PID参数: Kp={self.pid_config.kp}, "
f"Ki={self.pid_config.ki}, Kd={self.pid_config.kd}")
print(f"[CONFIG] 滤波参数: α={self.filter_config.alpha}, "
f"死区={self.filter_config.deadband}%, "
f"滞环={self.filter_config.hysteresis}%")
print(f"[CONFIG] 变化率限制: {self.filter_config.rate_limit}%/s")
print(f"[CONFIG] 自适应滤波: {'启用' if self.filter_config.adaptive_enabled else '禁用'}")
def compute(self, setpoint: float, process_variable: float,
filtered_pv: Optional[float] = None) -> Tuple[float, float]:
"""
计算PID输出
返回: (滤波后的PV, 最终控制输出)
"""
# 1. 信号滤波链
filtered = process_variable
for filter_obj in self.filters:
filtered = filter_obj.filter(filtered)
# 2. 计算误差（使用滤波后的PV）
error = setpoint - filtered
# 3. 死区+滞环处理（在误差层面）
error_processed = self.deadband_processor.process(error, 0.0)
# 4. PID计算
# 比例项
p_term = self.pid_config.kp * error_processed
# 积分项（带抗饱和）
if self.pid_config.anti_windup:
# 检查是否处于饱和状态
if not (self.prev_output >= self.pid_config.output_max and error_processed > 0) and \
not (self.prev_output <= self.pid_config.output_min and error_processed < 0):
self.integral += error_processed * self.filter_config.sample_time
else:
self.integral += error_processed * self.filter_config.sample_time
i_term = self.pid_config.ki * self.integral
# 微分项（对PV微分，避免设定值突变）
d_term = 0.0
if self.filter_config.sample_time > 0:
# 使用滤波后的PV进行微分
d_term = -self.pid_config.kd * \
(filtered - self.prev_pv) / self.filter_config.sample_time
# 总输出
output = p_term + i_term + d_term
# 5. 输出限幅
output = max(self.pid_config.output_min,
min(self.pid_config.output_max, output))
# 6. 变化率限制（保护阀门）
output_limited = self.rate_limiter.limit(output)
# 7. 统计阀门动作
if abs(output_limited - self.prev_output) > 0.01: # 超过0.01%才算动作
self.valve_action_count += 1
利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！
