550 lines
22 KiB
Python
550 lines
22 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 (
|
||
QComboBox,
|
||
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 ..ui.icons import ICON_COLOR_DARK, ICON_COLOR_MUTED, icon
|
||
from ..serial_config import SerialParams
|
||
|
||
|
||
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._health_timer: QTimer | None = None # таймер health-check
|
||
self._health_running = 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)
|
||
|
||
title = QLabel("Панель соединения")
|
||
title.setObjectName("SectionTitle")
|
||
root_layout.addWidget(title)
|
||
|
||
self.status_label = QLabel()
|
||
self.status_label.setObjectName("StatusBadge")
|
||
self._set_status("", PlateState.DISCONNECTED)
|
||
root_layout.addWidget(self.status_label)
|
||
|
||
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 = QComboBox()
|
||
self.port_combo.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed)
|
||
|
||
self.baud_combo = QComboBox()
|
||
self.baud_combo.addItems(["9600", "19200", "38400", "57600", "115200", "230400"])
|
||
self.baud_combo.setCurrentText("115200")
|
||
|
||
self.data_bits_combo = QComboBox()
|
||
self.data_bits_combo.addItems(["5", "6", "7", "8"])
|
||
self.data_bits_combo.setCurrentText("8")
|
||
|
||
self.stop_bits_combo = QComboBox()
|
||
self.stop_bits_combo.addItems(["1", "1.5", "2"])
|
||
self.stop_bits_combo.setCurrentText("1")
|
||
|
||
self.parity_combo = QComboBox()
|
||
self.parity_combo.addItems(["None", "Even", "Odd", "Mark", "Space"])
|
||
self.parity_combo.setCurrentText("None")
|
||
|
||
self.flow_control_combo = QComboBox()
|
||
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=ICON_COLOR_MUTED))
|
||
refresh_btn.clicked.connect(self._load_ports)
|
||
buttons_layout.addWidget(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=ICON_COLOR_MUTED))
|
||
cancel_btn.clicked.connect(self._close_this_tab)
|
||
buttons_layout.addWidget(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()
|
||
|
||
action_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),
|
||
)
|
||
for title, icon_name, handler in action_items:
|
||
btn = QPushButton(title)
|
||
btn.setMinimumHeight(34)
|
||
btn.setIcon(icon(icon_name, color=ICON_COLOR_DARK))
|
||
btn.clicked.connect(handler)
|
||
controls_layout.addWidget(btn)
|
||
|
||
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)
|
||
controls_layout.addWidget(command_input_btn)
|
||
controls_layout.addStretch(1)
|
||
root_layout.addWidget(self.controls_widget)
|
||
self.controls_widget.hide()
|
||
|
||
root_layout.addStretch(1)
|
||
|
||
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._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._start_health_timer()
|
||
|
||
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()
|
||
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)
|
||
|
||
# ── Открытие диалогов ───────────────────────────────────────────────
|
||
|
||
def _exec_dialog(self, dlg) -> None:
|
||
"""Открыть диалог с трекингом (увеличивает/уменьшает счётчик)."""
|
||
self._open_dialogs += 1
|
||
try:
|
||
dlg.exec()
|
||
finally:
|
||
self._open_dialogs = max(0, self._open_dialogs - 1)
|
||
|
||
def _open_input_test(self) -> None:
|
||
from .dialogs.input_test_dialog import InputTestDialog
|
||
self._exec_dialog(InputTestDialog(self._connection, parent=self))
|
||
|
||
def _open_led_test(self) -> None:
|
||
from .dialogs.led_test_dialog import LedTestDialog
|
||
self._exec_dialog(LedTestDialog(self._connection, parent=self))
|
||
|
||
def _open_output_test(self) -> None:
|
||
from .dialogs.output_test_dialog import OutputTestDialog
|
||
self._exec_dialog(OutputTestDialog(self._connection, parent=self))
|
||
|
||
def _open_firmware(self) -> None:
|
||
from .dialogs.firmware_dialog import FirmwareDialog
|
||
self._exec_dialog(FirmwareDialog(self._connection, parent=self))
|
||
|
||
def _open_role(self) -> None:
|
||
from .dialogs.role_dialog import RoleDialog
|
||
self._exec_dialog(RoleDialog(self._connection, parent=self))
|
||
|
||
def _open_terminal(self) -> None:
|
||
from .dialogs.command_terminal_dialog import CommandTerminalDialog
|
||
self._exec_dialog(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 ──────────────────────────────────────
|
||
|
||
_HEALTH_INTERVAL_MS = 10_000 # интервал таймера (10 сек)
|
||
_IDLE_THRESHOLD_SEC = 10.0 # минимальный простой перед проверкой
|
||
|
||
def _start_health_timer(self) -> None:
|
||
if self._health_timer is not None:
|
||
return
|
||
self._health_timer = QTimer(self)
|
||
self._health_timer.setInterval(self._HEALTH_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:
|
||
"""Проверка по таймеру: если нет открытых диалогов и
|
||
прошло ≥10 сек без команд — шлём 99 для проверки связи."""
|
||
if self._open_dialogs > 0:
|
||
return
|
||
if self._health_running:
|
||
return
|
||
if not self._connection.is_connected:
|
||
# Порт закрыт — пробуем переподключиться
|
||
self._try_reconnect()
|
||
return
|
||
if self._connection.seconds_since_last_command < self._IDLE_THRESHOLD_SEC:
|
||
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:
|
||
# Связь жива, плата авторизована
|
||
if self._plate_state != PlateState.READY:
|
||
self._update_plate_state(PlateState.READY)
|
||
elif status == AuthStatus.NEED_AUTH:
|
||
# Связь жива, но нужна авторизация
|
||
if self._plate_state != PlateState.NEED_AUTH:
|
||
self._update_plate_state(PlateState.NEED_AUTH)
|
||
else:
|
||
# Связь потеряна
|
||
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._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, таймаут 1 с)."""
|
||
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=1.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)
|