modbus_device/device/utils.py - master - DimerTrue/ecg_gui

modbus_device/device/utils.py
345 строк11 KB
dimer
fix: ui bug with rs485box, linux reconnect bug
30 июл 2026, 18:09
30 июл 2026, 18:09afafcc3
# utils.py
from core.app_logger import AppLogger

logger = AppLogger.get_logger(__name__)

import serial
import serial.tools.list_ports
import time
from typing import List, Optional, Tuple, Callable
import sys
import subprocess
from .manager import DeviceModbusRTU, DeviceConfig
from ..protocol.configs import (
    MAX_SLAVE_ID,
    READ_INTERVAL_MS,
    TIMEOUT_CONNECTION,
    TIMEOUT_CONNECTION_LONG,
    SIM_READ_INTERVAL_MS,
)


def is_port_available_windows(port_path: str) -> bool:
    """Проверка доступности порта на Windows"""
    try:
        test_ser = serial.Serial(port_path, baudrate=9600, timeout=0.1)
        test_ser.close()
        return True
    except (serial.SerialException, PermissionError, OSError):
        logger.exception(f"Порт {port_path} недоступен (Permission denied или занят)")
        return False
    except Exception as e:
        logger.exception(f"Ошибка подключения к {port_path}:{e}")
        return False


def is_port_busy_linux(port_path: str) -> bool:
    try:
        r = subprocess.run(
            ["lsof", "-t", port_path],
            capture_output=True,
            text=True,
            timeout=1,
        )
        return r.returncode == 0 and bool(r.stdout.strip())
    except Exception as e:
        logger.exception(f"Ошибка подключения к {port_path}:{e}")
        return False


def find_com_ports() -> List[str]:
    ports = serial.tools.list_ports.comports()
    found_ports = []

    for port in ports:
        if sys.platform == "win32":
            available = is_port_available_windows(port.device)
        else:
            available = not is_port_busy_linux(port.device)

        if available:
            found_ports.append(port.device)

    return found_ports


def connect_serial(
    port: str, baudrate: int, timeout: float = 1.0
) -> Optional[serial.Serial]:
    """Базовое подключение к Serial порту"""
    try:
        ser = serial.Serial(
            port=port,
            baudrate=baudrate,
            bytesize=8,
            parity="N",
            stopbits=1,
            timeout=timeout,
        )
        ser.reset_input_buffer()
        ser.reset_output_buffer()
        time.sleep(0.1)
        return ser
    except Exception as e:
        logger.exception(f"Не удалось открыть порт {port}: {e}")
        return None


def reconnect_serial(ser: Optional[serial.Serial]) -> Optional[serial.Serial]:
    try:
        port = ser.port
        baudrate = ser.baudrate
        timeout = 1.0
        ser.close()
        time.sleep(0.1)

        try:
            ser = serial.Serial(
                port=port,
                baudrate=baudrate,
                bytesize=8,
                parity="N",
                stopbits=1,
                timeout=timeout,
            )
            ser.reset_input_buffer()
            ser.reset_output_buffer()
            time.sleep(0.1)
            return ser
        except Exception as e:
            logger.exception(f"Не удалось открыть порт {port}: {e}")
            return None
    except Exception as e:
        logger.exception(f"Не удалось закрыть порт {port}: {e}")
        return None


def autodetect_baudrate(
    port: str,
    baudrate_list: List[int],
    max_devices: int,
    use_simulator: bool = False,
    status_callback: Optional[Callable] = None,
) -> Tuple[Optional[serial.Serial], Optional[int], Optional[List[DeviceConfig]]]:
    """
    Перебирает скорости и id, пытается прочитать данные.
    Возвращает рабочий device и найденную скорость.
    status_callback - пробрасывается глубже, чтобы мониторить текущий статус подключения
    """

    devices = None
    ser = None

    # Переменная необходима, чтобы достоверно проверить все порты и если не найдено реальных устройств,
    # то создавать симуляторы на последней доступной скорости
    use_simulator_local = False

    # for baudrate in baudrate_list:
    for i in range(len(baudrate_list)):
        logger.warning(f"Пробуем скорость: {baudrate_list[i]}...")

        try:
            # Открываем порт с поиском id
            if i == len(baudrate_list) - 1:
                use_simulator_local = use_simulator

            # TODO: При необходимости поиска N-устройств убираем id_null
            ser, devices = connect_with_finding_id(
                port,
                baudrate_list[i],
                max_devices,
                use_simulator_local,
                id_null=True,
                status_callback=status_callback,
            )

            if ser and devices:

                real_devices_count = len(devices)
                if use_simulator and real_devices_count < max_devices:
                    real_ids = [device.id for device in devices]

                    # Дополняем симуляторами до MAX_DEVICES
                    next_id = 1
                    while len(devices) < max_devices:
                        if next_id not in real_ids:

                            device = DeviceConfig(
                                name=f"Simdevice {next_id}",
                                id=next_id,
                                read_interval_ms=100,
                                simulator=True,
                            )
                            devices.append(device)
                        next_id += 1

                return ser, baudrate_list[i], devices
            else:
                logger.warning(
                    f"Не удалось подключиться к порту на скорости {baudrate_list[i]}"
                )

        except Exception as e:
            logger.exception(
                f"Не удалось подключиться к порту на скорости {baudrate_list[i]}"
            )
            # Безопасное закрытие ресурсов при ошибке
            if ser:
                try:
                    ser.close()
                except:
                    pass
            continue

    logger.warning(
        f"Не удалось найти устройство ни на одной скорости и ни по одному адресу"
    )

    if ser is None:
        return None, None, None
    else:
        return ser, None, None


