"""Диалог «Ввод команд» — мини-терминал для ручного общения с платой. Позволяет отправлять произвольные команды и видеть ответы. """ 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)