From 054bbd70f5e1058e191796b740ce949fcf54738d Mon Sep 17 00:00:00 2001 From: Ezra Maccabee Date: Mon, 29 Jun 2026 16:09:23 +0300 Subject: [PATCH] # firmware_test: TUI app skeleton + doc fixes --- .env.example | 2 + .vscode/settings.json | 6 + just/build.just | 2 +- just/host.just | 79 ++++-- tools/host/flash_swd.py | 45 ++-- tools/host/flash_usb.py | 345 +++++++++++++++++++----- tools/production/README.md | 286 ++++++++++++++++++++ tools/production/app/firmware_client.py | 243 +++++++++++++++++ tools/production/app/flasher.py | 233 ++++++++++++++++ tools/production/app/m5_client.py | 188 +++++++++++++ tools/production/app/models.py | 93 +++++++ tools/production/app/orchestrator.py | 345 ++++++++++++++++++++++++ tools/production/main.py | 0 tools/production/pyproject.toml | 21 ++ 14 files changed, 1768 insertions(+), 120 deletions(-) create mode 100644 .vscode/settings.json create mode 100644 tools/production/README.md create mode 100644 tools/production/app/firmware_client.py create mode 100644 tools/production/app/flasher.py create mode 100644 tools/production/app/m5_client.py create mode 100644 tools/production/app/models.py create mode 100644 tools/production/app/orchestrator.py create mode 100644 tools/production/main.py create mode 100644 tools/production/pyproject.toml diff --git a/.env.example b/.env.example index 7efa4d3..5b80699 100644 --- a/.env.example +++ b/.env.example @@ -13,6 +13,8 @@ BOOTROM_VID=1fc9 BOOTROM_PID=0130 FLASHLOADER_VID=15a2 FLASHLOADER_PID=0073 +SERVICE_CDC_VID=1996 +SERVICE_CDC_PID=00ad # --- Paths --- # BUILD_DIR и TOOLS_DIR задаются абсолютно в корневом justfile diff --git a/.vscode/settings.json b/.vscode/settings.json new file mode 100644 index 0000000..3b46aab --- /dev/null +++ b/.vscode/settings.json @@ -0,0 +1,6 @@ +{ +"[python]": { + "editor.formatOnSave": true, + "editor.defaultFormatter": "charliermarsh.ruff" + } +} diff --git a/just/build.just b/just/build.just index d2d95c7..58f2f95 100644 --- a/just/build.just +++ b/just/build.just @@ -1,6 +1,6 @@ # ============================================================================= # scripts/build.just — сборка, тесты, HAB-образы -# Выполняется ТОЛЬКО внутри devcontainer (из терминала VSCode или tasks). +# Выполняется внутри devcontainer # # Рабочая директория — корень репозитория. # ============================================================================= diff --git a/just/host.just b/just/host.just index 7b3de76..a95193a 100644 --- a/just/host.just +++ b/just/host.just @@ -11,6 +11,7 @@ _venv := if _uname =~ "MINGW|MSYS|CYGWIN" { ".venv-host-win" } else { ".venv-hos JUST_MIN := "1.36.0" UV_MIN := "0.4.0" DOCKER_MIN := "24.0.0" +PLATFORM := os() set working-directory := '..' @@ -31,14 +32,6 @@ check-deps: fail() { echo -e " ${RED}❌ ${1}${RESET}"; ERRORS=$((ERRORS+1)); } warn() { echo -e " ${YELLOW}⚠️ ${1}${RESET}"; WARNINGS=$((WARNINGS+1)); } - _uname="$(uname -s)" - case "${_uname}" in - Linux*) PLATFORM="linux" ;; - Darwin*) PLATFORM="macos" ;; - MINGW*|MSYS*|CYGWIN*) PLATFORM="windows" ;; - *) PLATFORM="unknown" ;; - esac - semver_ge() { local a="$1" b="$2" local a1 a2 a3 b1 b2 b3 @@ -56,7 +49,7 @@ check-deps: install_hint() { local tool=$1 - case "${PLATFORM}" in + case "{{ PLATFORM }}" in linux) case "$tool" in just) echo "uv tool install 'rust-just=={{ JUST_MIN }}'" ;; @@ -100,7 +93,7 @@ check-deps: } echo "" - echo -e "${BOLD} Platform: ${PLATFORM}${RESET}" + echo -e "${BOLD} Platform: {{ PLATFORM }}${RESET}" echo "" echo -e "${BOLD} Checking host dependencies...${RESET}" echo "" @@ -109,7 +102,7 @@ check-deps: check_tool "uv" "uv" "{{ UV_MIN }}" check_tool "docker" "docker" "{{ DOCKER_MIN }}" - if [[ "${PLATFORM}" == "linux" ]]; then + if [[ "{{ PLATFORM }}" == "linux" ]]; then echo "" echo -e "${BOLD} Checking udev rules (Linux only)...${RESET}" echo "" @@ -182,22 +175,16 @@ setup-tools: bootstrap: #!/usr/bin/env bash set -euo pipefail - _uname="$(uname -s)" - case "${_uname}" in - Linux*) PLATFORM="linux" ;; - Darwin*) PLATFORM="macos" ;; - MINGW*|MSYS*|CYGWIN*) PLATFORM="windows" ;; - *) PLATFORM="unknown" ;; - esac + echo "" echo " === Bootstrap: step 1/3 -- check-deps ===" just host::check-deps - if [[ "${PLATFORM}" == "linux" ]]; then + if [[ "{{ PLATFORM }}" == "linux" ]]; then echo " === Bootstrap: step 2/3 -- setup-udev ===" just host::setup-udev just host::setup-m5-udev else - echo " ℹ️ step 2/3 -- setup-udev skipped (${PLATFORM})" + echo " ℹ️ step 2/3 -- setup-udev skipped ({{ PLATFORM }})" fi echo " === Bootstrap: step 3/3 -- setup-tools ===" just host::setup-tools @@ -255,20 +242,23 @@ flash-ram project type="debug": [doc('Прошить firmware_test Debug через USB SDP')] [group('flash-sdp')] -flash-test-debug: - @just host::flash firmware_test debug +flash-test-debug: (flash "firmware_test" "debug") [doc('Прошить firmware_test Release через USB SDP')] [group('flash-sdp')] -flash-test-release: - @just host::flash firmware_test release +flash-test-release: (flash "firmware_test" "release") [confirm("Flash bootloader + app (Release)?")] [doc('Прошить bootloader + tft_app Release через USB SDP (производство)')] [group('flash-sdp')] -flash-production: - @just host::flash bootloader release - @just host::flash app release +flash-production: (flash "bootloader" "release") (flash "app" "release") + +[confirm("Стереть всю Flash (W25Q128) через USB SDP? Все прошивки будут удалены.")] +[doc('Стереть всю Flash (W25Q128) через USB SDP (chip erase, ~30 с)')] +[group('flash-sdp')] +flash-sdp-erase: + cd "{{ TOOLS_DIR }}" && UV_PROJECT_ENVIRONMENT={{ _venv }} uv run python flash_usb.py \ + --erase-chip # ============================================================================= # ГРУППА: flash-swd — прошивка через SWD (MCU-Link, без смены BOOT_MODE) @@ -319,7 +309,7 @@ flash-swd-erase: uv run --directory {{ HIL_DIR }} pyocd erase \ --target {{ env('PYOCD_TARGET', 'mimxrt1050_quadspi') }} \ --frequency {{ env('PYOCD_FREQUENCY', '4000000') }} \ - --chip + --chip -O resume_on_disconnect=False # ============================================================================= # ГРУППА: hil — запуск HIL-тестов на реальном железе @@ -543,3 +533,36 @@ incoming: flash-test-release [group('pipeline')] production: flash-production @echo " ✅ Production firmware flashed" + +# ============================================================================= +# ГРУППА: service — TUI сервисного инженера +# ============================================================================= + +PRODUCTION_DIR := justfile_directory() / 'tools/production' + +[doc('Установить/обновить зависимости TUI сервисного инженера')] +[group('service')] +service-setup: + #!/usr/bin/env bash + set -euo pipefail + echo " 📦 Syncing tools/production deps..." + cd "{{ PRODUCTION_DIR }}" && uv sync + echo " ✅ tools/production deps installed" + +[doc('Запустить TUI сервисного инженера')] +[group('service')] +service-tui: + uv run --directory {{ PRODUCTION_DIR }} python main.py + +[doc('Собрать standalone-бинарь TUI (PyInstaller → dist/service_tui)')] +[group('service')] +service-build: + #!/usr/bin/env bash + set -euo pipefail + echo " 🔨 Building standalone service-tui..." + cd "{{ PRODUCTION_DIR }}" && uv run pyinstaller \ + --onefile \ + --name service_tui \ + --add-data "../shared:shared" \ + main.py + echo " ✅ dist/service_tui ready" diff --git a/tools/host/flash_swd.py b/tools/host/flash_swd.py index fdf1b82..81213ac 100644 --- a/tools/host/flash_swd.py +++ b/tools/host/flash_swd.py @@ -48,16 +48,18 @@ REPO_ROOT = Path(__file__).resolve().parents[2] # ── Конфигурация из окружения (just экспортирует .env через set export) ─────── + def _env(key: str, default: str) -> str: return os.environ.get(key, default) -PYOCD_TARGET = _env("PYOCD_TARGET", "mimxrt1050_quadspi") + +PYOCD_TARGET = _env("PYOCD_TARGET", "mimxrt1050_quadspi") PYOCD_FREQUENCY = _env("PYOCD_FREQUENCY", "4000000") -BUILD_DIR = Path(_env("BUILD_DIR", str(REPO_ROOT / "build"))) -FCB_PATH = Path(_env("FCB_PATH", str(REPO_ROOT / "tools/host/dcd/w25q128_fdcb.bin"))) +BUILD_DIR = Path(_env("BUILD_DIR", str(REPO_ROOT / "build"))) +FCB_PATH = Path(_env("FCB_PATH", str(REPO_ROOT / "tools/host/dcd/w25q128_fdcb.bin"))) if not FCB_PATH.is_absolute(): FCB_PATH = REPO_ROOT / FCB_PATH -HIL_DIR = REPO_ROOT / "tools" / "hil" +HIL_DIR = REPO_ROOT / "tools" / "hil" # IVT располагается по смещению 0x1000 от начала Flash (ivtOffset в HAB yaml) IVT_OFFSET = 0x1000 @@ -65,12 +67,13 @@ IVT_OFFSET = 0x1000 # Маппинг имён прошивок → имена HAB-файлов (генерируются build::hab-*) HAB_NAMES = { "firmware_test": "firmware_test_hab.bin", - "bootloader": "bootloader_hab.bin", - "app": "app_hab.bin", + "bootloader": "bootloader_hab.bin", + "app": "app_hab.bin", } # ── Helpers ─────────────────────────────────────────────────────────────────── + def build_full_image(fcb_path: Path, hab_path: Path) -> bytes: """Объединить FCB + padding + HAB в единый образ для записи с 0x60000000.""" fcb = fcb_path.read_bytes() @@ -82,7 +85,7 @@ def build_full_image(fcb_path: Path, hab_path: Path) -> bytes: ) padding = b"\xff" * (IVT_OFFSET - len(fcb)) # 0xFF = erased flash value - image = fcb + padding + hab + image = fcb + padding + hab print(f" FCB: {len(fcb):>6} bytes @ 0x60000000") print(f" Padding: {len(padding):>6} bytes @ 0x{0x60000000 + len(fcb):08X}") @@ -95,13 +98,20 @@ def build_full_image(fcb_path: Path, hab_path: Path) -> bytes: def run_pyocd_flash(image_path: Path, target: str, frequency: str) -> int: """Запустить pyocd flash через uv run из tools/hil.""" cmd = [ - "uv", "run", - "--directory", str(HIL_DIR), - "pyocd", "flash", - "--target", target, - "--frequency", frequency, - "--base-address", "0x60000000", - "--erase", "sector", + "uv", + "run", + "--directory", + str(HIL_DIR), + "pyocd", + "flash", + "--target", + target, + "--frequency", + frequency, + "--base-address", + "0x60000000", + "--erase", + "sector", str(image_path), ] print(f"\n Running: {' '.join(cmd)}\n") @@ -110,10 +120,11 @@ def run_pyocd_flash(image_path: Path, target: str, frequency: str) -> int: # ── Main ────────────────────────────────────────────────────────────────────── + def main() -> int: parser = argparse.ArgumentParser( description="Flash MIMXRT1052 via SWD (MCU-Link). " - "Prepends FCB to HAB image before programming.", + "Prepends FCB to HAB image before programming.", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__, ) @@ -170,7 +181,7 @@ def main() -> int: if not hab_path.exists(): print(f" ❌ HAB image not found: {hab_path}", file=sys.stderr) fw_slug = args.firmware.replace("_", "-") - bt_slug = args.build_type.lower() + bt_slug = args.build_type.lower() print( f" Hint: run just build::hab-{fw_slug}-{bt_slug} " "inside devcontainer first.", @@ -216,4 +227,4 @@ def main() -> int: if __name__ == "__main__": - sys.exit(main()) \ No newline at end of file + sys.exit(main()) diff --git a/tools/host/flash_usb.py b/tools/host/flash_usb.py index c0c4fc5..642aa0b 100644 --- a/tools/host/flash_usb.py +++ b/tools/host/flash_usb.py @@ -7,17 +7,25 @@ flash_usb.py — прошивка MIMXRT1052 через USB (BootROM SDP → Fla python3 flash_usb.py --firmware app --build-type Release python3 flash_usb.py --firmware bootloader --build-type Release python3 flash_usb.py --firmware firmware_test --build-type Debug --ram-only + python3 flash_usb.py --bin-path /path/to/custom.bin + python3 flash_usb.py --erase-chip -Шаги: +Шаги (прошивка): 1. Устройство в SDP режиме (VID:PID 1FC9:0130) → sdphost загружает ivt_flashloader.bin в RAM → sdphost прыгает на flashloader 2. Flashloader запущен (VID:PID 15A2:0073) → blhost конфигурирует FlexSPI NOR (пишет FCB) → blhost стирает нужный регион Flash - → blhost пишет HAB образ начиная с 0x60002000 + → blhost пишет HAB образ начиная с 0x60001000 → blhost reset +Шаги (chip erase): + 1. Загрузка Flashloader (аналогично прошивке) + 2. blhost configure-memory (инициализация FlexSPI контроллера) + 3. blhost flash-erase-all (полная очистка W25Q, ~30 с) + 4. blhost reset + Конфигурация: Переменные окружения (задаются в .env, экспортируются через just): BOOTROM_VID — VID BootROM SDP (default: 1fc9) @@ -35,38 +43,40 @@ import time from pathlib import Path # ─── Пути ───────────────────────────────────────────────────────────────────── -SCRIPT_DIR = Path(__file__).parent.resolve() -REPO_ROOT = SCRIPT_DIR.parent.parent +SCRIPT_DIR = Path(__file__).parent.resolve() +REPO_ROOT = SCRIPT_DIR.parent.parent FLASHLOADER = SCRIPT_DIR / "dcd" / "ivt_flashloader.bin" -# BUILD_DIR: берём из окружения (just экспортирует из .env как абсолютный путь), -# fallback — рассчитываем от расположения скрипта BUILD_DIR = Path(os.environ.get("BUILD_DIR", str(REPO_ROOT / "build"))) -# ─── USB VID:PID — из окружения (.env → just set export → uv run) ───────────── +# ─── USB VID:PID ────────────────────────────────────────────────────────────── + + def _usb(vid_key: str, vid_default: str, pid_key: str, pid_default: str) -> str: - vid = os.environ.get(vid_key, vid_default).strip().upper().lstrip("0X") pid = os.environ.get(pid_key, pid_default).strip().upper().lstrip("0X") return f"0x{vid},0x{pid}" -SDP_USB = _usb("BOOTROM_VID", "1fc9", "BOOTROM_PID", "0130") # BootROM SDP -BLHOST_USB = _usb("FLASHLOADER_VID","15a2", "FLASHLOADER_PID","0073") # Flashloader -# ─── Аппаратные константы (часть логики прошивки, не конфигурация) ──────────── +SDP_USB = _usb("BOOTROM_VID", "1fc9", "BOOTROM_PID", "0130") +BLHOST_USB = _usb("FLASHLOADER_VID", "15a2", "FLASHLOADER_PID", "0073") + +# ─── Аппаратные константы ───────────────────────────────────────────────────── # FlexSPI NOR config option word: 0xC0000007 # bits[31:28]=0xC — tag (QuadSPI NOR) # bits[3:0]=0x7 — option size -FLEXSPI_OPTION_ADDR = "0x2000" +FLEXSPI_OPTION_ADDR = "0x2000" FLEXSPI_OPTION_VALUE = "0xC0000007" -FLEXSPI_MEMORY_ID = "9" # FlexSPI NOR memory interface ID +FLEXSPI_MEMORY_ID = "9" # FlexSPI NOR memory interface ID # Option word для записи FCB: tag=0xF → Write FCB command -FLEXSPI_FCB_VALUE = "0xF000000F" +FLEXSPI_FCB_VALUE = "0xF000000F" # Flash layout -FLASH_BASE = 0x60000000 -HAB_OFFSET = 0x1000 # IVT offset: write address = FLASH_BASE + HAB_OFFSET +FLASH_BASE = 0x60000000 +HAB_OFFSET = 0x1000 # IVT offset: write address = FLASH_BASE + HAB_OFFSET + +# ─── Helpers ────────────────────────────────────────────────────────────────── def run(cmd: list[str], check: bool = True) -> subprocess.CompletedProcess: @@ -79,22 +89,22 @@ def run(cmd: list[str], check: bool = True) -> subprocess.CompletedProcess: def step(msg: str) -> None: - print(f"\n{'─'*60}") + print(f"\n{'─' * 60}") print(f" {msg}") - print(f"{'─'*60}") + print(f"{'─' * 60}") def wait_for_flashloader(timeout: int = 10) -> bool: - """Ждём пока Flashloader поднимется""" + """Ждём пока Flashloader поднимется (опрашиваем blhost раз в секунду).""" print(f"\n Ожидание Flashloader (до {timeout}с)...", end="", flush=True) for i in range(timeout): time.sleep(1) result = subprocess.run( ["blhost", "-u", BLHOST_USB, "-j", "--", "get-property", "1", "0"], - capture_output=True + capture_output=True, ) if result.returncode == 0: - print(f" OK ({i+1}с)") + print(f" OK ({i + 1}с)") return True print(".", end="", flush=True) print(" TIMEOUT") @@ -102,67 +112,147 @@ def wait_for_flashloader(timeout: int = 10) -> bool: def load_flashloader() -> None: - """Загружает Flashloader через SDP если ещё не запущен""" + """Загружает Flashloader через SDP если ещё не запущен.""" result = subprocess.run( ["blhost", "-u", BLHOST_USB, "-j", "--", "get-property", "1", "0"], - capture_output=True + capture_output=True, ) if result.returncode == 0: print(" Flashloader уже запущен — пропускаем загрузку") return if not FLASHLOADER.exists(): - print(f"[ERROR] Не найден: {FLASHLOADER}") - print(" Скачай ivt_flashloader.bin из MCUXpresso Secure Provisioning Tool") - print(" и положи в tools/host/dcd/ivt_flashloader.bin") + print(f"[ERROR] Не найден: {FLASHLOADER}", file=sys.stderr) + print( + " Скачай ivt_flashloader.bin из MCUXpresso Secure Provisioning Tool\n" + " и положи в tools/host/dcd/ivt_flashloader.bin", + file=sys.stderr, + ) sys.exit(1) step("Загрузка Flashloader через SDP (1FC9:0130)") - run(["sdphost", "-u", SDP_USB, "-j", "--", - "write-file", "0x20001C00", str(FLASHLOADER)]) - run(["sdphost", "-u", SDP_USB, "-j", "--", - "jump-address", "0x20001C00"]) + run( + [ + "sdphost", + "-u", + SDP_USB, + "-j", + "--", + "write-file", + "0x20001C00", + str(FLASHLOADER), + ] + ) + run(["sdphost", "-u", SDP_USB, "-j", "--", "jump-address", "0x20001C00"]) if not wait_for_flashloader(): - print("[ERROR] Flashloader не ответил. Проверь BOOT_MOD пины и подключение.") + print( + "[ERROR] Flashloader не ответил. Проверь BOOT_MOD пины и подключение.", + file=sys.stderr, + ) sys.exit(1) def configure_flexspi() -> None: - """Инициализирует FlexSPI NOR контроллер через Flashloader""" + """Инициализирует FlexSPI NOR контроллер через Flashloader.""" step("Конфигурация FlexSPI NOR (инициализация контроллера)") - run(["blhost", "-u", BLHOST_USB, "--", - "fill-memory", FLEXSPI_OPTION_ADDR, "4", FLEXSPI_OPTION_VALUE, "word"]) - run(["blhost", "-u", BLHOST_USB, "--", - "configure-memory", FLEXSPI_MEMORY_ID, FLEXSPI_OPTION_ADDR]) + run( + [ + "blhost", + "-u", + BLHOST_USB, + "--", + "fill-memory", + FLEXSPI_OPTION_ADDR, + "4", + FLEXSPI_OPTION_VALUE, + "word", + ] + ) + run( + [ + "blhost", + "-u", + BLHOST_USB, + "--", + "configure-memory", + FLEXSPI_MEMORY_ID, + FLEXSPI_OPTION_ADDR, + ] + ) def write_fcb() -> None: - """Записывает Flash Configuration Block в 0x60000000 + """Записывает Flash Configuration Block в 0x60000000. - Отдельный шаг после erase! Flashloader генерирует FCB из параметров FlexSPI + Отдельный шаг после erase: Flashloader генерирует FCB из параметров FlexSPI и пишет его по адресу 0x60000000. Без FCB BootROM не знает как читать Flash. Option word 0xF000000F: tag=0xF → Write FCB command. """ step("Запись FCB в Flash[0x60000000]") - run(["blhost", "-u", BLHOST_USB, "--", - "fill-memory", FLEXSPI_OPTION_ADDR, "4", FLEXSPI_FCB_VALUE, "word"]) - run(["blhost", "-u", BLHOST_USB, "--", - "configure-memory", FLEXSPI_MEMORY_ID, FLEXSPI_OPTION_ADDR]) + run( + [ + "blhost", + "-u", + BLHOST_USB, + "--", + "fill-memory", + FLEXSPI_OPTION_ADDR, + "4", + FLEXSPI_FCB_VALUE, + "word", + ] + ) + run( + [ + "blhost", + "-u", + BLHOST_USB, + "--", + "configure-memory", + FLEXSPI_MEMORY_ID, + FLEXSPI_OPTION_ADDR, + ] + ) + + +# ─── Основные операции ──────────────────────────────────────────────────────── def flash(hab_bin: Path, ram_only: bool = False) -> None: + """Прошить HAB-образ в Flash или загрузить в RAM.""" if not hab_bin.exists(): - print(f"[ERROR] Файл не найден: {hab_bin}") - print(" Сначала собери образ: uv run nxpimage hab export -c ...") + print(f"[ERROR] Файл не найден: {hab_bin}", file=sys.stderr) + print( + " Сначала собери образ: uv run nxpimage hab export -c ...", file=sys.stderr + ) sys.exit(1) if ram_only: step(f"Загрузка в RAM (без записи во Flash): {hab_bin.name}") - run(["sdphost", "-u", SDP_USB, "-j", "--", - "write-file", f"0x{FLASH_BASE + HAB_OFFSET:08X}", str(hab_bin)]) - run(["sdphost", "-u", SDP_USB, "-j", "--", - "jump-address", f"0x{FLASH_BASE + HAB_OFFSET:08X}"]) + run( + [ + "sdphost", + "-u", + SDP_USB, + "-j", + "--", + "write-file", + f"0x{FLASH_BASE + HAB_OFFSET:08X}", + str(hab_bin), + ] + ) + run( + [ + "sdphost", + "-u", + SDP_USB, + "-j", + "--", + "jump-address", + f"0x{FLASH_BASE + HAB_OFFSET:08X}", + ] + ) return load_flashloader() @@ -177,47 +267,154 @@ def flash(hab_bin: Path, ram_only: bool = False) -> None: print(f" Адрес: {write_addr}") print(f" Стирание: 0x{FLASH_BASE:08X} .. +{erase_size} байт") - run(["blhost", "-u", BLHOST_USB, "--", - "flash-erase-region", f"0x{FLASH_BASE:08X}", - str(erase_size), "0"]) + run( + [ + "blhost", + "-u", + BLHOST_USB, + "--", + "flash-erase-region", + f"0x{FLASH_BASE:08X}", + str(erase_size), + "0", + ] + ) write_fcb() - run(["blhost", "-u", BLHOST_USB, "--", - "write-memory", write_addr, str(hab_bin), "0"]) + run( + [ + "blhost", + "-u", + BLHOST_USB, + "--", + "write-memory", + write_addr, + str(hab_bin), + "0", + ] + ) step("Reset") run(["blhost", "-u", BLHOST_USB, "--", "reset"]) - print("\n ✓ Прошивка завершена успешно") + print("\n ✅ Прошивка завершена успешно") + + +def erase_chip() -> None: + """Полная очистка Flash (chip erase) через Flashloader. + + Использует blhost flash-erase-all (memory ID 9 = FlexSPI NOR). + Время операции: ~30 с для W25Q128. + После erase FCB также стёрт — BootROM не сможет загрузить прошивку + до следующей прошивки (flash_usb.py запишет FCB автоматически). + """ + load_flashloader() + configure_flexspi() + + step("Полная очистка Flash (chip erase, ~30 с)") + print(" ⚠️ После chip erase BootROM не сможет загрузить прошивку.") + print(" ⚠️ Используй flash_usb.py для восстановления.\n") + run( + [ + "blhost", + "-t", + "100000", + "-u", + BLHOST_USB, + "--", + "flash-erase-all", + FLEXSPI_MEMORY_ID, + ] + ) + + step("Reset") + run(["blhost", "-u", BLHOST_USB, "--", "reset"]) + print("\n ✅ Chip erase завершён") + + +# ─── CLI ────────────────────────────────────────────────────────────────────── def main() -> None: - print(f"DEBUG SDP_USB = {repr(SDP_USB)}") - print(f"DEBUG BLHOST_USB = {repr(BLHOST_USB)}") - print(f"DEBUG BUILD_DIR = {repr(BUILD_DIR)}") - parser = argparse.ArgumentParser(description="Прошивка MIMXRT1052 через USB") - parser.add_argument("--firmware", required=True, - choices=["firmware_test", "bootloader", "app"], - help="Имя прошивки") - parser.add_argument("--build-type", required=True, - choices=["Debug", "Release"], - help="Тип сборки") - parser.add_argument("--ram-only", action="store_true", - help="Загрузить в RAM без записи во Flash") + parser = argparse.ArgumentParser( + description="Прошивка/очистка MIMXRT1052 через USB SDP", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=__doc__, + ) + + # Группа: что прошивать (взаимоисключающие варианты) + target_group = parser.add_mutually_exclusive_group() + target_group.add_argument( + "--firmware", + choices=["firmware_test", "bootloader", "app"], + help="Стандартная прошивка из BUILD_DIR (требует --build-type)", + ) + target_group.add_argument( + "--bin-path", + type=Path, + metavar="PATH", + help="Путь к произвольному HAB-бинарю (.bin) для прошивки", + ) + target_group.add_argument( + "--erase-chip", + action="store_true", + help="Полная очистка Flash (chip erase, ~30 с). Прошивка не выполняется.", + ) + + parser.add_argument( + "--build-type", + choices=["Debug", "Release"], + default="Release", + help="Тип сборки (только для --firmware, default: Release)", + ) + parser.add_argument( + "--ram-only", + action="store_true", + help="Загрузить в RAM без записи во Flash (только для --firmware / --bin-path)", + ) + args = parser.parse_args() - hab_bin = BUILD_DIR / args.build_type / f"{args.firmware}_hab.bin" + # Валидация: --firmware требует --build-type (уже есть default, но запомним) + # --bin-path: build-type игнорируется + # --erase-chip: несовместим с --ram-only + if args.erase_chip and args.ram_only: + parser.error("--erase-chip несовместим с --ram-only") - print(f"\n{'═'*60}") - print(f" MIMXRT1052 Flash Tool") - print(f" Прошивка: {args.firmware} [{args.build_type}]") - print(f" Образ: {hab_bin}") + if args.firmware is None and args.bin_path is None and not args.erase_chip: + parser.error("Укажи --firmware, --bin-path или --erase-chip") + + # ── Определить hab_bin ────────────────────────────────────────────────── + hab_bin: Path | None = None + + if args.firmware is not None: + hab_bin = BUILD_DIR / args.build_type / f"{args.firmware}_hab.bin" + elif args.bin_path is not None: + hab_bin = args.bin_path.resolve() + + # ── Шапка ────────────────────────────────────────────────────────────── + print(f"\n{'═' * 60}") + print(" MIMXRT1052 Flash Tool (USB SDP)") + if args.erase_chip: + print(" Операция: chip erase") + elif hab_bin is not None: + label = ( + f"{args.firmware} [{args.build_type}]" if args.firmware else str(hab_bin) + ) + print(f" Прошивка: {label}") + print(f" Образ: {hab_bin}") + if args.ram_only: + print(" Режим: RAM only (без записи во Flash)") print(f" SDP USB: {SDP_USB}") print(f" BL USB: {BLHOST_USB}") - print(f"{'═'*60}") + print(f"{'═' * 60}\n") - flash(hab_bin, ram_only=args.ram_only) + # ── Выполнить операцию ────────────────────────────────────────────────── + if args.erase_chip: + erase_chip() + elif hab_bin is not None: + flash(hab_bin, ram_only=args.ram_only) if __name__ == "__main__": - main() \ No newline at end of file + main() diff --git a/tools/production/README.md b/tools/production/README.md new file mode 100644 index 0000000..9bcd2e2 --- /dev/null +++ b/tools/production/README.md @@ -0,0 +1,286 @@ +# service-tui — TUI сервисного инженера + +TUI-приложение для диагностики и прошивки платы **MIMXRT1052CVJ5B** на сервисе. +Написано на Python + [Textual](https://textual.textualize.io/). Работает на Linux, macOS, Windows. + +--- + +## Концепция + +```mermaid +graph LR + subgraph PC["Сервисный ПК"] + TUI["service-tui\n(Textual App)"] + subgraph tools["tools/"] + FC["firmware_client.py\nUSB CDC ACM"] + M5["m5_client.py\nSerial JSON-lines"] + FL["flasher.py\n→ tools/host/flash_usb.py"] + end + TUI --> FC + TUI --> M5 + TUI --> FL + end + + subgraph Board["Плата TFT (MIMXRT1052)"] + FW["firmware_test\n(USB CDC)"] + ROM["BootROM\n(SDP 1FC9:0130)"] + end + + subgraph HIL["HIL стенд (опционально)"] + M5HW["M5StampPLC\nRLY1–4 + CAN"] + end + + FC <-->|"JSON-lines\nVID:PID 1996:00AD"| FW + FL -->|"sdphost + blhost\nVID:PID 1FC9:0130"| ROM + M5 <-->|"JSON-lines\nSerial"| M5HW + M5HW -->|"RLY1–4"| Board +``` + +--- + +## Два режима работы + +Режим определяется автодетектом USB при старте и меняется динамически. + +```mermaid +stateDiagram-v2 + [*] --> WAITING : запуск TUI + + WAITING --> FLASHING : обнаружен VID:PID 1FC9:0130\n(BootROM SDP) + WAITING --> DIAGNOSING : обнаружен VID:PID 1996:00AD\n+ session_start по CDC + + FLASHING --> WAITING : прошивка завершена / плата отключена + FLASHING --> DIAGNOSING: плата перезагружена в нормальный режим + + DIAGNOSING --> WAITING : плата отключена + DIAGNOSING --> FLASHING: плата переведена в SDP (перемычка) +``` + +### Режим A — Прошивка + +Триггер: обнаружен BootROM SDP (`1FC9:0130`). + +```bash +┌─ Прошивка платы ─────────────────────────────────┐ +│ Обнаружен BootROM (SDP режим) │ +│ │ +│ Что прошить? │ +│ ◉ firmware_test (диагностика) │ +│ ○ Production (bootloader + tft_app) │ +│ │ +│ ████████████░░░░░░ 64% Запись во Flash... │ +└────────────────────────────────────────────────────┘ +``` + +### Режим B — Диагностика + +Триггер: CDC-порт `1996:00AD` обнаружен и `ping→pong` прошёл. + +``` +┌─ Диагностика fw:0.1.4 UID:A1B2C3D4E5F60011 ─────┐ +│ M5StampPLC: ✓ │ Результаты: │ +├───────────────────────────┤ sdram ✓ PASS │ +│ ☑ SDRAM 32 MB │ qspi ✓ PASS │ +│ ☑ QSPI Flash │ usd ✗ FAIL │ +│ ☑ microSD │ mount failed: 5 │ +│ ☑ TFT Display │ display ✓ PASS │ +│ ☑ Кнопки │ buttons ✓ PASS │ +│ ☑ MQS Audio │ mqs ✓ PASS │ +│ ☑ CAN loopback [HIL] │ can … running │ +│ ☑ Оптовходы [HIL] │ opto ○ pending │ +├───────────────────────────┴─────────────────────────┤ +│ [ Запустить выбранные ] [ Все тесты ] │ +│ ████████████████░░░░ 80% Тест: can │ +├─────────────────────────────────────────────────────┤ +│ ⚠ Экран залит красным цветом? │ +│ [ ✓ Да ] [ ✗ Нет ] │ +└─────────────────────────────────────────────────────┘ +``` + +--- + +## Архитектура приложения + +```mermaid +graph TB + subgraph TUI["tui.py — Textual App"] + WS["WaitingScreen"] + FS["FlashScreen"] + DS["DiagScreen"] + end + + subgraph Core["app/"] + OR["orchestrator.py\nмаршрутизация confirm_request"] + FW["firmware_client.py\nasync CDC клиент"] + M5["m5_client.py\nasync M5 клиент"] + FL["flasher.py\nsubprocess flash_usb.py"] + MD["models.py\nTestInfo · TestResult\nSessionState · ConfirmRequest"] + end + + DS --> OR + OR --> FW + OR --> M5 + FS --> FL + DS --> MD + OR --> MD +``` + +--- + +## Обработка confirm_request + +Маршрутизация определяется по `id` поля `confirm_request`: + +```mermaid +flowchart TD + CR["confirm_request\nот firmware_test"] + + CR --> R{confirm_request.id} + + R -->|"opto_*"| HIL_OPTO["HIL: M5 relay_set\n→ settle → confirm"] + R -->|"can_rx_ready"| HIL_CAN_RX["HIL: M5 can_send\n→ confirm"] + R -->|"can_tx_verify"| HIL_CAN_TX["HIL: M5 can_recv\n→ verify → confirm"] + R -->|"btn*"| BTN["показать инструкцию\nне отправлять confirm\nждать test_result"] + R -->|"всё остальное"| OP["показать оператору\nprompt + OK/FAIL\n+ countdown"] + + HIL_OPTO --> AUTO["CONFIRM_RESOLVED\n(автоматически)"] + HIL_CAN_RX --> AUTO + HIL_CAN_TX --> AUTO + OP --> WAIT["CONFIRM_NEEDED\nждём resolve_operator_confirm()"] + WAIT --> SEND["send_confirm(id, confirmed)"] + AUTO --> SEND +``` + +| confirm id | Кто отвечает | Действие TUI | +| --------------- | ------------ | -------------------------------------- | +| `opto_*` | M5 авто | показать прогресс | +| `can_rx_ready` | M5 авто | показать прогресс | +| `can_tx_verify` | M5 авто | показать прогресс | +| `btn*` | физика | показать инструкцию, ждать test_result | +| всё остальное | оператор | prompt + OK/FAIL + countdown | + +--- + +## Структура + +```bash +tools/production/ +├── pyproject.toml ← зависимости: textual, pyserial, python-dotenv, pyinstaller +├── uv.lock +├── main.py ← точка входа: asyncio + Textual App +├── app/ +│ ├── tui.py ← Textual App, экраны (WaitingScreen, FlashScreen, DiagScreen) +│ ├── firmware_client.py ← async USB CDC клиент firmware_test +│ ├── m5_client.py ← async M5StampPLC клиент +│ ├── flasher.py ← subprocess → tools/host/flash_usb.py +│ ├── orchestrator.py ← confirm_request маршрутизатор +│ └── models.py ← AppMode, TestInfo, TestResult, SessionState, … +└── README.md ← этот файл +``` + +--- + +## Конфигурация (`.env`) + +Файл `.env` в корне репозитория — единый источник конфигурации. + +```ini +# USB VID:PID — BootROM SDP (менять нельзя, NXP ROM) +BOOTROM_VID=1fc9 +BOOTROM_PID=0130 + +# USB VID:PID — firmware_test CDC (наше устройство) +SERVICE_CDC_VID=1996 +SERVICE_CDC_PID=00ad + +# Пути к бинарям (опционально, TUI ищет в build/ автоматически) +FIRMWARE_TEST_BIN=build/Release/firmware_test_hab.bin +PRODUCTION_BIN_BOOT=build/Release/bootloader_hab.bin +PRODUCTION_BIN_APP=build/Release/tft_app_hab.bin +``` + +--- + +## Запуск + +### Из монорепозитория (разработчик) + +```bash +# Установить зависимости +just host::service-setup + +# Запустить TUI +just host::service-tui +``` + +### Standalone-бинарь (сервисник) + +Скачать `service_tui` из [GitHub Releases](https://github.com/OSabuser/tft_manufacture_test/releases) и запустить двойным кликом. +Для сборки из исходников: + +```bash +just host::service-build +# → tools/production/dist/service_tui +``` + +--- + +## Рабочий процесс сервисника + +```mermaid +sequenceDiagram + participant OP as Оператор + participant TUI as service-tui + participant FW as firmware_test (CDC) + participant M5 as M5StampPLC + + OP->>TUI: запустить service_tui + OP->>TUI: подключить плату USB (нормальный режим) + TUI->>FW: ping → pong (CDC автодетект) + TUI->>FW: list_tests + FW-->>TUI: TestInfo × 8 + TUI-->>OP: показать список тестов + + OP->>TUI: выбрать тесты → Запустить + TUI->>FW: run_selected([...]) + + loop Для каждого теста + FW-->>TUI: test_begin + TUI-->>OP: прогресс + + alt HIL тест (opto/can) + FW-->>TUI: confirm_request + TUI->>M5: relay_set / can_send + TUI->>FW: confirm(true) + else Интерактивный (display/mqs) + FW-->>TUI: confirm_request + TUI-->>OP: показать prompt + countdown + OP->>TUI: OK / FAIL + TUI->>FW: confirm(true/false) + else Кнопки + FW-->>TUI: confirm_request (инструкция) + TUI-->>OP: "Нажмите кнопку..." + OP->>FW: физическое нажатие + end + + FW-->>TUI: test_result + TUI-->>OP: результат теста + end + + FW-->>TUI: summary + TUI-->>OP: итог: PASS / FAIL +``` + +--- + +## Зависимости + +| Пакет | Версия | Назначение | +| --------------- | ------ | ------------------------ | +| `textual` | ≥ 0.80 | TUI фреймворк | +| `pyserial` | ≥ 3.5 | USB CDC ACM + M5 Serial | +| `python-dotenv` | ≥ 1.0 | загрузка `.env` | +| `pyinstaller` | ≥ 6.0 | сборка standalone-бинаря | + +**Runtime зависимость (не в pyproject.toml):** +`tools/host/flash_usb.py` вызывается через `subprocess` с `uv run` — `tools/host/` uv-проект должен быть инициализирован (`just host::setup-tools`). diff --git a/tools/production/app/firmware_client.py b/tools/production/app/firmware_client.py new file mode 100644 index 0000000..85e7b87 --- /dev/null +++ b/tools/production/app/firmware_client.py @@ -0,0 +1,243 @@ +""" +firmware_client.py — async CDC-клиент firmware_test. + +Транспорт: USB CDC ACM (pyserial в asyncio thread executor). +Протокол: JSON-lines v2 (один JSON-объект на строку, завершается '\\n'). + +Публичный API: + FirmwareClient.connect() — открыть порт, проверить ping→pong + FirmwareClient.disconnect() — закрыть порт + FirmwareClient.ping() — ping → pong, вернуть True/False + FirmwareClient.list_tests() — list_tests → list[TestInfo] + FirmwareClient.run_selected() — запустить тесты, вернуть AsyncGenerator событий + FirmwareClient.send_confirm() — отправить confirm + FirmwareClient.get_uid() — get_uid → str (hex UID) + FirmwareClient.send_cmd_raw() — отправить произвольную JSON-команду + +Все blocking-операции с pyserial выполняются в asyncio.get_event_loop().run_in_executor() +чтобы не блокировать event loop Textual. +""" + +from __future__ import annotations + +import asyncio +import json +import logging +from typing import AsyncGenerator, Optional + +import serial +import serial.tools.list_ports + +from .models import ConfirmRequest, TestInfo, TestResult, TestStatus + +logger = logging.getLogger(__name__) + +# Таймаут чтения одной строки (сек) +_READLINE_TIMEOUT_S = 0.1 +# Таймаут ping→pong при подключении (сек) +_PING_TIMEOUT_S = 5.0 +# Таймаут ожидания событий теста (сек) — длиннее самого долгого теста (SDRAM ~15 с) +_TEST_EVENT_TIMEOUT_S = 120.0 + + +def _find_cdc_port(vid: int, pid: int) -> Optional[str]: + """Найти первый CDC-порт с заданным VID/PID.""" + for info in serial.tools.list_ports.comports(): + if info.vid == vid and info.pid == pid: + return info.device + return None + + +def _parse_event(line: str) -> Optional[dict]: + """Распарсить JSON-строку. Вернуть None при ошибке.""" + line = line.strip() + if not line: + return None + try: + return json.loads(line) + except json.JSONDecodeError: + logger.warning("Bad JSON from firmware: %r", line) + return None + + +def _parse_test_status(s: str) -> TestStatus: + return { + "pass": TestStatus.PASS, + "fail": TestStatus.FAIL, + "skip": TestStatus.SKIP, + }.get(s, TestStatus.FAIL) + + +class FirmwareClient: + """ + Async-клиент для общения с firmware_test по USB CDC ACM. + + Пример использования:: + + client = FirmwareClient(port="/dev/ttyACM0", baudrate=115200) + await client.connect() + tests = await client.list_tests() + async for event in client.run_selected([t.id for t in tests]): + ... + await client.disconnect() + """ + + def __init__(self, port: str, baudrate: int = 115200) -> None: + self._port = port + self._baudrate = baudrate + self._ser: Optional[serial.Serial] = None + self._lock = asyncio.Lock() + + # ── Connection ────────────────────────────────────────────────────────── + + async def connect(self) -> None: + """Открыть порт и проверить связь через ping→pong.""" + loop = asyncio.get_event_loop() + await loop.run_in_executor(None, self._open) + ok = await self.ping() + if not ok: + await self.disconnect() + raise ConnectionError(f"firmware_test не отвечает на ping: {self._port}") + + def _open(self) -> None: + self._ser = serial.Serial( + port=self._port, + baudrate=self._baudrate, + timeout=_READLINE_TIMEOUT_S, + ) + # сбросить входной буфер — session_start уже ушёл при старте прошивки + self._ser.reset_input_buffer() + + async def disconnect(self) -> None: + """Закрыть порт.""" + loop = asyncio.get_event_loop() + await loop.run_in_executor(None, self._close) + + def _close(self) -> None: + if self._ser and self._ser.is_open: + self._ser.close() + self._ser = None + + @classmethod + async def auto_connect(cls, vid: int, pid: int, baudrate: int = 115200) -> "FirmwareClient": + """ + Найти CDC-порт по VID/PID и подключиться. + + :raises RuntimeError: если порт не найден. + :raises ConnectionError: если ping не прошёл. + """ + port = _find_cdc_port(vid, pid) + if port is None: + raise RuntimeError(f"CDC-порт с VID={vid:04X}:PID={pid:04X} не найден") + client = cls(port=port, baudrate=baudrate) + await client.connect() + return client + + # ── Low-level I/O ─────────────────────────────────────────────────────── + + def _write_line(self, obj: dict) -> None: + """Сериализовать dict в JSON и отправить строку (blocking).""" + assert self._ser is not None + line = json.dumps(obj, separators=(",", ":")) + "\n" + self._ser.write(line.encode("ascii")) + self._ser.flush() + + def _read_line(self) -> Optional[dict]: + """Прочитать одну строку и распарсить (blocking, таймаут _READLINE_TIMEOUT_S).""" + assert self._ser is not None + raw = self._ser.readline() + if not raw: + return None + return _parse_event(raw.decode("ascii", errors="replace")) + + async def _send(self, obj: dict) -> None: + """Отправить JSON-команду (async wrapper).""" + loop = asyncio.get_event_loop() + async with self._lock: + await loop.run_in_executor(None, self._write_line, obj) + + async def _recv_until( + self, + stop_types: set[str], + timeout_s: float = _TEST_EVENT_TIMEOUT_S, + ) -> AsyncGenerator[dict, None]: + """ + Читать события до получения одного из stop_types или таймаута. + Генератор — yield каждого полученного события. + """ + loop = asyncio.get_event_loop() + deadline = loop.time() + timeout_s + while loop.time() < deadline: + event = await loop.run_in_executor(None, self._read_line) + if event is None: + await asyncio.sleep(0) + continue + yield event + if event.get("type") in stop_types: + return + logger.warning("_recv_until timeout after %.1f s", timeout_s) + + # ── Public API ────────────────────────────────────────────────────────── + + async def ping(self) -> bool: + """Отправить ping, ждать pong. Вернуть True при успехе.""" + await self._send({"type": "cmd", "cmd": "ping"}) + loop = asyncio.get_event_loop() + deadline = loop.time() + _PING_TIMEOUT_S + while loop.time() < deadline: + event = await loop.run_in_executor(None, self._read_line) + if event and event.get("type") == "pong": + return True + await asyncio.sleep(0) + return False + + async def list_tests(self) -> list[TestInfo]: + """Запросить список тестов. Вернуть list[TestInfo].""" + await self._send({"type": "cmd", "cmd": "list_tests"}) + async for event in self._recv_until({"test_list"}, timeout_s=5.0): + if event.get("type") == "test_list": + return [ + TestInfo( + id=t["id"], + name=t["name"], + critical=t.get("critical", False), + requires_hil=t.get("requires_hil", False), + ) + for t in event.get("tests", []) + ] + return [] + + async def run_selected( + self, test_ids: list[str] + ) -> AsyncGenerator[dict, None]: + """ + Запустить выбранные тесты. Возвращает async generator событий: + test_begin, test_result, confirm_request, summary, error. + + Вызывающий код должен обрабатывать confirm_request и вызывать + send_confirm() не дожидаясь следующего события. + """ + await self._send({"type": "cmd", "cmd": "run_selected", "tests": test_ids}) + async for event in self._recv_until({"summary"}, timeout_s=_TEST_EVENT_TIMEOUT_S): + yield event + + async def send_confirm(self, confirm_id: str, confirmed: bool) -> None: + """Отправить confirm в ответ на confirm_request.""" + await self._send({"type": "confirm", "id": confirm_id, "confirmed": confirmed}) + + async def get_uid(self) -> str: + """Запросить OCOTP UID. Вернуть hex-строку (16 символов) или "".""" + await self._send({"type": "cmd", "cmd": "get_uid"}) + async for event in self._recv_until({"uid_response"}, timeout_s=3.0): + if event.get("type") == "uid_response": + return event.get("uid", "") + return "" + + async def send_cmd_raw(self, cmd: dict) -> None: + """Отправить произвольную команду (для отладки/расширения).""" + await self._send(cmd) + + @staticmethod + def find_port(vid: int, pid: int) -> Optional[str]: + """Найти CDC-порт по VID/PID. None если не найден.""" + return _find_cdc_port(vid, pid) \ No newline at end of file diff --git a/tools/production/app/flasher.py b/tools/production/app/flasher.py new file mode 100644 index 0000000..e4d4e37 --- /dev/null +++ b/tools/production/app/flasher.py @@ -0,0 +1,233 @@ +""" +flasher.py — обёртка над tools/host/flash_usb.py для TUI. + +Запускает flash_usb.py как subprocess, парсит stdout для прогресса, +пробрасывает события через asyncio.Queue в TUI. + +Не дублирует spsdk-окружение — переиспользует tools/host/ uv-проект. + +Публичный API: + Flasher.flash(target, bin_path, progress_cb) — async, прогресс через callback + Flasher.detect_sdp() — проверить наличие BootROM SDP + Flasher.detect_cdc() — проверить наличие CDC firmware_test +""" + +from __future__ import annotations + +import asyncio +import logging +import os +import re +from pathlib import Path +from typing import Awaitable, Callable, Optional + +from .models import 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) + +# Путь до flash_usb.py относительно корня репозитория +_FLASH_USB_SCRIPT = Path(__file__).parents[3] / "tools" / "host" / "flash_usb.py" +_HOST_TOOLS_DIR = _FLASH_USB_SCRIPT.parent + + +# Паттерны 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: + """Проверить наличие USB-устройства по VID/PID (синхронно).""" + 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 + + +class Flasher: + """ + Async-обёртка над flash_usb.py. + + Пример:: + + flasher = Flasher() + await flasher.flash( + target=FlashTarget.FIRMWARE_TEST, + progress_cb=lambda p: print(p.message), + ) + """ + + def __init__(self) -> None: + self._proc: Optional[asyncio.subprocess.Process] = None + + # ── Detection ─────────────────────────────────────────────────────────── + + @staticmethod + def detect_sdp() -> bool: + """True если виден BootROM SDP (1FC9:0130).""" + return _detect_usb(_BOOTROM_VID, _BOOTROM_PID) + + @staticmethod + def detect_cdc() -> bool: + """True если виден CDC firmware_test (1996:00AD).""" + return _detect_usb(_CDC_VID, _CDC_PID) + + # ── Erase ──────────────────────────────────────────────────────────────── + + async def erase_chip( + self, + progress_cb: Optional[ProgressCallback] = None, + ) -> bool: + """ + Chip erase Flash через USB SDP (flash_usb.py --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) + + # ── Flash ──────────────────────────────────────────────────────────────── + + async def flash( + self, + target: FlashTarget, + progress_cb: Optional[ProgressCallback] = None, + bin_path: Optional[Path] = None, + ) -> bool: + """ + Запустить прошивку через flash_usb.py. + + :param target: Что прошиваем (firmware_test или production). + :param progress_cb: Async callback с FlashProgress (может быть None). + :param bin_path: Путь к бинарю (опционально, если overriding default). + :return: True при успехе. + """ + 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", "Release", progress_cb) + elif target == FlashTarget.PRODUCTION: + ok = await self._run_flash("bootloader", "Release", progress_cb) + if ok: + ok = await self._run_flash("app", "Release", progress_cb) + return ok + elif target == FlashTarget.CUSTOM: + if bin_path is None: + raise ValueError("FlashTarget.CUSTOM требует bin_path") + return await self._run_flash_bin(bin_path, progress_cb) + return False + + async def _run_flash( # прошивка стандартного firmware по имени + self, + firmware: str, + build_type: str, + progress_cb: Optional[ProgressCallback], + ) -> bool: + """Запустить flash_usb.py для одного бинаря.""" + cmd = [ + "uv", + "run", + "--directory", + str(_HOST_TOOLS_DIR), + "python", + str(_FLASH_USB_SCRIPT), + "--firmware", + firmware, + "--build-type", + 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: + 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( # общий subprocess runner + self, + cmd: list[str], + label: str, + progress_cb: Optional[ProgressCallback], + ) -> bool: + 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()) diff --git a/tools/production/app/m5_client.py b/tools/production/app/m5_client.py new file mode 100644 index 0000000..a282622 --- /dev/null +++ b/tools/production/app/m5_client.py @@ -0,0 +1,188 @@ +""" +m5_client.py — клиент M5StampPLC для TUI. + +Тонкая async-обёртка над синхронным JSON-lines протоколом M5 агента. +Blocking-вызовы выполняются в executor чтобы не блокировать Textual. + +Поддерживаемые команды агента (JSON-lines): + {"cmd": "ping"} + {"cmd": "relay_set", "relay": 1..4, "state": true/false} + {"cmd": "relay_get", "relay": 1..4} + {"cmd": "can_send", "id": 0x100, "data": [0xDE, 0xAD, 0xBE, 0xEF]} + {"cmd": "can_recv", "timeout_ms": 500} + +Карта реле (из HIL_BENCH.md): + RLY1 — питание таргета (VIN) + RLY2 — RS_RX (оптовход RS) + RLY3 — EXT_IN1 (оптовход IN1) + RLY4 — EXT_IN2 (оптовход IN2) +""" + +from __future__ import annotations + +import asyncio +import json +import logging +from typing import Optional + +import serial +import serial.tools.list_ports + +logger = logging.getLogger(__name__) + +# VID/PID M5StampPLC (Espressif USB JTAG/serial) +_M5_VID = 0x303A +_M5_PID = 0x1001 + +_READLINE_TIMEOUT_S = 0.5 +_CMD_TIMEOUT_S = 3.0 + + +def _find_m5_port() -> Optional[str]: + """Автодетект M5StampPLC по VID/PID.""" + for info in serial.tools.list_ports.comports(): + if info.vid == _M5_VID and info.pid == _M5_PID: + return info.device + return None + + +class M5Client: + """ + Async-клиент для M5StampPLC. + + Пример:: + + m5 = await M5Client.auto_connect() + if m5: + await m5.relay_set(3, True) # RLY3 ON → EXT_IN1 ACTIVE + await m5.relay_set(3, False) # RLY3 OFF + await m5.disconnect() + """ + + def __init__(self, port: str, baudrate: int = 115200) -> None: + self._port = port + self._baudrate = baudrate + self._ser: Optional[serial.Serial] = None + self._lock = asyncio.Lock() + + # ── Connection ────────────────────────────────────────────────────────── + + @classmethod + async def auto_connect(cls, baudrate: int = 115200) -> Optional["M5Client"]: + """ + Попытаться найти и подключиться к M5StampPLC. + Вернуть None если не найден — M5 опционален. + """ + port = _find_m5_port() + if port is None: + logger.info("M5StampPLC не найден") + return None + client = cls(port=port, baudrate=baudrate) + try: + await client.connect() + return client + except Exception as exc: + logger.warning("M5StampPLC connect failed: %s", exc) + return None + + async def connect(self) -> None: + """Открыть порт и проверить связь через ping.""" + loop = asyncio.get_event_loop() + await loop.run_in_executor(None, self._open) + ok = await self.ping() + if not ok: + await self.disconnect() + raise ConnectionError(f"M5StampPLC не отвечает: {self._port}") + + def _open(self) -> None: + self._ser = serial.Serial( + port=self._port, + baudrate=self._baudrate, + timeout=_READLINE_TIMEOUT_S, + ) + self._ser.reset_input_buffer() + + async def disconnect(self) -> None: + loop = asyncio.get_event_loop() + await loop.run_in_executor(None, self._close) + + def _close(self) -> None: + if self._ser and self._ser.is_open: + self._ser.close() + self._ser = None + + @property + def connected(self) -> bool: + return self._ser is not None and self._ser.is_open + + # ── Low-level I/O ─────────────────────────────────────────────────────── + + def _send_recv(self, cmd: dict) -> Optional[dict]: + """Отправить команду, прочитать ответ (blocking).""" + assert self._ser is not None + line = json.dumps(cmd, separators=(",", ":")) + "\n" + self._ser.write(line.encode("ascii")) + self._ser.flush() + raw = self._ser.readline() + if not raw: + return None + try: + return json.loads(raw.decode("ascii", errors="replace").strip()) + except json.JSONDecodeError: + return None + + async def _cmd(self, cmd: dict) -> Optional[dict]: + """Async wrapper над _send_recv.""" + loop = asyncio.get_event_loop() + async with self._lock: + return await loop.run_in_executor(None, self._send_recv, cmd) + + # ── Public API ────────────────────────────────────────────────────────── + + async def ping(self) -> bool: + """Проверить связь с агентом.""" + resp = await self._cmd({"cmd": "ping"}) + return resp is not None and resp.get("status") == "ok" + + async def relay_set(self, relay: int, state: bool) -> bool: + """ + Переключить реле. + + :param relay: Номер реле 1..4. + :param state: True = ON, False = OFF. + :return: True при успехе. + """ + resp = await self._cmd({"cmd": "relay_set", "relay": relay, "state": state}) + return resp is not None and resp.get("status") == "ok" + + async def relay_get(self, relay: int) -> Optional[bool]: + """ + Прочитать состояние реле. + + :return: True/False или None при ошибке. + """ + resp = await self._cmd({"cmd": "relay_get", "relay": relay}) + if resp and resp.get("status") == "ok": + return bool(resp.get("state")) + return None + + async def can_send(self, can_id: int, data: list[int]) -> bool: + """Отправить CAN-фрейм через M5.""" + resp = await self._cmd({"cmd": "can_send", "id": can_id, "data": data}) + return resp is not None and resp.get("status") == "ok" + + async def can_recv(self, timeout_ms: int = 500) -> Optional[dict]: + """ + Принять CAN-фрейм. + + :return: {"id": int, "data": list[int]} или None при таймауте/ошибке. + """ + resp = await self._cmd({"cmd": "can_recv", "timeout_ms": timeout_ms}) + if resp and resp.get("status") == "ok": + return {"id": resp["id"], "data": resp["data"]} + return None + + @staticmethod + def find_port() -> Optional[str]: + """Найти порт M5StampPLC. None если не найден.""" + return _find_m5_port() \ No newline at end of file diff --git a/tools/production/app/models.py b/tools/production/app/models.py new file mode 100644 index 0000000..80ca804 --- /dev/null +++ b/tools/production/app/models.py @@ -0,0 +1,93 @@ +""" +models.py — типы данных TUI сервисного инженера. + +Все модели — frozen dataclasses или IntEnum. Без бизнес-логики. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import Enum, auto +from typing import Optional + + +class AppMode(Enum): + """Режим работы приложения — определяется автодетектом USB.""" + + WAITING = auto() # ждём подключения устройства + FLASHING = auto() # обнаружен BootROM SDP (1FC9:0130) + DIAGNOSING = auto() # получен session_start по CDC + + +class TestStatus(Enum): + """Статус выполнения теста.""" + + PENDING = auto() + RUNNING = auto() + PASS = auto() + FAIL = auto() + SKIP = auto() + + +class FlashTarget(Enum): + """Что прошиваем.""" + + FIRMWARE_TEST = "firmware_test" + PRODUCTION = "production" # bootloader + tft_app + CUSTOM = "custom" # произвольный HAB-бинарь, путь задаётся отдельно + + +@dataclass(frozen=True) +class TestInfo: + """Метаданные теста из list_tests.""" + + id: str + name: str + critical: bool + requires_hil: bool + + +@dataclass +class TestResult: + """Результат выполнения теста.""" + + id: str + status: TestStatus + duration_ms: int = 0 + detail: str = "" + + +@dataclass +class SessionState: + """Состояние текущей диагностической сессии.""" + + fw_version: str = "" + target: str = "" + chip_uid: str = "" + m5_connected: bool = False + tests: list[TestInfo] = field(default_factory=list) + results: dict[str, TestResult] = field(default_factory=dict) + + def get_result(self, test_id: str) -> Optional[TestResult]: + return self.results.get(test_id) + + def set_result(self, result: TestResult) -> None: + self.results[result.id] = result + + +@dataclass(frozen=True) +class ConfirmRequest: + """confirm_request от таргета.""" + + id: str + prompt: str + timeout_ms: int + + +@dataclass(frozen=True) +class FlashProgress: + """Прогресс прошивки.""" + + phase: str # "sdphost" | "blhost" | "done" | "error" + percent: int # 0..100 + message: str \ No newline at end of file diff --git a/tools/production/app/orchestrator.py b/tools/production/app/orchestrator.py new file mode 100644 index 0000000..748ba43 --- /dev/null +++ b/tools/production/app/orchestrator.py @@ -0,0 +1,345 @@ +""" +orchestrator.py — оркестратор confirm_request для TUI. + +Получает события от FirmwareClient, принимает решение кто должен +ответить на confirm_request, и отправляет confirm обратно. + +Три режима confirm (из закрытых решений Этапа 8): + + standalone/interactive (display, mqs, usd): + → TUI показывает prompt оператору + кнопки OK/FAIL + countdown + → ждёт действия оператора + → отправляет confirm + + standalone/buttons: + → TUI показывает инструкцию оператору + → НЕ отправляет confirm — таргет сам детектирует нажатие + → ждёт следующего confirm_request или test_result + + HIL (opto, can): + → TUI автоматически командует M5 через M5Client + → отправляет confirm без участия оператора + → оператор видит только прогресс + +Маршрутизация определяется по prefix confirm_request.id: + "opto_*" → HIL (M5 relay) + "can_rx_ready" → HIL (M5 can_send) + "can_tx_verify" → HIL (M5 can_recv + verify) + "btn*" → buttons (нет confirm, ждём test_result) + всё остальное → operator (показать prompt) + +Публичный API: + Orchestrator.run_tests(test_ids) — запустить тесты, yield OrchestratorEvent +""" + +from __future__ import annotations + +import asyncio +import logging +from dataclasses import dataclass +from enum import Enum, auto +from typing import AsyncGenerator, Callable, Coroutine, Optional + +from .firmware_client import FirmwareClient +from .m5_client import M5Client +from .models import ConfirmRequest, TestInfo, TestResult, TestStatus + +logger = logging.getLogger(__name__) + +# Задержки для HIL (из закрытых решений: RELAY_ON_S=0.15, RELAY_OFF_S=0.5) +_RELAY_ON_S = 0.15 +_RELAY_OFF_S = 0.50 + +# Карта confirm_id → реле M5 для opto-теста +# Формат: confirm_id_prefix → (relay_num, target_state) +_OPTO_RELAY_MAP: dict[str, tuple[int, bool]] = { + "opto_in1_active": (3, True), + "opto_in1_inactive": (3, False), + "opto_in2_active": (4, True), + "opto_in2_inactive": (4, False), + "opto_rs_active": (2, True), + "opto_rs_inactive": (2, False), +} + +# CAN параметры (из закрытых решений) +_CAN_RX_ID = 0x100 +_CAN_RX_DATA = [0xDE, 0xAD, 0xBE, 0xEF] +_CAN_TX_ID = 0x200 +_CAN_TX_DATA = [0xCA, 0xFE, 0xBA, 0xBE] + + +# ── Типы событий оркестратора ──────────────────────────────────────────── + +class OrchestratorEventType(Enum): + TEST_BEGIN = auto() # тест начался + TEST_RESULT = auto() # тест завершился + CONFIRM_NEEDED = auto() # нужен ответ оператора (standalone) + CONFIRM_RESOLVED = auto() # HIL confirm выполнен автоматически + BUTTONS_PROMPT = auto() # показать инструкцию для buttons (без confirm) + SUMMARY = auto() # итог всей сессии + ERROR = auto() # ошибка протокола или M5 + + +@dataclass +class OrchestratorEvent: + """Событие от оркестратора — передаётся в TUI через asyncio.Queue.""" + + type: OrchestratorEventType + test_id: str = "" + test_name: str = "" + result: Optional[TestResult] = None + confirm: Optional[ConfirmRequest] = None + summary: Optional[dict] = None + message: str = "" + + +# Тип callback для ответа оператора: +# TUI вызывает его с True (OK) или False (FAIL) после показа prompt +OperatorConfirmCallback = Callable[[bool], Coroutine] + + +class Orchestrator: + """ + Оркестратор confirm_request. + + Пример:: + + orch = Orchestrator(firmware_client, m5_client) + async for event in orch.run_tests(["sdram", "opto", "display"]): + if event.type == OrchestratorEventType.CONFIRM_NEEDED: + # TUI показывает prompt, получает ответ оператора + confirmed = await tui.ask_operator(event.confirm) + await orch.resolve_operator_confirm(confirmed) + elif event.type == OrchestratorEventType.TEST_RESULT: + tui.update_result(event.result) + """ + + def __init__( + self, + firmware: FirmwareClient, + m5: Optional[M5Client] = None, + ) -> None: + self._fw = firmware + self._m5 = m5 + # Queue для ответов оператора на CONFIRM_NEEDED + self._operator_queue: asyncio.Queue[bool] = asyncio.Queue(maxsize=1) + + async def resolve_operator_confirm(self, confirmed: bool) -> None: + """ + TUI вызывает этот метод когда оператор нажал OK или FAIL. + Разблокирует ожидание внутри run_tests(). + """ + await self._operator_queue.put(confirmed) + + async def run_tests( + self, test_ids: list[str] + ) -> AsyncGenerator[OrchestratorEvent, None]: + """ + Запустить тесты и оркестрировать confirm_request. + + Yields OrchestratorEvent в порядке поступления событий от firmware_test. + Блокируется на CONFIRM_NEEDED до вызова resolve_operator_confirm(). + """ + async for raw in self._fw.run_selected(test_ids): + event_type = raw.get("type", "") + + if event_type == "test_begin": + yield OrchestratorEvent( + type=OrchestratorEventType.TEST_BEGIN, + test_id=raw.get("id", ""), + test_name=raw.get("name", ""), + ) + + elif event_type == "test_result": + status_str = raw.get("status", "fail") + status = { + "pass": TestStatus.PASS, + "fail": TestStatus.FAIL, + "skip": TestStatus.SKIP, + }.get(status_str, TestStatus.FAIL) + + result = TestResult( + id=raw.get("id", ""), + status=status, + duration_ms=raw.get("ms", 0), + detail=raw.get("detail", ""), + ) + yield OrchestratorEvent( + type=OrchestratorEventType.TEST_RESULT, + test_id=result.id, + result=result, + ) + + elif event_type == "confirm_request": + confirm = ConfirmRequest( + id=raw.get("id", ""), + prompt=raw.get("prompt", ""), + timeout_ms=raw.get("timeout_ms", 30000), + ) + async for orch_event in self._handle_confirm(confirm): + yield orch_event + + elif event_type == "summary": + yield OrchestratorEvent( + type=OrchestratorEventType.SUMMARY, + summary=raw, + ) + + elif event_type in ("error", None): + yield OrchestratorEvent( + type=OrchestratorEventType.ERROR, + message=raw.get("error", "unknown error"), + ) + + # ── Маршрутизация confirm ──────────────────────────────────────────── + + async def _handle_confirm( + self, confirm: ConfirmRequest + ) -> AsyncGenerator[OrchestratorEvent, None]: + """Определить тип confirm и обработать соответственно.""" + cid = confirm.id + + # HIL: opto-тест + if cid in _OPTO_RELAY_MAP: + async for ev in self._handle_hil_opto(confirm): + yield ev + return + + # HIL: CAN RX + if cid == "can_rx_ready": + async for ev in self._handle_hil_can_rx(confirm): + yield ev + return + + # HIL: CAN TX verify + if cid == "can_tx_verify": + async for ev in self._handle_hil_can_tx(confirm): + yield ev + return + + # buttons: нет JSON-confirm, только инструкция оператору + if cid.startswith("btn"): + yield OrchestratorEvent( + type=OrchestratorEventType.BUTTONS_PROMPT, + confirm=confirm, + ) + return + + # standalone interactive: показать оператору + async for ev in self._handle_operator_confirm(confirm): + yield ev + + # ── HIL opto ──────────────────────────────────────────────────────── + + async def _handle_hil_opto( + self, confirm: ConfirmRequest + ) -> AsyncGenerator[OrchestratorEvent, None]: + relay_num, relay_state = _OPTO_RELAY_MAP[confirm.id] + + if self._m5 is None: + logger.error("HIL confirm без M5: %s", confirm.id) + await self._fw.send_confirm(confirm.id, False) + yield OrchestratorEvent( + type=OrchestratorEventType.ERROR, + message=f"M5 не подключён для HIL confirm: {confirm.id}", + ) + return + + settle_s = _RELAY_ON_S if relay_state else _RELAY_OFF_S + ok = await self._m5.relay_set(relay_num, relay_state) + await asyncio.sleep(settle_s) + + await self._fw.send_confirm(confirm.id, ok) + + yield OrchestratorEvent( + type=OrchestratorEventType.CONFIRM_RESOLVED, + test_id="opto", + message=f"RLY{relay_num} {'ON' if relay_state else 'OFF'} → {confirm.id}", + ) + + # ── HIL CAN RX ────────────────────────────────────────────────────── + + async def _handle_hil_can_rx( + self, confirm: ConfirmRequest + ) -> AsyncGenerator[OrchestratorEvent, None]: + if self._m5 is None: + await self._fw.send_confirm(confirm.id, False) + yield OrchestratorEvent( + type=OrchestratorEventType.ERROR, + message="M5 не подключён для CAN RX confirm", + ) + return + + # M5 отправляет CAN-фрейм, таргет принимает после confirmed:true + ok = await self._m5.can_send(_CAN_RX_ID, _CAN_RX_DATA) + await self._fw.send_confirm(confirm.id, ok) + + yield OrchestratorEvent( + type=OrchestratorEventType.CONFIRM_RESOLVED, + test_id="can", + message=f"M5 CAN TX id=0x{_CAN_RX_ID:03X} data={_CAN_RX_DATA}", + ) + + # ── HIL CAN TX verify ─────────────────────────────────────────────── + + async def _handle_hil_can_tx( + self, confirm: ConfirmRequest + ) -> AsyncGenerator[OrchestratorEvent, None]: + if self._m5 is None: + await self._fw.send_confirm(confirm.id, False) + yield OrchestratorEvent( + type=OrchestratorEventType.ERROR, + message="M5 не подключён для CAN TX verify", + ) + return + + # Таргет уже отправил фрейм, M5 принимает и верифицирует + frame = await self._m5.can_recv(timeout_ms=500) + verified = ( + frame is not None + and frame["id"] == _CAN_TX_ID + and frame["data"] == _CAN_TX_DATA + ) + await self._fw.send_confirm(confirm.id, verified) + + msg = ( + f"M5 CAN RX ok: id=0x{_CAN_TX_ID:03X}" + if verified + else "M5 CAN RX: фрейм не получен или не совпадает" + ) + yield OrchestratorEvent( + type=OrchestratorEventType.CONFIRM_RESOLVED, + test_id="can", + message=msg, + ) + + # ── Operator confirm ───────────────────────────────────────────────── + + async def _handle_operator_confirm( + self, confirm: ConfirmRequest + ) -> AsyncGenerator[OrchestratorEvent, None]: + """ + Передать confirm оператору. Блокируется до resolve_operator_confirm(). + TUI должен показать prompt и дать возможность ответить. + """ + # очистить очередь на случай устаревшего ответа + while not self._operator_queue.empty(): + self._operator_queue.get_nowait() + + yield OrchestratorEvent( + type=OrchestratorEventType.CONFIRM_NEEDED, + confirm=confirm, + ) + + # ждём ответа от оператора (TUI вызовет resolve_operator_confirm) + timeout_s = confirm.timeout_ms / 1000.0 + try: + confirmed = await asyncio.wait_for( + self._operator_queue.get(), + timeout=timeout_s, + ) + except asyncio.TimeoutError: + logger.warning("Operator confirm timeout: %s", confirm.id) + confirmed = False + + await self._fw.send_confirm(confirm.id, confirmed) \ No newline at end of file diff --git a/tools/production/main.py b/tools/production/main.py new file mode 100644 index 0000000..e69de29 diff --git a/tools/production/pyproject.toml b/tools/production/pyproject.toml new file mode 100644 index 0000000..b7f6ac8 --- /dev/null +++ b/tools/production/pyproject.toml @@ -0,0 +1,21 @@ +[project] +name = "service-tui" +version = "0.1.0" +description = "TUI сервисного инженера для диагностики платы MIMXRT1052" +requires-python = ">=3.11" +dependencies = [ + "textual>=0.80.0", + "pyserial>=3.5", + "python-dotenv>=1.0.0", + "pyinstaller>=6.0.0", +] + +[project.scripts] +service-tui = "main:main" + +[tool.uv] +package = false + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" \ No newline at end of file