Danil Iarosh
add8aff0af
- Добавлена обратная совместимость с прошивками, содержащими маркер no_auth. - Повторная авторизация после команды 04 и успешного обновления прошивки приведена к требуемому порядку. - В терминал команд добавлена ручная кнопка авторизации. - Реализовано OTA-обновление приложения с загрузкой платформенного артефакта, безопасной заменой и перезапуском. - Проверка обновлений учитывает только более новые версии по схеме major, minor и patch. - Обновлены имена Windows и Linux сборок, документация и тестовое покрытие.
279 lines
10 KiB
Python
279 lines
10 KiB
Python
"""Диалог «Ввод команд» — мини-терминал для ручного общения с платой.
|
||
|
||
Позволяет отправлять произвольные команды и видеть ответы.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from PySide6.QtCore import QThread, Qt, Signal
|
||
from PySide6.QtGui import QKeyEvent
|
||
from PySide6.QtWidgets import (
|
||
QHBoxLayout,
|
||
QLineEdit,
|
||
QPushButton,
|
||
QTextEdit,
|
||
QVBoxLayout,
|
||
QSplitter,
|
||
QWidget,
|
||
)
|
||
|
||
from ...services.com_service import ComConnection
|
||
from ...services.auth_service import (
|
||
AuthResult,
|
||
AuthStatus,
|
||
perform_required_handshake,
|
||
)
|
||
from ...ui.icons import get_icon_color, get_icon_muted, icon
|
||
from .styled_dialog import StyledDialog
|
||
|
||
|
||
class CommandTerminalDialog(StyledDialog):
|
||
authorization_finished = Signal(object)
|
||
|
||
def __init__(self, connection: ComConnection, parent=None) -> None:
|
||
super().__init__(parent)
|
||
self._conn = connection
|
||
self._auth_thread: _TerminalHandshakeThread | None = None
|
||
self._command_controls: list[QWidget] = []
|
||
self.setWindowTitle("Ввод команд")
|
||
self.setMinimumSize(540, 400)
|
||
|
||
self._setup_ui()
|
||
self._conn.line_received.connect(
|
||
self._on_line_received, Qt.ConnectionType.QueuedConnection
|
||
)
|
||
|
||
# История команд (макс. 50)
|
||
self._history: list[str] = []
|
||
self._history_index: int = -1
|
||
self._history_draft: str = ""
|
||
self._MAX_HISTORY = 50
|
||
|
||
# Перехватываем клавиши ↑↓ в поле ввода
|
||
self._input.installEventFilter(self)
|
||
|
||
# Фокус на поле ввода при открытии
|
||
self._input.setFocus()
|
||
|
||
def _setup_ui(self) -> None:
|
||
root = self.containerLayout()
|
||
root.setSpacing(8)
|
||
|
||
# Вертикальный сплиттер для изменения высоты лога
|
||
splitter = QSplitter(Qt.Orientation.Vertical)
|
||
|
||
# ── Лог ──
|
||
self._log = QTextEdit()
|
||
self._log.setReadOnly(True)
|
||
self._log.setStyleSheet(
|
||
"QTextEdit { background: #1e293b; color: #e2e8f0; font-family: 'Consolas', 'Courier New', monospace; font-size: 12px; border-radius: 8px; padding: 8px; }"
|
||
)
|
||
splitter.addWidget(self._log)
|
||
|
||
# ── Нижняя панель управления ──
|
||
bottom_widget = QWidget()
|
||
bottom_layout = QVBoxLayout(bottom_widget)
|
||
bottom_layout.setContentsMargins(0, 0, 0, 0)
|
||
bottom_layout.setSpacing(8)
|
||
|
||
# ── Ввод ──
|
||
input_layout = QHBoxLayout()
|
||
self._input = QLineEdit()
|
||
self._input.setPlaceholderText("Введите команду (напр. 99, 05, 91, 8003)…")
|
||
self._input.returnPressed.connect(self._send_command)
|
||
input_layout.addWidget(self._input, 1)
|
||
|
||
self._send_btn = QPushButton("Отправить")
|
||
self._send_btn.setProperty("role", "primary")
|
||
self._send_btn.setIcon(icon("fa5s.paper-plane", color="#ecfdf5"))
|
||
self._send_btn.clicked.connect(self._send_command)
|
||
input_layout.addWidget(self._send_btn)
|
||
|
||
clear_btn = QPushButton("Очистить")
|
||
clear_btn.setIcon(icon("fa5s.eraser", color=get_icon_muted()))
|
||
clear_btn.clicked.connect(self._log.clear)
|
||
input_layout.addWidget(clear_btn)
|
||
|
||
bottom_layout.addLayout(input_layout)
|
||
|
||
# ── Быстрые команды ──
|
||
quick_layout = QHBoxLayout()
|
||
self._auth_btn = QPushButton("Авторизация")
|
||
self._auth_btn.setMinimumHeight(30)
|
||
self._auth_btn.setIcon(icon("fa5s.key", color=get_icon_color()))
|
||
self._auth_btn.clicked.connect(self._start_authorization)
|
||
quick_layout.addWidget(self._auth_btn)
|
||
|
||
quick_commands = [
|
||
("Статус (99)", "99"),
|
||
("Версия (05)", "05"),
|
||
("MAC (06)", "06"),
|
||
("Входы (91)", "91"),
|
||
("Сброс (00)", "00"),
|
||
]
|
||
for label, cmd in quick_commands:
|
||
btn = QPushButton(label)
|
||
btn.setMinimumHeight(30)
|
||
btn.setStyleSheet("padding: 4px 10px;")
|
||
btn.clicked.connect(lambda checked=False, c=cmd: self._send(c))
|
||
quick_layout.addWidget(btn)
|
||
self._command_controls.append(btn)
|
||
quick_layout.addStretch()
|
||
bottom_layout.addLayout(quick_layout)
|
||
|
||
# ── Закрыть ──
|
||
btn_layout = QHBoxLayout()
|
||
btn_layout.addStretch()
|
||
close_btn = QPushButton("Закрыть")
|
||
close_btn.setIcon(icon("fa5s.times", color=get_icon_muted()))
|
||
close_btn.clicked.connect(self.close)
|
||
btn_layout.addWidget(close_btn)
|
||
bottom_layout.addLayout(btn_layout)
|
||
|
||
splitter.addWidget(bottom_widget)
|
||
|
||
# Начальное распределение высоты (75% на лог, 25% на нижнюю панель)
|
||
splitter.setStretchFactor(0, 75)
|
||
splitter.setStretchFactor(1, 25)
|
||
|
||
root.addWidget(splitter, 1)
|
||
self._command_controls.extend((self._input, self._send_btn, self._auth_btn))
|
||
|
||
def _start_authorization(self) -> None:
|
||
if not self._conn.is_connected:
|
||
self._log.append('<span style="color:#f87171;">✗ Порт не подключён</span>')
|
||
return
|
||
if self._auth_thread is not None:
|
||
return
|
||
|
||
self._set_command_controls_enabled(False)
|
||
self._auth_btn.setText("Авторизация…")
|
||
self._log.append('<span style="color:#facc15;">Проверка авторизации…</span>')
|
||
self._conn.pause_drain()
|
||
|
||
thread = _TerminalHandshakeThread(self._conn, parent=self)
|
||
thread.result_ready.connect(self._on_authorization_done)
|
||
thread.finished.connect(self._cleanup_auth_thread)
|
||
self._auth_thread = thread
|
||
thread.start()
|
||
|
||
def _on_authorization_done(self, result: AuthResult) -> None:
|
||
self._conn.resume_drain()
|
||
self._set_command_controls_enabled(True)
|
||
self._auth_btn.setText("Авторизация")
|
||
if result.status == AuthStatus.OK:
|
||
if result.auth_skipped:
|
||
self._log.append(
|
||
'<span style="color:#4ade80;">✓ Прошивка no_auth: '
|
||
'авторизация не требуется</span>'
|
||
)
|
||
else:
|
||
self._log.append(
|
||
'<span style="color:#4ade80;">✓ Авторизация успешно завершена</span>'
|
||
)
|
||
else:
|
||
self._log.append(
|
||
f'<span style="color:#f87171;">✗ Авторизация: {result.detail}</span>'
|
||
)
|
||
self.authorization_finished.emit(result)
|
||
|
||
def _cleanup_auth_thread(self) -> None:
|
||
if self._auth_thread is not None:
|
||
self._auth_thread.deleteLater()
|
||
self._auth_thread = None
|
||
|
||
def _set_command_controls_enabled(self, enabled: bool) -> None:
|
||
for control in self._command_controls:
|
||
control.setEnabled(enabled)
|
||
|
||
def _send_command(self) -> None:
|
||
text = self._input.text().strip()
|
||
if not text:
|
||
return
|
||
# Сохранить в историю
|
||
if not self._history or self._history[-1] != text:
|
||
self._history.append(text)
|
||
if len(self._history) > self._MAX_HISTORY:
|
||
self._history.pop(0)
|
||
self._history_index = -1
|
||
self._history_draft = ""
|
||
|
||
self._send(text)
|
||
self._input.clear()
|
||
|
||
def eventFilter(self, watched, event): # type: ignore[override]
|
||
if watched is self._input and isinstance(event, QKeyEvent):
|
||
if event.type() == QKeyEvent.Type.KeyPress:
|
||
key = event.key()
|
||
if key == Qt.Key.Key_Up:
|
||
self._history_navigate(-1)
|
||
return True
|
||
if key == Qt.Key.Key_Down:
|
||
self._history_navigate(1)
|
||
return True
|
||
return super().eventFilter(watched, event)
|
||
|
||
def _history_navigate(self, direction: int) -> None:
|
||
"""Перемещение по истории: direction=-1 (назад), +1 (вперёд)."""
|
||
if not self._history:
|
||
return
|
||
|
||
if self._history_index == -1:
|
||
# Сохраняем текущий ввод как черновик
|
||
self._history_draft = self._input.text()
|
||
|
||
new_index = self._history_index - direction
|
||
if new_index < -1:
|
||
new_index = -1
|
||
elif new_index >= len(self._history):
|
||
new_index = len(self._history) - 1
|
||
|
||
self._history_index = new_index
|
||
if new_index == -1:
|
||
self._input.setText(self._history_draft)
|
||
else:
|
||
self._input.setText(self._history[len(self._history) - 1 - new_index])
|
||
|
||
def _send(self, cmd: str) -> None:
|
||
if not self._conn.is_connected:
|
||
self._log.append('<span style="color:#f87171;">✗ Порт не подключён</span>')
|
||
return
|
||
try:
|
||
self._conn.send_async(cmd)
|
||
self._log.append(f'<span style="color:#4ade80;">→ {cmd}</span>')
|
||
except Exception as exc:
|
||
self._log.append(f'<span style="color:#f87171;">✗ {exc}</span>')
|
||
|
||
def _on_line_received(self, line: str) -> None:
|
||
stripped = line.strip()
|
||
if stripped:
|
||
self._log.append(f'<span style="color:#93c5fd;">← {stripped}</span>')
|
||
|
||
def closeEvent(self, event) -> None:
|
||
if self._auth_thread is not None and self._auth_thread.isRunning():
|
||
self._log.append(
|
||
'<span style="color:#facc15;">Дождитесь завершения авторизации.</span>'
|
||
)
|
||
event.ignore()
|
||
return
|
||
try:
|
||
self._conn.line_received.disconnect(self._on_line_received)
|
||
except (RuntimeError, TypeError):
|
||
pass
|
||
super().closeEvent(event)
|
||
|
||
|
||
class _TerminalHandshakeThread(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:
|
||
try:
|
||
result = perform_required_handshake(self._conn, timeout=2.0)
|
||
except Exception as exc:
|
||
result = AuthResult(AuthStatus.FAILED, str(exc))
|
||
self.result_ready.emit(result)
|