档案库房净化恒湿一体机调试：环境参数闭环控制与稳定性测试

风机启动 → 确认转向、电流、振动、噪声
电加热/再热投入 → 确认逐级投入、电流平稳
压缩机启动 → 确认吸排气压力、运行电流、无液击异响
加湿器投入 → 确认补水阀开、水位上升、加湿量输出
冷凝水提升泵 → 手动注水测试启停和排水
UV/杀菌 → 确认灯管点亮（可见光/臭氧检测）

"""
point_check.py - 通信点表逐项核对脚本
上电后逐点读写，确认地址、数据类型、缩放因子正确
"""

import asyncio
from pymodbus.client import AsyncModbusTcpClient

POINTS = {
    # 只读点：回风温湿度、运行状态、故障字、过滤器压差
    "ret_temp":      {"type": "read", "fc": 3, "addr": 0, "scale": 0.1, "unit": "℃"},
    "ret_hum":       {"type": "read", "fc": 3, "addr": 1, "scale": 0.1, "unit": "%RH"},
    "supply_temp":   {"type": "read", "fc": 3, "addr": 2, "scale": 0.1, "unit": "℃"},
    "supply_hum":    {"type": "read", "fc": 3, "addr": 3, "scale": 0.1, "unit": "%RH"},
    "fan_status":    {"type": "read", "fc": 1, "addr": 0, "scale": 1, "unit": "bool"},
    "comp_status":   {"type": "read", "fc": 1, "addr": 1, "scale": 1, "unit": "bool"},
    "heat_status":   {"type": "read", "fc": 1, "addr": 2, "scale": 1, "unit": "bool"},
    "hum_status":    {"type": "read", "fc": 1, "addr": 3, "scale": 1, "unit": "bool"},
    "fault_word":    {"type": "read", "fc": 3, "addr": 10, "scale": 1, "unit": "hex"},
    "filter_dp":     {"type": "read", "fc": 3, "addr": 20, "scale": 1.0, "unit": "Pa"},
    # 可写点：设定值、启停、模式
    "sp_temp":       {"type": "write", "fc": 6, "addr": 100, "scale": 0.1, "unit": "℃"},
    "sp_hum":        {"type": "write", "fc": 6, "addr": 101, "scale": 0.1, "unit": "%RH"},
    "run_cmd":       {"type": "write", "fc": 5, "addr": 0, "scale": 1, "unit": "bool"},
    "mode_cmd":      {"type": "write", "fc": 6, "addr": 1, "scale": 1, "unit": "enum"},
}

async def check_read(c, point, cfg):
    """读单个点并验证合理性"""
    if cfg["fc"] == 3:  # 保持寄存器
        rr = await c.read_holding_registers(cfg["addr"], 1, slave=1)
    else:  # 线圈
        rr = await c.read_coils(cfg["addr"], 1, slave=1)
    if rr.isError():
        return False, "error"
    raw = rr.registers[0] if hasattr(rr, 'registers') else rr.bits[0]
    val = raw * cfg["scale"] if isinstance(raw, (int, float)) else raw
    return True, val

async def check_write_readback(c, point, cfg, test_val):
    """写设定值并回读验证"""
    scaled = int(test_val / cfg["scale"])
    await c.write_registers(cfg["addr"], [scaled], slave=1)
    await asyncio.sleep(1)
    ok, val = await check_read(c, point, cfg)
    return ok and abs(val - test_val) < 0.2

async def main():
    c = AsyncModbusTcpClient("10.20.1.50", port=502)
    await c.connect()
    
    print("=== 只读点核对 ===")
    for point, cfg in POINTS.items():
        if cfg["type"] == "read":
            ok, val = await check_read(c, point, cfg)
            status = "✓" if ok else "✗"
            print(f"  {status} {point}: {val} {cfg['unit']}")
    
    print("\n=== 可写点回读验证 ===")
    for point, cfg in POINTS.items():
        if cfg["type"] == "write":
            if point.startswith("sp_"):
                test_val = 22.0 if "temp" in point else 50.0
                ok = await check_write_readback(c, point, cfg, test_val)
                status = "✓" if ok else "✗"
                print(f"  {status} {point}: write {test_val} → readback OK={ok}")

asyncio.run(main())

步骤：
1. 机组手动模式，风机运行
2. 加湿器从 0% → 50% 输出，记录回风湿度上升曲线
3. 加湿器从 50% → 0%，记录湿度回落曲线
4. 电加热从 0% → 30%，记录回风温度上升曲线
5. 压缩机启动，记录温湿度联合变化

