399 lines
18 KiB
Python
399 lines
18 KiB
Python
"""Flash firmware tab: factory image or imported developer builds; App-only by default."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
from PySide6.QtCore import QObject, Qt, QTimer, Signal
|
|
from PySide6.QtWidgets import (
|
|
QButtonGroup,
|
|
QCheckBox,
|
|
QFileDialog,
|
|
QInputDialog,
|
|
QComboBox,
|
|
QGroupBox,
|
|
QHBoxLayout,
|
|
QLabel,
|
|
QMessageBox,
|
|
QPlainTextEdit,
|
|
QProgressBar,
|
|
QPushButton,
|
|
QRadioButton,
|
|
QVBoxLayout,
|
|
QWidget,
|
|
)
|
|
|
|
from ..core.session import SESSIONS_DIR
|
|
from ..flash.image import FirmwareImage, FirmwareLibrary, ImageError, bundled_images
|
|
from ..flash.plan import Mode, build_plan
|
|
from ..flash.runner import FlashRunner, Result
|
|
from ..serialio.ports import PortInfo, list_serial_ports
|
|
|
|
BACKUP_DIR = SESSIONS_DIR.parent / "backups"
|
|
VERIFY_TIMEOUT_MS = 45_000
|
|
|
|
HELP = (
|
|
"<b>If flashing cannot connect:</b> use a data-capable USB cable directly on the laptop, close "
|
|
"other programs using the port, then retry. Last resort (case must be opened): hold <b>BOOT</b>, "
|
|
"press and release <b>RESET</b>, release <b>BOOT</b>, and press Start again."
|
|
)
|
|
|
|
|
|
class _Bridge(QObject):
|
|
step = Signal(int, str)
|
|
progress = Signal(int)
|
|
output = Signal(str)
|
|
done = Signal(object)
|
|
|
|
|
|
class FlashTab(QWidget):
|
|
def __init__(self, serial_tab, parent: QWidget | None = None):
|
|
super().__init__(parent)
|
|
self.serial_tab = serial_tab
|
|
self._images: list[FirmwareImage] = []
|
|
self._ports: list[PortInfo] = [] # before any signal can call _update()
|
|
self._runner: FlashRunner | None = None
|
|
self._port: PortInfo | None = None
|
|
self._image: FirmwareImage | None = None
|
|
|
|
# 1. box
|
|
box = QGroupBox("1. Sensorbox")
|
|
self.port_combo = QComboBox(minimumWidth=360)
|
|
refresh = QPushButton("Refresh")
|
|
refresh.clicked.connect(self.refresh_ports)
|
|
self.box_fw = QLabel(wordWrap=True)
|
|
bl = QVBoxLayout(box)
|
|
row = QHBoxLayout()
|
|
row.addWidget(self.port_combo)
|
|
row.addWidget(refresh)
|
|
row.addStretch(1)
|
|
bl.addLayout(row)
|
|
bl.addWidget(self.box_fw)
|
|
|
|
# 2. firmware
|
|
fwbox = QGroupBox("2. Firmware")
|
|
self.image_combo = QComboBox(minimumWidth=360)
|
|
self.import_btn = QPushButton("Import firmware file...")
|
|
self.import_btn.setToolTip("Add a newer build the firmware team sent you (.bin)")
|
|
self.import_btn.clicked.connect(self._import)
|
|
self.remove_btn = QPushButton("Remove from list")
|
|
self.remove_btn.clicked.connect(self._remove)
|
|
self.image_info = QLabel(wordWrap=True)
|
|
self.image_info.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse)
|
|
fl = QVBoxLayout(fwbox)
|
|
fl.addWidget(self.image_combo)
|
|
brow = QHBoxLayout()
|
|
brow.addWidget(self.import_btn)
|
|
brow.addWidget(self.remove_btn)
|
|
brow.addStretch(1)
|
|
fl.addLayout(brow)
|
|
fl.addWidget(self.image_info)
|
|
self._library = FirmwareLibrary()
|
|
self._reference: FirmwareImage | None = None
|
|
self._problems: list[str] = []
|
|
|
|
# 3. mode
|
|
modebox = QGroupBox("3. What to write")
|
|
self.rb_app = QRadioButton("App only (recommended for repairs)")
|
|
self.rb_full = QRadioButton("Full factory image (new or completely wiped boards)")
|
|
self.mode_group = QButtonGroup(self)
|
|
self.mode_group.addButton(self.rb_app)
|
|
self.mode_group.addButton(self.rb_full)
|
|
self.rb_app.setChecked(True)
|
|
app_note = QLabel("Replaces the firmware only. Keeps the Matter pairing, the OTA registration and the "
|
|
"settings stored on the box.", wordWrap=True)
|
|
full_note = QLabel("Writes everything like the factory. <b>Erases the Matter pairing and the OTA "
|
|
"registration</b> - the box must be paired again afterwards.", wordWrap=True)
|
|
for n in (app_note, full_note):
|
|
n.setContentsMargins(24, 0, 0, 6)
|
|
n.setStyleSheet("color:#555")
|
|
self.confirm_full = QCheckBox("I understand the pairing and OTA registration will be erased")
|
|
self.backup = QCheckBox("Back up the box's complete flash first (8 MB, about 3 minutes)")
|
|
ml = QVBoxLayout(modebox)
|
|
for w in (self.rb_app, app_note, self.rb_full, full_note, self.confirm_full, self.backup):
|
|
ml.addWidget(w)
|
|
|
|
# 4. run
|
|
runbox = QGroupBox("4. Flash")
|
|
self.plan_label = QLabel(wordWrap=True)
|
|
self.start_btn = QPushButton("Start flashing")
|
|
self.start_btn.setMinimumHeight(36)
|
|
self.start_btn.clicked.connect(self._start)
|
|
self.step_label = QLabel()
|
|
self.bar = QProgressBar()
|
|
self.result = QLabel(wordWrap=True)
|
|
self.result.setMargin(8)
|
|
self.result.setTextFormat(Qt.TextFormat.RichText)
|
|
self.details_btn = QPushButton("Show esptool output")
|
|
self.details_btn.setCheckable(True)
|
|
self.output = QPlainTextEdit(readOnly=True)
|
|
self.output.setVisible(False)
|
|
self.details_btn.toggled.connect(self.output.setVisible)
|
|
rl = QVBoxLayout(runbox)
|
|
for w in (self.plan_label, self.start_btn, self.step_label, self.bar, self.result, self.details_btn, self.output):
|
|
rl.addWidget(w)
|
|
help_lbl = QLabel(HELP, wordWrap=True)
|
|
help_lbl.setStyleSheet("color:#555")
|
|
rl.addWidget(help_lbl)
|
|
rl.addStretch(1) # keep the run column packed at the top
|
|
|
|
left = QVBoxLayout()
|
|
for w in (box, fwbox, modebox):
|
|
left.addWidget(w)
|
|
left.addStretch(1)
|
|
lay = QHBoxLayout(self)
|
|
lay.addLayout(left, 1)
|
|
lay.addWidget(runbox, 1)
|
|
|
|
self._bridge = _Bridge()
|
|
self._bridge.step.connect(self._on_step)
|
|
self._bridge.progress.connect(self.bar.setValue)
|
|
self._bridge.output.connect(self.output.appendPlainText)
|
|
self._bridge.done.connect(self._on_done)
|
|
self._verify_timer = QTimer(self, interval=500)
|
|
self._verify_timer.timeout.connect(self._check_verify)
|
|
|
|
for sig in (self.rb_app.toggled, self.confirm_full.toggled, self.backup.toggled):
|
|
sig.connect(self._update)
|
|
self.rb_full.toggled.connect(lambda on: self.backup.setChecked(on)) # backup default on for Full
|
|
self.image_combo.currentIndexChanged.connect(self._update)
|
|
self.port_combo.currentIndexChanged.connect(self._update)
|
|
|
|
self._load_images()
|
|
self.refresh_ports()
|
|
self._show_result("", None)
|
|
|
|
# ------------------------------------------------------------ setup
|
|
def _load_images(self, select_sha: str | None = None) -> None:
|
|
self._problems = []
|
|
try:
|
|
bundled = bundled_images()
|
|
except ImageError as exc:
|
|
bundled = []
|
|
self._problems.append(f"bundled image: {exc}")
|
|
factory = [i for i in bundled if not i.is_developer]
|
|
self._reference = factory[0] if factory else None
|
|
dev, problems = self._library.images(self._reference)
|
|
dev += [i for i in bundled if i.is_developer]
|
|
self._problems += problems
|
|
# Newest developer builds first, then the factory image.
|
|
dev.sort(key=lambda i: i.app.built_at or dt.datetime.min, reverse=True)
|
|
self._images = dev + factory
|
|
self.image_combo.blockSignals(True)
|
|
self.image_combo.clear()
|
|
for img in self._images:
|
|
self.image_combo.addItem(img.label())
|
|
if not self._images:
|
|
self.image_combo.addItem("No firmware image available - import one")
|
|
idx = next((i for i, im in enumerate(self._images) if im.sha256 == select_sha), None)
|
|
if idx is None:
|
|
idx = len(dev) if factory else 0 # default: the factory image
|
|
self.image_combo.setCurrentIndex(idx)
|
|
self.image_combo.blockSignals(False)
|
|
self._update()
|
|
|
|
def _import(self) -> None:
|
|
path, _ = QFileDialog.getOpenFileName(self, "Import firmware from the firmware team", str(Path.home()),
|
|
"Firmware (*.bin);;All files (*)")
|
|
if not path:
|
|
return
|
|
built_by, ok = QInputDialog.getText(self, "Who built this firmware?",
|
|
"Name of the developer who built this image (shown next to it):")
|
|
if not ok:
|
|
return
|
|
try:
|
|
img = self._library.import_file(Path(path), self._reference, built_by=built_by)
|
|
except (ImageError, OSError) as exc:
|
|
QMessageBox.warning(self, "Cannot use this file", f"This file cannot be flashed:\n\n{exc}")
|
|
return
|
|
self._load_images(select_sha=img.sha256)
|
|
kind = "complete image" if img.kind == "merged" else "app only"
|
|
QMessageBox.information(
|
|
self, "Firmware imported",
|
|
f"{img.name}\nbuild {img.app.elf_sha[:9]}, {kind}\n\nFile fingerprint (SHA-256):\n{img.sha256}\n\n"
|
|
"Compare it with the fingerprint the firmware team sent you before flashing.")
|
|
|
|
def _remove(self) -> None:
|
|
img = self._selected_image()
|
|
if img and img.is_developer and QMessageBox.question(
|
|
self, "Remove firmware", f"Remove {img.name} from this computer?") == QMessageBox.StandardButton.Yes:
|
|
self._library.remove(img.sha256)
|
|
self._load_images()
|
|
|
|
def refresh_ports(self) -> None:
|
|
current = self.port_combo.currentData()
|
|
self._ports = list_serial_ports()
|
|
self.port_combo.clear()
|
|
for p in self._ports:
|
|
self.port_combo.addItem(p.label, p.device)
|
|
if not self._ports:
|
|
self.port_combo.addItem("No sensorbox found - check USB cable and power switch", None)
|
|
self.port_combo.setCurrentIndex(max(self.port_combo.findData(current), 0))
|
|
self._update()
|
|
|
|
def showEvent(self, event): # noqa: N802 - refresh when the tab is opened
|
|
self.refresh_ports()
|
|
super().showEvent(event)
|
|
|
|
# ------------------------------------------------------------ state
|
|
def _selected_image(self) -> FirmwareImage | None:
|
|
i = self.image_combo.currentIndex()
|
|
return self._images[i] if 0 <= i < len(self._images) else None
|
|
|
|
def _selected_port(self) -> PortInfo | None:
|
|
dev = self.port_combo.currentData()
|
|
return next((p for p in self._ports if p.device == dev), None)
|
|
|
|
def _update(self) -> None:
|
|
img = self._selected_image()
|
|
info = ""
|
|
if img:
|
|
a = img.app
|
|
origin = ("<b>Factory image</b> (bundled with the tool)" if not img.is_developer else
|
|
"<b style='color:#c77c02'>Developer build</b> - only flash builds you received "
|
|
"from the firmware team")
|
|
kind = "complete image (bootloader + layout + app)" if img.kind == "merged" else "app only"
|
|
info = (f"{origin}<br>{a.project} {a.version}, built {a.built}, ESP-IDF {a.idf}, {kind}<br>"
|
|
f"<small>build {a.elf_sha[:9]} · file SHA-256 {img.sha256[:16]}... "
|
|
f"· checksum OK</small>" + (f"<br><small>{img.notes}</small>" if img.notes else ""))
|
|
if self._problems:
|
|
info += ("<br><span style='color:#b03a2e'><b>Not offered:</b> "
|
|
+ "; ".join(self._problems) + "</span>")
|
|
self.image_info.setText(info)
|
|
self.remove_btn.setEnabled(bool(img and img.is_developer) and self._runner is None)
|
|
# An app-only file has no bootloader/layout: Full mode is impossible.
|
|
full_ok = bool(img and img.kind == "merged")
|
|
self.rb_full.setEnabled(full_ok and self._runner is None)
|
|
if not full_ok and self.rb_full.isChecked():
|
|
self.rb_app.setChecked(True)
|
|
self.box_fw.setText(self._box_text(img))
|
|
|
|
self.confirm_full.setVisible(self.rb_full.isChecked())
|
|
ok = bool(img and self._selected_port() and self._runner is None
|
|
and (self.rb_app.isChecked() or self.confirm_full.isChecked()))
|
|
self.start_btn.setEnabled(ok)
|
|
if img and self._selected_port():
|
|
mode = Mode.APP_ONLY if self.rb_app.isChecked() else Mode.FULL
|
|
steps = ["Back up the complete flash (~3 min)"] if self.backup.isChecked() else []
|
|
if mode == Mode.APP_ONLY:
|
|
steps += ["Check the box has the same partition layout", "Write the firmware app (~20 s)"]
|
|
else:
|
|
steps += ["Write the complete factory image (~25 s)"]
|
|
steps += ["Restart the box and verify the new firmware boots"]
|
|
self.plan_label.setText("<b>Plan:</b><ol>" + "".join(f"<li>{s}</li>" for s in steps) + "</ol>")
|
|
else:
|
|
self.plan_label.setText("Select a sensorbox and a firmware image.")
|
|
|
|
def _box_text(self, img: FirmwareImage | None) -> str:
|
|
fw = self.serial_tab.engine.firmware
|
|
if not fw.elf_sha:
|
|
return "Firmware on the box: unknown - connect in the Serial monitor and press Reset & capture to see it."
|
|
msg = f"Box currently runs: {fw.describe()}" + (f", built {fw.built}" if fw.built else "")
|
|
if not img:
|
|
return msg
|
|
if fw.elf_sha.startswith(img.app.elf_sha[:9]) or img.app.elf_sha.startswith(fw.elf_sha):
|
|
return msg + "<br><i>It already has this firmware. Re-flash only if you suspect the firmware is damaged.</i>"
|
|
box_built, img_built = _parse_built(fw.built), img.app.built_at
|
|
if box_built and img_built and img_built > box_built:
|
|
return msg + f"<br><b style='color:#2e6da4'>The selected firmware is newer (built {img.app.built}).</b>"
|
|
if box_built and img_built and img_built < box_built:
|
|
return msg + ("<br><b style='color:#c77c02'>The selected firmware is OLDER than the box's (the box was "
|
|
"probably updated over OTA). The box updates itself again once online.</b>")
|
|
return msg + "<br><b style='color:#c77c02'>The box runs a different build.</b>"
|
|
|
|
# ------------------------------------------------------------ flash
|
|
def _start(self) -> None:
|
|
img, port = self._selected_image(), self._selected_port()
|
|
if not img or not port:
|
|
return
|
|
mode = Mode.APP_ONLY if self.rb_app.isChecked() else Mode.FULL
|
|
what = ("the firmware app (pairing and settings are kept)" if mode == Mode.APP_ONLY
|
|
else "the COMPLETE factory image (pairing and OTA registration are erased)")
|
|
if QMessageBox.question(self, "Flash firmware",
|
|
f"Write {what}\n\nImage: {img.name}\nBox: {port.label}\n\n"
|
|
"Keep the box powered and the USB cable connected until it finishes.",
|
|
QMessageBox.StandardButton.Ok | QMessageBox.StandardButton.Cancel
|
|
) != QMessageBox.StandardButton.Ok:
|
|
return
|
|
backup_to = None
|
|
if self.backup.isChecked():
|
|
BACKUP_DIR.mkdir(parents=True, exist_ok=True)
|
|
stamp = dt.datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
backup_to = BACKUP_DIR / f"{(port.mac or port.device).replace(':', '')}_{stamp}.bin"
|
|
plan = build_plan(img, mode, port.device, Path(tempfile.mkdtemp(prefix="kota_flash_")), backup_to)
|
|
|
|
self._port, self._image = port, img
|
|
self.serial_tab.release_port() # esptool needs the port exclusively
|
|
self.output.clear()
|
|
self.output.appendPlainText("\n".join(plan.summary()))
|
|
self._show_result("", None)
|
|
self._runner = FlashRunner(plan, self._bridge.step.emit, self._bridge.progress.emit,
|
|
self._bridge.output.emit, self._bridge.done.emit)
|
|
self._set_busy(True)
|
|
self._runner.start()
|
|
|
|
def _set_busy(self, busy: bool) -> None:
|
|
for w in (self.port_combo, self.image_combo, self.rb_app, self.backup, self.confirm_full, self.import_btn):
|
|
w.setEnabled(not busy)
|
|
self._update()
|
|
|
|
def _on_step(self, i: int, title: str) -> None:
|
|
n = len(self._runner.plan.steps) if self._runner else 0
|
|
self.step_label.setText(f"Step {i + 1} of {n + 1}: {title}")
|
|
|
|
def _on_done(self, res: Result) -> None:
|
|
self._runner = None
|
|
if not res.ok:
|
|
self._set_busy(False)
|
|
self.step_label.setText("Stopped.")
|
|
hint = f"<p>{res.hint}</p>" if res.hint else ""
|
|
self._show_result(f"<b>Flashing failed.</b> {res.message}{hint}", "error")
|
|
self.details_btn.setChecked(True)
|
|
return
|
|
self.step_label.setText("Final step: restarting the box and checking which firmware it boots...")
|
|
self.bar.setValue(0)
|
|
self._show_result("Firmware written. Waiting for the box to start...", "info")
|
|
self.serial_tab.connect_port(self._port, capture=True)
|
|
self._verify_ms = 0
|
|
self._verify_timer.start()
|
|
|
|
def _check_verify(self) -> None:
|
|
self._verify_ms += self._verify_timer.interval()
|
|
self.bar.setValue(min(100, int(100 * self._verify_ms / VERIFY_TIMEOUT_MS)))
|
|
fw = self.serial_tab.engine.firmware
|
|
want = self._image.app.elf_sha[:9]
|
|
done = fw.elf_sha is not None and fw.elf_sha.startswith(want)
|
|
if done or self._verify_ms >= VERIFY_TIMEOUT_MS:
|
|
self._verify_timer.stop()
|
|
self._set_busy(False)
|
|
self.bar.setValue(100)
|
|
if done:
|
|
self.step_label.setText("Done.")
|
|
self._show_result(f"<b>Success.</b> The box boots the new firmware (build {want}). "
|
|
"Check the Diagnosis panel in the Serial monitor.", "ok")
|
|
else:
|
|
seen = fw.elf_sha[:9] if fw.elf_sha else "no boot seen"
|
|
self._show_result(f"<b>Written, but not verified.</b> Expected build {want}, saw: {seen}. "
|
|
"Open the Serial monitor and press Reset & capture.", "warning")
|
|
|
|
def _show_result(self, html: str, kind: str | None) -> None:
|
|
colours = {"ok": ("#2e9e44", "#e3f4e7"), "info": ("#2e6da4", "#e3edf7"),
|
|
"warning": ("#c77c02", "#fdf0d6"), "error": ("#d9534f", "#f9e0df")}
|
|
self.result.setVisible(bool(html))
|
|
if html:
|
|
border, bg = colours[kind or "info"]
|
|
self.result.setStyleSheet(f"QLabel {{ border:2px solid {border}; background:{bg}; border-radius:6px; }}")
|
|
self.result.setText(html)
|
|
|
|
|
|
def _parse_built(s: str | None) -> dt.datetime | None:
|
|
"""'Jun 29 2026 11:03:32' (app_init "Compile time") -> datetime."""
|
|
if not s:
|
|
return None
|
|
try:
|
|
return dt.datetime.strptime(" ".join(s.split()), "%b %d %Y %H:%M:%S")
|
|
except ValueError:
|
|
return None
|