恒温恒湿的自动化闭环：基于 PID 算法的档案库房环境调控系统落地手记

┌──────────────────────────────┐
│ 感知层                       │
│ 温湿度 / 回风温湿度 / 漏水   │
│ 阀门开度 / 设备电流 / 滤网   │
└────────────┬─────────────────┘
             ▼
┌──────────────────────────────┐
│ 决策层                       │
│ 数据滤波 → 偏差计算          │
│ 安全联锁 → PID运算 → 解耦    │
└────────────┬─────────────────┘
             ▼
┌──────────────────────────────┐
│ 执行层                       │
│ 冷水阀 / 电加热 / 加湿       │
│ 除湿压缩机 / 新风机 / 风机   │
└────────────┬─────────────────┘
             ▼
┌──────────────────────────────┐
│ 审计层                       │
│ 设定值 / 动作 / 能耗         │
│ 报警 / 校准 / 操作员审批     │
└──────────────────────────────┘

e(k) = SP(k) − PV(k)
u(k) = Kp·e(k)
     + Ki·Σe(k)·Δt
     + Kd·[PV(k−1) − PV(k)] / Δt

predicted_temp_drop = k × dehumidifier_capacity
reheat_output = predicted_temp_drop × temp_gain

from dataclasses import dataclass
from time import time

@dataclass
class PIDConfig:
    kp: float = 1.0
    ki: float = 0.05
    kd: float = 0.0
    out_min: float = 0.0
    out_max: float = 100.0
    out_step: float = 15.0     # 单周期最大输出变化
    integral_limit: float = 200.0
    derivative_filter: float = 0.2  # 0~1，越大越平滑

class PIDController:
    def __init__(self, cfg: PIDConfig):
        self.cfg = cfg
        self.integral = 0.0
        self.prev_error = 0.0
        self.prev_measurement = None
        self.prev_output = 0.0
        self.last_time = None

    def update(self, setpoint: float, measurement: float, now: float = None) -> float:
        if now is None:
            now = time()
        if measurement is None or measurement != measurement:
            return self.prev_output  # NaN 保护

        dt = now - self.last_time if self.last_time else 0.0
        self.last_time = now

        error = setpoint - measurement

        # 对测量值微分，避免设定值阶跃时产生微分冲击
        if self.prev_measurement is not None and dt > 0:
            raw_d = (self.prev_measurement - measurement) / dt
            raw_d = raw_d * 0.0 + error - self.prev_error  # 可切换为测量值微分
        else:
            raw_d = 0.0

        derivative = self._low_pass(raw_d)

        self.integral = self._clamp(
            self.integral + error * (dt if dt > 0 else 0.0),
            -self.cfg.integral_limit,
            self.cfg.integral_limit
        )

        output = (
            self.cfg.kp * error
            + self.cfg.ki * self.integral
            + self.cfg.kd * derivative
        )

        # 抗积分饱和：输出越限时，只累加有助于退出饱和的误差
        if output > self.cfg.out_max and error > 0:
            self.integral -= error * dt
        elif output < self.cfg.out_min and error < 0:
            self.integral -= error * dt

        output = self._clamp(output, self.cfg.out_min, self.cfg.out_max)
        output = self._clamp_step(output)

        self.prev_error = error
        self.prev_measurement = measurement
        self.prev_output = output
        return round(output, 2)

    def _clamp(self, value, lo, hi):
        return max(lo, min(hi, value))

    def _clamp_step(self, value):
        step = self.cfg.out_step
        return self._clamp(value, self.prev_output - step, self.prev_output + step)

    def _low_pass(self, value):
        if self.prev_measurement is None:
            return value
        alpha = self.cfg.derivative_filter
        self._filtered_d = getattr(self, '_filtered_d', value)
        self._filtered_d = alpha * value + (1 - alpha) * self._filtered_d
        return self._filtered_d

pid = PIDController(PIDConfig(
    kp=1.2,
    ki=0.03,
    kd=0.1,
    out_min=0,
    out_max=100,
    out_step=8
))

for pv in [55, 56, 58, 60, 59, 57, 54, 53]:
    print(pid.update(52, pv))

who + when + zone + before + after + reason + approved_by