"""
step_response.py - 阶跃响应数据采集
手动操作设备后，采集并标注时间，用于后续 PID 参数估算
"""

import time
from datetime import datetime
import csv

class StepResponseRecorder:
    def __init__(self, device_host, csv_path):
        self.device_host = device_host
        self.csv_path = csv_path
        self.data = []
    
    def record(self, sp_hum_pct, actual_hum, fan_rpm=None):
        """每次调用记录一行"""
        self.data.append({
            "ts": datetime.now().isoformat(),
            "elapsed_s": time.time() - self.t_start,
            "output_pct": sp_hum_pct,
            "ret_hum": actual_hum,
            "fan_rpm": fan_rpm or 0
        })
    
    def start(self):
        self.t_start = time.time()
    
    def save(self):
        with open(self.csv_path, "w", newline="") as f:
            writer = csv.DictWriter(f, fieldnames=self.data[0].keys())
            writer.writeheader()
            writer.writerows(self.data)
        print(f"阶跃响应数据已保存: {self.csv_path}")

1. 置 Ki=0, Kd=0，Kp 从 0.5 开始逐步增大
2. 人为注入扰动（开门 30s），观察系统响应
3. 找到临界增益 Ku（出现持续等幅振荡时的 Kp）
4. 记录临界周期 Tu（振荡周期）
5. 按以下参数设定：
   Kp = 0.45 × Ku
   Ki = 0.54 × Ku / Tu
   Kd = 0.075 × Ku × Tu

"""
humidity_pid.py - 湿度 PID 回路参考实现
运行在边缘 DDC/网关上，与一体机通信
"""

import time
from dataclasses import dataclass

@dataclass
class PIDParams:
    Kp: float = 3.0
    Ki: float = 0.4
    Kd: float = 1.5
    Ts: float = 60.0        # 采样周期(s)
    out_min: float = 0.0    # 输出下限(%)
    out_max: float = 100.0  # 输出上限(%)
    anti_windup: bool = True

class HumidityPID:
    def __init__(self, params: PIDParams):
        self.p = params
        self.integral = 0.0
        self.prev_error = 0.0
        self.last_output = 0.0
        self.last_time = time.time()
    
    def compute(self, setpoint: float, process_var: float) -> float:
        now = time.time()
        dt = now - self.last_time
        if dt < self.p.Ts * 0.5:  # 防抖：距上次计算不足半个周期则跳过
            return self.last_output
        
        error = setpoint - process_var
        
        # P
        P_term = self.p.Kp * error
        
        # I（带抗积分饱和）
        self.integral += error * dt / 60.0  # 积分项单位：%·min
        if self.p.anti_windup:
            # 回算抗饱和
            I_term = self.p.Ki * self.integral
            output_unsaturated = P_term + I_term + self.p.Kd * (error - self.prev_error) / (dt / 60.0)
            if output_unsaturated > self.p.out_max:
                self.integral = (self.p.out_max - P_term) / self.p.Ki if self.p.Ki != 0 else 0
            elif output_unsaturated < self.p.out_min:
                self.integral = (self.p.out_min - P_term) / self.p.Ki if self.p.Ki != 0 else 0
        I_term = self.p.Ki * self.integral
        
        # D（对 PV 微分，避免设定值阶跃时微分冲击）
        D_term = self.p.Kd * (-(process_var - (process_var - self.prev_error))) / (dt / 60.0)
        # 等价于 -Kd * d(PV)/dt
        derivative = -(process_var - (process_var - (error - self.prev_error))) / (dt / 60.0)
        D_term = self.p.Kd * derivative
        
        output = P_term + I_term + D_term
        output = max(self.p.out_min, min(self.p.out_max, output))
        
        self.prev_error = error
        self.last_output = output
        self.last_time = now
        
        return output
    
    def reset(self):
        """模式切换/故障恢复时重置"""
        self.integral = 0.0
        self.prev_error = 0.0
        self.last_output = 0.0

优先级：
1. 湿度越限 → 湿度回路全权控制（压缩机/加湿器）
2. 湿度在带内 → 温度回路微调（电加热/风机转速）
3. 两者冲突时，湿度优先（档案保护核心）

"""
mode_manager.py - 运行模式管理
"""

