恒湿净化一体机长期稳定运行：RS485通讯链路定期检测与平台校验方案

总负载 = 120Ω // 120Ω // (Σ 接收器输入阻抗)
       = 60Ω  //  Rin_eq

测法：断电，断开一端，万用表测 A-B：
- 约 120Ω → 只有一个终端（漏接一个）
- 约 60Ω  → 两端正常
- 约 40Ω  → 三个终端（中间某台误开）
- 约 30Ω  → 四个终端（严重错误）

# rs485_health_check.py
import serial
import time
from pymodbus.client import ModbusSerialClient
from pymodbus.exceptions import ModbusException

class RS485HealthCheck:
    def __init__(self, port, baudrate=9600, timeout=1.0):
        self.client = ModbusSerialClient(
            port=port, baudrate=baudrate,
            parity='N', stopbits=1, timeout=timeout
        )
        self.client.connect()

    def check_node(self, unit_id, addr, count=2, trials=10):
        """对单节点发 N 次读请求，统计质量指标"""
        stats = {"ok": 0, "crc_err": 0, "timeout": 0, "exc_resp": 0,
                 "rtt_list": []}
        for _ in range(trials):
            t0 = time.perf_counter()
            try:
                rr = self.client.read_holding_registers(addr, count, slave=unit_id)
                rtt = (time.perf_counter() - t0) * 1000
                if rr.isError():
                    if "CRC" in str(rr):
                        stats["crc_err"] += 1
                    else:
                        stats["exc_resp"] += 1
                else:
                    stats["ok"] += 1
                    stats["rtt_list"].append(rtt)
            except Exception:
                stats["timeout"] += 1
        return stats

    def bus_scan(self, unit_ids, addr=0, count=2):
        """全链路扫描，输出体检报告"""
        report = {}
        for uid in unit_ids:
            s = self.check_node(uid, addr, count)
            rate = s["ok"] / (s["ok"]+s["crc_err"]+s["timeout"]+s["exc_resp"])
            avg_rtt = sum(s["rtt_list"])/len(s["rtt_list"]) if s["rtt_list"] else 0
            report[uid] = {
                "ok": s["ok"], "crc": s["crc_err"], "to": s["timeout"],
                "exc": s["exc_resp"], "success_rate": round(rate*100, 1),
                "avg_rtt_ms": round(avg_rtt, 1),
            }
        return report

# platform_verify.py

class PlatformVerifier:
    def __init__(self, modbus_client, platform_api):
        self.mb = modbus_client      # 直连 RS485 读真实值
        self.api = platform_api      # 平台查询接口

    def verify_live(self, unit_id, node_id):
        """实时值比对：Modbus 直读 vs 平台显示"""
        mb_val = self.mb.read_holding_registers(0, 2, slave=unit_id)
        plat_val = self.api.get_latest(node_id)
        t_mb = mb_val.registers[0] * 0.1
        t_pf = plat_val["temp"]
        return {
            "node": node_id,
            "direct": t_mb,
            "platform": t_pf,
            "diff": abs(t_mb - t_pf),
            "pass": abs(t_mb - t_pf) <= 0.3    # 0.3℃ 容差
        }

    def verify_threshold(self, unit_id, node_id):
        """阈值一致性：平台设定值 vs 设备实际寄存器"""
        mb_th = self.mb.read_holding_registers(0x0064, 1, slave=unit_id)
        plat_th = self.api.get_threshold(node_id)
        return mb_th.registers[0] * 0.1 == plat_th

    def verify_continuity(self, node_id, window_min=60):
        """历史连续性：检查平台数据是否有空洞"""
        pts = self.api.get_history(node_id, window_min)
        ts_list = sorted(p[1] for p in pts)
        gaps = [ts_list[i] - ts_list[i-1] for i in range(1, len(ts_list))
                if ts_list[i] - ts_list[i-1] > 120]   # 超过 2 分钟空洞
        return {"gaps": len(gaps), "gap_list": gaps}

    def verify_quality_flag(self, node_id, window_min=60):
        """质量位：离线/STALE 期间平台是否停止画图"""
        pts = self.api.get_history(node_id, window_min)
        bad_during_stale = [p for p in pts
                           if p["quality"] != "GOOD" and p["value"] is not None]
        return len(bad_during_stale) == 0