def connect_with_finding_id(
    port: str,
    baudrate: int,
    max_devices: int,
    use_simulator: bool = False,
    id_null=False,
    status_callback: Optional[Callable] = None,
) -> Tuple[Optional[serial.Serial], Optional[List[DeviceConfig]]]:
    """
    Подключение к устройству с автоматизированным поиском id
    """
    ser = connect_serial(port, baudrate)

    if not ser:
        return None, None

    ser.reset_input_buffer()
    ser.reset_output_buffer()

    # Сканируем id на шине
    devices = autodetect_slave_address(
        ser=ser,
        max_id=MAX_SLAVE_ID,
        max_devices_count=max_devices,
        id_null=id_null,
        use_simulator=use_simulator,
        status_callback=status_callback,
    )

    if devices:
        logger.warning(f"Устройство найдено на скорости {baudrate}!")

        return ser, devices
    else:
        ser.close()
        return ser, None


# TODO: Подумать, как ускорить...
def autodetect_slave_address(
    ser: serial.Serial,
    max_id: int,
    max_devices_count: int,
    id_null: bool = False,
    use_simulator: bool = False,
    status_callback: Optional[Callable] = None,
) -> Optional[List[DeviceConfig]]:
    """
    Сканирует порт на наличие slave id.
    Возвращает список найденных устройств.
    """
    devices = []
    start_id = 1

    # attempts = 3
    attempts = 1
    timeout = TIMEOUT_CONNECTION
    real_ids = []

    if id_null:
        timeout = TIMEOUT_CONNECTION_LONG
        start_id = 0

    for slave_id in range(start_id, max_id + 1):
        if len(devices) >= max_devices_count:
            break

        if status_callback:
            if ser is None:
                status_callback("---", "---", slave_id)
            else:
                status_callback(ser.port, ser.baudrate, slave_id)

        attempt = 0
        success = False

        ser.reset_input_buffer()

        device_rtu = DeviceModbusRTU(ser=ser, slave_id=slave_id, use_simulator=False)

        while attempt < attempts and not success:
            logger.warning(f"slave_id:{slave_id}, attempt:{attempt}")

            is_connected = device_rtu.ping(timeout)
            if is_connected:
                if slave_id == 0:
                    logger.warning(f"Осторожно, адрес устройства равен {slave_id}")

                device = DeviceConfig(
                    name=f"device slave {slave_id}",
                    id=slave_id,
                    read_interval_ms=READ_INTERVAL_MS,
                    simulator=False,
                )
                devices.append(device)
                real_ids.append(slave_id)
                success = True

            else:
                attempt += 1

        # Если указан id_null, то ищем только одно устройство
        if id_null:
            break
        # device_rtu.disconnect()

    # Если используем симулятор и устройств недостает, то добавляем устройства-симуляторы
    if use_simulator:
        # Дополняем симуляторами до MAX_DEVICES
        next_id = 1
        while len(devices) < max_devices_count:
            if next_id not in real_ids:

                device = DeviceConfig(
                    name=f"Simdevice {next_id}",
                    id=next_id,
                    read_interval_ms=SIM_READ_INTERVAL_MS,
                    simulator=True,
                )
                devices.append(device)
            next_id += 1

    if len(devices) == 0:
        ser.close()
        return None

    return devices


def create_simulator_devices(device_count: int) -> List[DeviceConfig]:
    devices = []

    for i in range(1, device_count + 1):
        device = DeviceConfig(
            name=f"Simdevice {i}",
            id=i,
            read_interval_ms=100,
            simulator=True,
        )
        devices.append(device)

    return devices