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

# firmware_test: TUI app major refactoring

Browse source
This commit is contained in:
Dmitry Akimov 2026-06-30 15:09:47 +03:00
parent 55cd9e8273
commit 6e42970fef
27 changed files with 1893 additions and 333 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
.env.example
View file

@ -16,10 +16,6 @@ FLASHLOADER_PID=0073
SERVICE_CDC_VID=1996
SERVICE_CDC_PID=00ad
# Тип сборки firmware_test для TUI (Debug | Release)
FIRMWARE_BUILD_TYPE=Debug
# --- Paths ---
# BUILD_DIR и TOOLS_DIR задаются абсолютно в корневом justfile
# через justfile_directory(), здесь можно переопределить если нужно
@ -38,9 +34,10 @@ GDB_EXECUTABLE=arm-none-eabi-gdb
PYOCD_TARGET=mimxrt1050_quadspi
PYOCD_FREQUENCY=4000000
# FCB-бинарник для flash_swd.py (Flash Configuration Block, W25Q128 Quad SPI)
# FCB-бинарник для flash_swd.py (W25Q128 Quad SPI)
FCB_PATH=tools/host/dcd/w25q128_fdcb.bin
# --- HIL (аппаратный стенд) ---
HIL_PYOCD_FREQUENCY=1000000
HIL_BUILD_DIR=build/target-debug
@ -62,3 +59,6 @@ HIL_M5_TIMEOUT=3.0
HIL_USB_CDC_PORT=/dev/cu.usbmodem11301
HIL_USB_CDC_BAUD=115200
HIL_USB_CDC_TIMEOUT=5.0
# Тип сборки firmware_test для TUI (Debug | Release)
FIRMWARE_BUILD_TYPE=Release

4
cmake/toolchain_arm.cmake
View file

@ -109,10 +109,10 @@ set(CMAKE_ASM_FLAGS_DEBUG
CACHE INTERNAL "")
set(CMAKE_C_FLAGS_RELEASE
"-O3 -DNDEBUG"
"-O1 -DNDEBUG"
CACHE INTERNAL "")
set(CMAKE_CXX_FLAGS_RELEASE
"-O3 -DNDEBUG"
"-O1 -DNDEBUG"
CACHE INTERNAL "")
set(CMAKE_ASM_FLAGS_RELEASE
""

2
firmware/test/CMakeLists.txt
View file

@ -84,7 +84,7 @@ target_link_options(
-Wl,-Map=${CMAKE_BINARY_DIR}/firmware_test.map
-Wl,--defsym=__stack_size__=0x2000
-Wl,--defsym=__heap_size__=0x2000
-T${PROJECT_SOURCE_DIR}/cmake/linker/MIMXRT1052xxxxx_flexspi_nor_sdram.ld)
-T${CMAKE_SOURCE_DIR}/cmake/linker/MIMXRT1052xxxxx_flexspi_nor_sdram.ld)
set_target_properties(${TARGET_NAME} PROPERTIES RUNTIME_OUTPUT_DIRECTORY
${CMAKE_BINARY_DIR})

2
firmware/test/src/tests/test_buttons.c
View file

