增加手工rtu测试

backend/.env

@@ -16,7 +16,7 @@ EMCP_MODBUS_FLOAT_BYTE_ORDER=big
 EMCP_MODBUS_FLOAT_WORD_ORDER=big
 
 # RTU settings
-EMCP_MODBUS_SERIAL_DEVICE=/dev/ttyS1
+EMCP_MODBUS_SERIAL_DEVICE=/dev/ttyS4
 EMCP_MODBUS_SERIAL_BAUD=115200
 EMCP_MODBUS_SERIAL_PARITY=N
 EMCP_MODBUS_SERIAL_DATA_BITS=8

backend/scripts/manual_rtu_read.py

@@ -0,0 +1,111 @@
+from __future__ import annotations
+
+import sys
+import traceback
+from pathlib import Path
+
+
+BASE_DIR = Path(__file__).resolve().parents[1]
+if str(BASE_DIR) not in sys.path:
+    sys.path.insert(0, str(BASE_DIR))
+
+from app.adapters.device_client import CDeviceClient
+from app.core.config import settings
+
+
+def configure_runtime() -> None:
+    """手工联机脚本固定走 RTU，直接复用 .env 中的串口参数。"""
+    settings.use_mock_device = False
+    settings.modbus_transport = "rtu"
+
+
+def print_runtime_config() -> None:
+    print("==== EMCP RTU 手工联机读取 ====")
+    print(f"串口设备: {settings.modbus_serial_device}")
+    print(f"波特率:   {settings.modbus_serial_baud}")
+    print(f"校验位:   {settings.modbus_serial_parity}")
+    print(f"数据位:   {settings.modbus_serial_data_bits}")
+    print(f"停止位:   {settings.modbus_serial_stop_bits}")
+    print(f"从机地址: {settings.modbus_slave_id}")
+    print(f"超时时间: {settings.modbus_timeout_seconds}s")
+    print()
+
+
+def print_realtime_summary(client: CDeviceClient) -> None:
+    realtime = client.read_realtime_data()
+    print("---- 实时数据摘要 ----")
+    for line in realtime.line_list:
+        sec = line.sec_val
+        print(
+            f"线路{line.line_no}: "
+            f"Ua={sec.Ua:.3f}, Ub={sec.Ub:.3f}, Uc={sec.Uc:.3f}, "
+            f"Ia={sec.Ia:.3f}, Ib={sec.Ib:.3f}, Ic={sec.Ic:.3f}, "
+            f"Pt={sec.Pt:.3f}, Qt={sec.Qt:.3f}, St={sec.St:.3f}, Frq={sec.frq:.3f}"
+        )
+    print()
+
+    print("---- AI 采集 ----")
+    ai_text = ", ".join(f"{key}={value:.3f}" for key, value in realtime.ai_collect.items())
+    print(ai_text)
+    print()
+
+    print("---- 开入/开出状态 ----")
+    switch_text = ", ".join(f"{key}={value}" for key, value in realtime.switch.items())
+    print(switch_text)
+    print()
+
+
+def print_status_summary(client: CDeviceClient) -> None:
+    status = client.read_device_status()
+    print("---- 装置状态 ----")
+    print(
+        f"自检={status.self_check}, "
+        f"网口1={status.net1}, 网口2={status.net2}, "
+        f"串口1={status.uart1}, 串口2={status.uart2}"
+    )
+    print()
+
+
+def print_alarm_summary(client: CDeviceClient) -> None:
+    alarms = client.read_alarm_events()
+    print("---- 最新事件 ----")
+    if not alarms:
+        print("当前无新事件")
+    else:
+        for alarm in alarms:
+            print(
+                f"time={alarm.time}, no={alarm.no}, type={alarm.type}, "
+                f"level={alarm.level}, content={alarm.content}"
+            )
+    print()
+
+
+def close_client(client: CDeviceClient) -> None:
+    """尽量关闭底层 pymodbus 连接，避免串口被长期占用。"""
+    transport = getattr(client, "transport", None)
+    raw_client = getattr(transport, "_client", None)
+    close = getattr(raw_client, "close", None)
+    if callable(close):
+        close()
+
+
+def main() -> int:
+    configure_runtime()
+    print_runtime_config()
+    client = CDeviceClient()
+    try:
+        print_status_summary(client)
+        print_realtime_summary(client)
+        print_alarm_summary(client)
+        print("RTU 联机读取完成。")
+        return 0
+    except Exception as exc:
+        print(f"RTU 联机读取失败: {exc}", file=sys.stderr)
+        traceback.print_exc()
+        return 1
+    finally:
+        close_client(client)
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())