Danil Iarosh
c34cb0cd7e
- добавлена полная повторная авторизация перед командой 04 - добавлена авторизация после успешного обновления и перезагрузки платы - общий OTA-сценарий подключён к базовому и PRO-интерфейсам - в терминал команд добавлена кнопка ручной авторизации - состояние авторизации синхронизировано с основной вкладкой - добавлены тесты порядка авторизации и обработки ошибок OTA - версия приложения повышена до 1.5.0 - обновлена документация
210 lines
8.6 KiB
Python
210 lines
8.6 KiB
Python
"""Встраиваемая секция «Обновление прошивки» — загрузка и 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()
|