@ -165,7 +165,7 @@ static test_result_t buttons_test_run(void)
/** @brief Дескриптор тест-модуля кнопок для реестра test_runner. */
const test_module_t K_TEST_BUTTONS = {
.id = "buttons",
.name = "Test Buttons",
.name = "Buttons",
.critical = false,
.requires_hil = false,
.pre_confirm_prompt = NULL,

2
firmware/test/src/tests/test_can.c
View file

@ -197,7 +197,7 @@ static test_result_t can_run(void)
/** @brief Дескриптор для регистрации в test_runner. */
const test_module_t K_TEST_CAN = {
.id = "can",
.name = "CAN loopback",
.name = "CAN Network",
.critical = false,
.requires_hil = true,
.pre_confirm_prompt = NULL,

2
firmware/test/src/tests/test_display.c
View file

@ -308,7 +308,7 @@ static void display_test_deinit(void)
const test_module_t K_TEST_DISPLAY = {
.id = "display",
.name = "TFT Display RGB888",
.name = "TFT Display",
.critical = false,
.requires_hil = false,
.pre_confirm_prompt = NULL,

2
firmware/test/src/tests/test_mqs.c
View file

@ -220,7 +220,7 @@ static void test_mqs_deinit(void)
const test_module_t K_TEST_MQS = {
.id = "mqs",
.name = "MQS Audio Out",
.name = "Audio Out",
.critical = false,
.requires_hil = false,
.pre_confirm_prompt = NULL,

2
firmware/test/src/tests/test_qspi.c
View file

@ -380,7 +380,7 @@ static void qspi_test_deinit(void)
const test_module_t K_TEST_QSPI = {
.id = "qspi",
.name = "QSPI Flash W25Qxx",
.name = "NOR Flash W25Qxx",
.critical = true,
.requires_hil = false,
.pre_confirm_prompt = NULL,

2
firmware/test/src/tests/test_sdram.c
View file

@ -367,7 +367,7 @@ static void sdram_test_deinit(void)
const test_module_t K_TEST_SDRAM = {
.id = "sdram",
.name = "SDRAM 32 MB",
.name = "SDRAM MT48LC16x",
.critical = true,
.requires_hil = false,
.pre_confirm_prompt = NULL,

2
firmware/test/src/tests/test_usd.c
View file

@ -343,7 +343,7 @@ static void usd_deinit(void)
const test_module_t K_TEST_USD = {
.id = "usd",
.name = "microSD (SDIO)",
.name = "microSD Card",
.critical = false,
.requires_hil = false,
.pre_confirm_prompt = "Insert microSD card and press OK",

34
tools/production/app/app.py
View file

@ -17,8 +17,8 @@ from textual.binding import Binding
from .firmware_client import FirmwareClient
from .m5_client import M5Client
from .models import AppMode
from .screens import DiagScreen, FlashScreen, WaitingScreen
from .models import AppMode, FlashTarget
from .screens import DiagScreen, FlashScreen, PostFlashScreen, WaitingScreen
logger = logging.getLogger(__name__)
@ -30,11 +30,12 @@ class ServiceApp(App):
"""Корневое приложение service-tui."""
TITLE = "TFT Board Service Tool"
CSS_PATH = "app.tcss" # относительно app/app.py → app/app.tcss
BINDINGS = [
Binding("ctrl+c", "quit", "Выход", show=True),
Binding("ctrl+q", "quit", "Выход"),
]
CSS_PATH = "app.tcss"
def __init__(self) -> None:
super().__init__()
@ -54,8 +55,20 @@ class ServiceApp(App):
self._connect_and_diagnose()
@on(FlashScreen.FlashDone)
def _on_flash_done(self, _event: FlashScreen.FlashDone) -> None:
"""После прошивки — вернуться на WaitingScreen."""
def _on_flash_done(self, event: FlashScreen.FlashDone) -> None:
"""
После прошивки:
- firmware_test + успех → PostFlashScreen (промпт смены BootMode)
- всё остальное (production/custom, либо ошибка) → сразу WaitingScreen
"""
if event.success and event.target == FlashTarget.FIRMWARE_TEST:
self.switch_screen(PostFlashScreen())
else:
self.switch_screen(WaitingScreen())
@on(PostFlashScreen.Done)
def _on_post_flash_done(self) -> None:
"""Оператор подтвердил смену BootMode или истёк таймаут."""
self.switch_screen(WaitingScreen())
@on(DiagScreen.DiagDone)
@ -76,7 +89,16 @@ class ServiceApp(App):
return
self._m5 = await M5Client.auto_connect()
self.switch_screen(DiagScreen(firmware=self._fw, m5=self._m5))
fw_version = ""
try:
fw_version = await self._fw.get_version()
except Exception as exc:
logger.warning("get_version failed: %s", exc)
self.switch_screen(
DiagScreen(firmware=self._fw, m5=self._m5, fw_version=fw_version)
)
def _disconnect(self) -> None:
if self._fw is not None:

181
tools/production/app/app.tcss
View file

@ -1,13 +1,31 @@
/* ── WaitingScreen ───────────────────────────────────────────────────────── */
/* ═══════════════════════════════════════════════════════════
service-tui — единый файл стилей
Путь резолвится относительно app/app.py → app/app.tcss
═══════════════════════════════════════════════════════════ */
WaitingScreen {
/* ── Общая рамка приложения (AppFrame) ──────────────────── */
/* Все три экрана центрируют один AppFrame фиксированного */
/* размера — единообразный визуальный каркас независимо от */
/* размера окна терминала (как в ratatui-приложениях). */
Screen {
align: center middle;
layout: vertical;
}
WaitingScreen > Static {
content-align: center middle;
AppFrame {
width: 100%;
height: 100%;
max-width: 160;
max-height: 50;
border: heavy $primary;
background: $surface;
padding: 1 2;
}
/* ── WaitingScreen ──────────────────────────────────────── */
#waiting-frame {
align: center middle;
}
#waiting-logo {
@ -16,9 +34,8 @@ WaitingScreen > Static {
border: round $primary;
padding: 1 2;
content-align: center middle;
text-style: bold;
color: $primary;
margin-bottom: 0;
text-style: bold;
}
#waiting-hint {
@ -33,11 +50,7 @@ WaitingScreen > Static {
content-align: center middle;
}
/* ── FlashScreen ─────────────────────────────────────────────────────────── */
FlashScreen {
padding: 1 2;
}
/* ── FlashScreen ────────────────────────────────────────── */
#flash-title {
text-style: bold;
@ -52,7 +65,6 @@ FlashScreen {
height: auto;
}
#flash-custom-path {
margin-top: 1;
height: auto;
@ -72,12 +84,14 @@ FlashScreen {
align: left middle;
}
#flash-btn-flash {
#flash-btn-flash,
#flash-btn-erase {
margin-right: 2;
}
#flash-btn-flash:disabled,
#flash-btn-erase:disabled {
#flash-btn-erase:disabled,
#flash-btn-quit:disabled {
background: $panel;
color: $text-muted;
}
@ -86,8 +100,8 @@ FlashScreen {
margin-top: 1;
}
#flash-progress-label {
color: $text-muted;
#flash-progress-bar.hidden {
display: none;
}
#flash-log {
@ -98,15 +112,51 @@ FlashScreen {
overflow-y: auto;
}
/* ── DiagScreen ──────────────────────────────────────────────────────────── */
/* ── PostFlashScreen ────────────────────────────────────── */
DiagScreen {
layout: grid;
grid-size: 1;
grid-rows: 3 1fr 1 1 3 auto;
#post-flash-frame {
align: center middle;
}
#post-flash-title {
text-style: bold;
color: $success;
content-align: center middle;
margin-bottom: 2;
}
#post-flash-instruction {
border: round $warning;
padding: 1 3;
color: $text;
text-style: bold;
content-align: center middle;
width: auto;
}
#post-flash-countdown {
margin-top: 2;
color: $text-muted;
content-align: center middle;
}
#post-flash-btn-row {
margin-top: 2;
height: auto;
align: center middle;
}
#post-flash-btn-ok {
margin-right: 2;
}
/* ── DiagScreen ─────────────────────────────────────────── */
#diag-frame {
layout: vertical;
padding: 0;
}
/* Шапка */
#diag-header {
height: 3;
background: $panel;
@ -115,16 +165,21 @@ DiagScreen {
}
#diag-header-fw {
width: auto;
min-width: 14;
color: $text-muted;
margin-right: 3;
}
#diag-header-uid {
width: auto;
min-width: 28;
color: $text-muted;
margin-right: 3;
}
#diag-header-m5 {
width: auto;
color: $success;
}
@ -132,24 +187,30 @@ DiagScreen {
color: $text-muted;
}
/* Рабочая зона */
#diag-main {
height: 1fr;
min-height: 10;
}
#diag-progress-row {
height: auto;
padding: 0 2;
}
#diag-progress-row.hidden {
display: none;
}
/* Прогресс */
#diag-progress-bar {
height: 1;
margin: 0 2;
margin-top: 1;
}
#diag-progress-label {
height: 1;
color: $text-muted;
padding: 0 2;
}
/* Кнопки запуска */
#diag-btn-row {
height: 3;
align: left middle;
@ -157,7 +218,8 @@ DiagScreen {
background: $panel;
}
#diag-btn-run-selected {
#diag-btn-run-selected,
#diag-btn-run-all {
margin-right: 2;
}
@ -167,7 +229,7 @@ DiagScreen {
color: $text-muted;
}
/* ── TestListPanel ───────────────────────────────────────────────────────── */
/* ── TestListPanel ──────────────────────────────────────── */
TestListPanel {
width: 38;
@ -182,9 +244,19 @@ TestListPanel .section-title {
padding: 0 0 1 0;
}
.test-row {
align: left middle;
#test-list-select-row {
height: auto;
margin-bottom: 1;
}
#test-list-select-row Button {
margin-right: 1;
min-width: 12;
}
.test-row {
height: auto;
align: left middle;
}
.test-row-hil-badge {
@ -196,10 +268,10 @@ TestListPanel .section-title {
color: $text-muted;
}
/* ── ResultsPanel ────────────────────────────────────────────────────────── */
/* ── ResultsPanel ───────────────────────────────────────── */
ResultsPanel {
width: 1fr;
width: 70;
padding: 1 2;
overflow-y: auto;
}
@ -210,45 +282,26 @@ ResultsPanel .section-title {
padding: 0 0 1 0;
}
.result-row {
height: 1;
align: left middle;
#results-empty {
height: 1fr;
width: 100%;
}
.result-id {
#results-empty-text {
color: $text-muted;
width: 12;
text-align: center;
}
.result-status-pass {
color: $success;
width: 8;
text-style: bold;
#results-table {
height: auto;
max-height: 100%;
}
.result-status-fail {
color: $error;
width: 8;
text-style: bold;
#results-table.hidden {
display: none;
}
.result-status-running {
color: $warning;
width: 8;
}
.result-status-skip {
color: $text-muted;
width: 8;
}
.result-detail {
color: $text-muted;
width: 1fr;
overflow: hidden;
}
/* ── ConfirmPanel ────────────────────────────────────────────────────────── */
/* ── ConfirmPanel ───────────────────────────────────────── */
ConfirmPanel {
height: auto;

15
tools/production/app/firmware_client.py
View file

@ -12,9 +12,10 @@ firmware_client.py — async CDC-клиент firmware_test.
FirmwareClient.run_selected() — запустить тесты, вернуть AsyncGenerator событий
FirmwareClient.send_confirm() — отправить confirm
FirmwareClient.get_uid() — get_uid → str (hex UID)
FirmwareClient.get_version() — get_version → str (X.Y.Z)
FirmwareClient.send_cmd_raw() — отправить произвольную JSON-команду
Все blocking-операции с pyserial выполняются в asyncio.get_event_loop().run_in_executor()
Все blocking-операции с pyserial выполняются в run_in_executor()
чтобы не блокировать event loop Textual.
"""
@ -28,7 +29,7 @@ from typing import AsyncGenerator, Optional
import serial
import serial.tools.list_ports
from .models import TestInfo, TestStatus
from .models import TestInfo
logger = logging.getLogger(__name__)
@ -60,14 +61,6 @@ def _parse_event(line: str) -> Optional[dict]:
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.
@ -194,7 +187,7 @@ class FirmwareClient:
return False
async def get_version(self) -> str:
"""Запросить версию firmware_test. Вернуть строку X.Y.Z или ''."""
"""Запросить версию firmware_test. Вернуть строку 'X.Y.Z' или ''."""
await self._send({"type": "cmd", "cmd": "get_version"})
async for event in self._recv_until({"version_response"}, timeout_s=3.0):
if event.get("type") == "version_response":

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

@ -8,6 +8,7 @@ flasher.py — обёртка над tools/host/flash_usb.py для TUI.
Публичный API:
Flasher.flash(target, bin_path, progress_cb) — async, прогресс через callback
Flasher.erase_chip(progress_cb) — async chip erase
Flasher.detect_sdp() — проверить наличие BootROM SDP
Flasher.detect_cdc() — проверить наличие CDC firmware_test
"""
@ -31,17 +32,18 @@ _BOOTROM_PID = int(os.environ.get("BOOTROM_PID", "0x0130"), 16)
_CDC_VID = int(os.environ.get("SERVICE_CDC_VID", "0x1996"), 16)
_CDC_PID = int(os.environ.get("SERVICE_CDC_PID", "0x00ad"), 16)
# Тип сборки firmware_test для прошивки (Debug | Release).
# Release временно нестабилен (см. отчёт о тестировании) — по умолчанию Debug.
_FIRMWARE_BUILD_TYPE = os.environ.get("FIRMWARE_BUILD_TYPE", "Debug")
# Путь до flash_usb.py относительно корня репозитория
_FLASH_USB_SCRIPT = Path(__file__).parents[3] / "tools" / "host" / "flash_usb.py"
_HOST_TOOLS_DIR = _FLASH_USB_SCRIPT.parent
# Паттерны 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)
_FIRMWARE_BUILD_TYPE = os.environ.get("FIRMWARE_BUILD_TYPE", "Debug")
ProgressCallback = Callable[[FlashProgress], Awaitable[None]]
@ -50,10 +52,10 @@ def _detect_usb(vid: int, pid: int) -> bool:
Проверить наличие USB-устройства по VID/PID (синхронно).
Два метода детекта:
1. pyusb (usb.core) — видит все USB-устройства включая SDP bulk/HID
(на macOS SDP не создаёт serial-порт и невидим через list_ports).
2. serial.tools.list_ports — fallback для CDC ACM устройств
если pyusb недоступен.
1. pyusb (usb.core) — видит все USB-устройства включая SDP bulk/HID.
На macOS SDP-устройство (1FC9:0130) не создаёт serial-порт
и невидимо через serial.tools.list_ports.
2. serial.tools.list_ports — fallback для CDC ACM устройств.
"""
# Метод 1: pyusb — работает для SDP и CDC
try:
@ -140,9 +142,9 @@ class Flasher:
"""
Запустить прошивку через flash_usb.py.
:param target: Что прошиваем (firmware_test или production).
:param target: Что прошиваем (firmware_test, production или custom).
:param progress_cb: Async callback с FlashProgress (может быть None).
:param bin_path: Путь к бинарю (опционально, если overriding default).
:param bin_path: Путь к бинарю (обязателен для CUSTOM).
:return: True при успехе.
"""
if target == FlashTarget.FIRMWARE_TEST:
@ -160,12 +162,12 @@ class Flasher:
return await self._run_flash_bin(bin_path, progress_cb)
return False
async def _run_flash( # прошивка стандартного firmware по имени
async def _run_flash(
self,
firmware: str,
progress_cb: Optional[ProgressCallback],
) -> bool:
"""Запустить flash_usb.py для одного бинаря."""
"""Запустить flash_usb.py для одного бинаря (стандартный firmware)."""
cmd = [
"uv",
"run",
@ -180,11 +182,12 @@ class Flasher:
]
return await self._run_cmd(cmd, firmware, progress_cb)
async def _run_flash_bin( # прошивка произвольного бинаря
async def _run_flash_bin(
self,
bin_path: Path,
progress_cb: Optional[ProgressCallback],
) -> bool:
"""Запустить flash_usb.py для произвольного бинаря."""
cmd = [
"uv",
"run",
@ -197,12 +200,13 @@ class Flasher:
]
return await self._run_cmd(cmd, bin_path.stem, progress_cb)
async def _run_cmd( # общий subprocess runner
async def _run_cmd(
self,
cmd: list[str],
label: str,
progress_cb: Optional[ProgressCallback],
) -> bool:
"""Общий subprocess runner для всех операций flash_usb.py."""
logger.info("Running: %s", " ".join(cmd))
try:
self._proc = await asyncio.create_subprocess_exec(

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

@ -14,9 +14,9 @@ from typing import Optional
class AppMode(Enum):
"""Режим работы приложения — определяется автодетектом USB."""
WAITING = auto() # ждём подключения устройства
FLASHING = auto() # обнаружен BootROM SDP (1FC9:0130)
DIAGNOSING = auto() # получен session_start по CDC
WAITING = auto() # ждём подключения устройства
FLASHING = auto() # обнаружен BootROM SDP (1FC9:0130)
DIAGNOSING = auto() # получен session_start по CDC
class TestStatus(Enum):
@ -34,7 +34,7 @@ class FlashTarget(Enum):
FIRMWARE_TEST = "firmware_test"
PRODUCTION = "production" # bootloader + tft_app
CUSTOM = "custom" # произвольный HAB-бинарь, путь задаётся отдельно
CUSTOM = "custom" # произвольный HAB-бинарь, путь задаётся отдельно
@dataclass(frozen=True)
@ -88,6 +88,6 @@ class ConfirmRequest:
class FlashProgress:
"""Прогресс прошивки."""
phase: str # "sdphost" | "blhost" | "done" | "error"
percent: int # 0..100
phase: str # "sdphost" | "blhost" | "done" | "error"
percent: int # 0..100
message: str

65
tools/production/app/orchestrator.py
View file

@ -4,7 +4,7 @@ orchestrator.py — оркестратор confirm_request для TUI.
Получает события от FirmwareClient, принимает решение кто должен
ответить на confirm_request, и отправляет confirm обратно.
Три режима confirm (из закрытых решений Этапа 8):
Три режима confirm:
standalone/interactive (display, mqs, usd):
→ TUI показывает prompt оператору + кнопки OK/FAIL + countdown
@ -21,7 +21,7 @@ orchestrator.py — оркестратор confirm_request для TUI.
→ отправляет confirm без участия оператора
→ оператор видит только прогресс
Маршрутизация определяется по prefix confirm_request.id:
Маршрутизация определяется по id confirm_request:
"opto_*" → HIL (M5 relay)
"can_rx_ready" → HIL (M5 can_send)
"can_tx_verify" → HIL (M5 can_recv + verify)
@ -42,47 +42,48 @@ from typing import AsyncGenerator, Callable, Coroutine, Optional
from .firmware_client import FirmwareClient
from .m5_client import M5Client
from .models import ConfirmRequest, TestInfo, TestResult, TestStatus
from .models import ConfirmRequest, TestResult, TestStatus
logger = logging.getLogger(__name__)
# Задержки для HIL (из закрытых решений: RELAY_ON_S=0.15, RELAY_OFF_S=0.5)
# Задержки для HIL
_RELAY_ON_S = 0.15
_RELAY_OFF_S = 0.50
# Карта confirm_id → реле M5 для opto-теста
# Формат: confirm_id_prefix → (relay_num, target_state)
# Формат: confirm_id → (relay_num, target_state)
_OPTO_RELAY_MAP: dict[str, tuple[int, bool]] = {
"opto_in1_active": (3, True),
"opto_in1_active": (3, True),
"opto_in1_inactive": (3, False),
"opto_in2_active": (4, True),
"opto_in2_active": (4, True),
"opto_in2_inactive": (4, False),
"opto_rs_active": (2, True),
"opto_rs_inactive": (2, False),
"opto_rs_active": (2, True),
"opto_rs_inactive": (2, False),
}
# CAN параметры (из закрытых решений)
_CAN_RX_ID = 0x100
# CAN параметры
_CAN_RX_ID = 0x100
_CAN_RX_DATA = [0xDE, 0xAD, 0xBE, 0xEF]
_CAN_TX_ID = 0x200
_CAN_TX_ID = 0x200
_CAN_TX_DATA = [0xCA, 0xFE, 0xBA, 0xBE]
# ── Типы событий оркестратора ────────────────────────────────────────────
class OrchestratorEventType(Enum):
TEST_BEGIN = auto() # тест начался
TEST_RESULT = auto() # тест завершился
CONFIRM_NEEDED = auto() # нужен ответ оператора (standalone)
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
BUTTONS_PROMPT = auto() # показать инструкцию для buttons (без confirm)
SUMMARY = auto() # итог всей сессии
ERROR = auto() # ошибка протокола или M5
@dataclass
class OrchestratorEvent:
"""Событие от оркестратора — передаётся в TUI через asyncio.Queue."""
"""Событие от оркестратора — передаётся в TUI."""
type: OrchestratorEventType
test_id: str = ""
@ -93,8 +94,7 @@ class OrchestratorEvent:
message: str = ""
# Тип callback для ответа оператора:
# TUI вызывает его с True (OK) или False (FAIL) после показа prompt
# Тип callback для ответа оператора
OperatorConfirmCallback = Callable[[bool], Coroutine]
@ -107,7 +107,6 @@ class Orchestrator:
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:
@ -121,7 +120,6 @@ class Orchestrator:
) -> 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:
@ -151,12 +149,11 @@ class Orchestrator:
)
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)
}.get(raw.get("status", "fail"), TestStatus.FAIL)
result = TestResult(
id=raw.get("id", ""),
@ -176,8 +173,8 @@ class Orchestrator:
prompt=raw.get("prompt", ""),
timeout_ms=raw.get("timeout_ms", 30000),
)
async for orch_event in self._handle_confirm(confirm):
yield orch_event
async for ev in self._handle_confirm(confirm):
yield ev
elif event_type == "summary":
yield OrchestratorEvent(
@ -185,10 +182,10 @@ class Orchestrator:
summary=raw,
)
elif event_type in ("error", None):
else:
yield OrchestratorEvent(
type=OrchestratorEventType.ERROR,
message=raw.get("error", "unknown error"),
message=raw.get("error", f"unknown event: {event_type}"),
)
# ── Маршрутизация confirm ────────────────────────────────────────────
@ -199,25 +196,21 @@ class Orchestrator:
"""Определить тип 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,
@ -225,7 +218,6 @@ class Orchestrator:
)
return
# standalone interactive: показать оператору
async for ev in self._handle_operator_confirm(confirm):
yield ev
@ -248,7 +240,6 @@ class Orchestrator:
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(
@ -270,7 +261,6 @@ class Orchestrator:
)
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)
@ -293,7 +283,6 @@ class Orchestrator:
)
return
# Таргет уже отправил фрейм, M5 принимает и верифицирует
frame = await self._m5.can_recv(timeout_ms=500)
verified = (
frame is not None
@ -322,7 +311,6 @@ class Orchestrator:
Передать confirm оператору. Блокируется до resolve_operator_confirm().
TUI должен показать prompt и дать возможность ответить.
"""
# очистить очередь на случай устаревшего ответа
while not self._operator_queue.empty():
self._operator_queue.get_nowait()
@ -331,7 +319,6 @@ class Orchestrator:
confirm=confirm,
)
# ждём ответа от оператора (TUI вызовет resolve_operator_confirm)
timeout_s = confirm.timeout_ms / 1000.0
try:
confirmed = await asyncio.wait_for(

3
tools/production/app/screens/__init__.py
View file

@ -2,6 +2,7 @@
from .diag import DiagScreen
from .flash import FlashScreen
from .post_flash import PostFlashScreen
from .waiting import WaitingScreen
__all__ = ["WaitingScreen", "FlashScreen", "DiagScreen"]
__all__ = ["WaitingScreen", "FlashScreen", "PostFlashScreen", "DiagScreen"]

108
tools/production/app/screens/diag/__init__.py
View file

@ -17,7 +17,7 @@ from typing import Optional
from textual import on, work
from textual.app import ComposeResult
from textual.binding import Binding
from textual.containers import Horizontal
from textual.containers import Horizontal, Vertical
from textual.css.query import NoMatches
from textual.message import Message
from textual.screen import Screen
@ -27,6 +27,7 @@ from ...firmware_client import FirmwareClient
from ...m5_client import M5Client
from ...models import SessionState
from ...orchestrator import Orchestrator, OrchestratorEvent, OrchestratorEventType
from ...widgets import AppFrame
from .confirm_panel import ConfirmPanel
from .results import ResultsPanel
from .test_list import TestListPanel
@ -55,48 +56,55 @@ class DiagScreen(Screen):
self,
firmware: FirmwareClient,
m5: Optional[M5Client] = None,
fw_version: str = "",
**kwargs,
) -> None:
super().__init__(**kwargs)
self._fw = firmware
self._m5 = m5
self._orchestrator = Orchestrator(firmware, m5)
self._session = SessionState()
self._running = False
self._session = SessionState(fw_version=fw_version)
self._tests_running = False
def compose(self) -> ComposeResult:
# Шапка
with Horizontal(id="diag-header"):
yield Static("fw: —", id="diag-header-fw")
yield Static("UID: —", id="diag-header-uid")
yield Static("M5: —", id="diag-header-m5")
with AppFrame(id="diag-frame"):
# Шапка — фиксированная высота 3
with Horizontal(id="diag-header"):
yield Static("firmware_test: —\nFw: —", id="diag-header-fw")
yield Static("MCU ID: —", id="diag-header-uid")
yield Static("M5: —", id="diag-header-m5")
# Рабочая зона: список тестов + результаты
with Horizontal(id="diag-main"):
yield TestListPanel(id="diag-test-list")
yield ResultsPanel(id="diag-results")
# Рабочая зона: список тестов + результаты — занимает всё
# оставшееся место (1fr), сама прокручивается при переполнении
with Horizontal(id="diag-main"):
yield TestListPanel(id="diag-test-list")
yield ResultsPanel(id="diag-results")
# Прогресс
yield ProgressBar(id="diag-progress-bar", show_eta=False)
yield Label("", id="diag-progress-label")
# Прогресс — скрыт пока нет активного прогона. Контейнер
# имеет classes="hidden" по умолчанию — height: auto + display:
# none даёт нулевую высоту, не отнимая место у остального layout.
with Vertical(id="diag-progress-row", classes="hidden"):
yield ProgressBar(id="diag-progress-bar", show_eta=False)
yield Label("", id="diag-progress-label")
# Кнопки запуска
with Horizontal(id="diag-btn-row"):
yield Button(
"▶ Запустить выбранные",
id="diag-btn-run-selected",
variant="primary",
disabled=True,
)
yield Button(
"▶▶ Все тесты",
id="diag-btn-run-all",
variant="default",
disabled=True,
)
# Кнопки запуска — фиксированная высота 3 (под border Button)
with Horizontal(id="diag-btn-row"):
yield Button(
"▶ Запустить выбранные",
id="diag-btn-run-selected",
variant="primary",
disabled=True,
)
yield Button(
"▶▶ Все тесты",
id="diag-btn-run-all",
variant="default",
disabled=True,
)
yield Button("✕ Выйти", id="diag-btn-quit", variant="default")
# Панель confirm
yield ConfirmPanel(id="diag-confirm", classes="hidden")
# Панель confirm — auto-высота, видна только когда есть запрос
yield ConfirmPanel(id="diag-confirm", classes="hidden")
def on_mount(self) -> None:
self._init_session()
@ -108,7 +116,6 @@ class DiagScreen(Screen):
try:
tests = await self._fw.list_tests()
uid = await self._fw.get_uid()
version = await self._fw.get_version()
except Exception as exc:
logger.error("Session init failed: %s", exc)
self.post_message(self.DiagDone())
@ -116,7 +123,6 @@ class DiagScreen(Screen):
self._session.tests = tests
self._session.chip_uid = uid
self._session.fw_version = version
self._session.m5_connected = self._m5 is not None
self._update_header()
@ -129,17 +135,17 @@ class DiagScreen(Screen):
def _update_header(self) -> None:
self.query_one("#diag-header-fw", Static).update(
f"fw: {self._session.fw_version or '?'}"
f"firmware_test: {self._session.fw_version or '?'}"
)
uid_short = self._session.chip_uid[:16] if self._session.chip_uid else "—"
self.query_one("#diag-header-uid", Static).update(f"UID: {uid_short}")
uid = self._session.chip_uid or "—"
self.query_one("#diag-header-uid", Static).update(f"MCU ID: {uid}")
m5_widget = self.query_one("#diag-header-m5", Static)
if self._session.m5_connected:
m5_widget.update("M5: ✓ подключён")
m5_widget.remove_class("m5-absent")
else:
m5_widget.update("M5: — нет")
m5_widget.update("M5: ✕ не подключен")
m5_widget.add_class("m5-absent")
# ── Кнопки ───────────────────────────────────────────────────────────────
@ -156,8 +162,13 @@ class DiagScreen(Screen):
if ids:
self._start_run(ids)
@on(Button.Pressed, "#diag-btn-quit")
def _on_quit_pressed(self) -> None:
if not self._tests_running:
self.app.exit()
def action_go_back(self) -> None:
if not self._running:
if not self._tests_running:
self.post_message(self.DiagDone())
def action_run_all(self) -> None:
@ -181,9 +192,11 @@ class DiagScreen(Screen):
results = self.query_one("#diag-results", ResultsPanel)
results.reset()
self._session.results.clear()
self._show_progress(True)
self._update_progress(0, len(test_ids), "")
self._set_run_buttons(enabled=False)
self._running = True
self.query_one("#diag-test-list", TestListPanel).set_enabled(False)
self._tests_running = True
self._run_worker(test_ids)
@work(exclusive=True, thread=False)
@ -200,8 +213,9 @@ class DiagScreen(Screen):
except Exception as exc:
logger.error("run_worker error: %s", exc)
finally:
self._running = False
self._tests_running = False
self._set_run_buttons(enabled=True)
self.query_one("#diag-test-list", TestListPanel).set_enabled(True)
try:
self.query_one("#diag-confirm", ConfirmPanel).hide()
except NoMatches:
@ -251,9 +265,21 @@ class DiagScreen(Screen):
# ── Утилиты ───────────────────────────────────────────────────────────────
def _show_progress(self, visible: bool) -> None:
"""Показать/скрыть строку прогресса. Скрыта в простое — без анимации."""
row = self.query_one("#diag-progress-row")
bar = self.query_one("#diag-progress-bar", ProgressBar)
if visible:
bar.update(total=100, progress=0)
row.remove_class("hidden")
else:
row.add_class("hidden")
def _update_progress(self, done: int, total: int, msg: str) -> None:
pct = int(done / total * 100) if total else 0
self.query_one("#diag-progress-bar", ProgressBar).update(progress=pct)
self.query_one("#diag-progress-bar", ProgressBar).update(
total=100, progress=pct
)
self.query_one("#diag-progress-label", Label).update(msg)
def _set_run_buttons(self, enabled: bool) -> None:

11
tools/production/app/screens/diag/confirm_panel.py
View file

@ -28,7 +28,7 @@ class ConfirmPanel(Widget):
Использование::
panel = ConfirmPanel()
panel.show(prompt="Экран залит красным?", timeout_ms=30000)
panel.show_operator(prompt="Экран залит красным?", timeout_ms=30000)
# Слушать ConfirmPanel.Confirmed в родительском экране
"""
@ -39,13 +39,10 @@ class ConfirmPanel(Widget):
super().__init__()
self.confirmed = confirmed
DEFAULT_CSS = "" # стили в diag.tcss
def __init__(self, **kwargs) -> None:
super().__init__(**kwargs)
self._timer: Timer | None = None
self._remaining: int = 0
self._operator_mode: bool = True # False → режим buttons (нет кнопок)
def compose(self) -> ComposeResult:
yield Static("", id="confirm-prompt")
@ -57,9 +54,8 @@ class ConfirmPanel(Widget):
def show_operator(self, prompt: str, timeout_ms: int) -> None:
"""Показать панель с кнопками OK/Нет и countdown."""
self._operator_mode = True
self._remaining = timeout_ms // 1000
self._set_prompt(prompt)
self._set_prompt(f"⚠ {prompt}")
self._set_countdown(self._remaining)
self._show_buttons(True)
self.remove_class("hidden")
@ -70,7 +66,6 @@ class ConfirmPanel(Widget):
Показать инструкцию для теста кнопок.
Без кнопок OK/Нет — оператор только читает, нажимает физическую кнопку.
"""
self._operator_mode = False
self._stop_timer()
self._set_prompt(f"⌨ {prompt}")
self._set_countdown("")
@ -117,7 +112,7 @@ class ConfirmPanel(Widget):
def _set_prompt(self, text: str) -> None:
try:
self.query_one("#confirm-prompt", Static).update(f"⚠ {text}")
self.query_one("#confirm-prompt", Static).update(text)
except NoMatches:
pass

241
tools/production/app/screens/diag/results.py
View file

@ -1,115 +1,212 @@
"""
results.py — виджет правой колонки DiagScreen.
Отображает строки результатов тестов: id | статус | detail.
Обновляется по событиям от Orchestrator через DiagScreen.
DataTable с колонками: Тест | HIL | Статус | Время | Детали.
До первого запуска показывает empty-state placeholder вместо таблицы.
Сортировка: FAIL всегда наверху (естественно бросается в глаза),
внутри групп статусов — исходный порядок реестра firmware_test.
Строка с FAIL дополнительно подсвечивается красным фоном целиком.
Публичный API:
ResultsPanel.populate(tests) — инициализировать пустые строки
ResultsPanel.set_running(test_id) — показать "выполняется"
ResultsPanel.populate(tests) — показать пустую таблицу (PENDING)
ResultsPanel.set_running(test_id) — пометить тест как выполняющийся
ResultsPanel.set_result(result) — показать финальный результат
ResultsPanel.reset() — сбросить все строки
ResultsPanel.reset() — сбросить все строки в PENDING
"""
from __future__ import annotations
from rich.style import Style
from rich.text import Text
from textual.app import ComposeResult
from textual.containers import Horizontal
from textual.css.query import NoMatches
from textual.containers import Center, Middle
from textual.widget import Widget
from textual.widgets import Label, Static
from textual.widgets import DataTable, Static
from ...models import TestResult, TestStatus
from ...models import TestInfo, TestResult, TestStatus
# Порядок сортировки: чем меньше число — тем выше строка в таблице.
# FAIL всегда наверху, PASS/SKIP внизу — внутри групп сохраняется
# исходный порядок реестра (стабильная сортировка).
_SORT_RANK: dict[TestStatus, int] = {
TestStatus.FAIL: 0,
TestStatus.RUNNING: 1,
TestStatus.PENDING: 2,
TestStatus.SKIP: 3,
TestStatus.PASS: 4,
}
def _status_display(status: TestStatus) -> tuple[str, str]:
"""Вернуть (текст, css-класс) для статуса."""
return {
TestStatus.PASS: ("✓ PASS", "result-status-pass"),
TestStatus.FAIL: ("✗ FAIL", "result-status-fail"),
TestStatus.RUNNING: ("…", "result-status-running"),
TestStatus.SKIP: ("SKIP", "result-status-skip"),
TestStatus.PENDING: ("", "result-status-skip"),
}[status]
_STATUS_DISPLAY: dict[TestStatus, tuple[str, str]] = {
TestStatus.PASS: ("✓ PASS", "green"),
TestStatus.FAIL: ("✗ FAIL", "bold white"),
TestStatus.RUNNING: ("…", "yellow"),
TestStatus.SKIP: ("SKIP", "grey50"),
TestStatus.PENDING: ("", "grey50"),
}
# CSS-классы статусов — для очистки перед сменой
_STATUS_CLASSES = (
"result-status-pass",
"result-status-fail",
"result-status-running",
"result-status-skip",
)
_FAIL_BG = "dark_red"
class ResultsPanel(Widget):
"""Правая колонка DiagScreen — результаты тестов."""
"""Правая колонка DiagScreen — таблица результатов тестов."""
DEFAULT_CSS = "" # стили в diag.tcss
def __init__(self, **kwargs) -> None:
super().__init__(**kwargs)
# test_id → TestInfo, нужно для перестроения строк при сортировке
self._tests: dict[str, TestInfo] = {}
# test_id → текущий статус (для сортировки и повторных update)
self._statuses: dict[str, TestStatus] = {}
self._order: list[str] = [] # исходный порядок реестра
def compose(self) -> ComposeResult:
yield Label("Результаты", classes="section-title")
with Center(id="results-empty"):
with Middle():
yield Static(
"Выберите тесты и нажмите\n«Запустить выбранные»",
id="results-empty-text",
)
yield DataTable(id="results-table", cursor_type="row", classes="hidden")
def on_mount(self) -> None:
table = self.query_one("#results-table", DataTable)
table.add_columns("Тест", "M5", "Статус", "Время", "Детали")
# ── Public API ────────────────────────────────────────────────────────────
def populate(self, tests: list) -> None:
def populate(self, tests: list[TestInfo]) -> None:
"""
Инициализировать пустые строки результатов.
Заполнить таблицу тестами в состоянии PENDING.
До первого вызова populate() с непустым списком виден empty-state.
"""
table = self.query_one("#results-table", DataTable)
table.clear()
self._tests.clear()
self._statuses.clear()
self._order = [t.id for t in tests]
:param tests: list[TestInfo]
"""
for child in list(self.children):
if not child.has_class("section-title"):
child.remove()
if not tests:
self._show_empty(True)
return
for test in tests:
row = Horizontal(classes="result-row", id=f"result-row-{test.id}")
id_lbl = Static(test.id, classes="result-id")
status_lbl = Static(
"", classes="result-status-skip", id=f"result-status-{test.id}"
)
detail_lbl = Static(
"", classes="result-detail", id=f"result-detail-{test.id}"
)
self.mount(row)
row.mount(id_lbl)
row.mount(status_lbl)
row.mount(detail_lbl)
self._tests[test.id] = test
self._statuses[test.id] = TestStatus.PENDING
self._add_row(test, TestStatus.PENDING, duration_ms=0, detail="")
self._show_empty(False)
def set_running(self, test_id: str) -> None:
"""Пометить тест как выполняющийся."""
self._update(test_id, TestStatus.RUNNING, "")
self._update_row(test_id, TestStatus.RUNNING, duration_ms=0, detail="")
def set_result(self, result: TestResult) -> None:
"""Показать финальный результат теста."""
detail = result.detail if result.status == TestStatus.FAIL else ""
self._update(result.id, result.status, detail)
self._update_row(result.id, result.status, result.duration_ms, detail)
def reset(self) -> None:
"""Сбросить все строки в пустое состояние."""
for widget in self.query(Static):
wid = widget.id or ""
if wid.startswith("result-status-"):
for cls in _STATUS_CLASSES:
widget.remove_class(cls)
widget.add_class("result-status-skip")
widget.update("")
elif wid.startswith("result-detail-"):
widget.update("")
"""Сбросить все строки в PENDING (перед повторным запуском)."""
for test_id in list(self._tests.keys()):
self._update_row(test_id, TestStatus.PENDING, duration_ms=0, detail="")
# ── Internal ──────────────────────────────────────────────────────────────
# ── Internal: empty state ───────────────────────────────────────────────
def _update(self, test_id: str, status: TestStatus, detail: str) -> None:
try:
status_w = self.query_one(f"#result-status-{test_id}", Static)
detail_w = self.query_one(f"#result-detail-{test_id}", Static)
except NoMatches:
def _show_empty(self, visible: bool) -> None:
empty = self.query_one("#results-empty")
table = self.query_one("#results-table", DataTable)
if visible:
empty.remove_class("hidden")
table.add_class("hidden")
else:
empty.add_class("hidden")
table.remove_class("hidden")
# ── Internal: строки таблицы ────────────────────────────────────────────
def _add_row(
self, test: TestInfo, status: TestStatus, duration_ms: int, detail: str
) -> None:
table = self.query_one("#results-table", DataTable)
cells = self._row_cells(test, status, duration_ms, detail)
table.add_row(*cells, key=test.id)
def _update_row(
self, test_id: str, status: TestStatus, duration_ms: int, detail: str
) -> None:
test = self._tests.get(test_id)
if test is None:
return
self._statuses[test_id] = status
for cls in _STATUS_CLASSES:
status_w.remove_class(cls)
table = self.query_one("#results-table", DataTable)
cells = self._row_cells(test, status, duration_ms, detail)
col_keys = [c.key for c in table.ordered_columns]
for col_key, value in zip(col_keys, cells):
table.update_cell(test_id, col_key, value)
text, css = _status_display(status)
status_w.update(text)
status_w.add_class(css)
detail_w.update(detail)
self._resort(table)
def _row_cells(
self, test: TestInfo, status: TestStatus, duration_ms: int, detail: str
) -> tuple:
"""Собрать пять ячеек строки с учётом подсветки FAIL."""
status_text, status_color = _STATUS_DISPLAY[status]
is_fail = status == TestStatus.FAIL
bg = _FAIL_BG if is_fail else None
fg = "white" if is_fail else status_color
name_style = Style(bgcolor=bg, bold=is_fail)
hil_style = Style(bgcolor=bg, color="white" if is_fail else "cyan")
status_style = Style(bgcolor=bg, color=fg, bold=True)
time_style = Style(bgcolor=bg, color="white" if is_fail else "grey70")
detail_style = Style(bgcolor=bg, color="white" if is_fail else "grey50")
time_text = f"{duration_ms / 1000:.1f}с" if duration_ms > 0 else ""
hil_text = "[*]" if test.requires_hil else "[-]"
return (
Text(test.name, style=name_style),
Text(hil_text, style=hil_style),
Text(status_text, style=status_style),
Text(time_text, style=time_style),
Text(detail, style=detail_style),
)
def _resort(self, table: DataTable) -> None:
"""
Пересортировать: FAIL наверх, дальше RUNNING/PENDING/SKIP/PASS.
Внутри групп — исходный порядок реестра (стабильная сортировка).
DataTable.sort(*columns, key=...) передаёт в key() кортеж значений
ЯЧЕЕК (не row_key) для указанных columns — поэтому сортируем по
содержимому самой ячейки "Тест" (используем как индекс в self._order)
и по тексту статуса, который мы сами туда пишем и полностью
контролируем — никаких приватных атрибутов DataTable не трогаем.
"""
name_col = table.ordered_columns[0].key
status_col = table.ordered_columns[2].key
# text -> status enum, обратное к _STATUS_DISPLAY
status_by_text = {text: status for status, (text, _) in _STATUS_DISPLAY.items()}
# имя теста -> индекс в исходном реестре (для стабильности внутри группы)
name_to_order = {
self._tests[tid].name: idx
for idx, tid in enumerate(self._order)
if tid in self._tests
}
def sort_key(cells: tuple) -> tuple:
name_cell, status_cell = cells
name_plain = (
name_cell.plain if hasattr(name_cell, "plain") else str(name_cell)
)
status_plain = (
status_cell.plain if hasattr(status_cell, "plain") else str(status_cell)
)
status = status_by_text.get(status_plain, TestStatus.PENDING)
order_idx = name_to_order.get(name_plain, 0)
return (_SORT_RANK.get(status, 99), order_idx)
table.sort(name_col, status_col, key=sort_key)

75
tools/production/app/screens/diag/test_list.py
View file

@ -12,46 +12,63 @@ test_list.py — виджет списка тестов с чекбоксами.
from __future__ import annotations
from textual import on
from textual.app import ComposeResult
from textual.containers import Horizontal
from textual.widget import Widget
from textual.widgets import Checkbox, Label
from textual.widgets import Button, Checkbox, Label
class TestListPanel(Widget):
"""Левая колонка DiagScreen — список тестов с чекбоксами."""
DEFAULT_CSS = "" # стили в diag.tcss
def __init__(self, **kwargs) -> None:
super().__init__(**kwargs)
# test_id → Checkbox для быстрого доступа
self._checkboxes: dict[str, Checkbox] = {}
# test_id → True если HIL-тест недоступен без M5 (постоянное состояние,
# не зависящее от прогона). Отдельно от Checkbox.disabled, который
# временно перещёлкивается на время прогона тестов через set_enabled().
self._hil_unavailable: dict[str, bool] = {}
def compose(self) -> ComposeResult:
yield Label("Тесты", classes="section-title")
with Horizontal(id="test-list-select-row"):
yield Button(
"Выбрать все", id="test-list-select-all", classes="-textual-compact"
)
yield Button(
"Снять все", id="test-list-select-none", classes="-textual-compact"
)
# ── Public API ────────────────────────────────────────────────────────────
def populate(self, tests: list, m5_connected: bool) -> None:
"""
Заполнить список тестами.
Заполнить список тестами. Изначально все тесты НЕ выбраны
(сервисник выбирает явно что запускать) — кроме HIL без M5,
которые всегда disabled.
:param tests: list[TestInfo]
:param m5_connected: True если M5StampPLC подключён
"""
# Удалить старые строки (кроме заголовка)
# Удалить старые строки (кроме заголовка и select-row)
for child in list(self.children):
if not child.has_class("section-title"):
if child.id != "test-list-select-row" and not child.has_class(
"section-title"
):
child.remove()
self._checkboxes.clear()
self._hil_unavailable.clear()
rows = []
for test in tests:
hil_unavailable = test.requires_hil and not m5_connected
self._hil_unavailable[test.id] = hil_unavailable
cb = Checkbox(
test.name,
value=not hil_unavailable,
value=False,
disabled=hil_unavailable,
id=f"cb-{test.id}",
classes="-textual-compact",
@ -65,10 +82,16 @@ class TestListPanel(Widget):
)
badge_text = "[HIL]" if test.requires_hil else ""
row = Horizontal(classes="test-row")
self.mount(row)
row.mount(cb)
row.mount(Label(badge_text, classes=hil_css))
rows.append(
Horizontal(
cb,
Label(badge_text, classes=hil_css),
classes="test-row",
)
)
if rows:
self.mount_all(rows)
def get_selected_ids(self) -> list[str]:
"""Вернуть список id выбранных (checked + not disabled) тестов."""
@ -77,7 +100,31 @@ class TestListPanel(Widget):
]
def set_enabled(self, enabled: bool) -> None:
"""Разрешить/запретить изменение чекбоксов во время прогона."""
for cb in self._checkboxes.values():
if not cb.disabled: # не трогать серые HIL-без-M5
"""
Разрешить/запретить изменение чекбоксов во время прогона.
HIL-чекбоксы без M5 остаются disabled всегда — их постоянное
состояние хранится в self._hil_unavailable и не зависит от прогона.
"""
for tid, cb in self._checkboxes.items():
if self._hil_unavailable.get(tid, False):
cb.disabled = True # HIL без M5 — всегда недоступен
else:
cb.disabled = not enabled
self.query_one("#test-list-select-all", Button).disabled = not enabled
self.query_one("#test-list-select-none", Button).disabled = not enabled
# ── Обработчики ───────────────────────────────────────────────────────────
@on(Button.Pressed, "#test-list-select-all")
def _on_select_all(self) -> None:
for tid, cb in self._checkboxes.items():
if not self._hil_unavailable.get(tid, False):
cb.value = True
@on(Button.Pressed, "#test-list-select-none")
def _on_select_none(self) -> None:
for tid, cb in self._checkboxes.items():
if not self._hil_unavailable.get(tid, False):
cb.value = False

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

