"""Встраиваемая секция «Обновление прошивки» — загрузка и OTA. Повторяет функциональность FirmwareDialog, но как обычный QWidget. """ from __future__ import annotations import os from PySide6.QtCore import QThread, Signal, Qt from PySide6.QtWidgets import ( QFileDialog, QGroupBox, QHBoxLayout, QLabel, QMessageBox, QProgressBar, QPushButton, QTextEdit, QVBoxLayout, QWidget, QSplitter, ) from ...services.com_service import ComConnection from ...services.firmware_service import perform_authenticated_ota from ...ui.icons import get_icon_color, icon # Переиспользуем воркер из диалога from ..dialogs.firmware_dialog import _UploadWorker class FirmwareSection(QWidget): """Inline-секция обновления прошивки (для базового режима).""" upload_started = Signal() upload_finished = Signal() def __init__(self, connection: ComConnection, parent: QWidget | None = None) -> None: super().__init__(parent) self._conn = connection self._firmware_path: str | None = None self._upload_thread: QThread | None = None self._worker: _UploadWorker | None = None self._need_ota: bool = False self._setup_ui() # ── UI ──────────────────────────────────────────────────────────── def _setup_ui(self) -> None: root = QVBoxLayout(self) root.setContentsMargins(0, 0, 0, 0) root.setSpacing(10) # Вертикальный сплиттер для изменения высоты лога splitter = QSplitter(Qt.Orientation.Vertical) # ── Обновление прошивки ── fw_group = QGroupBox("Обновление прошивки") fw_layout = QVBoxLayout(fw_group) desc = QLabel( "Выберите файл прошивки и запустите обновление. " "Дождитесь завершения прогресса и сообщения о перезагрузке платы. " "Во время обновления остальные проверки временно недоступны." ) desc.setWordWrap(True) desc.setObjectName("HelpText") fw_layout.addWidget(desc) file_layout = QHBoxLayout() self._file_label = QLabel("Файл не выбран") self._file_label.setWordWrap(True) file_layout.addWidget(self._file_label, 1) self._pick_btn = QPushButton("Выбрать .bin") self._pick_btn.setIcon(icon("fa5s.folder-open", color=get_icon_color())) self._pick_btn.clicked.connect(self._pick_file) file_layout.addWidget(self._pick_btn) fw_layout.addLayout(file_layout) self._progress = QProgressBar() self._progress.setValue(0) fw_layout.addWidget(self._progress) self._update_btn = QPushButton("Обновить прошивку") self._update_btn.setProperty("role", "primary") self._update_btn.setIcon(icon("fa5s.upload", color="#ecfdf5")) self._update_btn.clicked.connect(self._start_update) fw_layout.addWidget(self._update_btn) splitter.addWidget(fw_group) # ── Лог ── self._log = QTextEdit() self._log.setReadOnly(True) splitter.addWidget(self._log) # Задаем начальное распределение размеров (70% на верхнюю часть, 30% на лог) splitter.setStretchFactor(0, 70) splitter.setStretchFactor(1, 30) root.addWidget(splitter, 1) # ── Public ──────────────────────────────────────────────────────── def refresh_icons(self) -> None: clr = get_icon_color() self._pick_btn.setIcon(icon("fa5s.folder-open", color=clr)) def cancel_upload(self) -> None: """Прервать загрузку (если идёт) — вызывается при скрытии секции.""" if self._upload_thread is not None and self._upload_thread.isRunning(): self._upload_thread.quit() self._upload_thread.wait(3000) # ── Загрузка файла ──────────────────────────────────────────────── def _pick_file(self) -> None: path, _ = QFileDialog.getOpenFileName(self, "Выбрать файл прошивки", "", "BIN Files (*.bin)") if path: self._firmware_path = path size = os.path.getsize(path) self._file_label.setText(f"{os.path.basename(path)} ({size:,} байт)") self._progress.setValue(0) def _start_update(self) -> None: """Загрузить файл на плату (02), а затем автоматически прошить (04).""" if not self._firmware_path: QMessageBox.warning(self, "Ошибка", "Сначала выберите файл прошивки.") return if not self._conn.is_connected: QMessageBox.warning(self, "Ошибка", "Порт не подключён.") return self._update_btn.setEnabled(False) self._pick_btn.setEnabled(False) self._progress.setValue(0) self._log.append("Запуск загрузки…") self.upload_started.emit() # Очистить serial-буфер от остатков предыдущего опроса try: ser = self._conn.serial if ser is not None and ser.is_open: ser.reset_input_buffer() except Exception: pass self._worker = _UploadWorker(self._conn, self._firmware_path) self._upload_thread = QThread() self._worker.moveToThread(self._upload_thread) self._worker.progress.connect(self._on_upload_progress) self._worker.log_message.connect(self._log.append) self._worker.finished.connect(self._on_upload_done) self._worker.finished.connect(self._upload_thread.quit) self._upload_thread.finished.connect(self._cleanup_upload) self._upload_thread.started.connect(self._worker.run) self._upload_thread.start() def _cleanup_upload(self) -> None: if self._worker is not None: self._worker.deleteLater() self._worker = None if self._upload_thread is not None: self._upload_thread.deleteLater() self._upload_thread = None # Worker thread полностью завершён, finally выполнен, lock свободен. # Теперь безопасно запускать OTA. if self._need_ota: self._need_ota = False self._flash_ota() def _on_upload_progress(self, percent: int) -> None: self._progress.setValue(percent) def _on_upload_done(self, success: bool) -> None: if success: self._log.append("✓ Загрузка завершена. Запуск OTA (04)…") # Не вызываем _flash_ota сразу: worker thread ещё в finally, # io_lock может быть занят. Отложим до _cleanup_upload, # который сработает после реального завершения потока. self._need_ota = True else: self._need_ota = False self._log.append("✗ Загрузка не удалась.") self._update_btn.setEnabled(True) self._pick_btn.setEnabled(True) self.upload_finished.emit() def _flash_ota(self) -> None: if not self._conn.is_connected: self._log.append("✗ Порт не подключён.") self._update_btn.setEnabled(True) self._pick_btn.setEnabled(True) self.upload_finished.emit() return result = perform_authenticated_ota( self._conn, log_message=self._log.append, ) if result.success: self._log.append(f"✓ {result.detail}.") else: self._log.append(f"✗ {result.detail}") self._update_btn.setEnabled(True) self._pick_btn.setEnabled(True) self.upload_finished.emit()