diff --git a/tools/production/app/flasher.py b/tools/production/app/flasher.py index ef041c2..879d539 100644 --- a/tools/production/app/flasher.py +++ b/tools/production/app/flasher.py @@ -1,10 +1,26 @@ """ -flasher.py — обёртка над tools/host/flash_usb.py для TUI. +flasher.py — async-обёртка над app/flash_backend.py для TUI. -Запускает flash_usb.py как subprocess, парсит stdout для прогресса, -пробрасывает события через asyncio.Queue в TUI. +Фаза 2: внутренности переведены с subprocess (flash_usb.py) на прямые +вызовы синхронного app/flash_backend.py через asyncio.to_thread. Публичный +API класса Flasher не менялся с Фазы 0 — сигнатуры те же, что в исходной +subprocess-версии. -Не дублирует spsdk-окружение — переиспользует tools/host/ uv-проект. +Мост sync → async для прогресса: flash_backend отдаёт события через +синхронный callback (вызывается из worker-потока asyncio.to_thread), здесь +он оборачивается в asyncio.run_coroutine_threadsafe(...).result() — +согласовано с автором проекта (Фаза 2, вопрос 1): блокирующий .result() +внутри worker-потока гарантирует, что события прогресса приходят в TUI +строго по порядку. + +Отмена прошивки посреди операции НЕ поддерживается и не должна +поддерживаться (согласовано, Фаза 2, вопрос 2) — если стирание/запись уже +начались, они обязаны докрутиться до конца (успешно или с ошибкой). +Специального кода для этого не потребовалось: Python-поток, запущенный +через asyncio.to_thread, нельзя прервать снаружи — если ожидающая +корутина в TUI получит CancelledError, сам поток всё равно продолжит +работу в фоне до естественного завершения flash_backend.flash()/ +erase_chip(). Это и есть требуемое поведение, а не обходной путь. Публичный API: Flasher.flash(target, bin_path, progress_cb) — async, прогресс через callback @@ -18,31 +34,30 @@ from __future__ import annotations import asyncio import logging import os -import re +import shutil import sys -import tempfile from pathlib import Path from typing import Awaitable, Callable, Optional +from . import flash_backend from .models import FcbVariant, FlashProgress, FlashTarget logger = logging.getLogger(__name__) -# VID/PID констант — читаются из env, fallback на известные значения -_BOOTROM_VID = int(os.environ.get("BOOTROM_VID", "0x1fc9"), 16) -_BOOTROM_PID = int(os.environ.get("BOOTROM_PID", "0x0130"), 16) -_CDC_VID = int(os.environ.get("SERVICE_CDC_VID", "0x1996"), 16) -_CDC_PID = int(os.environ.get("SERVICE_CDC_PID", "0x00ad"), 16) - # Тип сборки firmware_test для прошивки (Debug | Release). # Release временно нестабилен (см. отчёт о тестировании) — по умолчанию Debug. _FIRMWARE_BUILD_TYPE = os.environ.get("FIRMWARE_BUILD_TYPE", "Debug") -# Путь до flash_usb.py относительно корня репозитория -_FLASH_USB_SCRIPT = Path(__file__).parents[3] / "tools" / "host" / "flash_usb.py" -_HOST_TOOLS_DIR = _FLASH_USB_SCRIPT.parent -_HAB_DIR = _HOST_TOOLS_DIR / "hab" -_DCD_DIR = _HOST_TOOLS_DIR / "dcd" +# tools/host/build (или переопределено через BUILD_DIR) — та же логика, +# что раньше жила внутри flash_usb.py::main() и была ему видна только +# через CLI-аргументы --firmware/--build-type. Теперь резолвим сами. +_BUILD_DIR = Path(os.environ.get("BUILD_DIR", str(flash_backend.REPO_ROOT / "build"))) + + +def _firmware_hab_path(firmware: str) -> Path: + """Путь к готовому HAB-образу штатной прошивки (см. build.just — тот же + файл, что собирает `nxpimage hab export` при обычной сборке).""" + return _BUILD_DIR / _FIRMWARE_BUILD_TYPE / f"{firmware}_hab.bin" def _resolve_custom_binaries_dir() -> Path: @@ -67,103 +82,114 @@ def _resolve_custom_binaries_dir() -> Path: CUSTOM_BINARIES_DIR = _resolve_custom_binaries_dir() -# Паттерны stdout flash_usb.py для извлечения прогресса -_RE_PERCENT = re.compile(r"(\d{1,3})\s*%") -_RE_PHASE = re.compile(r"(sdphost|blhost|Writing|Erasing|Verifying)", re.IGNORECASE) - ProgressCallback = Callable[[FlashProgress], Awaitable[None]] -def _detect_usb(vid: int, pid: int) -> bool: +def _make_sync_progress_cb( + progress_cb: Optional[ProgressCallback], + loop: asyncio.AbstractEventLoop, +) -> Optional[flash_backend.ProgressCallback]: + """Мост sync (flash_backend, вызывается из worker-потока) → async (TUI). + + .result() — намеренно блокирующий вызов внутри worker-потока: гарантирует + доставку событий прогресса в порядке их возникновения (см. docstring + модуля). Исключения из progress_cb (например, если экран уже закрыт) + логируются и не прерывают саму операцию прошивки — см. договорённость + по отмене (Фаза 2, вопрос 2): прошивка должна докрутиться до конца + независимо от состояния UI. """ - Проверить наличие USB-устройства по VID/PID (синхронно). + if progress_cb is None: + return None - Два метода детекта: - 1. pyusb (usb.core) — видит все USB-устройства включая SDP bulk/HID. - На macOS SDP-устройство (1FC9:0130) не создаёт serial-порт - и невидимо через serial.tools.list_ports. - 2. serial.tools.list_ports — fallback для CDC ACM устройств. - """ - # Метод 1: pyusb — работает для SDP и CDC - try: - import usb.core + def _sync_cb(progress: FlashProgress) -> None: + future = asyncio.run_coroutine_threadsafe(progress_cb(progress), loop) + try: + future.result() + except Exception: + logger.exception("progress_cb выбросил исключение из TUI-потока") - dev = usb.core.find(idVendor=vid, idProduct=pid) - if dev is not None: - return True - except Exception: - pass - - # Метод 2: serial list_ports — fallback для CDC ACM - try: - import serial.tools.list_ports - - for info in serial.tools.list_ports.comports(): - if info.vid == vid and info.pid == pid: - return True - except Exception: - pass - - return False + return _sync_cb class Flasher: """ - Async-обёртка над flash_usb.py. + Async-обёртка над app/flash_backend.py. Пример:: flasher = Flasher() - await flasher.flash( - target=FlashTarget.FIRMWARE_TEST, - progress_cb=lambda p: print(p.message), - ) + + async def on_progress(p: FlashProgress) -> None: + print(p.message) + + await flasher.flash(target=FlashTarget.FIRMWARE_TEST, progress_cb=on_progress) """ - def __init__(self) -> None: - self._proc: Optional[asyncio.subprocess.Process] = None - - # ── Detection ─────────────────────────────────────────────────────────── + # ── Detection (делегируется в flash_backend, см. Р7) ────────────────── @staticmethod def detect_sdp() -> bool: - """True если виден BootROM SDP (1FC9:0130).""" - return _detect_usb(_BOOTROM_VID, _BOOTROM_PID) + """True если виден BootROM SDP.""" + return flash_backend.detect_sdp() @staticmethod def detect_cdc() -> bool: - """True если виден CDC firmware_test (1996:00AD).""" - return _detect_usb(_CDC_VID, _CDC_PID) + """True если виден CDC firmware_test.""" + return flash_backend.detect_cdc() @staticmethod def list_custom_binaries() -> list[Path]: """Отсканировать custom_binaries/ на *.bin, отсортировано по имени.""" return sorted(CUSTOM_BINARIES_DIR.glob("*.bin")) - # ── Erase ──────────────────────────────────────────────────────────────── + # ── Общий раннер: flash_backend-операция в потоке + перевод ошибок ──── + + async def _run_flash_op( + self, + func: Callable[..., None], + *args: object, + async_progress_cb: Optional[ProgressCallback], + sync_progress_cb: Optional[flash_backend.ProgressCallback], + **kwargs: object, + ) -> bool: + """Выполнить flash_backend.flash()/erase_chip() в потоке. + + flash_backend поднимает FlashBackendError вместо возврата False — + здесь это конвертируется обратно в контракт Flasher (bool + событие + phase="error"), который был у subprocess-версии (ненулевой returncode). + """ + try: + await asyncio.to_thread(func, *args, progress_cb=sync_progress_cb, **kwargs) + return True + except flash_backend.FlashBackendError as exc: + logger.error("%s: %s", getattr(func, "__name__", func), exc) + if async_progress_cb is not None: + await async_progress_cb( + FlashProgress(phase="error", percent=0, message=str(exc)) + ) + return False + + # ── Erase ──────────────────────────────────────────────────────────── async def erase_chip( self, progress_cb: Optional[ProgressCallback] = None, ) -> bool: """ - Chip erase Flash через USB SDP (flash_usb.py --erase-chip). + Chip erase Flash через USB SDP (flash_backend.erase_chip). Занимает ~30 с для W25Q128. FCB будет стёрт. :return: True при успехе. """ - cmd = [ - "uv", - "run", - "--directory", - str(_HOST_TOOLS_DIR), - "python", - str(_FLASH_USB_SCRIPT), - "--erase-chip", - ] - return await self._run_cmd(cmd, "erase", progress_cb) + loop = asyncio.get_running_loop() + sync_cb = _make_sync_progress_cb(progress_cb, loop) + return await self._run_flash_op( + flash_backend.erase_chip, + async_progress_cb=progress_cb, + sync_progress_cb=sync_cb, + ) - # ── Flash ──────────────────────────────────────────────────────────────── + # ── Flash ──────────────────────────────────────────────────────────── async def flash( self, @@ -174,257 +200,126 @@ class Flasher: fcb_variant: FcbVariant = FcbVariant.W25Q128, ) -> bool: """ - Запустить прошивку через flash_usb.py. + Запустить прошивку через flash_backend. :param target: Что прошиваем (firmware_test, production или custom). :param progress_cb: Async callback с FlashProgress (может быть None). - :param bin_path: Путь к бинарю (обязателен для CUSTOM). + :param bin_path: Путь к бинарю (обязателен для CUSTOM; для + FIRMWARE_TEST — опционален, уже готовый HAB-образ + вместо штатной сборки из BUILD_DIR). :param use_dcd: Только для CUSTOM — включить DCDFilePath (SDRAM-init) - при сборке HAB-образа через nxpimage. + при сборке HAB-образа через HabImage. :param fcb_variant: Только для CUSTOM — какой явный FCB-блоб (dcd/*_fdcb.bin) записать в Flash[0x60000000] вместо auto-config. :return: True при успехе. """ + loop = asyncio.get_running_loop() + sync_cb = _make_sync_progress_cb(progress_cb, loop) + if target == FlashTarget.FIRMWARE_TEST: - if bin_path is not None: - return await self._run_flash_bin(bin_path, progress_cb) - return await self._run_flash("firmware_test", progress_cb) + hab_bin = ( + bin_path.resolve() + if bin_path is not None + else _firmware_hab_path("firmware_test") + ) + return await self._run_flash_op( + flash_backend.flash, + hab_bin, + async_progress_cb=progress_cb, + sync_progress_cb=sync_cb, + ) + elif target == FlashTarget.PRODUCTION: - ok = await self._run_flash("bootloader", progress_cb) + ok = await self._run_flash_op( + flash_backend.flash, + _firmware_hab_path("bootloader"), + async_progress_cb=progress_cb, + sync_progress_cb=sync_cb, + ) if ok: - ok = await self._run_flash("app", progress_cb) + ok = await self._run_flash_op( + flash_backend.flash, + _firmware_hab_path("app"), + async_progress_cb=progress_cb, + sync_progress_cb=sync_cb, + ) return ok + elif target == FlashTarget.CUSTOM: if bin_path is None: raise ValueError("FlashTarget.CUSTOM требует bin_path") - return await self._run_flash_custom( - bin_path, use_dcd, fcb_variant, progress_cb + return await self._flash_custom( + bin_path, use_dcd, fcb_variant, progress_cb, sync_cb ) + return False - async def _run_flash_custom( + async def _flash_custom( self, raw_bin_path: Path, use_dcd: bool, fcb_variant: FcbVariant, progress_cb: Optional[ProgressCallback], + sync_cb: Optional[flash_backend.ProgressCallback], ) -> bool: """ Прошить «сырой» (не-HAB) кастомный бинарник из custom_binaries/. Два шага: 1. Собрать HAB-образ (IVT + опционально DCD, БЕЗ FCB) через - nxpimage — так же, как just build::hab-* собирает штатные - прошивки, только конфиг генерируется на лету под выбранный файл. - 2. Прошить получившийся HAB-образ через flash_usb.py --bin-path, - подставив явный FCB-блоб (--fcb-path) под выбранный тип памяти — - см. FcbVariant. - """ - if progress_cb is not None: - await progress_cb( - FlashProgress( - phase="nxpimage", percent=0, message="Сборка HAB-образа..." - ) - ) + flash_backend.build_custom_hab() (HabImage, см. Фазу 0) — сам + эмитит phase="hab_build" 0%/100% через sync_cb, отдельно + дублировать здесь не нужно. + 2. Прошить получившийся HAB-образ через flash_backend.flash(), + подставив явный FCB-блоб под выбранный тип памяти (FcbVariant). - hab_bin = await self._build_custom_hab(raw_bin_path, use_dcd, progress_cb) + Временная директория build_custom_hab() (tempfile.mkdtemp) целиком + удаляется в finally — не только *.bin, как было в subprocess-версии. + """ + hab_bin = await self._build_custom_hab( + raw_bin_path, use_dcd, progress_cb, sync_cb + ) if hab_bin is None: + return False + + fcb_path = flash_backend.fcb_blob_path(fcb_variant.fcb_filename) + try: + return await self._run_flash_op( + flash_backend.flash, + hab_bin, + fcb_path=fcb_path, + async_progress_cb=progress_cb, + sync_progress_cb=sync_cb, + ) + finally: + shutil.rmtree(hab_bin.parent, ignore_errors=True) + + async def _build_custom_hab( + self, + raw_bin_path: Path, + use_dcd: bool, + progress_cb: Optional[ProgressCallback], + sync_cb: Optional[flash_backend.ProgressCallback], + ) -> Optional[Path]: + """Обёртка над flash_backend.build_custom_hab() в потоке. + + :return: путь к собранному *.hab.bin, либо None при ошибке. + """ + try: + return await asyncio.to_thread( + flash_backend.build_custom_hab, + raw_bin_path, + use_dcd, + progress_cb=sync_cb, + ) + except flash_backend.FlashBackendError as exc: + logger.error("build_custom_hab: %s", exc) if progress_cb is not None: await progress_cb( FlashProgress( phase="error", percent=0, - message="Ошибка сборки HAB-образа (nxpimage)", + message=f"Ошибка сборки HAB-образа: {exc}", ) ) - return False - - fcb_path = _DCD_DIR / fcb_variant.fcb_filename - try: - cmd = [ - "uv", - "run", - "--directory", - str(_HOST_TOOLS_DIR), - "python", - str(_FLASH_USB_SCRIPT), - "--bin-path", - str(hab_bin), - "--fcb-path", - str(fcb_path), - ] - return await self._run_cmd(cmd, raw_bin_path.stem, progress_cb) - finally: - hab_bin.unlink(missing_ok=True) - - async def _build_custom_hab( - self, - raw_bin: Path, - use_dcd: bool, - progress_cb: Optional[ProgressCallback] = None, - ) -> Optional[Path]: - """ - Собрать HAB-образ из сырого бинарника через nxpimage. - - Временный YAML пишется прямо в tools/host/hab/ (как и штатные - hab_*.yaml) и nxpimage запускается с cwd=tools/host/hab/ — это - обязательно: относительный DCDFilePath ("../dcd/dcd.bin") в - существующих конфигах резолвится именно так (см. build.just, - группа hab_image_gen — `cd tools/host/hab && uv run nxpimage ...`). - Отходить от этой схемы рискованно — nxpimage не документирует - явно, от чего резолвит относительные пути. - - :return: путь к собранному *.hab.bin, либо None при ошибке nxpimage. - """ - yaml_f = tempfile.NamedTemporaryFile( - dir=_HAB_DIR, suffix=".yaml", prefix="_tui_custom_", delete=False - ) - yaml_path = Path(yaml_f.name) - out_path = yaml_path.with_suffix(".hab.bin") - - lines = [ - "options:", - " flags: 0x00", - " startAddress: 0x60000000", - " ivtOffset: 0x1000", - " initialLoadSize: 0x2000", - " family: mimxrt1050", - ] - if use_dcd: - lines.append(" DCDFilePath: ../dcd/dcd.bin") - lines.append(f'inputImageFile: "{raw_bin.resolve()}"') - lines.append("sections: []") - - try: - yaml_f.write("\n".join(lines).encode("utf-8")) - yaml_f.close() - - cmd = [ - "uv", - "run", - "nxpimage", - "hab", - "export", - "--force", - "-c", - str(yaml_path), - "-o", - str(out_path), - ] - logger.info("Building custom HAB image: %s", " ".join(cmd)) - proc = await asyncio.create_subprocess_exec( - *cmd, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.STDOUT, - cwd=str(_HAB_DIR), - ) - assert proc.stdout is not None - async for raw_line in proc.stdout: - line = raw_line.decode("utf-8", errors="replace").rstrip() - logger.debug("nxpimage: %s", line) - if progress_cb is not None and line: - await progress_cb( - FlashProgress(phase="nxpimage", percent=0, message=line) - ) - await proc.wait() - - if proc.returncode != 0 or not out_path.exists(): - logger.error("nxpimage hab export failed (rc=%s)", proc.returncode) - return None - return out_path - finally: - yaml_path.unlink(missing_ok=True) - - async def _run_flash( - self, - firmware: str, - progress_cb: Optional[ProgressCallback], - ) -> bool: - """Запустить flash_usb.py для одного бинаря (стандартный firmware).""" - cmd = [ - "uv", - "run", - "--directory", - str(_HOST_TOOLS_DIR), - "python", - str(_FLASH_USB_SCRIPT), - "--firmware", - firmware, - "--build-type", - _FIRMWARE_BUILD_TYPE, - ] - return await self._run_cmd(cmd, firmware, progress_cb) - - async def _run_flash_bin( - self, - bin_path: Path, - progress_cb: Optional[ProgressCallback], - ) -> bool: - """Запустить flash_usb.py для произвольного бинаря.""" - cmd = [ - "uv", - "run", - "--directory", - str(_HOST_TOOLS_DIR), - "python", - str(_FLASH_USB_SCRIPT), - "--bin-path", - str(bin_path.resolve()), - ] - return await self._run_cmd(cmd, bin_path.stem, progress_cb) - - async def _run_cmd( - self, - cmd: list[str], - label: str, - progress_cb: Optional[ProgressCallback], - ) -> bool: - """Общий subprocess runner для всех операций flash_usb.py.""" - logger.info("Running: %s", " ".join(cmd)) - try: - self._proc = await asyncio.create_subprocess_exec( - *cmd, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.STDOUT, - cwd=str(_HOST_TOOLS_DIR), - ) - assert self._proc.stdout is not None - async for raw_line in self._proc.stdout: - line = raw_line.decode("utf-8", errors="replace").rstrip() - logger.debug("flash_usb [%s]: %s", label, line) - if progress_cb is not None: - progress = _parse_progress(line) - if progress is not None: - await progress_cb(progress) - await self._proc.wait() - success = self._proc.returncode == 0 - if progress_cb is not None: - phase = "done" if success else "error" - msg = "Завершено" if success else "Ошибка" - await progress_cb(FlashProgress(phase=phase, percent=100, message=msg)) - return success - except Exception as exc: - logger.error("Subprocess error [%s]: %s", label, exc) - if progress_cb is not None: - await progress_cb( - FlashProgress(phase="error", percent=0, message=str(exc)) - ) - return False - finally: - self._proc = None - - -def _parse_progress(line: str) -> Optional[FlashProgress]: - """ - Извлечь прогресс из строки stdout flash_usb.py. - Возвращает None если строка не несёт прогресс-информации. - """ - percent_m = _RE_PERCENT.search(line) - phase_m = _RE_PHASE.search(line) - - if percent_m is None and phase_m is None: - return None - - percent = int(percent_m.group(1)) if percent_m else 0 - phase = phase_m.group(1).lower() if phase_m else "flash" - return FlashProgress(phase=phase, percent=percent, message=line.strip()) + return None