adam_di_akimow/lift_indicator_suite
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

# Phase 4: connection loss handling with spsdk

Browse source
This commit is contained in:
Dmitry Akimov 2026-07-03 17:19:16 +03:00
parent ad77c897e3
commit 234f1a60cf
8 changed files with 301 additions and 62 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

BIN
tools/production/app/TFT_BOOTLOADER_OLD.bin Normal file
View file

Binary file not shown.
Side by side

After

Width:  |  Height:  |  Size: 46 KiB

192
tools/production/app/flash_backend.py
View file

@ -9,12 +9,18 @@ Zero Textual/asyncio импортов — модуль полностью син
без event loop (см. test_flash_backend.py). Async-обвязка (asyncio.to_thread
+ run_coroutine_threadsafe) — забота Flasher (Фаза 2), не этого модуля.
Открытые вопросы/допущения (см. сопроводительное сообщение в чате):
Открытые вопросы/допущения:
- Одна сессия McuBoot на весь flash()/erase_chip(), а не переоткрытие
на каждую операцию, как в CLI flash_usb.py.
- Имена FlashProgress.phase свои (не парсинг stdout blhost).
- Иерархия исключений — предложение автора модуля, не зафиксирована
отдельно в MONOLITH_APP_PLAN.md.
Фаза 4: SPSDKConnectionError оборачивается в ConnectionLostError на всех
трёх точках отказа (SDP write, McuBoot handshake+команды, McuBoot chip erase),
что позволяет Flasher/TUI отличить обрыв USB от логической ошибки через
поле connection_lost. USB-интерфейс, полученный из load_flashloader(),
закрывается в finally на любом исходе (защита от утечки HID-хэндла в
редком окне «wait_for_flashloader вернул интерфейс → USB выдернут →
McuBoot.__enter__ упал»).
"""
from __future__ import annotations
@ -22,11 +28,13 @@ from __future__ import annotations
import hashlib
import logging
import os
import sys
import tempfile
import time
from pathlib import Path
from typing import Callable, Optional
from spsdk.exceptions import SPSDKConnectionError
from spsdk.image.hab.hab_image import HabImage
from spsdk.mboot import McuBoot, MbootUSBInterface
from spsdk.sdp import SDP, SdpUSBInterface
@ -52,6 +60,15 @@ def fcb_blob_path(fcb_filename: str) -> Path:
return _HOST_DCD_DIR / fcb_filename
def firmware_hab_path(firmware: str, build_type: str) -> Path:
"""Путь к готовому HAB-образу штатной прошивки (Р6, двухрежимный резолв)."""
if getattr(sys, "frozen", False):
base = Path(sys.executable).resolve().parent / "firmware"
else:
base = Path(os.environ.get("BUILD_DIR", str(REPO_ROOT / "build")))
return base / build_type / f"{firmware}_hab.bin"
# ─── USB VID:PID (те же переменные окружения, что уже приняты в проекте) ──
_BOOTROM_VID = int(os.environ.get("BOOTROM_VID", "0x1fc9"), 16)
@ -92,7 +109,38 @@ _HAB_OPTIONS_TEMPLATE = [
class FlashBackendError(Exception):
"""Базовая ошибка flash_backend."""
"""Базовая ошибка flash_backend.
:cvar connection_lost: True если ошибка связана с потерей физического
соединения (USB выдернут, устройство пропало с шины). Позволяет
Flasher/TUI отличить обрыв от логической ошибки без парсинга
текста сообщения. По умолчанию False; подкласс ConnectionLostError
переопределяет на True.
"""
connection_lost: bool = False
def __init__(self, message: str, *, connection_lost: bool = False) -> None:
super().__init__(message)
# instance-level override — на случай, если базовый класс поднимается
# напрямую с connection_lost=True без использования ConnectionLostError
if connection_lost:
self.connection_lost = True
class ConnectionLostError(FlashBackendError):
"""Потеря USB-соединения посреди операции (обёртка над SPSDKConnectionError).
Обёртка над spsdk.exceptions.SPSDKConnectionError, поднимается при
исчезновении устройства с шины во время выполнения SDP/McuBoot команд.
Отличается от DeviceNotFoundError, который возникает ДО начала операции
(устройство никогда не было подключено).
"""
connection_lost: bool = True
def __init__(self, message: str) -> None:
super().__init__(message, connection_lost=True)
class DeviceNotFoundError(FlashBackendError):
@ -127,6 +175,21 @@ def _emit(
progress_cb(FlashProgress(phase=phase, percent=percent, message=message))
def _close_iface_quiet(iface: MbootUSBInterface) -> None:
"""Закрыть интерфейс в finally, глотая любые ошибки.
Внутри McuBoot() как context-manager закрытие уже происходит, поэтому
повторное close() на закрытом интерфейсе может выкинуть исключение
из libusbsio — нам это не важно, мы просто хотим гарантию, что если
McuBoot.__enter__ упал (окно между load_flashloader и with McuBoot()),
интерфейс не остался висеть с открытым HID-хэндлом.
"""
try:
iface.close()
except Exception:
pass
# ─── Flashloader bring-up ───────────────────────────────────────────────────
@ -155,6 +218,7 @@ def load_flashloader(
:raises DeviceNotFoundError: SDP-устройство не найдено (и Flashloader тоже не поднят).
:raises FlashLoaderTimeoutError: см. wait_for_flashloader().
:raises ConnectionLostError: USB-соединение потеряно во время SDP-обмена.
"""
already = MbootUSBInterface.scan(device_id=_FLASHLOADER_DEVICE_ID)
if already:
@ -176,9 +240,14 @@ def load_flashloader(
f"Загрузка Flashloader через SDP ({_SDP_DEVICE_ID})",
)
data = FLASHLOADER_BIN.read_bytes()
with SDP(sdp_devices[0]) as sdp:
sdp.write_file(FLASHLOADER_LOAD_ADDR, data)
sdp.jump_and_run(FLASHLOADER_LOAD_ADDR)
try:
with SDP(sdp_devices[0]) as sdp:
sdp.write_file(FLASHLOADER_LOAD_ADDR, data)
sdp.jump_and_run(FLASHLOADER_LOAD_ADDR)
except SPSDKConnectionError as exc:
raise ConnectionLostError(
f"USB-соединение потеряно при загрузке Flashloader: {exc}"
) from exc
iface = wait_for_flashloader()
_emit(progress_cb, "load_flashloader", 100, "Flashloader готов")
@ -239,6 +308,7 @@ def flash(
:param fcb_path: Явный FCB-блоб (custom-бинари). None → auto-config
(write_fcb_auto, только для штатных firmware_test/production).
:raises FlashBackendError: и подклассы — на любой ошибке.
:raises ConnectionLostError: обрыв USB посреди операции.
"""
if not hab_bin.exists():
raise FlashBackendError(f"Файл не найден: {hab_bin}")
@ -249,9 +319,14 @@ def flash(
if not sdp_devices:
raise DeviceNotFoundError(f"SDP-устройство не найдено ({_SDP_DEVICE_ID})")
addr = FLASH_BASE + HAB_OFFSET
with SDP(sdp_devices[0]) as sdp:
sdp.write_file(addr, hab_bin.read_bytes())
sdp.jump_and_run(addr)
try:
with SDP(sdp_devices[0]) as sdp:
sdp.write_file(addr, hab_bin.read_bytes())
sdp.jump_and_run(addr)
except SPSDKConnectionError as exc:
raise ConnectionLostError(
f"USB-соединение потеряно при RAM-загрузке: {exc}"
) from exc
_emit(progress_cb, "done", 100, "Загружено в RAM")
return
@ -261,59 +336,82 @@ def flash(
hab_size = hab_bin.stat().st_size
erase_size = ((HAB_OFFSET + hab_size + 0xFFF) // 0x1000) * 0x1000
with McuBoot(iface) as mboot:
_emit(progress_cb, "configure", 0, "Конфигурация FlexSPI NOR")
configure_flexspi(mboot)
# try/finally гарантирует закрытие iface даже если McuBoot.__enter__ падает
# ДО того, как SDP context-manager отработает (тонкое окно, но реальное:
# wait_for_flashloader вернул интерфейс, USB выдернут до McuBoot handshake).
try:
with McuBoot(iface) as mboot:
_emit(progress_cb, "configure", 0, "Конфигурация FlexSPI NOR")
configure_flexspi(mboot)
_emit(
progress_cb, "erase", 0, f"Стирание 0x{FLASH_BASE:08X} + {erase_size} байт"
)
ok = mboot.flash_erase_region(FLASH_BASE, erase_size, mem_id=0)
if not ok:
raise FlashBackendError("flash_erase_region вернул False")
_emit(
progress_cb,
"erase",
0,
f"Стирание 0x{FLASH_BASE:08X} + {erase_size} байт",
)
ok = mboot.flash_erase_region(FLASH_BASE, erase_size, mem_id=0)
if not ok:
raise FlashBackendError("flash_erase_region вернул False")
_emit(progress_cb, "fcb", 0, "Запись FCB")
if fcb_path is not None:
write_fcb_explicit(mboot, fcb_path)
else:
write_fcb_auto(mboot)
_emit(progress_cb, "fcb", 0, "Запись FCB")
if fcb_path is not None:
write_fcb_explicit(mboot, fcb_path)
else:
write_fcb_auto(mboot)
_emit(progress_cb, "write", 0, f"Запись {hab_bin.name} ({hab_size} байт)")
_emit(progress_cb, "write", 0, f"Запись {hab_bin.name} ({hab_size} байт)")
def _on_progress(current: int, total: int) -> None:
percent = int(current * 100 / total) if total else 0
_emit(progress_cb, "write", percent, f"{current}/{total} байт")
def _on_progress(current: int, total: int) -> None:
percent = int(current * 100 / total) if total else 0
_emit(progress_cb, "write", percent, f"{current}/{total} байт")
data = hab_bin.read_bytes()
ok = mboot.write_memory(
write_addr, data, mem_id=0, progress_callback=_on_progress
)
if not ok:
raise FlashBackendError("write_memory (HAB-образ) вернул False")
data = hab_bin.read_bytes()
ok = mboot.write_memory(
write_addr, data, mem_id=0, progress_callback=_on_progress
)
if not ok:
raise FlashBackendError("write_memory (HAB-образ) вернул False")
_emit(progress_cb, "reset", 0, "Reset")
mboot.reset(reopen=False)
_emit(progress_cb, "reset", 0, "Reset")
mboot.reset(reopen=False)
except SPSDKConnectionError as exc:
raise ConnectionLostError(
f"USB-соединение потеряно во время прошивки: {exc}"
) from exc
finally:
_close_iface_quiet(iface)
_emit(progress_cb, "done", 100, "Прошивка завершена успешно")
def erase_chip(progress_cb: Optional[ProgressCallback] = None) -> None:
"""Полная очистка Flash (см. flash_usb.py::erase_chip). После erase FCB
тоже стёрт — плата не загрузится до следующей прошивки."""
тоже стёрт — плата не загрузится до следующей прошивки.
:raises ConnectionLostError: обрыв USB посреди chip erase.
"""
iface = load_flashloader(progress_cb)
with McuBoot(iface) as mboot:
_emit(progress_cb, "configure", 0, "Конфигурация FlexSPI NOR")
configure_flexspi(mboot)
try:
with McuBoot(iface) as mboot:
_emit(progress_cb, "configure", 0, "Конфигурация FlexSPI NOR")
configure_flexspi(mboot)
_emit(progress_cb, "erase", 0, "Полная очистка Flash (~30с)")
iface.device.timeout = ERASE_ALL_TIMEOUT_MS # см. Фазу 0, ⚠В2
ok = mboot.flash_erase_all(mem_id=FLEXSPI_MEMORY_ID)
if not ok:
raise FlashBackendError("flash_erase_all вернул False")
_emit(progress_cb, "erase", 0, "Полная очистка Flash (~30с)")
iface.device.timeout = ERASE_ALL_TIMEOUT_MS # см. Фазу 0, ⚠В2
ok = mboot.flash_erase_all(mem_id=FLEXSPI_MEMORY_ID)
if not ok:
raise FlashBackendError("flash_erase_all вернул False")
_emit(progress_cb, "reset", 0, "Reset")
mboot.reset(reopen=False)
_emit(progress_cb, "reset", 0, "Reset")
mboot.reset(reopen=False)
except SPSDKConnectionError as exc:
raise ConnectionLostError(
f"USB-соединение потеряно во время chip erase: {exc}"
) from exc
finally:
_close_iface_quiet(iface)
_emit(progress_cb, "done", 100, "Chip erase завершён")

68
tools/production/app/flasher.py
View file

@ -48,16 +48,10 @@ logger = logging.getLogger(__name__)
# Release временно нестабилен (см. отчёт о тестировании) — по умолчанию Debug.
_FIRMWARE_BUILD_TYPE = os.environ.get("FIRMWARE_BUILD_TYPE", "Debug")
# 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"
"""Делегирование в backend (Р6): dev/frozen резолв живёт там."""
return flash_backend.firmware_hab_path(firmware, _FIRMWARE_BUILD_TYPE)
def _resolve_custom_binaries_dir() -> Path:
@ -111,6 +105,20 @@ def _make_sync_progress_cb(
return _sync_cb
def _format_error_message(exc: BaseException) -> str:
"""Человекочитаемое сообщение для FlashProgress(phase="error").
Для ConnectionLostError (и любого FlashBackendError с connection_lost=True)
добавляет явный префикс — вариант А (согласовано): специального перехода
экрана нет, но в #flash-log причина должна читаться однозначно, без
необходимости лезть в общий лог-файл за трейсбеком.
"""
connection_lost = getattr(exc, "connection_lost", False)
if connection_lost:
return f"Соединение с платой потеряно: {exc}"
return str(exc)
class Flasher:
"""
Async-обёртка над app/flash_backend.py.
@ -154,9 +162,17 @@ class Flasher:
) -> bool:
"""Выполнить flash_backend.flash()/erase_chip() в потоке.
flash_backend поднимает FlashBackendError вместо возврата False —
здесь это конвертируется обратно в контракт Flasher (bool + событие
phase="error"), который был у subprocess-версии (ненулевой returncode).
flash_backend поднимает FlashBackendError (включая ConnectionLostError,
см. Фазу 4) вместо возврата False — здесь это конвертируется обратно
в контракт Flasher (bool + событие phase="error"), который был у
subprocess-версии (ненулевой returncode).
Отдельный except Exception — safety net (Фаза 4, согласовано):
любое непредвиденное исключение из worker-потока (не только
FlashBackendError) обязано вернуть управление в TUI с ok=False,
а не оставить кнопки заблокированными навсегда. KeyboardInterrupt/
SystemExit/CancelledError не перехватываются — это BaseException,
не Exception, пробрасываются как есть
"""
try:
await asyncio.to_thread(func, *args, progress_cb=sync_progress_cb, **kwargs)
@ -165,7 +181,22 @@ class Flasher:
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))
FlashProgress(
phase="error", percent=0, message=_format_error_message(exc)
)
)
return False
except Exception as exc: # noqa: BLE001 — safety net, см. docstring
logger.exception(
"%s: непредвиденная ошибка", getattr(func, "__name__", func)
)
if async_progress_cb is not None:
await async_progress_cb(
FlashProgress(
phase="error",
percent=0,
message=f"Непредвиденная ошибка: {exc}",
)
)
return False
@ -319,7 +350,18 @@ class Flasher:
FlashProgress(
phase="error",
percent=0,
message=f"Ошибка сборки HAB-образа: {exc}",
message=f"Ошибка сборки HAB-образа: {_format_error_message(exc)}",
)
)
return None
except Exception as exc: # noqa: BLE001 — safety net, см. _run_flash_op
logger.exception("build_custom_hab: непредвиденная ошибка")
if progress_cb is not None:
await progress_cb(
FlashProgress(
phase="error",
percent=0,
message=f"Непредвиденная ошибка сборки HAB-образа: {exc}",
)
)
return None

3
tools/production/app/models.py
View file

@ -131,6 +131,7 @@ class ConfirmRequest:
class FlashProgress:
"""Прогресс прошивки."""
phase: str # "sdphost" | "blhost" | "done" | "error"
phase: str # "load_flashloader" | "configure" | "erase" | "fcb" | "write"
# | "reset" | "hab_build" | "done" | "error"
percent: int # 0..100
message: str

5
tools/production/app/screens/flash.py
View file

@ -49,7 +49,8 @@ class FlashScreen(Screen, ConnectionWatcherMixin):
Если плата физически отключена в простое — сессия считается
недостоверной, экран сразу уходит на WaitingScreen (см. замечание
№4 отчёта). Во время самой прошивки/erase мониторинг приостановлен —
обрыв в этом случае обнаружит и обработает сам flash_usb.py subprocess.
обрыв в этом случае обнаруживает сам flash_backend (SPSDKConnectionError
→ ConnectionLostError, см. Фазу 4) и репортит через #flash-log.
"""
BINDINGS = [
@ -154,6 +155,8 @@ class FlashScreen(Screen, ConnectionWatcherMixin):
def _check_sdp_present(self) -> bool:
# Не считаем потерей соединения, если идёт активная операция —
# flash_usb.py сам обработает реальный обрыв через subprocess.
# обрыв в этом случае обнаружит и обработает сам flash_backend
# (ConnectionLostError, см. Фазу 4), не watcher.
if self._flashing:
return True
return Flasher.detect_sdp()

0
tools/production/app/firmware_test_hab.bin → tools/production/custom_binaries/firmware_test_hab.bin
View file

70
tools/production/spike/spike_readback.py Normal file
View file

@ -0,0 +1,70 @@
#!/usr/bin/env python3
"""
spike_readback.py — диагностика Гейта 3: читает Flash обратно и сравнивает
с ожидаемыми файлами (FCB-блоб @0x60000000, HAB-образ @0x60001000).
Запуск (плата в SDP или с уже поднятым Flashloader):
uv run python spike/spike_readback.py \
--expect-fcb ../host/dcd/w25q128_fdcb.bin \
--expect-image /путь/к/собранному.hab.bin
Дампы кладёт рядом: readback_fcb.bin, readback_image.bin.
"""
from __future__ import annotations
import argparse
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from app import flash_backend as fb # noqa: E402
from spsdk.mboot import McuBoot # noqa: E402
def _cmp(label: str, actual: bytes, expected: bytes) -> bool:
if actual == expected:
print(f" {label}: ✅ ПОБАЙТНО СОВПАДАЕТ ({len(actual)} байт)")
return True
n = min(len(actual), len(expected))
diff_at = next((i for i in range(n) if actual[i] != expected[i]), n)
print(
f" {label}: ❌ РАСХОЖДЕНИЕ с офсета 0x{diff_at:X} "
f"(len actual={len(actual)}, expected={len(expected)})"
)
print(f" actual [{diff_at:#x}]: {actual[diff_at : diff_at + 16].hex()}")
print(f" expected[{diff_at:#x}]: {expected[diff_at : diff_at + 16].hex()}")
return False
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--expect-fcb", type=Path, required=True)
ap.add_argument("--expect-image", type=Path, required=True)
args = ap.parse_args()
exp_fcb = args.expect_fcb.read_bytes()
exp_img = args.expect_image.read_bytes()
iface = fb.load_flashloader(progress_cb=lambda p: print(f" {p.message}"))
ok = True
with McuBoot(iface) as mboot:
fb.configure_flexspi(mboot)
fcb = mboot.read_memory(fb.FLASH_BASE, max(len(exp_fcb), 512), mem_id=0)
img = mboot.read_memory(fb.FLASH_BASE + fb.HAB_OFFSET, len(exp_img), mem_id=0)
if fcb is None or img is None:
print("❌ read_memory вернул None")
return 1
Path("readback_fcb.bin").write_bytes(fcb)
Path("readback_image.bin").write_bytes(img)
ok = _cmp("FCB @0x60000000", fcb[: len(exp_fcb)], exp_fcb) and ok
ok = _cmp("Image @0x60001000", img, exp_img) and ok
return 0 if ok else 1
if __name__ == "__main__":
sys.exit(main())

25
tools/production/tests/test_flash_backend.py
View file

@ -421,3 +421,28 @@ def test_build_custom_hab_bytes_match_golden(tmp_path):
out1 = fb.build_custom_hab(raw, use_dcd=False)
out2 = fb.build_custom_hab(raw, use_dcd=False)
assert out1.read_bytes() == out2.read_bytes()
# ─── firmware_hab_path() — двухрежимный резолв (Р6) ──────────────────────
def test_firmware_hab_path_dev_default(monkeypatch):
monkeypatch.delenv("BUILD_DIR", raising=False)
p = fb.firmware_hab_path("firmware_test", "Debug")
assert p == fb.REPO_ROOT / "build" / "Debug" / "firmware_test_hab.bin"
def test_firmware_hab_path_dev_env_override(monkeypatch, tmp_path):
monkeypatch.setenv("BUILD_DIR", str(tmp_path))
p = fb.firmware_hab_path("bootloader", "Release")
assert p == tmp_path / "Release" / "bootloader_hab.bin"
def test_firmware_hab_path_frozen(monkeypatch, tmp_path):
exe = tmp_path / "dist" / "service_tui"
exe.parent.mkdir(parents=True)
exe.touch()
monkeypatch.setattr(fb.sys, "frozen", True, raising=False)
monkeypatch.setattr(fb.sys, "executable", str(exe))
p = fb.firmware_hab_path("firmware_test", "Debug")
assert p == exe.parent / "firmware" / "Debug" / "firmware_test_hab.bin"