# firmware_test: TUI app skeleton + doc fixes
This commit is contained in:
parent
9be4a2aecc
commit
054bbd70f5
14 changed files with 1768 additions and 120 deletions
|
|
@ -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
|
||||
|
|
|
|||
6
.vscode/settings.json
vendored
Normal file
6
.vscode/settings.json
vendored
Normal file
|
|
@ -0,0 +1,6 @@
|
|||
{
|
||||
"[python]": {
|
||||
"editor.formatOnSave": true,
|
||||
"editor.defaultFormatter": "charliermarsh.ruff"
|
||||
}
|
||||
}
|
||||
|
|
@ -1,6 +1,6 @@
|
|||
# =============================================================================
|
||||
# scripts/build.just — сборка, тесты, HAB-образы
|
||||
# Выполняется ТОЛЬКО внутри devcontainer (из терминала VSCode или tasks).
|
||||
# Выполняется внутри devcontainer
|
||||
#
|
||||
# Рабочая директория — корень репозитория.
|
||||
# =============================================================================
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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.",
|
||||
|
|
|
|||
|
|
@ -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) ─────────────
|
||||
def _usb(vid_key: str, vid_default: str, pid_key: str, pid_default: str) -> str:
|
||||
# ─── 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,46 +267,153 @@ 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__":
|
||||
|
|
|
|||
286
tools/production/README.md
Normal file
286
tools/production/README.md
Normal file
|
|
@ -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`).
|
||||
243
tools/production/app/firmware_client.py
Normal file
243
tools/production/app/firmware_client.py
Normal file
|
|
@ -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)
|
||||
233
tools/production/app/flasher.py
Normal file
233
tools/production/app/flasher.py
Normal file
|
|
@ -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())
|
||||
188
tools/production/app/m5_client.py
Normal file
188
tools/production/app/m5_client.py
Normal file
|
|
@ -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()
|
||||
93
tools/production/app/models.py
Normal file
93
tools/production/app/models.py
Normal file
|
|
@ -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
|
||||
345
tools/production/app/orchestrator.py
Normal file
345
tools/production/app/orchestrator.py
Normal file
|
|
@ -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)
|
||||
0
tools/production/main.py
Normal file
0
tools/production/main.py
Normal file
21
tools/production/pyproject.toml
Normal file
21
tools/production/pyproject.toml
Normal file
|
|
@ -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"
|
||||
Loading…
Reference in a new issue