恒湿一体机与监控平台联动控制：RS485 标准协议下发启停指令实操

01 05 00 00 FF 00 8C 3A

01 05 00 00 00 00 CD CA

01 06 00 01 00 01 19 CA

平台点击“启动”
   ↓
1. 白名单校验：该寄存器是否允许写
   ↓
2. 构造 RTU 帧（地址+功能码+地址+值+CRC）
   ↓
3. 网关串口发送（RS485 半双工，发完切接收）
   ↓
4. 等从站回执
   ↓
5. 回执校验：
   - 原样回显 → 写成功
   - 返回 8X + 功能码 → 异常（01不支持/02地址错/03数据非法）
   ↓
6. 回读确认：再用 03 读一次运行模式/状态字
   ↓
7. 平台状态机更新：running / stopped / fault

# hvac_start_stop.py
import time
from pymodbus.client import ModbusSerialClient

class PurifierController:
    def __init__(self, port, slave=1, baudrate=9600, parity="N", timeout=1.5):
        self.slave = slave
        self.client = ModbusSerialClient(
            port=port, baudrate=baudrate, parity=parity,
            stopbits=1, bytesize=8, timeout=timeout
        )
        # 点表：不同厂家改这里
        self.REG = {
            "run_mode": 0,      # 40001
            "remote_enable": 14, # 40015
            "power_ctrl": 1,     # 40002（若厂家用保持寄存器做启停）
        }
        self.COIL = {
            "power_coil": 0,     # 线圈 00001
        }

    def connect(self):
        return self.client.connect()

    # ---- 线圈启停 ----
    def start_by_coil(self):
        rr = self.client.write_coil(self.COIL["power_coil"], True, slave=self.slave)
        if rr.is_error():
            raise RuntimeError(f"coil start error: {rr}")
        return self._confirm_running()

    def stop_by_coil(self):
        rr = self.client.write_coil(self.COIL["power_coil"], False, slave=self.slave)
        if rr.is_error():
            raise RuntimeError(f"coil stop error: {rr}")
        return self._confirm_stopped()

    # ---- 保持寄存器启停 ----
    def start_by_register(self):
        # 先开远程权限
        self.client.write_register(self.REG["remote_enable"], 1, slave=self.slave)
        # 写运行模式=自动
        rr = self.client.write_register(self.REG["run_mode"], 1, slave=self.slave)
        if rr.is_error():
            raise RuntimeError(f"register start error: {rr}")
        return self._confirm_running()

    def stop_by_register(self):
        rr = self.client.write_register(self.REG["run_mode"], 0, slave=self.slave)
        if rr.is_error():
            raise RuntimeError(f"register stop error: {rr}")
        return self._confirm_stopped()

    # ---- 回读确认 ----
    def _confirm_running(self, retry=3):
        for _ in range(retry):
            rr = self.client.read_holding_registers(self.REG["run_mode"], 1, slave=self.slave)
            if not rr.is_error():
                mode = rr.registers[0]
                status = self.read_status_word()
                return {
                    "mode": mode,
                    "running": mode != 0 and "running" in status,
                    "status": status,
                }
            time.sleep(0.5)
        return {"running": False, "status": ["read_back_failed"]}

    def _confirm_stopped(self, retry=3):
        for _ in range(retry):
            rr = self.client.read_holding_registers(self.REG["run_mode"], 1, slave=self.slave)
            if not rr.is_error():
                mode = rr.registers[0]
                return {"mode": mode, "stopped": mode == 0}
            time.sleep(0.5)
        return {"stopped": False, "status": ["read_back_failed"]}

    def read_status_word(self):
        rr = self.client.read_holding_registers(8, 1, slave=self.slave)  # 40009 状态字
        if rr.is_error():
            return ["read_error"]
        w = rr.registers[0]
        bits = {0:"running",1:"fault",2:"humidifying",3:"dehumidifying",
                4:"purifying",5:"disinfecting"}
        return [v for k,v in bits.items() if w & (1<<k)]

湿度 > 60%RH  → 启动除湿
湿度 < 52%RH  → 停止除湿
52~60 之间：保持现状

湿度 > 60%RH 且 持续 5 分钟 → 才下发启动
湿度 < 52%RH 且 持续 10 分钟 → 才下发停止

单台最少运行 10 分钟
两次启停间隔 ≥ 3 分钟
多台同时动作 ≤ 3 台，其余按地址错峰 30s

if rh > 60 and since_over_high > 300:
    if now - last_switch > 180:
        controller.start_by_register()
        last_switch = now

if rh < 52 and since_under_low > 600:
    if now - last_switch > 180:
        controller.stop_by_register()
        last_switch = now