113 lines
3.4 KiB
Python
113 lines
3.4 KiB
Python
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"event_time={alarm.event_time}, line_code={alarm.line_code}, "
|
||
f"event_type={alarm.event_type}, event_code={alarm.event_code}, "
|
||
f"event_value={alarm.event_value}, 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())
|