恒湿消毒净化一体机 RS485 通讯调试：寄存器定义与平台数据采集实践

0x 线圈       00001~09999   读写位
1x 离散输入   10001~19999   只读位
3x 输入寄存器 30001~39999   只读 16bit
4x 保持寄存器 40001~49999   读写 16bit

协议地址 = 手册点号 - 区基址
40001 → 0
40002 → 1
30001 → 0

大端：reg0 高16位，reg1 低16位
小端：reg0 低16位，reg1 高16位

import struct

def combine_u32(high: int, low: int, big_endian: bool = True) -> int:
    raw = struct.pack('>HH', high, low) if big_endian else struct.pack('<HH', high, low)
    return struct.unpack('>I' if big_endian else '<I', raw)[0]

恒湿消毒净化一体机（RS485/Modbus RTU）
        │
        ▼
RS485→串口/以太网网关（Modbus RTU→TCP 或 MQTT）
        │
        ▼
边缘采集服务（点表解析 / 字节序 / 缩放 / 合理性校验）
        │
        ▼
动环平台（设备模型 + 实时值 + 告警 + 控制下发）

# hvac_purifier_collector.py
import time
import struct
from typing import Dict, Any
from pymodbus.client import ModbusSerialClient

POINT_TABLE = {
    "run_mode":        {"addr": 0,  "fc": 3, "type": "u16", "scale": 1,   "unit": ""},
    "target_rh":       {"addr": 1,  "fc": 3, "type": "u16", "scale": 0.1, "unit": "%RH"},
    "room_rh":         {"addr": 6,  "fc": 3, "type": "u16", "scale": 0.1, "unit": "%RH"},
    "room_temp":       {"addr": 7,  "fc": 3, "type": "i16", "scale": 0.1, "unit": "°C"},
    "status_word":     {"addr": 8,  "fc": 3, "type": "u16", "scale": 1,   "unit": ""},
    "fault_code":      {"addr": 9,  "fc": 3, "type": "u16", "scale": 1,   "unit": ""},
    "disinfect_mode":  {"addr": 15, "fc": 3, "type": "u16", "scale": 1,   "unit": ""},
}

STATUS_BITS = {
    0: "running",
    1: "fault",
    2: "humidifying",
    3: "dehumidifying",
    4: "purifying",
    5: "disinfecting",
}

class PurifierCollector:
    def __init__(self, port: str, slave: int, baudrate: int = 9600,
                 parity: str = "N", stopbits: int = 1, timeout: float = 1.5):
        self.slave = slave
        self.client = ModbusSerialClient(
            port=port, baudrate=baudrate, parity=parity,
            stopbits=stopbits, bytesize=8, timeout=timeout
        )

    def connect(self) -> bool:
        return self.client.connect()

    def read_point(self, name: str) -> Any:
        cfg = POINT_TABLE[name]
        rr = self.client.read_holding_registers(cfg["addr"], count=1, slave=self.slave)
        if rr.is_error():
            raise RuntimeError(f"{name} read error: {rr}")

        raw = rr.registers[0]
        if cfg["type"] == "i16" and raw >= 32768:
            raw -= 65536

        return raw * cfg["scale"]

    def read_all(self) -> Dict[str, Any]:
        out = {}
        for name in POINT_TABLE:
            try:
                out[name] = self.read_point(name)
            except Exception as e:
                out[name] = None
                out[f"{name}__err"] = str(e)
        out["status_bits"] = self.decode_status(out.get("status_word") or 0)
        return out

    def decode_status(self, word: int) -> list:
        return [label for bit, label in STATUS_BITS.items() if word & (1 << bit)]

    def set_target_rh(self, rh: float):
        """下发目标湿度：%RH → 0.1 精度整数"""
        val = int(rh / 0.1)
        rr = self.client.write_register(POINT_TABLE["target_rh"]["addr"], val, slave=self.slave)
        if rr.is_error():
            raise RuntimeError(f"write target_rh error: {rr}")

- name: room_rh
  slave: 1
  function: 3
  address: 6          # 0-based
  register_type: holding
  data_type: u16
  scale: 0.1
  byte_order: AB
  word_order: ABCD    # 32位才用
  unit: '%RH'
  min: 0
  max: 100
  valid_range: [0, 100]
  poll_interval: 2
  alarm:
    warning_high: 60
    warning_low: 45
    critical_high: 65
    critical_low: 40