RS485总线布线注意事项：档案库房恒湿净化一体机部署与平台通讯优化

RS485 最大距离（理论值，实际打 7 折）：

线径       9600bps    19200bps   38400bps
0.5mm²    1200m      800m       500m
0.75mm²   1500m      1000m      700m
1.0mm²    1800m      1200m      900m

档案库房实际建议：
  <100m   → 0.5mm² 足够
  100~200m → 0.75mm²
  >200m   → 1.0mm² 或加分光中继器

正确 ✅：
  网关 ── 设备1 ── 设备2 ── 设备3 ── 设备4 ── 终端电阻
        (分支<0.5m)

错误 ❌：
  网关 ──┬── 设备1
         ├── 设备2    （星型，反射严重）
         └── 设备3

  网关 ──────── 设备1
       └────── 设备2   （长分支 >3m，信号反射）

[边缘网关]（位于消控室）
  │ S/FTP 干线，沿桥架
  ▼ 穿防火封堵（防火泥+金属套管）
[分区A：库房1] 设备1─设备2─设备3─终端120Ω
  ▼ 穿墙
[分区B：库房2] 设备4─设备5─设备6─终端120Ω
  ▼ 穿墙
[分区C：特藏库] 设备7─设备8─终端120Ω
  ▼
[分区D：缓冲间] 设备9─设备10─设备11─设备12─终端120Ω

每段独立终端电阻，段间通过隔离中继器连接

正确接法：
  物理最远端设备 A-B 之间并联 120Ω（±1%精度）
  次远端设备 A-B 之间并联 120Ω
  中间所有设备 A-B 不接电阻

测量验证：
  断电，在总线任意位置测 A-B 电阻：
  - 两端终端 = 60Ω（120∥120）✅
  - 全关 = 开路（∞）✅（短距离）
  - 全开 = <60Ω ❌ 需逐个排查

正确 ✅：单端接地
  屏蔽层 ── 接主站侧（网关/串口服务器）保护地
  从站侧 ── 屏蔽层悬空（用热缩管包裹）

错误 ❌：两端接地
  不同接地点电位差 → 地环流 → 干扰信号叠加在 A-B 差分上

规划表模板：

设备编号  │ 物理位置      │ RS485地址 │ 波特率  │ 校验  │ 分区
──────────┼───────────────┼──────────┼────────┼──────┼──────
TH-01     │ 库房1-门口    │ 11       │ 9600   │ 8N1  │ A
TH-02     │ 库房1-中部    │ 12       │ 9600   │ 8N1  │ A
TH-03     │ 库房1-深处    │ 13       │ 9600   │ 8N1  │ A
TH-04     │ 库房2-门口    │ 21       │ 9600   │ 8N1  │ B
...

def scan_bus(port, baudrate=9600):
    """广播探测法排查地址冲突"""
    found = {}
    for addr in range(1, 248):
        try:
            resp = modbus_read(port, addr, baudrate, reg=0, count=1, timeout=0.3)
            if resp:
                if resp in found.values():
                    print(f"⚠️ 地址 {addr} 响应与已有设备相同，可能冲突")
                found[addr] = resp
        except:
            pass
    return found

class RobustPoller:
    def __init__(self, timeout=0.5, retries=2, backoff=1.5):
        self.timeout = timeout
        self.retries = retries
        self.backoff = backoff
        self.stats = {'total': 0, 'success': 0, 'crc_err': 0, 'timeout': 0}

    async def poll_with_retry(self, device, reg, count):
        self.stats['total'] += 1
        for attempt in range(self.retries + 1):
            try:
                result = await asyncio.wait_for(
                    self._do_poll(device, reg, count),
                    timeout=self.timeout
                )
                if result and self._crc_check(result):
                    self.stats['success'] += 1
                    return result
                else:
                    self.stats['crc_err'] += 1
            except asyncio.TimeoutError:
                self.stats['timeout'] += 1
            if attempt < self.retries:
                await asyncio.sleep(self.backoff ** attempt * 0.1)
        return None

class BusHealthMonitor:
    def __init__(self, window=1000):
        self.window = window
        self.crc_errors = 0
        self.total_frames = 0
        self.last_reset = time.time()

    def record(self, success: bool, crc_err: bool = False):
        self.total_frames += 1
        if crc_err:
            self.crc_errors += 1

    def get_health(self) -> dict:
        if self.total_frames == 0:
            return {'crc_error_rate': 0, 'status': 'UNKNOWN'}
        rate = self.crc_errors / self.total_frames
        status = 'HEALTHY' if rate < 0.001 else 'DEGRADED' if rate < 0.01 else 'CRITICAL'
        return {
            'crc_error_rate': round(rate * 100, 3),
            'total_frames': self.total_frames,
            'status': status
        }