core/connection_manager.py - master - DimerTrue/ecg_gui

core/connection_manager.py
207 строк9 KB
dimer
add status connection bar
21 июл 2026, 15:22
21 июл 2026, 15:22245fab6
# connection_manager.py
from core.app_logger import AppLogger

logger = AppLogger.get_logger(__name__)

from typing import List, Tuple
from modbus_device.device import utils
from models.common import MessageType
from config.write_data_config import VALID_BAUDRATES


class ConnectionManager:
    def __init__(self, status_callback=None):
        self.ser = None
        self.devices = []
        self.com_port = None

        # status_callback пробрасывается глубже, чтобы мониторить текущий статус подключения
        self.status_callback = status_callback

    def _update_status(self, port: str, baudrate: str, id: str):
        """Обновление статуса через колбэк"""
        if self.status_callback:
            self.status_callback(port, baudrate, id)

    def try_manual_connect(
        self, com_port, baudrate, device_count, use_simulator
    ) -> Tuple[bool, str, MessageType]:
        """Подключение к устройству с выбранными скоростью и портом"""
        message = ""
        self.com_port = com_port

        try:
            if self.com_port:
                self.ser, devices = utils.connect_with_finding_id(
                    com_port,
                    baudrate,
                    device_count,
                    use_simulator,
                    status_callback=self._update_status,
                )

                if self.ser is None:
                    message = f"Порт {com_port} занят!"
                    logger.warning(f"Порт {com_port} занят!")

                    return False, message, MessageType.ERROR

                if devices is None:
                    message = f"Не удалось найти устройство! Проверьте подключение."
                    logger.warning(
                        f"Не удалось найти устройство! Проверьте подключение."
                    )
                    return False, message, MessageType.ERROR

                message_type = None
                # Проверяем количество найденных устройств
                if len(devices) != device_count:
                    actual = len(devices)
                    expected = device_count

                    if actual > expected:
                        message = f"Найдено {actual} устройств, используется {expected}"
                        logger.warning(
                            f"Найдено {actual} устройств, используется {expected}"
                        )

                        devices = devices[:expected]
                    else:
                        message = f"Найдено {actual} устройств из {expected} ожидаемых"
                        logger.warning(
                            f"Найдено {actual} устройств из {expected} ожидаемых"
                        )

                    message_type = MessageType.INFO
                else:
                    message_type = MessageType.SUCCESS

                # Успешное подключение
                self.devices = devices

                logger.warning(f"Успешно подключено к {com_port} ({baudrate} бод)")

                return True, message, message_type
            # Порт не найден, создаем симуляторы
            else:
                if use_simulator:
                    self.devices = utils.create_simulator_devices(device_count)
                    message = f"Реальные устройства не найдены. Запуск {device_count} симуляторов."
                    logger.warning(
                        f"Реальные устройства не найдены. Запуск {device_count} симуляторов."
                    )

                    return True, message, MessageType.INFO
                else:
                    message = f"Порт не обнаружен!"
                    logger.warning(f"Порт не обнаружен и симулятор отключен!")
                    return False, message, MessageType.ERROR

        except Exception as e:
            message = str(e)
            logger.exception(f"Ошибка запуска")
            return False, message, MessageType.ERROR

    def try_auto_connect(
        self, found_ports: List[str], device_count: int, use_simulator: bool
    ) -> Tuple[bool, str, MessageType]:
        """Подключение к устройству со сканированием всех портов и скоростей"""
        message = ""

        try:
            baudrates_list = [int(key) for key in VALID_BAUDRATES]

            if found_ports:
                for port in found_ports:
                    logger.info(f"Сканирование порта: {port}")

                    # Пытаемся найти устройства на порту
                    ser, baudrate, devices = utils.autodetect_baudrate(
                        port,
                        baudrates_list,
                        device_count,
                        use_simulator=use_simulator,
                        status_callback=self._update_status,
                    )

                    if ser is None:
                        logger.warning(f"Не удалось открыть порт {port}")
                        continue

                    if not devices:
                        logger.warning(f"Устройства не найдены на порту {port}")
                        ser.close()
                        continue

                    self.ser = ser

                    # Нормализация количества устройств
                    actual_count = len(devices)
                    if actual_count != device_count:
                        if actual_count > device_count:
                            message = f"Найдено {actual_count} устройств, используется {device_count}"
                            devices = devices[:device_count]
                            logger.warning(
                                f"Найдено {actual_count} устройств, используется {device_count}"
                            )

                        else:
                            message = f"Найдено {actual_count} из {device_count}. Остальные будут симуляторами."
                            logger.warning(
                                f"Найдено {actual_count} из {device_count}. Остальные будут симуляторами."
                            )

                        message_type = MessageType.INFO
                    else:
                        message = f"Успешно подключено к {port} ({baudrate} бод)"
                        logger.warning(f"Успешно подключено к {port} ({baudrate} бод)")

                        message_type = MessageType.SUCCESS

                    self.devices = devices
                    self.com_port = port

                    logger.info(f"Количество устройств: {device_count}")
                    return True, message, message_type

                # Если цикл закончился, значит ни один порт не подошел
                if use_simulator:
                    self.devices = utils.create_simulator_devices(device_count)
                    self.com_port = None
                    logger.warning(
                        f"Реальные устройства не найдены. Запуск симуляторов."
                    )
                    return (
                        True,
                        "Реальные устройства не найдены. Запуск симуляторов.",
                        MessageType.INFO,
                    )
                else:
                    logger.warning(f"Не удалось подключиться ни к одному порту!")
                    return (
                        False,
                        "Не удалось подключиться ни к одному порту!",
                        MessageType.ERROR,
                    )

            # Портов вообще нет в системе
            else:
                if use_simulator:
                    self.devices = utils.create_simulator_devices(device_count)
                    self.com_port = None
                    return (
                        True,
                        "COM-порты не обнаружены. Запуск симуляторов.",
                        MessageType.INFO,
                    )
                else:
                    return False, "COM-порты не обнаружены!", MessageType.ERROR

        except Exception as e:
            logger.exception(f"Критическая ошибка при сканировании!")

            return (
                False,
                f"Критическая ошибка при сканировании: {str(e)}",
                MessageType.ERROR,
            )