from __future__ import annotations from collections.abc import Callable from enum import Enum, auto from PySide6.QtCore import Qt, QThread, QTimer, Signal from PySide6.QtWidgets import ( QCheckBox, QDialog, QFrame, QFormLayout, QGroupBox, QHBoxLayout, QLabel, QMessageBox, QPushButton, QScrollArea, QSizePolicy, QVBoxLayout, QWidget, ) from ..services.com_service import ( PARITY_SHORT_MAP, ComConnection, list_available_ports, params_from_ui, ) from ..services.auth_service import ( AuthResult, AuthStatus, check_status, perform_required_handshake, perform_startup_handshake, ) from ..services import settings_service from ..ui.icons import get_icon_color, get_icon_muted, icon from ..serial_config import SerialParams from .sections.led_test_section import LedTestSection from .sections.button_backlight_section import ButtonBacklightSection from .sections.input_test_section import InputTestSection from .sections.firmware_section import FirmwareSection from .sections.payment_section import PaymentSection from .no_wheel_combo_box import NoWheelComboBox class _CollapsibleGroup(QWidget): """Сворачиваемая группа с заголовком-кнопкой и контентом.""" def __init__(self, title: str, *, expanded: bool = False, parent: QWidget | None = None) -> None: super().__init__(parent) layout = QVBoxLayout(self) layout.setContentsMargins(0, 0, 0, 0) layout.setSpacing(0) self._header = QPushButton(title) self._header.setObjectName("CollapsibleHeader") self._header.setCursor(Qt.CursorShape.PointingHandCursor) self._header.clicked.connect(self._toggle) layout.addWidget(self._header) self._content = QWidget() self._content_layout = QVBoxLayout(self._content) self._content_layout.setContentsMargins(0, 6, 0, 0) self._content_layout.setSpacing(0) layout.addWidget(self._content) self._expanded = expanded self._content.setVisible(expanded) self._update_arrow() def add_widget(self, widget: QWidget) -> None: self._content_layout.addWidget(widget) def set_expanded(self, expanded: bool) -> None: self._expanded = expanded self._content.setVisible(expanded) self._update_arrow() def _toggle(self) -> None: self.set_expanded(not self._expanded) def _update_arrow(self) -> None: arrow = "▼" if self._expanded else "▶" text = self._header.text() # strip any existing arrow prefix for prefix in ("▼ ", "▶ "): if text.startswith(prefix): text = text[len(prefix):] break self._header.setText(f"{arrow} {text}") class PlateState(Enum): """Состояние подключения к плате.""" DISCONNECTED = auto() # красный NEED_AUTH = auto() # жёлтый READY = auto() # зелёный class ConnectionTab(QWidget): def __init__( self, tab_index_getter: Callable[[], int], set_tab_title: Callable[[int, str], None], close_tab: Callable[[int], None], set_tab_indicator: Callable[[int, str], None] | None = None, ) -> None: super().__init__() self._tab_index_getter = tab_index_getter self._set_tab_title = set_tab_title self._close_tab = close_tab self._set_tab_indicator = set_tab_indicator self._connection = ComConnection() self._plate_state = PlateState.DISCONNECTED self._active_threads: list[QThread] = [] self._open_dialogs: int = 0 # кол-во открытых диалоговых окон self._open_dialog_windows: list[QDialog] = [] self._open_dialog_windows_by_key: dict[str, QDialog] = {} self._health_timer: QTimer | None = None # таймер health-check self._health_running = False # идёт ли сейчас проверка self._health_fail_count = 0 # подряд неудачных проверок self._payment_observers_bound = False self._startup_ready = False self._setup_ui() self._load_ports() # Отложить до момента, когда вкладка уже добавлена в QTabBar QTimer.singleShot(0, self._sync_indicator) def _setup_ui(self) -> None: outer_layout = QVBoxLayout(self) outer_layout.setContentsMargins(0, 0, 0, 0) outer_layout.setSpacing(0) scroll = QScrollArea() scroll.setWidgetResizable(True) scroll.setFrameShape(QFrame.Shape.NoFrame) outer_layout.addWidget(scroll) inner = QWidget() scroll.setWidget(inner) root_layout = QVBoxLayout(inner) root_layout.setContentsMargins(18, 18, 18, 18) root_layout.setSpacing(14) # ── Заголовок + PRO-режим ────────────────────────────────── header_row = QHBoxLayout() title = QLabel("Панель соединения") title.setObjectName("SectionTitle") header_row.addWidget(title) header_row.addStretch(1) self._pro_toggle = QCheckBox("PRO-режим") self._pro_toggle.setObjectName("ProToggle") self._pro_toggle.setCursor(Qt.CursorShape.PointingHandCursor) self._pro_toggle.setChecked(False) self._pro_toggle.toggled.connect(self._on_pro_toggled) header_row.addWidget(self._pro_toggle) root_layout.addLayout(header_row) self.status_label = QLabel() self.status_label.setObjectName("StatusBadge") self.status_label.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed) self._set_status("", PlateState.DISCONNECTED) self._reconnect_btn = QPushButton() self._reconnect_btn.setObjectName("ReconnectBtn") self._reconnect_btn.setToolTip("Переподключиться") self._reconnect_btn.setIcon(icon("fa5s.sync-alt", color=get_icon_color())) self._reconnect_btn.setFixedSize(34, 34) self._reconnect_btn.setCursor(Qt.CursorShape.PointingHandCursor) self._reconnect_btn.clicked.connect(self._do_reconnect) self._reconnect_btn.hide() status_row = QHBoxLayout() status_row.setContentsMargins(0, 0, 0, 0) status_row.setSpacing(6) status_row.addWidget(self.status_label) status_row.addWidget(self._reconnect_btn) root_layout.addLayout(status_row) self.setup_widget = QFrame() self.setup_widget.setObjectName("PanelCard") setup_layout = QVBoxLayout(self.setup_widget) setup_layout.setContentsMargins(14, 14, 14, 14) setup_layout.setSpacing(10) settings_group = QGroupBox("Настройки соединения") settings_form = QFormLayout(settings_group) settings_form.setLabelAlignment(Qt.AlignmentFlag.AlignLeft) self.port_combo = NoWheelComboBox() self.port_combo.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed) self.baud_combo = NoWheelComboBox() self.baud_combo.addItems(["9600", "19200", "38400", "57600", "115200", "230400"]) self.baud_combo.setCurrentText("115200") self.data_bits_combo = NoWheelComboBox() self.data_bits_combo.addItems(["5", "6", "7", "8"]) self.data_bits_combo.setCurrentText("8") self.stop_bits_combo = NoWheelComboBox() self.stop_bits_combo.addItems(["1", "1.5", "2"]) self.stop_bits_combo.setCurrentText("1") self.parity_combo = NoWheelComboBox() self.parity_combo.addItems(["None", "Even", "Odd", "Mark", "Space"]) self.parity_combo.setCurrentText("None") self.flow_control_combo = NoWheelComboBox() self.flow_control_combo.addItems(["None", "XON/XOFF", "RTS/CTS"]) self.flow_control_combo.setCurrentText("None") settings_form.addRow("Port", self.port_combo) settings_form.addRow("Baud rate", self.baud_combo) settings_form.addRow("Data bits", self.data_bits_combo) settings_form.addRow("Stop bits", self.stop_bits_combo) settings_form.addRow("Parity", self.parity_combo) settings_form.addRow("Flow control", self.flow_control_combo) setup_layout.addWidget(settings_group) buttons_layout = QHBoxLayout() buttons_layout.addStretch(1) refresh_btn = QPushButton("Обновить порты") refresh_btn.setProperty("role", "ghost") refresh_btn.setIcon(icon("fa5s.sync-alt", color=get_icon_muted())) refresh_btn.clicked.connect(self._load_ports) buttons_layout.addWidget(refresh_btn) self._refresh_btn = refresh_btn connect_btn = QPushButton("OK") connect_btn.setProperty("role", "primary") connect_btn.setIcon(icon("fa5s.check", color="#ecfdf5")) connect_btn.clicked.connect(self._connect) buttons_layout.addWidget(connect_btn) cancel_btn = QPushButton("Cancel") cancel_btn.setProperty("role", "ghost") cancel_btn.setIcon(icon("fa5s.times", color=get_icon_muted())) cancel_btn.clicked.connect(self._close_this_tab) buttons_layout.addWidget(cancel_btn) self._cancel_btn = cancel_btn setup_layout.addLayout(buttons_layout) root_layout.addWidget(self.setup_widget) self.controls_widget = QFrame() self.controls_widget.setObjectName("PanelCard") controls_layout = QVBoxLayout(self.controls_widget) controls_layout.setContentsMargins(14, 14, 14, 14) controls_layout.setSpacing(8) controls_title = QLabel("Действия с платой") controls_title.setObjectName("SectionTitle") controls_layout.addWidget(controls_title) # ── Кнопка авторизации (скрыта по умолчанию, показывается если нужна) ── self.auth_btn = QPushButton("Handshake (Авторизация)") self.auth_btn.setMinimumHeight(38) self.auth_btn.setProperty("role", "primary") self.auth_btn.setIcon(icon("fa5s.key", color="#ecfdf5")) self.auth_btn.clicked.connect(self._do_handshake) controls_layout.addWidget(self.auth_btn) self.auth_btn.hide() # ── Базовые секции (PRO выключен) ── каждая в сворачиваемой группе ── self._firmware_section = FirmwareSection(self._connection) self._firmware_section.upload_started.connect(self._on_fw_upload_started) self._firmware_section.upload_finished.connect(self._on_fw_upload_finished) self._led_section = LedTestSection(self._connection) self._button_backlight_section = ButtonBacklightSection(self._connection) self._input_section = InputTestSection(self._connection) self._payment_section = PaymentSection( connection=self._connection, send_raw=lambda cmd: self._input_section._send_raw(cmd), ) self._firmware_group = _CollapsibleGroup("Обновление прошивки", expanded=True) self._firmware_group.add_widget(self._firmware_section) controls_layout.addWidget(self._firmware_group) self._led_group = _CollapsibleGroup("Тест подсветки", expanded=False) self._led_group.add_widget(self._led_section) self._led_group.add_widget(self._button_backlight_section) controls_layout.addWidget(self._led_group) self._input_group = _CollapsibleGroup("Тест ввода", expanded=False) self._input_group.add_widget(self._input_section) controls_layout.addWidget(self._input_group) self._payment_group = _CollapsibleGroup("Тест оплаты и призов", expanded=False) self._payment_group.add_widget(self._payment_section) controls_layout.addWidget(self._payment_group) self._basic_groups: list[_CollapsibleGroup] = [ self._firmware_group, self._led_group, self._input_group, self._payment_group, ] self._basic_sections: list[QWidget] = [ self._firmware_section, self._led_section, self._button_backlight_section, self._input_section, self._payment_section, ] # ── PRO-кнопки (все действия, скрыты по умолчанию) ───────── pro_items = ( ("Тест ввода", "fa5s.keyboard", self._open_input_test), ("Тест подсветки", "fa5s.lightbulb", self._open_led_test), ("Тест силовых выходов", "fa5s.bolt", self._open_output_test), ("Обновление прошивки", "fa5s.upload", self._open_firmware), ("Задание роли", "fa5s.user-tag", self._open_role), ("Витрина призов", "fa5s.gift", self._open_prize_showcase), ) self._pro_buttons: list[tuple[QPushButton, str]] = [] for title, icon_name, handler in pro_items: btn = QPushButton(title) btn.setMinimumHeight(34) btn.setIcon(icon(icon_name, color=get_icon_color())) btn.clicked.connect(handler) btn.hide() controls_layout.addWidget(btn) self._pro_buttons.append((btn, icon_name)) command_input_btn = QPushButton("Ввод команд") command_input_btn.setMinimumHeight(34) command_input_btn.setProperty("role", "primary") command_input_btn.setIcon(icon("fa5s.terminal", color="#ecfdf5")) command_input_btn.clicked.connect(self._open_terminal) self._command_input_btn = command_input_btn self._command_input_btn.hide() controls_layout.addWidget(command_input_btn) controls_layout.addStretch(1) root_layout.addWidget(self.controls_widget) self.controls_widget.hide() root_layout.addStretch(1) # ── Public helpers ──────────────────────────────────────────── def refresh_icons(self) -> None: """Re-create icons with the current theme colours.""" clr = get_icon_color() muted = get_icon_muted() self._refresh_btn.setIcon(icon("fa5s.sync-alt", color=muted)) self._cancel_btn.setIcon(icon("fa5s.times", color=muted)) self._reconnect_btn.setIcon(icon("fa5s.sync-alt", color=clr)) for btn, icon_name in self._pro_buttons: btn.setIcon(icon(icon_name, color=clr)) self._led_section.refresh_icons() self._button_backlight_section.refresh_icons() self._input_section.refresh_icons() self._firmware_section.refresh_icons() self._payment_section.refresh_icons() def apply_runtime_settings(self) -> None: """Apply updated runtime settings (e.g. health-check) immediately.""" self._apply_health_settings() # ── PRO-режим ───────────────────────────────────────────────── def _start_basic_polling(self) -> None: """Запустить подписку на входы и подключить PaymentSection к reader-событиям.""" if self._plate_state != PlateState.READY: return self._input_section.start_polling() if not self._payment_observers_bound: self._connection.input_received.connect( self._payment_section.on_digital, Qt.ConnectionType.QueuedConnection ) self._connection.info_received.connect( self._payment_section.on_info_line, Qt.ConnectionType.QueuedConnection ) self._payment_observers_bound = True def _on_fw_upload_started(self) -> None: """Остановить опрос входов перед прошивкой — порт нужен целиком.""" self._input_section.stop_polling() # На время прошивки полностью останавливаем фоновое чтение, # чтобы UploadWorker имел эксклюзивный доступ к порту. self._connection.pause_drain() def _on_fw_upload_finished(self) -> None: """Возобновить опрос входов после завершения прошивки.""" if not self._pro_toggle.isChecked(): self._start_basic_polling() else: # В PRO-режиме poll не запускается — вернуть drain self._connection.resume_drain() def _on_pro_toggled(self, checked: bool) -> None: """Переключить видимость базовых секций / PRO кнопок.""" # Базовые inline-группы for group in self._basic_groups: group.setVisible(not checked) # Управление poll-потоком InputTestSection if not checked and self._plate_state == PlateState.READY: self._start_basic_polling() else: self._input_section.stop_polling() # PRO-кнопки for btn, _ in self._pro_buttons: btn.setVisible(checked) self._command_input_btn.setVisible(checked) def _load_ports(self) -> None: current = self.port_combo.currentText() self.port_combo.clear() ports = list_available_ports() if not ports: self.port_combo.addItem("COM1") else: self.port_combo.addItems(ports) if current and self.port_combo.findText(current) != -1: self.port_combo.setCurrentText(current) elif self.port_combo.findText("COM1") != -1: self.port_combo.setCurrentText("COM1") def _collect_params(self) -> SerialParams: return params_from_ui( baudrate=self.baud_combo.currentText(), data_bits=self.data_bits_combo.currentText(), stop_bits=self.stop_bits_combo.currentText(), parity=self.parity_combo.currentText(), flow_control=self.flow_control_combo.currentText(), ) def _connect(self) -> None: port = self.port_combo.currentText().strip() if not port: QMessageBox.warning(self, "Подключение", "Выберите COM-порт.") return params = self._collect_params() self._startup_ready = False try: self._connection.open( port=port, params=params, drain_paused=True, reset_device=True, ) except Exception as exc: QMessageBox.critical(self, "Ошибка подключения", f"Не удалось открыть {port}\n{exc}") return self.setup_widget.hide() self.controls_widget.show() self._reconnect_btn.show() self._conn_info = ( f"{port} ({self.baud_combo.currentText()}-" f"{self.data_bits_combo.currentText()}-" f"{PARITY_SHORT_MAP[self.parity_combo.currentText()]}-" f"{self.stop_bits_combo.currentText()})" ) self._update_plate_state(PlateState.NEED_AUTH) tab_index = self._tab_index_getter() if tab_index >= 0: self._set_tab_title(tab_index, port) # До стартового banner и READY! никакие другие команды не отправляются. self._start_startup_handshake() def _close_this_tab(self) -> None: index = self._tab_index_getter() if index >= 0: self._close_tab(index) def close_serial(self) -> None: self._stop_health_timer() # Закрыть все открытые немодальные диалоги этой вкладки for dlg in list(self._open_dialog_windows): try: dlg.close() except Exception: pass self._open_dialog_windows.clear() self._open_dialog_windows_by_key.clear() self._open_dialogs = 0 # Остановить inline-секции self._input_section.stop_polling() self._button_backlight_section.shutdown() self._firmware_section.cancel_upload() self._connection.close() self._startup_ready = False # Дождаться завершения всех фоновых потоков for thread in self._active_threads: if thread.isRunning(): thread.quit() thread.wait(1000) self._active_threads.clear() self._set_board_actions_enabled(False) self._update_plate_state(PlateState.DISCONNECTED) self._reconnect_btn.hide() def _do_reconnect(self) -> None: """Переподключиться к порту: закрыть → переоткрыть → проверить авторизацию.""" if not self._connection.is_connected: return # Остановить опрос / health self._stop_health_timer() self._input_section.stop_polling() self._set_board_actions_enabled(False) port = self._connection.serial.port if self._connection.serial else None if port is None: return params = self._collect_params() self._connection.close() self._startup_ready = False self._update_plate_state(PlateState.DISCONNECTED) try: self._connection.open( port=port, params=params, drain_paused=True, reset_device=True, ) except Exception as exc: QMessageBox.critical(self, "Переподключение", f"Не удалось открыть {port}\n{exc}") return self._update_plate_state(PlateState.NEED_AUTH) self._start_startup_handshake() # ── Открытие диалогов ─────────────────────────────────────────────── def _apply_dialog_title_with_port(self, dlg: QWidget) -> None: serial = self._connection.serial port = serial.port if serial is not None else "" base_title = dlg.property("_base_title") if not isinstance(base_title, str) or not base_title.strip(): base_title = dlg.windowTitle().strip() dlg.setProperty("_base_title", base_title) base_title = base_title.strip() if not base_title: return if port: dlg.setWindowTitle(f"{base_title} ({port})") else: dlg.setWindowTitle(base_title) def _on_dialog_destroyed(self, _obj=None) -> None: sender = self.sender() if sender is not None and sender in self._open_dialog_windows: self._open_dialog_windows.remove(sender) # type: ignore[arg-type] for key, dlg in list(self._open_dialog_windows_by_key.items()): if dlg is sender: self._open_dialog_windows_by_key.pop(key, None) self._open_dialogs = len(self._open_dialog_windows_by_key) def _exec_dialog(self, key: str, factory: Callable[[], QDialog]) -> None: """Открыть немодальный диалог с трекингом открытых окон.""" existing = self._open_dialog_windows_by_key.get(key) if existing is not None: try: self._apply_dialog_title_with_port(existing) existing.show() existing.raise_() existing.activateWindow() return except RuntimeError: self._open_dialog_windows_by_key.pop(key, None) if existing in self._open_dialog_windows: self._open_dialog_windows.remove(existing) dlg = factory() self._apply_dialog_title_with_port(dlg) dlg.setModal(False) dlg.setWindowModality(Qt.WindowModality.NonModal) dlg.setAttribute(Qt.WidgetAttribute.WA_DeleteOnClose, True) dlg.destroyed.connect(self._on_dialog_destroyed) self._open_dialog_windows.append(dlg) self._open_dialog_windows_by_key[key] = dlg self._open_dialogs = len(self._open_dialog_windows_by_key) dlg.show() dlg.raise_() dlg.activateWindow() def _open_input_test(self) -> None: from .dialogs.input_test_dialog import InputTestDialog self._exec_dialog( "input_test", lambda: InputTestDialog(self._connection, parent=self), ) def _open_led_test(self) -> None: from .dialogs.led_test_dialog import LedTestDialog self._exec_dialog( "led_test", lambda: LedTestDialog(self._connection, parent=self), ) def _open_output_test(self) -> None: from .dialogs.output_test_dialog import OutputTestDialog self._exec_dialog( "output_test", lambda: OutputTestDialog(self._connection, parent=self), ) def _open_firmware(self) -> None: from .dialogs.firmware_dialog import FirmwareDialog self._exec_dialog( "firmware", lambda: FirmwareDialog(self._connection, parent=self), ) def _open_role(self) -> None: from .dialogs.role_dialog import RoleDialog self._exec_dialog( "role", lambda: RoleDialog(self._connection, parent=self), ) def _open_prize_showcase(self) -> None: from .dialogs.prize_showcase_dialog import PrizeShowcaseDialog self._exec_dialog( "prize_showcase", lambda: PrizeShowcaseDialog(self._connection, parent=self), ) def _open_terminal(self) -> None: from .dialogs.command_terminal_dialog import CommandTerminalDialog def create_terminal() -> CommandTerminalDialog: dialog = CommandTerminalDialog(self._connection, parent=self) dialog.authorization_finished.connect(self._on_terminal_authorization) return dialog self._exec_dialog( "terminal", create_terminal, ) def _on_terminal_authorization(self, result: AuthResult) -> None: if result.status == AuthStatus.OK: self._update_plate_state(PlateState.READY) self._set_board_actions_enabled(True) self._health_fail_count = 0 self._apply_health_settings() if not self._pro_toggle.isChecked(): self._start_basic_polling() return state = ( PlateState.NEED_AUTH if self._connection.is_connected else PlateState.DISCONNECTED ) self._input_section.stop_polling() self._set_board_actions_enabled(False) self._update_plate_state(state) def _not_implemented(self) -> None: QMessageBox.information(self, "В разработке", "Эта функция пока не реализована.") def _set_status(self, text: str, state: PlateState) -> None: _STATE_LABELS = { PlateState.DISCONNECTED: "Отключено", PlateState.NEED_AUTH: "Требуется авторизация", PlateState.READY: "Готово", } if state == PlateState.READY and self._connection.auth_required is False: _STATE_LABELS[PlateState.READY] = "Готово без авторизации" _STATE_CSS = { PlateState.DISCONNECTED: "disconnected", PlateState.NEED_AUTH: "need-auth", PlateState.READY: "ready", } status_text = _STATE_LABELS[state] full = f"[{status_text}] {text}" if text else f"[{status_text}]" self.status_label.setText(full) self.status_label.setProperty("state", _STATE_CSS[state]) style = self.status_label.style() style.unpolish(self.status_label) style.polish(self.status_label) # ── Индикация состояния платы ─────────────────────────────────────── def _sync_indicator(self) -> None: """Обновить цвет индикатора в табе.""" if self._set_tab_indicator is None: return idx = self._tab_index_getter() if idx < 0: return color_map = { PlateState.READY: "green", PlateState.NEED_AUTH: "yellow", PlateState.DISCONNECTED: "red", } self._set_tab_indicator(idx, color_map[self._plate_state]) def _update_plate_state(self, state: PlateState) -> None: self._plate_state = state self._sync_indicator() # Обновить строку статуса conn_info = getattr(self, "_conn_info", "") if state == PlateState.DISCONNECTED: self._set_status("", PlateState.DISCONNECTED) else: self._set_status(conn_info, state) # Показать/скрыть кнопку авторизации if state == PlateState.NEED_AUTH: self.auth_btn.show() else: self.auth_btn.hide() def _cleanup_thread(self, thread: QThread) -> None: """Убрать завершённый поток из списка.""" if thread in self._active_threads: self._active_threads.remove(thread) thread.deleteLater() def _start_startup_handshake(self) -> None: """Дождаться загрузочной строки платы и автоматически авторизоваться.""" self._run_handshake(wait_for_startup=True) def _set_board_actions_enabled(self, enabled: bool) -> None: """Запретить команды UI до подтверждённой строки READY!.""" for group in self._basic_groups: group.setEnabled(enabled) for button, _ in self._pro_buttons: button.setEnabled(enabled) self._command_input_btn.setEnabled(enabled) def _do_handshake(self) -> None: """Выполнить авторизацию (двойное рукопожатие AES-256) в фоновом потоке.""" self._run_handshake(wait_for_startup=not self._startup_ready) def _run_handshake(self, *, wait_for_startup: bool) -> None: if not self._connection.is_connected: QMessageBox.warning(self, "Авторизация", "Порт не подключён.") return self._stop_health_timer() self._input_section.stop_polling() self._connection.pause_drain() self._set_board_actions_enabled(False) self.auth_btn.setEnabled(False) if wait_for_startup: self.auth_btn.setText("Ожидание платы…") else: self.auth_btn.setText("Авторизация…") thread = _HandshakeThread( self._connection, wait_for_startup=wait_for_startup, parent=self, ) thread.result_ready.connect( lambda result, automatic=wait_for_startup: self._on_handshake_done( result, automatic=automatic, ) ) thread.finished.connect(lambda: self._cleanup_thread(thread)) self._active_threads.append(thread) thread.start() def _on_handshake_done(self, result: AuthResult, *, automatic: bool) -> None: self._connection.resume_drain() if automatic: self._startup_ready = result.startup_ready self.auth_btn.setEnabled(True) self.auth_btn.setText("Handshake (Авторизация)") if result.status == AuthStatus.OK: self._update_plate_state(PlateState.READY) self._set_board_actions_enabled(True) self._health_fail_count = 0 self._apply_health_settings() if not self._pro_toggle.isChecked(): self._start_basic_polling() if not automatic: QMessageBox.information(self, "Авторизация", f"Успешно!\n{result.detail}") else: state = ( PlateState.NEED_AUTH if self._connection.is_connected else PlateState.DISCONNECTED ) self._update_plate_state(state) QMessageBox.warning( self, "Авторизация", f"Ошибка: {result.detail}\n\n" "Возможно, подключённое устройство не поддерживает " "данную систему команд.", ) # ── Периодический health-check ────────────────────────────────────── _MAX_HEALTH_FAILURES = 3 # закрывать порт после N неудач подряд def _apply_health_settings(self) -> None: """Прочитать настройки и запустить/остановить health-check.""" if settings_service.health_check_enabled(): interval_ms = settings_service.health_check_interval() * 1000 if self._health_timer is not None: # Обновить интервал на лету self._health_timer.setInterval(interval_ms) else: self._start_health_timer(interval_ms) else: self._stop_health_timer() def _start_health_timer(self, interval_ms: int | None = None) -> None: if self._health_timer is not None: return if interval_ms is None: interval_ms = settings_service.health_check_interval() * 1000 self._health_timer = QTimer(self) self._health_timer.setInterval(interval_ms) self._health_timer.timeout.connect(self._on_health_tick) self._health_timer.start() def _stop_health_timer(self) -> None: if self._health_timer is not None: self._health_timer.stop() self._health_timer.deleteLater() self._health_timer = None def _on_health_tick(self) -> None: """Проверка по таймеру: если нет открытых диалогов и прошло достаточно без команд — шлём 99 для проверки связи.""" # Если проверка отключена в настройках — остановить таймер if not settings_service.health_check_enabled(): self._stop_health_timer() return if self._open_dialogs > 0: return # Если InputTestSection активен — не мешаем проверками связи if self._input_section._polling: return if self._health_running: return if not self._connection.is_connected: # Порт закрыт — пробуем переподключиться self._try_reconnect() return idle_threshold = float(settings_service.health_check_interval()) if self._connection.seconds_since_last_command < idle_threshold: return self._health_running = True thread = _HealthCheckThread(self._connection, parent=self) thread.result_ready.connect(self._on_health_result) thread.finished.connect(lambda: self._cleanup_thread(thread)) self._active_threads.append(thread) thread.start() def _on_health_result(self, status: AuthStatus) -> None: self._health_running = False if status == AuthStatus.OK: # Только READY! подтверждает готовность платы. self._health_fail_count = 0 if self._plate_state != PlateState.READY: self._update_plate_state(PlateState.READY) self._set_board_actions_enabled(True) if not self._pro_toggle.isChecked(): self._start_basic_polling() elif status == AuthStatus.NEED_AUTH: # Связь жива, но READY! не получен. self._health_fail_count = 0 self._input_section.stop_polling() self._set_board_actions_enabled(False) if self._plate_state != PlateState.NEED_AUTH: self._update_plate_state(PlateState.NEED_AUTH) else: # Связь, вероятно, потеряна — увеличиваем счётчик self._health_fail_count += 1 if self._health_fail_count >= self._MAX_HEALTH_FAILURES: self._health_fail_count = 0 self._set_board_actions_enabled(False) self._startup_ready = False self._update_plate_state(PlateState.DISCONNECTED) self._connection.close() def _try_reconnect(self) -> None: """Попытаться переподключиться к порту.""" if self._health_running: return self._health_running = True thread = _ReconnectThread(self._connection, parent=self) thread.result_ready.connect(self._on_reconnect_result) thread.finished.connect(lambda: self._cleanup_thread(thread)) self._active_threads.append(thread) thread.start() def _on_reconnect_result(self, success: bool) -> None: self._health_running = False if success: self._health_fail_count = 0 self._startup_ready = False self._update_plate_state(PlateState.NEED_AUTH) self._start_startup_handshake() # Если не удалось — ничего не делаем, следующий тик попробует снова class _HealthCheckThread(QThread): """Быстрая проверка связи с платой (команда 99, таймаут 2 с).""" result_ready = Signal(AuthStatus) def __init__(self, connection: ComConnection, parent: QWidget | None = None) -> None: super().__init__(parent) self._conn = connection def run(self) -> None: # noqa: D401 try: status = check_status(self._conn, timeout=2.0) except Exception: status = AuthStatus.DISCONNECTED self.result_ready.emit(status) class _ReconnectThread(QThread): """Попытка переподключения к порту в фоновом потоке.""" result_ready = Signal(bool) def __init__(self, connection: ComConnection, parent: QWidget | None = None) -> None: super().__init__(parent) self._conn = connection def run(self) -> None: # noqa: D401 try: self._conn.reconnect(drain_paused=True, reset_device=True) self.result_ready.emit(True) except Exception: self.result_ready.emit(False) class _HandshakeThread(QThread): """Фоновый поток стартовой последовательности или повторной авторизации.""" result_ready = Signal(object) def __init__( self, connection: ComConnection, *, wait_for_startup: bool, parent: QWidget | None = None, ) -> None: super().__init__(parent) self._conn = connection self._wait_for_startup = wait_for_startup def run(self) -> None: # noqa: D401 if self._wait_for_startup: result = perform_startup_handshake(self._conn) else: result = perform_required_handshake(self._conn, timeout=2.0) self.result_ready.emit(result)