emcp/backend/scripts/manual_rtu_read.py

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())