@ -30,6 +30,7 @@ from textual.widgets import (
from ..flasher import Flasher
from ..models import FlashProgress, FlashTarget
from ..widgets import AppFrame
logger = logging.getLogger(__name__)
@ -39,7 +40,7 @@ class FlashScreen(Screen):
Экран прошивки.
Messages:
FlashDone(success) — прошивка завершена
FlashDone(success, target) — прошивка завершена
"""
BINDINGS = [
@ -47,9 +48,10 @@ class FlashScreen(Screen):
]
class FlashDone(Message):
def __init__(self, success: bool) -> None:
def __init__(self, success: bool, target: Optional[FlashTarget] = None) -> None:
super().__init__()
self.success = success
self.target = target
def __init__(self, **kwargs) -> None:
super().__init__(**kwargs)
@ -57,48 +59,42 @@ class FlashScreen(Screen):
self._flashing = False
def compose(self) -> ComposeResult:
with Vertical():
with AppFrame(id="flash-frame"):
yield Label(
"⚡ Прошивка платы — BootROM SDP обнаружен",
"⚡ Загрузка прошивки на плату (BootROM SDP обнаружен)",
id="flash-title",
)
with Vertical(id="flash-target-group"):
yield Label("Что прошить?", classes="section-title")
yield Label("Выберите файл для загрузки", classes="section-title")
with RadioSet(id="flash-radio"):
yield RadioButton(
"firmware_test (диагностическая прошивка)",
"Диагностическая прошивка (firmware_test)",
id="radio-fw-test",
value=True,
)
yield RadioButton(
"Production (bootloader + tft_app)",
"Серийная прошивка (bootloader + tft_app)",
id="radio-production",
)
yield RadioButton(
"Кастомный бинарь...",
"Другое (подготовленный бинарный файл)",
id="radio-custom",
)
with Horizontal(id="flash-custom-path", classes="hidden"):
yield Input(
placeholder="Путь к HAB-бинарю (.bin)",
placeholder="Путь к бинарному файлу (.bin)",
id="flash-custom-input",
)
with Horizontal(id="flash-btn-row"):
yield Button("▶ Загрузить ПО", id="flash-btn-flash", variant="warning")
yield Button("⚠ Очистить память", id="flash-btn-erase", variant="error")
yield Button(
"▶ Прошить",
id="flash-btn-flash",
variant="warning",
)
yield Button(
"⚠ Chip Erase",
id="flash-btn-erase",
variant="error",
"✕ Выйти из программы", id="flash-btn-quit", variant="default"
)
yield ProgressBar(id="flash-progress-bar", show_eta=False)
yield Label("", id="flash-progress-label")
yield ProgressBar(id="flash-progress-bar", show_eta=False, classes="hidden")
yield Log(id="flash-log", auto_scroll=True)
# ── Обработчики ───────────────────────────────────────────────────────────
@ -118,15 +114,19 @@ class FlashScreen(Screen):
return
target, bin_path = self._resolve_target()
if target is None:
self._log("⚠ Укажите путь к бинарю")
self._log("⚠ Укажите путь к бинарному файлу!")
return
self._do_flash(target, bin_path)
@on(Button.Pressed, "#flash-btn-erase")
def _on_erase_pressed(self) -> None:
if self._flashing:
return
self._do_erase()
if not self._flashing:
self._do_erase()
@on(Button.Pressed, "#flash-btn-quit")
def _on_quit_pressed(self) -> None:
if not self._flashing:
self.app.exit()
def action_go_back(self) -> None:
if not self._flashing:
@ -137,6 +137,7 @@ class FlashScreen(Screen):
@work(exclusive=True, thread=False)
async def _do_flash(self, target: FlashTarget, bin_path: Optional[Path]) -> None:
self._set_busy(True)
self._show_progress(True)
self._log(f"▶ Прошивка: {target.value}")
ok = await self._flasher.flash(
target=target,
@ -144,16 +145,23 @@ class FlashScreen(Screen):
progress_cb=self._on_progress,
)
self._set_busy(False)
self._finish_progress(ok)
self._log("✅ Готово" if ok else "❌ Ошибка")
self.post_message(self.FlashDone(success=ok))
self.post_message(self.FlashDone(success=ok, target=target))
@work(exclusive=True, thread=False)
async def _do_erase(self) -> None:
self._set_busy(True)
self._log("⚠ Chip erase (~30 с)...")
self._show_progress(True)
self._log("⚠ Очистка флеш памяти (~30 с)...")
ok = await self._flasher.erase_chip(progress_cb=self._on_progress)
self._set_busy(False)
self._log("✅ Chip erase завершён" if ok else "❌ Chip erase: ошибка")
self._finish_progress(ok)
self._log(
"✅ Очистка флеш памяти завершена"
if ok
else "❌ Очистка флеш памяти: ошибка"
)
# ── Вспомогательные ───────────────────────────────────────────────────────
@ -177,18 +185,29 @@ class FlashScreen(Screen):
return None, None
async def _on_progress(self, progress: FlashProgress) -> None:
self.query_one("#flash-progress-bar", ProgressBar).update(
progress=progress.percent
)
self.query_one("#flash-progress-label", Label).update(
f"{progress.phase} {progress.percent}%"
)
bar = self.query_one("#flash-progress-bar", ProgressBar)
bar.update(total=100, progress=progress.percent)
self._log(progress.message)
def _show_progress(self, visible: bool) -> None:
"""Показать/скрыть прогресс-бар. Скрыт в простое — без анимации."""
bar = self.query_one("#flash-progress-bar", ProgressBar)
if visible:
bar.update(total=100, progress=0)
bar.remove_class("hidden")
else:
bar.add_class("hidden")
def _finish_progress(self, success: bool) -> None:
"""Зафиксировать прогресс-бар на 100% с финальным статусом."""
bar = self.query_one("#flash-progress-bar", ProgressBar)
bar.update(total=100, progress=100)
def _set_busy(self, busy: bool) -> None:
self._flashing = busy
self.query_one("#flash-btn-flash", Button).disabled = busy
self.query_one("#flash-btn-erase", Button).disabled = busy
self.query_one("#flash-btn-quit", Button).disabled = busy
def _log(self, msg: str) -> None:
try:

88
tools/production/app/screens/post_flash.py Normal file
View file

@ -0,0 +1,88 @@
"""
post_flash.py — экран-промпт после успешной прошивки firmware_test.
Показывается только для FlashTarget.FIRMWARE_TEST (диагностическая прошивка) —
после неё плата должна быть переведена в нормальный режим (BOOT_MOD → GND)
чтобы попасть в DiagScreen. Без этого промпта TUI зацикливался: SDP всё ещё
виден → WaitingScreen снова детектит SDP → снова FlashScreen.
Для PRODUCTION/CUSTOM (либо когда прошивка не удалась) промпт не нужен —
сразу WaitingScreen с тем же общим поведением автодетекта.
Три способа покинуть экран:
1. Оператор успел сменить BootMode и нажал "Готово" → WaitingScreen
(далее автодетект сам поймает CDC, если плата уже перезагружена)
2. Оператор нажал "Выйти" → app.exit()
3. Таймаут (по умолчанию 40 с) → автоматически WaitingScreen
"""
from __future__ import annotations
from textual.app import ComposeResult
from textual.containers import Horizontal
from textual.message import Message
from textual.screen import Screen
from textual.timer import Timer
from textual.widgets import Button, Label, Static
from ..widgets import AppFrame
_AUTO_TIMEOUT_S = 40
_TICK_S = 1.0
class PostFlashScreen(Screen):
"""
Промпт смены BootMode после прошивки firmware_test.
Messages:
Done() — оператор подтвердил или истёк таймаут → пора в WaitingScreen
"""
class Done(Message):
"""Готово к переходу на WaitingScreen."""
def __init__(self, **kwargs) -> None:
super().__init__(**kwargs)
self._remaining: int = _AUTO_TIMEOUT_S
self._timer: Timer | None = None
def compose(self) -> ComposeResult:
with AppFrame(id="post-flash-frame"):
yield Label("✅ firmware_test успешно записан", id="post-flash-title")
yield Static(
"Переведите плату в нормальный режим:\nBOOT_MOD_1 → GND → Reset",
id="post-flash-instruction",
)
yield Static("", id="post-flash-countdown")
with Horizontal(id="post-flash-btn-row"):
yield Button("✓ Готово", id="post-flash-btn-ok", variant="success")
yield Button("✕ Выйти", id="post-flash-btn-quit", variant="default")
def on_mount(self) -> None:
self._update_countdown()
self._timer = self.set_interval(_TICK_S, self._tick)
def on_unmount(self) -> None:
if self._timer is not None:
self._timer.stop()
def _tick(self) -> None:
self._remaining -= 1
self._update_countdown()
if self._remaining <= 0:
self.post_message(self.Done())
def _update_countdown(self) -> None:
self.query_one("#post-flash-countdown", Static).update(
f"Автопереход через: {self._remaining}с"
)
def on_button_pressed(self, event: Button.Pressed) -> None:
if event.button.id == "post-flash-btn-ok":
self.post_message(self.Done())
elif event.button.id == "post-flash-btn-quit":
self.app.exit()

16
tools/production/app/screens/waiting.py
View file

@ -9,12 +9,14 @@ from __future__ import annotations
from textual.app import ComposeResult
from textual.css.query import NoMatches
from textual.message import Message
from textual.screen import Screen
from textual.timer import Timer
from textual.widgets import Static
from textual.message import Message
from ..flasher import Flasher
from ..models import AppMode
from ..widgets import AppFrame
_DETECT_INTERVAL_S = 1.5
_SPIN_INTERVAL_S = 0.1
@ -43,19 +45,17 @@ class WaitingScreen(Screen):
self._spin_timer: Timer | None = None
def compose(self) -> ComposeResult:
yield Static("TFT Indicator Board\nService Tool", id="waiting-logo")
yield Static("Подключите плату к USB...", id="waiting-hint")
yield Static(_SPINNER_FRAMES[0], id="waiting-spinner")
with AppFrame(id="waiting-frame"):
yield Static("TFT Indicator Board\nService Tool", id="waiting-logo")
yield Static("Подключите плату к USB...", id="waiting-hint")
yield Static(_SPINNER_FRAMES[0], id="waiting-spinner")
def on_mount(self) -> None:
self._detect_timer = self.set_interval(_DETECT_INTERVAL_S, self._poll_usb)
self._spin_timer = self.set_interval(_SPIN_INTERVAL_S, self._spin)
def on_unmount(self) -> None:
if self._detect_timer:
self._detect_timer.stop()
if self._spin_timer:
self._spin_timer.stop()
self._stop_timers()
# ── Internal ──────────────────────────────────────────────────────────────

5
tools/production/app/widgets/__init__.py Normal file
View file

@ -0,0 +1,5 @@
"""widgets — общие переиспользуемые виджеты TUI."""
from .app_frame import AppFrame
__all__ = ["AppFrame"]

40
tools/production/app/widgets/app_frame.py Normal file
View file

@ -0,0 +1,40 @@
"""
app_frame.py — общий контейнер фиксированного размера для всех экранов.
Решает две задачи:
1. UX: визуальная рамка фиксированного размера (как в ratatui-приложениях,
напр. binsider), центрированная в терминале вне зависимости от размера окна.
2. Баг Textual 8.x: AssertionError при mouse drag, когда Screen выступает
content_widget напрямую (assert isinstance(content_widget.parent, Widget)
падает, потому что в этом сценарии родитель не является валидным Widget).
AppFrame как промежуточный контейнер между Screen и содержимым устраняет
этот сценарий — content_widget при drag теперь всегда AppFrame или его
потомок, у которых .parent валиден.
Использование в экране::
def compose(self) -> ComposeResult:
with AppFrame():
yield Static("...")
yield Button("...")
"""
from __future__ import annotations
from textual.containers import Container
class AppFrame(Container):
"""Контейнер фиксированного размера, центрированный в Screen."""
DEFAULT_CSS = """
AppFrame {
width: 100%;
height: 100%;
max-width: 160;
max-height: 50;
border: heavy $primary;
background: $surface;
padding: 1 2;
}
"""

1183
tools/production/service_tui.log
View file

File diff suppressed because it is too large Load diff