Danil Iarosh
448b9bb07a
- Добавлена поддержка выбора произвольного цвета (QColorDialog) с отправкой команды 8008RRRGGGBBB. - Реализована панель управления GPIO25/GPIO26 (команда 50) в output_test_dialog.py. - Добавлена поддержка direct-объектов и эффектов Бродяги морей (команда 81) в led_test_dialog.py. - Реализованы тесты матриц и direct-шкалы (команда 85) в led_test_dialog.py. - Создан новый диалог prize_showcase_dialog.py для ячеек призовой витрины (команда 25). - Добавлена кнопка «Витрина призов» в PRO-кнопки connection_tab.py. - В payment_section.py добавлен показ накопительного счетчика монет global_summa (команда 93).
892 lines
38 KiB
Python
892 lines
38 KiB
Python
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_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._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-событиям."""
|
||
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:
|
||
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()
|
||
try:
|
||
self._connection.open(port=port, params=params)
|
||
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._set_status(self._conn_info, PlateState.NEED_AUTH)
|
||
|
||
tab_index = self._tab_index_getter()
|
||
if tab_index >= 0:
|
||
self._set_tab_title(tab_index, port)
|
||
|
||
# Проверяем авторизацию один раз при подключении
|
||
self._check_auth_on_connect()
|
||
|
||
# Запускаем периодический health-check (если включён в настройках)
|
||
self._apply_health_settings()
|
||
|
||
# Если PRO выключен — запускаем опрос входов в inline-секции
|
||
if not self._pro_toggle.isChecked():
|
||
self._start_basic_polling()
|
||
|
||
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()
|
||
# Дождаться завершения всех фоновых потоков
|
||
for thread in self._active_threads:
|
||
if thread.isRunning():
|
||
thread.quit()
|
||
thread.wait(1000)
|
||
self._active_threads.clear()
|
||
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()
|
||
port = self._connection.serial.port if self._connection.serial else None
|
||
if port is None:
|
||
return
|
||
params = self._collect_params()
|
||
self._connection.close()
|
||
self._update_plate_state(PlateState.DISCONNECTED)
|
||
try:
|
||
self._connection.open(port=port, params=params)
|
||
except Exception as exc:
|
||
QMessageBox.critical(self, "Переподключение", f"Не удалось открыть {port}\n{exc}")
|
||
return
|
||
self._set_status(self._conn_info, PlateState.NEED_AUTH)
|
||
self._check_auth_on_connect()
|
||
self._apply_health_settings()
|
||
if not self._pro_toggle.isChecked():
|
||
self._start_basic_polling()
|
||
|
||
# ── Открытие диалогов ───────────────────────────────────────────────
|
||
|
||
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
|
||
self._exec_dialog(
|
||
"terminal",
|
||
lambda: CommandTerminalDialog(self._connection, parent=self),
|
||
)
|
||
|
||
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: "Готово",
|
||
}
|
||
_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 _check_auth_on_connect(self) -> None:
|
||
"""Проверить статус авторизации сразу после подключения (в фоновом потоке)."""
|
||
self._run_status_check(timeout=3)
|
||
|
||
def _run_status_check(self, timeout: float = 0.5) -> None:
|
||
"""Запуск проверки статуса в фоновом потоке."""
|
||
thread = _StatusThread(self._connection, timeout, parent=self)
|
||
thread.result_ready.connect(self._on_status_checked)
|
||
thread.finished.connect(lambda: self._cleanup_thread(thread))
|
||
self._active_threads.append(thread)
|
||
thread.start()
|
||
|
||
def _cleanup_thread(self, thread: QThread) -> None:
|
||
"""Убрать завершённый поток из списка."""
|
||
if thread in self._active_threads:
|
||
self._active_threads.remove(thread)
|
||
thread.deleteLater()
|
||
|
||
def _on_status_checked(self, status: AuthStatus) -> None:
|
||
if status == AuthStatus.OK:
|
||
self._update_plate_state(PlateState.READY)
|
||
elif status == AuthStatus.NEED_AUTH:
|
||
self._update_plate_state(PlateState.NEED_AUTH)
|
||
else:
|
||
self._update_plate_state(PlateState.DISCONNECTED)
|
||
|
||
def _do_handshake(self) -> None:
|
||
"""Выполнить авторизацию (двойное рукопожатие AES-256) в фоновом потоке."""
|
||
if not self._connection.is_connected:
|
||
QMessageBox.warning(self, "Авторизация", "Порт не подключён.")
|
||
return
|
||
self.auth_btn.setEnabled(False)
|
||
self.auth_btn.setText("Авторизация…")
|
||
|
||
thread = _HandshakeThread(self._connection, parent=self)
|
||
thread.result_ready.connect(self._on_handshake_done)
|
||
thread.finished.connect(lambda: self._cleanup_thread(thread))
|
||
self._active_threads.append(thread)
|
||
thread.start()
|
||
|
||
def _on_handshake_done(self, result: AuthResult) -> None:
|
||
self.auth_btn.setEnabled(True)
|
||
self.auth_btn.setText("Handshake (Авторизация)")
|
||
if result.status == AuthStatus.OK:
|
||
self._update_plate_state(PlateState.READY)
|
||
QMessageBox.information(self, "Авторизация", f"Успешно!\n{result.detail}")
|
||
else:
|
||
self._update_plate_state(PlateState.DISCONNECTED)
|
||
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:
|
||
# Связь жива, плата авторизована (или не требует авторизацию)
|
||
self._health_fail_count = 0
|
||
if self._plate_state != PlateState.READY:
|
||
self._update_plate_state(PlateState.READY)
|
||
elif status == AuthStatus.NEED_AUTH:
|
||
# Связь жива, но нужна авторизация (полный таймаут на 99)
|
||
self._health_fail_count = 0
|
||
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._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._check_auth_on_connect()
|
||
# Если не удалось — ничего не делаем, следующий тик попробует снова
|
||
|
||
|
||
class _StatusThread(QThread):
|
||
"""Фоновый поток для проверки статуса платы."""
|
||
result_ready = Signal(AuthStatus)
|
||
|
||
def __init__(self, connection: ComConnection, timeout: float, parent: QWidget | None = None) -> None:
|
||
super().__init__(parent)
|
||
self._conn = connection
|
||
self._timeout = timeout
|
||
|
||
def run(self) -> None: # noqa: D401
|
||
status = check_status(self._conn, timeout=self._timeout)
|
||
self.result_ready.emit(status)
|
||
|
||
|
||
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()
|
||
self.result_ready.emit(True)
|
||
except Exception:
|
||
self.result_ready.emit(False)
|
||
|
||
|
||
class _HandshakeThread(QThread):
|
||
"""Фоновый поток для авторизации."""
|
||
result_ready = Signal(object)
|
||
|
||
def __init__(self, connection: ComConnection, parent: QWidget | None = None) -> None:
|
||
super().__init__(parent)
|
||
self._conn = connection
|
||
|
||
def run(self) -> None: # noqa: D401
|
||
from ..services.auth_service import perform_handshake as _handshake
|
||
result = _handshake(self._conn, timeout=2.0)
|
||
self.result_ready.emit(result)
|