"""Диалог «Ввод команд» — мини-терминал для ручного общения с платой.
Позволяет отправлять произвольные команды и видеть ответы.
"""
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('✗ Порт не подключён')
return
if self._auth_thread is not None:
return
self._set_command_controls_enabled(False)
self._auth_btn.setText("Авторизация…")
self._log.append('Проверка авторизации…')
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(
'✓ Прошивка no_auth: '
'авторизация не требуется'
)
else:
self._log.append(
'✓ Авторизация успешно завершена'
)
else:
self._log.append(
f'✗ Авторизация: {result.detail}'
)
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('✗ Порт не подключён')
return
try:
self._conn.send_async(cmd)
self._log.append(f'→ {cmd}')
except Exception as exc:
self._log.append(f'✗ {exc}')
def _on_line_received(self, line: str) -> None:
stripped = line.strip()
if stripped:
self._log.append(f'← {stripped}')
def closeEvent(self, event) -> None:
if self._auth_thread is not None and self._auth_thread.isRunning():
self._log.append(
'Дождитесь завершения авторизации.'
)
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)