Danil Iarosh
e163b63666
- удалена повторная авторизация перед отправкой команды 04 - полный AES-256 handshake запускается только после успешного OTA и перезагрузки контроллера - обновлены сообщения интерфейса и документация порядка обновления - тесты теперь проверяют, что команда 04 предшествует команде авторизации 07 - версия приложения повышена до 1.5.1 - локальные каталоги релизных артефактов исключены из Git
454 lines
18 KiB
Python
454 lines
18 KiB
Python
"""Диалог «Обновление прошивки» — загрузка и прошивка контроллера.
|
||
|
||
Команды:
|
||
05 — версия прошивки
|
||
06 — MAC-адрес
|
||
03 — содержимое SPIFFS
|
||
02<size> — подготовка к приёму файла → пересылка .bin блоками по 4096
|
||
04 — запуск OTA-обновления
|
||
|
||
Протокол загрузки (команда 02):
|
||
1. Отправляем «02<size>\r\n»
|
||
2. Контроллер ~15 сек форматирует SPIFFS, затем шлёт строку-подтверждение
|
||
3. После подтверждения всё, что приходит в порт, записывается в файл update.bin
|
||
4. На каждый принятый блок (4096 байт) контроллер отвечает размером принятого
|
||
5. Когда принято size байт — переход обратно в режим команд
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import time
|
||
|
||
from PySide6.QtCore import Qt, QThread, Signal, QObject
|
||
from PySide6.QtWidgets import (
|
||
QFileDialog,
|
||
QGroupBox,
|
||
QHBoxLayout,
|
||
QLabel,
|
||
QMessageBox,
|
||
QProgressBar,
|
||
QPushButton,
|
||
QTextEdit,
|
||
QVBoxLayout,
|
||
QSplitter,
|
||
QWidget,
|
||
)
|
||
|
||
from ...services.com_service import ComConnection
|
||
from ...services.firmware_service import perform_authenticated_ota
|
||
from ...ui.icons import get_icon_color, get_icon_muted, icon
|
||
from .styled_dialog import StyledDialog
|
||
|
||
_BLOCK_SIZE = 4096
|
||
|
||
|
||
class FirmwareDialog(StyledDialog):
|
||
def __init__(self, connection: ComConnection, parent=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.setWindowTitle("Обновление прошивки")
|
||
self.setMinimumSize(520, 440)
|
||
self._setup_ui()
|
||
|
||
def _setup_ui(self) -> None:
|
||
root = self.containerLayout()
|
||
root.setSpacing(12)
|
||
|
||
# Вертикальный сплиттер для изменения высоты лога
|
||
splitter = QSplitter(Qt.Orientation.Vertical)
|
||
|
||
# ── Верхняя панель (информация + загрузка) ──
|
||
top_widget = QWidget()
|
||
top_layout = QVBoxLayout(top_widget)
|
||
top_layout.setContentsMargins(0, 0, 0, 0)
|
||
top_layout.setSpacing(12)
|
||
|
||
# ── Информация о плате ──
|
||
info_group = QGroupBox("Информация о плате")
|
||
info_layout = QVBoxLayout(info_group)
|
||
|
||
info_btn_layout = QHBoxLayout()
|
||
ver_btn = QPushButton("Версия (05)")
|
||
ver_btn.setIcon(icon("fa5s.info-circle", color=get_icon_color()))
|
||
ver_btn.clicked.connect(self._query_version)
|
||
info_btn_layout.addWidget(ver_btn)
|
||
|
||
mac_btn = QPushButton("MAC-адрес (06)")
|
||
mac_btn.setIcon(icon("fa5s.fingerprint", color=get_icon_color()))
|
||
mac_btn.clicked.connect(self._query_mac)
|
||
info_btn_layout.addWidget(mac_btn)
|
||
|
||
spiffs_btn = QPushButton("SPIFFS (03)")
|
||
spiffs_btn.setIcon(icon("fa5s.hdd", color=get_icon_color()))
|
||
spiffs_btn.clicked.connect(self._query_spiffs)
|
||
info_btn_layout.addWidget(spiffs_btn)
|
||
|
||
info_layout.addLayout(info_btn_layout)
|
||
|
||
self._info_label = QLabel("—")
|
||
self._info_label.setWordWrap(True)
|
||
self._info_label.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse)
|
||
info_layout.addWidget(self._info_label)
|
||
|
||
top_layout.addWidget(info_group)
|
||
|
||
# ── Загрузка файла ──
|
||
upload_group = QGroupBox("Загрузка прошивки")
|
||
upload_layout = QVBoxLayout(upload_group)
|
||
|
||
file_layout = QHBoxLayout()
|
||
self._file_label = QLabel("Файл не выбран")
|
||
self._file_label.setWordWrap(True)
|
||
file_layout.addWidget(self._file_label, 1)
|
||
|
||
pick_btn = QPushButton("Выбрать .bin")
|
||
pick_btn.setIcon(icon("fa5s.folder-open", color=get_icon_color()))
|
||
pick_btn.clicked.connect(self._pick_file)
|
||
file_layout.addWidget(pick_btn)
|
||
upload_layout.addLayout(file_layout)
|
||
|
||
self._progress = QProgressBar()
|
||
self._progress.setValue(0)
|
||
upload_layout.addWidget(self._progress)
|
||
|
||
upload_btn_layout = QHBoxLayout()
|
||
self._upload_btn = QPushButton("Загрузить на плату (02)")
|
||
self._upload_btn.setProperty("role", "primary")
|
||
self._upload_btn.setIcon(icon("fa5s.upload", color="#ecfdf5"))
|
||
self._upload_btn.clicked.connect(self._start_upload)
|
||
upload_btn_layout.addWidget(self._upload_btn)
|
||
|
||
self._flash_btn = QPushButton("Прошить (04)")
|
||
self._flash_btn.setIcon(icon("fa5s.bolt", color=get_icon_color()))
|
||
self._flash_btn.clicked.connect(self._flash_ota)
|
||
self._flash_btn.setEnabled(False)
|
||
upload_btn_layout.addWidget(self._flash_btn)
|
||
upload_layout.addLayout(upload_btn_layout)
|
||
|
||
top_layout.addWidget(upload_group)
|
||
splitter.addWidget(top_widget)
|
||
|
||
# ── Лог (нижняя часть сплиттера) ──
|
||
self._log = QTextEdit()
|
||
self._log.setReadOnly(True)
|
||
splitter.addWidget(self._log)
|
||
|
||
# Начальное распределение размеров (65% на верхнюю часть, 35% на лог)
|
||
splitter.setStretchFactor(0, 65)
|
||
splitter.setStretchFactor(1, 35)
|
||
|
||
root.addWidget(splitter, 1)
|
||
|
||
# ── Закрыть ──
|
||
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)
|
||
root.addLayout(btn_layout)
|
||
|
||
# ── Запросы информации ──
|
||
|
||
def _query(self, cmd: str) -> str | None:
|
||
if not self._conn.is_connected:
|
||
QMessageBox.warning(self, "Ошибка", "Порт не подключён.")
|
||
return None
|
||
try:
|
||
self._conn.send_command(cmd)
|
||
resp = self._conn.read_response(timeout=1.0)
|
||
self._conn.release()
|
||
return resp
|
||
except Exception as exc:
|
||
self._conn.release()
|
||
self._log.append(f"Ошибка: {exc}")
|
||
return None
|
||
|
||
def _query_version(self) -> None:
|
||
resp = self._query("05")
|
||
self._info_label.setText(f"Версия: {resp or '—'}")
|
||
self._log.append(f"→ 05\n← {resp or 'таймаут'}")
|
||
|
||
def _query_mac(self) -> None:
|
||
resp = self._query("06")
|
||
self._info_label.setText(f"MAC: {resp or '—'}")
|
||
self._log.append(f"→ 06\n← {resp or 'таймаут'}")
|
||
|
||
def _query_spiffs(self) -> None:
|
||
resp = self._query("03")
|
||
self._info_label.setText(f"SPIFFS:\n{resp or '—'}")
|
||
self._log.append(f"→ 03\n← {resp or 'таймаут'}")
|
||
|
||
# ── Загрузка файла ──
|
||
|
||
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._flash_btn.setEnabled(False)
|
||
self._progress.setValue(0)
|
||
|
||
def _start_upload(self) -> None:
|
||
if not self._firmware_path:
|
||
QMessageBox.warning(self, "Ошибка", "Сначала выберите файл прошивки.")
|
||
return
|
||
if not self._conn.is_connected:
|
||
QMessageBox.warning(self, "Ошибка", "Порт не подключён.")
|
||
return
|
||
|
||
self._upload_btn.setEnabled(False)
|
||
self._progress.setValue(0)
|
||
self._log.append("Запуск загрузки…")
|
||
|
||
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
|
||
|
||
def _on_upload_progress(self, percent: int) -> None:
|
||
self._progress.setValue(percent)
|
||
|
||
def _on_upload_done(self, success: bool) -> None:
|
||
self._upload_btn.setEnabled(True)
|
||
if success:
|
||
self._flash_btn.setEnabled(True)
|
||
self._log.append("✓ Загрузка завершена. Можно прошивать (04).")
|
||
else:
|
||
self._log.append("✗ Загрузка не удалась.")
|
||
|
||
def _flash_ota(self) -> None:
|
||
if not self._conn.is_connected:
|
||
self._log.append("✗ Порт не подключён.")
|
||
return
|
||
|
||
self._flash_btn.setEnabled(False)
|
||
self._upload_btn.setEnabled(False)
|
||
self._log.append("Запуск OTA (04)…")
|
||
|
||
from PySide6.QtCore import QCoreApplication
|
||
|
||
def append_log(message: str) -> None:
|
||
self._log.append(message)
|
||
QCoreApplication.processEvents()
|
||
|
||
result = perform_authenticated_ota(self._conn, log_message=append_log)
|
||
if result.success:
|
||
self._log.append(f"✓ {result.detail}.")
|
||
else:
|
||
self._log.append(f"✗ {result.detail}")
|
||
self._flash_btn.setEnabled(True)
|
||
self._upload_btn.setEnabled(True)
|
||
|
||
def closeEvent(self, event) -> None:
|
||
if self._upload_thread is not None and self._upload_thread.isRunning():
|
||
self._upload_thread.quit()
|
||
self._upload_thread.wait(3000)
|
||
super().closeEvent(event)
|
||
|
||
|
||
class _UploadWorker(QObject):
|
||
progress = Signal(int)
|
||
log_message = Signal(str)
|
||
finished = Signal(bool)
|
||
|
||
def __init__(self, conn: ComConnection, path: str) -> None:
|
||
super().__init__()
|
||
self._conn = conn
|
||
self._path = path
|
||
|
||
def run(self) -> None: # noqa: C901
|
||
serial = None
|
||
old_timeout = None
|
||
# Приостанавливаем фоновый дренаж на время прошивки
|
||
self._conn._io_lock.acquire()
|
||
self._conn._cmd_owner = True
|
||
try:
|
||
file_size = os.path.getsize(self._path)
|
||
serial = self._conn.serial
|
||
if serial is None:
|
||
self.log_message.emit("✗ Serial-объект не доступен")
|
||
self.finished.emit(False)
|
||
return
|
||
|
||
old_timeout = serial.timeout
|
||
|
||
# ── Шаг 1: очистить буферы ──
|
||
serial.reset_input_buffer()
|
||
serial.reset_output_buffer()
|
||
self.log_message.emit("Буферы очищены")
|
||
|
||
# ── Шаг 2: отправить 02<size> ──
|
||
cmd = f"02{file_size}"
|
||
payload = (cmd + "\r\n").encode("ascii")
|
||
self.log_message.emit(f"→ {cmd} ({len(payload)} байт)")
|
||
written = serial.write(payload)
|
||
serial.flush()
|
||
self.log_message.emit(f" write() вернул {written}")
|
||
|
||
# ── Шаг 3: ждать «file ready to write» от контроллера ──
|
||
# Последовательность ответов контроллера:
|
||
# ← SPIFFS Mount OK
|
||
# ← SPIFFS formatting .... WAIT (~15 сек)
|
||
# ← format OK
|
||
# ← file ready to write ← ТОЛЬКО после этого можно слать данные
|
||
self.log_message.emit("⏳ Ожидание форматирования SPIFFS...")
|
||
deadline = time.time() + 30.0
|
||
buf = b""
|
||
ready = False
|
||
last_report = time.time()
|
||
|
||
while time.time() < deadline:
|
||
waiting = serial.in_waiting
|
||
if waiting > 0:
|
||
chunk = serial.read(waiting)
|
||
buf += chunk
|
||
# Разбираем полные строки
|
||
while b"\n" in buf:
|
||
line_raw, buf = buf.split(b"\n", 1)
|
||
line = line_raw.decode("ascii", errors="replace").strip()
|
||
if line:
|
||
self.log_message.emit(f"← {line}")
|
||
# Ждём конкретно эту строку
|
||
if "file ready" in line.lower():
|
||
ready = True
|
||
break
|
||
if ready:
|
||
break
|
||
else:
|
||
now = time.time()
|
||
if now - last_report >= 2.0:
|
||
elapsed = int(now - (deadline - 30.0))
|
||
self.log_message.emit(f" ожидание... {elapsed} сек")
|
||
last_report = now
|
||
time.sleep(0.1)
|
||
|
||
# Если в буфере остались данные без \n
|
||
if buf:
|
||
tail = buf.decode("ascii", errors="replace").strip()
|
||
if tail:
|
||
self.log_message.emit(f"← {tail}")
|
||
if "file ready" in tail.lower():
|
||
ready = True
|
||
|
||
if not ready:
|
||
self.log_message.emit(
|
||
"✗ Не получено «file ready to write» (таймаут 30 сек)"
|
||
)
|
||
serial.timeout = old_timeout
|
||
self.finished.emit(False)
|
||
return
|
||
|
||
# Небольшая пауза, чтобы контроллер был готов принимать данные
|
||
time.sleep(0.2)
|
||
# Очистим входной буфер — всё что пришло до этого момента уже обработано
|
||
serial.reset_input_buffer()
|
||
|
||
self.log_message.emit("✓ Контроллер готов к приёму файла")
|
||
|
||
# ── Шаг 4: отправляем файл блоками ──
|
||
sent = 0
|
||
block_num = 0
|
||
total_blocks = (file_size + _BLOCK_SIZE - 1) // _BLOCK_SIZE
|
||
self.log_message.emit(
|
||
f"Отправка {file_size} байт ({total_blocks} блоков по {_BLOCK_SIZE})"
|
||
)
|
||
|
||
serial.timeout = 10.0 # для readline ACK
|
||
|
||
with open(self._path, "rb") as f:
|
||
while True:
|
||
block = f.read(_BLOCK_SIZE)
|
||
if not block:
|
||
break
|
||
block_num += 1
|
||
|
||
serial.write(block)
|
||
serial.flush()
|
||
sent += len(block)
|
||
percent = int(sent * 100 / file_size)
|
||
self.progress.emit(percent)
|
||
|
||
# Ждём ACK — polling, чтобы не зависнуть навечно
|
||
ack_deadline = time.time() + 10.0
|
||
ack_buf = b""
|
||
got_ack = False
|
||
while time.time() < ack_deadline:
|
||
w = serial.in_waiting
|
||
if w > 0:
|
||
ack_buf += serial.read(w)
|
||
if b"\n" in ack_buf:
|
||
ack_line, ack_buf = ack_buf.split(b"\n", 1)
|
||
ack = ack_line.decode("ascii", errors="replace").strip()
|
||
self.log_message.emit(
|
||
f" [{block_num}/{total_blocks}] "
|
||
f"{sent}/{file_size}: ← {ack}"
|
||
)
|
||
got_ack = True
|
||
break
|
||
else:
|
||
time.sleep(0.05)
|
||
|
||
if not got_ack:
|
||
# Может ACK без \n
|
||
if ack_buf:
|
||
ack = ack_buf.decode("ascii", errors="replace").strip()
|
||
self.log_message.emit(
|
||
f" [{block_num}/{total_blocks}] "
|
||
f"{sent}/{file_size}: ← {ack}"
|
||
)
|
||
else:
|
||
self.log_message.emit(
|
||
f" [{block_num}/{total_blocks}] "
|
||
f"{sent}/{file_size}: (нет ACK)"
|
||
)
|
||
|
||
serial.timeout = old_timeout
|
||
self.progress.emit(100)
|
||
|
||
# ── Шаг 5: финальный ответ ──
|
||
time.sleep(0.5)
|
||
final_deadline = time.time() + 5.0
|
||
while time.time() < final_deadline:
|
||
if serial.in_waiting > 0:
|
||
raw = serial.read(serial.in_waiting)
|
||
final = raw.decode("ascii", errors="replace").strip()
|
||
if final:
|
||
self.log_message.emit(f"← {final}")
|
||
break
|
||
time.sleep(0.1)
|
||
|
||
self.log_message.emit(f"✓ Отправлено {sent} байт ({block_num} блоков)")
|
||
self.finished.emit(True)
|
||
|
||
except Exception as exc:
|
||
self.log_message.emit(f"Ошибка: {exc}")
|
||
self.finished.emit(False)
|
||
finally:
|
||
if serial and old_timeout is not None:
|
||
try:
|
||
serial.timeout = old_timeout
|
||
except Exception:
|
||
pass
|
||
self._conn.release()
|