diff --git a/backend/.env b/backend/.env index 6f01075..d8cab6d 100644 --- a/backend/.env +++ b/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 diff --git a/backend/scripts/manual_rtu_read.py b/backend/scripts/manual_rtu_read.py new file mode 100644 index 0000000..c1f1d3d --- /dev/null +++ b/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())