智慧档案馆十防系统实战：温湿度感知闭环调控与现场故障排查

采样 → 归一化 → 控制决策 → 执行下发 → 执行反馈 → 再采样
  │         │         │           │          │
  └─传感器   └─点表    └─DDC/PLC   └─执行器   └─状态回读

"""
ddc_control_loop.py - 现场 DDC 侧闭环调控参考实现（运行于边缘/嵌入式节点）
说明：生产环境多由 PLC/DDC 固件实现，此处用 Python 描述控制逻辑与状态机，
     实际部署时替换为实时循环 + 硬件 IO 驱动。
"""

import time
from dataclasses import dataclass

@dataclass
class LoopState:
    kp: float = 0.8
    ki: float = 0.05
    kd: float = 0.1
    integ: float = 0.0
    prev_err: float = 0.0
    last_out: float = 0.0
    mode: str = "summer"          # summer / winter / transition
    reheat_hold: bool = False

def clamp(x, lo, hi):
    return max(lo, min(hi, x))

def pid_update(s: LoopState, setpoint, pv, dt):
    err = setpoint - pv
    s.integ = clamp(s.integ + err * dt, -100.0, 100.0)
    deriv = (err - s.prev_err) / dt if dt > 1e-3 else 0.0
    out = s.kp * err + s.ki * s.integ + s.kd * deriv
    s.prev_err = err
    s.last_out = clamp(out, 0.0, 100.0)
    return s.last_out

class HVACLoop:
    def __init__(self, t_set=20.0, h_set=50.0):
        self.t_set = t_set
        self.h_set = h_set
        self.t_loop = LoopState()
        self.h_loop = LoopState(kp=0.6, ki=0.03, kd=0.05)
        self.exec_state = {"ac": 0, "humidifier": 0, "dehumidifier": 0, "reheat": 0, "damper": 0}

    def step(self, env, dt):
        # env: dict with keys t_ret, h_ret, dew_point, surf_temp, co2, outdoor_t
        # ---- 防冷凝保护：独立于 PID，无条件介入 ----
        cond_risk = env["surf_temp"] < env["dew_point"] + 2.0
        if cond_risk:
            self.exec_state["reheat"] = 100
            # 抑制除湿输出
            dehum_out = 0.0
        else:
            self.exec_state["reheat"] = 0
            dehum_out = pid_update(self.h_loop, self.h_set, env["h_ret"], dt)

        t_out = pid_update(self.t_loop, self.t_set, env["t_ret"], dt)

        # ---- 模式相关执行映射 ----
        if env["outdoor_t"] > 22:
            self.t_loop.mode = "summer"
        elif env["outdoor_t"] < 12:
            self.t_loop.mode = "winter"
        else:
            self.t_loop.mode = "transition"

        if self.t_loop.mode == "summer":
            self.exec_state["ac"] = t_out                      # 制冷需求
            if env["h_ret"] > self.h_set + 5:
                self.exec_state["dehumidifier"] = dehum_out
            else:
                self.exec_state["dehumidifier"] = 0
        else:
            # 加热模式：ac 输出映射为加热阀/电加热驱动
            self.exec_state["ac"] = t_out
            if env["h_ret"] < self.h_set - 5:
                self.exec_state["humidifier"] = dehum_out
            else:
                self.exec_state["humidifier"] = 0

        # 新风阀随 CO2
        self.exec_state["damper"] = clamp((env["co2"] - 800) / 4.0, 0.0, 100.0) if env["co2"] > 800 else 0.0
        return self.exec_state

# 部署注意：
# - 实际 DDC 周期 0.5~1s，此处 sleep 仅为示意
# - 写执行器前做行程校验，回读状态字确认，失败则降级
def run(loop, get_env, write_exec, poll=1.0):
    last = time.time()
    while True:
        now = time.time()
        dt = now - last
        last = now
        env = get_env()
        cmd = loop.step(env, dt)
        ok = write_exec(cmd)
        if not ok:
            # 本地降级：维持上一状态，触发告警，不激进切换
            loop.step(env, dt)
        time.sleep(poll)

L0 物理层   供电/PoE/线缆/接头/安装位置
L1 网络层   VLAN/IP/端口/ARP/生成树/PoE功率预算
L2 协议层   Modbus TCP会话/事务ID/寄存器表/BACnet对象/点表映射
L3 数据层   质量位/时标/单位/量程/滤波/去抖
L4 控制层   DDC状态机/模式/限位/防冷凝旁路/执行器反馈
L5 业务层   联动规则/告警分发/报表/审计

"""
field_diag.py - 现场快速诊断片段
"""
import subprocess, json, sys

def modbus_probe(host, port=502, unit=1, addr=0x0000, count=8):
    from pymodbus.client import ModbusTcpClient
    c = ModbusTcpClient(host, port=port)
    if not c.connect():
        return {"ok": False, "reason": "connect_fail"}
    rr = c.read_holding_registers(addr, count, slave=unit)
    if rr.isError():
        return {"ok": False, "reason": rr.message}
    # 注意：缩放/字序依厂家表而定，此处仅透出原始值
    return {"ok": True, "regs": rr.registers}

def net_probe(host):
    try:
        out = subprocess.check_output(["ping","-c","3",host], stderr=subprocess.STDOUT).decode()
    except subprocess.CalledProcessError as e:
        out = e.output.decode()
    return out

def capture_modbus(iface, host, dur=10):
    # 现场抓包辅助定位，需在授权网络内执行
    pcap = f"/tmp/modbus_{host}.pcap"
    proc = subprocess.Popen(["tcpdump","-i",iface,"-w",pcap,f"host {host} and port 502"])
    import time; time.sleep(dur); proc.terminate()
    return pcap

if __name__ == "__main__":
    host = sys.argv[1] if len(sys.argv)>1 else "192.168.10.21"
    print(json.dumps(modbus_probe(host), indent=2))