class ModeManager:
    def __init__(self):
        self.current_mode = "normal"
        self.mode_hold_time = {}  # 模式保持计时
    
    def evaluate(self, ret_t, ret_h, sp_t, sp_h, fault_active, schedule):
        """根据当前状态评估应进入的模式"""
        # 故障优先
        if fault_active:
            return "safe"
        
        # 时间调度
        if schedule.get("is_off_hours"):
            return "maintenance"
        
        # 湿度偏差判定
        hum_dev = ret_h - sp_h
        temp_dev = ret_t - sp_t
        
        if hum_dev > 2.0:
            return "dehumidify_priority"
        elif hum_dev < -2.0:
            return "humidify_priority"
        elif temp_dev > 3.0:
            return "cool_fast"
        else:
            return "normal"
    
    def get_pid_params(self, mode):
        """不同模式下使用不同的 PID 参数"""
        params_map = {
            "normal":             {"hum": PIDParams(Kp=3.0, Ki=0.4, Kd=1.5)},
            "dehumidify_priority":{"hum": PIDParams(Kp=4.0, Ki=0.3, Kd=2.0)},  # 更激进
            "humidify_priority":  {"hum": PIDParams(Kp=2.5, Ki=0.5, Kd=1.0)},  # 稍温和
            "cool_fast":          {"hum": PIDParams(Kp=2.0, Ki=0.2, Kd=0.5)},  # 弱湿度控制
            "maintenance":        {"hum": PIDParams(Kp=2.0, Ki=0.3, Kd=1.0)},
            "safe":               {"hum": PIDParams(Kp=1.0, Ki=0.1, Kd=0.5)},  # 保守
        }
        return params_map.get(mode, params_map["normal"])

条件：无人为扰动，门窗关闭，无人进出
持续：24h
判定：
  - 温度波动 ≤ ±1.0℃（设定值 20℃）
  - 湿度波动 ≤ ±3.0%RH（设定值 50%RH）
  - 数据完整率 ≥ 99.5%

场景：开门（湿度冲击）+ 同时注入加湿阶跃（控制对抗）
目的：验证 PID 在对抗性输入下的稳定性，不发散、不持续振荡
判定：30min 内回到设定带，无持续振荡（振幅递减）

持续：7 天连续运行
条件：模拟日常使用（每天定时开门 2 次、人员进出）
监测：
  - 稳态精度趋势（逐日对比）
  - 执行器动作频次（压缩机启停次数/天）
  - 过滤器压差趋势（是否异常上升）
  - 能耗（可选）
判定：7 天内稳态指标不劣化，执行器无异常频繁动作

"""
stability_test.py - 稳定性测试数据记录与分析
"""

import statistics
from collections import defaultdict

class StabilityAnalyzer:
    def __init__(self, setpoint_t=20.0, setpoint_h=50.0):
        self.sp_t = setpoint_t
        self.sp_h = setpoint_h
        self.samples = []
    
    def add_sample(self, ts, temp, hum, mode, actuator_state):
        self.samples.append({
            "ts": ts, "temp": temp, "hum": hum,
            "mode": mode, "actuator": actuator_state
        })
    
    def analyze(self):
        if len(self.samples) < 10:
            return {"error": "样本不足"}
        
        temps = [s["temp"] for s in self.samples]
        hums = [s["hum"] for s in self.samples]
        
        results = {
            "temp": {
                "mean": statistics.mean(temps),
                "stdev": statistics.stdev(temps) if len(temps)>1 else 0,
                "min": min(temps), "max": max(temps),
                "range": max(temps) - min(temps),
                "within_band": sum(1 for t in temps if abs(t-self.sp_t)<=1.0) / len(temps)
            },
            "hum": {
                "mean": statistics.mean(hums),
                "stdev": statistics.stdev(hums) if len(hums)>1 else 0,
                "min": min(hums), "max": max(hums),
                "range": max(hums) - min(hums),
                "within_band": sum(1 for h in hums if abs(h-self.sp_h)<=3.0) / len(hums)
            },
            "actuator_stats": self._actuator_stats()
        }
        return results
    
    def _actuator_stats(self):
        """统计执行器动作频次"""
        stats = defaultdict(lambda: {"on_count": 0, "total_time": 0, "last_state": None, "last_ts": None})
        for s in self.samples:
            for act, state in s["actuator"].items():
                if state and not stats[act]["last_state"]:
                    stats[act]["on_count"] += 1
                if state:
                    stats[act]["total_time"] += 1  # 采样间隔的倍数
                stats[act]["last_state"] = state
        return dict(stats)