档案库房多台恒湿净化一体机，基于 RS485 标准协议批量接入平台方案

┌──────────────┐
平台 ◄──TCP──► 串口服务器 ├──┐
                    └──────────────┘  │
                                      ▼
                         A ──── B ──── C ──── D ──── E   ← 手拉手
                        [1#]   [2#]   [3#]   [4#]   [5#]
                        终端电阻120Ω只挂在链路两端(A和E)

理论：最大32个单元负载，最大1200米（9600bps）
实际建议（档案库房项目经验值）：

设备数量 ≤ 16台   → 1200米内无压力，9600bps足够
设备数量 17~24台  → 需确认每台是1/4或1/8负载，否则加485中继器
设备数量 > 24台   → 拆分为2条总线（如按区域或楼层）

波特率选择：
  9600  → 默认，1200米内稳定
  19200 → 50台以内、距离<400米可考虑
   >19200 → 不建议，抗干扰能力差，调试痛苦

规则：每条总线内地址唯一，1~247
建议：地址 = 区域号×10 + 序号
  1#库房：11, 12, 13 ...
  2#库房：21, 22, 23 ...

⚠️ 现场常踩的坑：
  □ 多台设备出厂默认地址都是1 → 总线全是错帧
  □ 地址拨码开关是二进制但读反了
  □ 改完地址没断电重启，实际未生效
  → 规范：上电前用Modbus Poll逐台改地址，改完断电再级联

# rs485_bus.py - RS485总线管理器（精简版）

import socket
import struct
import time
import threading
from typing import Dict, Optional

def crc16_rtu(data: bytes) -> bytes:
    crc = 0xFFFF
    for b in data:
        crc ^= b
        for _ in range(8):
            crc = (crc >> 1) ^ 0xA001 if crc & 1 else crc >> 1
    return struct.pack('<H', crc)

class RS485Bus:
    """一条RS485总线 = 1个TCP连接（串口服务器透明模式）"""

    def __init__(self, host, port, timeout=1.5, inter_frame=0.02):
        self.host, self.port = host, port
        self.timeout = timeout
        self.inter_frame = inter_frame  # 帧间隔，防粘包
        self.sock: Optional[socket.socket] = None
        self._lock = threading.Lock()

    def connect(self):
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.sock.settimeout(self.timeout)
        self.sock.connect((self.host, self.port))

    def close(self):
        if self.sock:
            self.sock.close()
            self.sock = None

    def _request(self, slave, fc, addr, count=1, value=None) -> Optional[bytes]:
        with self._lock:  # 整条总线串行访问
            if not self.sock:
                self.connect()
            if fc == 0x03:
                body = struct.pack('>HHBHH', 0, 0, 6, slave, fc, addr, count)
            elif fc == 0x06:
                body = struct.pack('>HHHHBHH', 0, 0, 6, slave, fc, addr, value)
            else:
                return None
            frame = body + crc16_rtu(body)
            self.sock.send(frame)
            time.sleep(self.inter_frame)
            # RTU靠时间判定帧尾，收不够就超时
            buf = b''
            deadline = time.time() + self.timeout
            while time.time() < deadline:
                try:
                    chunk = self.sock.recv(256)
                    if not chunk:
                        break
                    buf += chunk
                    if len(buf) >= 5 and len(buf) >= (5 + buf[2]):
                        break
                except socket.timeout:
                    break
            if len(buf) < 5:
                return None
            if buf[-2:] != crc16_rtu(buf[:-2]):
                return None
            return buf

    def read_holding(self, slave, addr, count=1):
        return self._request(slave, 0x03, addr, count)

    def write_single(self, slave, addr, value):
        return self._request(slave, 0x06, addr, value=value) is not None

# polling_scheduler.py - 批量轮询（关键在调度策略）

import asyncio
from dataclasses import dataclass, field

@dataclass
class RTUDevice:
    slave_id: int
    room: str
    model: str
    reg_map: dict = field(default_factory=dict)
    online: bool = True
    last_seen: float = 0.0
    fail_count: int = 0

class PollingScheduler:
    """多总线、多设备批量轮询，带失败重试与离线判定"""

    def __init__(self, buses: Dict[str, 'RS485Bus'], devices: list):
        self.buses = buses          # bus_id -> RS485Bus
        self.devices: Dict[str, RTUDevice] = {d.slave_id: d for d in devices}
        self.offline_threshold = 3  # 连续3次失败判离线
        self.scan_interval = 10     # 全量扫描周期(秒)

    async def poll_device(self, bus_id: str, dev: RTUDevice):
        bus = self.buses[bus_id]
        try:
            resp = bus.read_holding(dev.slave_id, dev.reg_map['hum_return'], 2)
            if resp:
                dev.online = True
                dev.fail_count = 0
                dev.last_seen = time.time()
                # 解析：return (value, )
                return self._parse(resp, dev.reg_map)
            else:
                raise IOError("no response")
        except Exception:
            dev.fail_count += 1
            if dev.fail_count >= self.offline_threshold:
                dev.online = False
            return None

    async def run(self):
        """全量轮询循环；设备多时分片扫描避免单次周期过长"""
        while True:
            tasks = []
            for bus_id, bus in self.buses.items():
                for dev in self._devices_on_bus(bus_id):
                    tasks.append(self.poll_device(bus_id, dev))
            await asyncio.gather(*tasks, return_exceptions=True)
            await asyncio.sleep(self.scan_interval)

    def _devices_on_bus(self, bus_id):
        return [d for d in self.devices.values() if d.bus_id == bus_id]

    def _parse(self, resp, reg_map):
        # 简化：按寄存器表解析，实际需处理字节序
        byte_count = resp[2]
        return struct.unpack(f'>{byte_count//2}H', resp[3:3+byte_count])

Step 1 单台验证（脱离总线）
  用Modbus Poll直连单台设备，核对全部寄存器读/写
  目的：确认寄存器表正确、设备本身正常

Step 2 地址规划与改址
  逐台上电改从站地址，改完断电、记录地址贴标签
  目的：杜绝地址冲突

Step 3 总线级联
  从首端一台一台往下挂，每挂一台用Modbus Poll扫描全部
  目的：实时发现接反、终端电阻错误

Step 4 平台联调
  跑全量轮询，核对数据一致性、写入回读、告警联动
  目的：验证批量接入逻辑