# 13 Слой bsp_opto

В процессе доработки. Сейчас невозможно понять, что сигнал на входе перешел с ACTIVE в INACTIVE из-за того, что прерывание включено только на Rising Edge и только там взводится флаг pending
This commit is contained in:
Dmitry Akimov 2026-03-26 17:54:54 +03:00
parent b608d43bfc
commit f037bea91f
22 changed files with 694 additions and 100 deletions

View file

@ -80,7 +80,7 @@ CheckOptions:
- key: readability-identifier-naming.PointerParameterPrefix
value: "p_" # uint8_t *p_buffer
- key: readability-function-size.LineThreshold
value: '50'
value: '60'
- key: readability-function-size.StatementThreshold
value: '30'
- key: readability-magic-numbers.IgnoredIntegerValues

View file

@ -1,7 +1,10 @@
# =============================================================================
# .env — единый источник конфигурации проекта
# Читается: just (set dotenv-load) → экспортируется в окружение (set export)
# → наследуется uv run python3 автоматически
# → наследуется uv run / python3 автоматически
#
# Скопируйте в .env и настройте под свою машину:
# cp .env.example .env
# =============================================================================
# --- Hardware ---
@ -28,7 +31,6 @@ TARGET_CFG=target/imxrt.cfg
GDB_PORT=3333
GDB_EXECUTABLE=arm-none-eabi-gdb
# pyOCD — таргет и частота для gdbserver и flash_swd.py
PYOCD_TARGET=mimxrt1050_quadspi
PYOCD_FREQUENCY=4000000
@ -37,8 +39,12 @@ PYOCD_FREQUENCY=4000000
FCB_PATH=tools/host/dcd/w25q128_fdcb.bin
# --- HIL (аппаратный стенд) ---
HIL_VCOM_PORT=/dev/tty.usbmodemGUXFBWDJBWTGQ3
HIL_VCOM_BAUD=115200 # default: 115200
HIL_READY_TIMEOUT=5.0 # default: 5.0 сек
HIL_PYOCD_FREQUENCY=1000000 # default: 1 МГц
HIL_BUILD_DIR=build/target-debug # default:
# Порты: macOS = /dev/cu.usbmodem*, Linux = /dev/ttyACM*
HIL_VCOM_PORT=/dev/ttyACM0
HIL_M5_PORT=/dev/ttyACM1
HIL_VCOM_BAUD=115200
HIL_M5_BAUD=115200
HIL_READY_TIMEOUT=5.0
HIL_M5_TIMEOUT=3.0
HIL_PYOCD_FREQUENCY=1000000
HIL_BUILD_DIR=build/target-debug

1
.gitignore vendored
View file

@ -69,6 +69,7 @@ compile_commands.json
__pycache__/
*.pyc
.env
.claude
tools/host/.venv-host/
tools/host/.venv-host-win/

View file

@ -165,7 +165,8 @@
"displayName": "Target Tests — Debug (HIL)",
"configurePreset": "target-debug",
"targets": [
"test_host_uart"
"test_host_uart",
"test_hil_opto"
]
}
],

View file

@ -23,8 +23,9 @@
#define OPTO_IN2_PIN 21U
#define OPTO_RS_PIN 23U
/* Active-low: оптопара тянет пин к GND → ACTIVE = LOW = 0 */
#define OPTO_PIN_TO_STATE(raw) ((raw) == 0U ? BSP_OPTO_STATE_ACTIVE : BSP_OPTO_STATE_INACTIVE)
//FIXME: fixed (оптопара неинтвертирующая: когда на входе 1, на выходе тоже будет 1)
/* Active-HIGH: оптопара тянет пин к VCC → ACTIVE = HIGH = 1 */
#define OPTO_PIN_TO_STATE(raw) ((raw) == 1U ? BSP_OPTO_STATE_ACTIVE : BSP_OPTO_STATE_INACTIVE)
/* ── Состояние канала ────────────────────────────────────────────────────── */
@ -71,6 +72,7 @@ static void enable_irq(uint8_t pin, bsp_opto_edge_t edge)
* Общий обработчик для GPIO1[16..31].
* Все три канала (пины 21, 22, 23) попадают сюда.
*/
void GPIO1_Combined_16_31_IRQHandler(void) // NOLINT(readability-identifier-naming)
{
uint32_t flags = GPIO_GetPinsInterruptFlags(OPTO_GPIO_BASE);
@ -185,11 +187,11 @@ void bsp_opto_process(void)
for (bsp_opto_ch_t ch = 0U; ch < BSP_OPTO_CH_COUNT; ch++)
{
opto_ch_state_t *p_ch = &g_s_opto.channels[ch];
if (!p_ch->enabled || !p_ch->pending)
{
continue;
}
// TODO: разобраться как ловить ситуацию, когда сигнал отключается (Возможно нужно включить, как RISING, так и FALLING (кроме RS))
// if (!p_ch->enabled || !p_ch->pending)
// {
// continue;
// }
uint32_t elapsed = now - p_ch->last_edge_ms;
if (elapsed < g_s_opto.debounce_ms)

View file

@ -20,6 +20,7 @@ target_link_libraries(
bsp_tick
bsp_uart_host
bsp_boot_xip
bsp_opto
port_log_uart
lib_external # SEGGER RTT если включён через SEGGER_RTT_ENABLED
)

View file

@ -1,15 +1,43 @@
#include "board.h"
#include "bsp/led.h"
#include "bsp/opto.h"
#include "bsp/tick.h"
#include "bsp/uart_host.h"
#include "log/log.h"
#include "port/log_uart.h"
#include <stdbool.h>
#include <stdint.h>
static volatile bool is_in1_activated = false;
static volatile bool is_in2_activated = false;
static volatile bool is_rs_activated = false;
static void on_opto_change(bsp_opto_ch_t ch, bsp_opto_state_t state)
{
if (ch == BSP_OPTO_CH_IN1 && state == BSP_OPTO_STATE_ACTIVE)
{
/* IN1 активирован */
is_in1_activated = true;
}
if (ch == BSP_OPTO_CH_IN2 && state == BSP_OPTO_STATE_ACTIVE)
{
/* IN2 активирован */
is_in2_activated = true;
}
if (ch == BSP_OPTO_CH_RS && state == BSP_OPTO_STATE_ACTIVE)
{
/* RS активирован */
is_rs_activated = true;
}
}
int main(void)
{
const uint16_t DELAY_MS = 1000;
const uint16_t DELAY_MS = 10;
const uint32_t UART_BAUDRATE = 115200;
board_hw_init();
bsp_led_init();
@ -17,16 +45,58 @@ int main(void)
bsp_uart_host_init(UART_BAUDRATE);
log_uart_init();
/* --- Opto init --- */
bsp_opto_config_t opto_cfg = {
.callbacks = { on_opto_change, on_opto_change, on_opto_change },
.edges = { BSP_OPTO_EDGE_RISING, BSP_OPTO_EDGE_RISING, BSP_OPTO_EDGE_RISING },
.rs_as_gpio = true,
.debounce_ms = 0, /* instant — для быстрого тестирования */
};
bsp_opto_init(&opto_cfg);
bsp_led_on(LED_APP);
LOG_I("BOOT", "firmware_test started, tick=%lu", (unsigned long) bsp_tick_get_ms());
LOG_D("BOOT", "RX buffer ready, waiting for host...");
uint32_t cycle = 0;
bool in1_enable = false;
bool in2_enable = false;
bool rs_enable = false;
while (1)
{
bsp_led_on(LED_HEARTBEAT);
LOG_D("CLI", "Running in the loop: '%d'", cycle % UINT32_MAX);
bsp_delay(DELAY_MS);
bsp_led_off(LED_HEARTBEAT);
bsp_opto_process();
// Контроль включения
if (is_in1_activated)
{
is_in1_activated = false;
in1_enable = true;
LOG_D("INPUT", "CH1: ACTIVE!");
}
if (is_in2_activated)
{
is_in2_activated = false;
in2_enable = true;
LOG_D("INPUT", "CH2: ACTIVE!");
}
if (is_rs_activated)
{
is_rs_activated = false;
rs_enable = true;
LOG_D("INPUT", "RS: ACTIVE!");
}
// Контроль выключения
if (in1_enable && bsp_opto_read(BSP_OPTO_CH_IN1) == BSP_OPTO_STATE_INACTIVE)
{
in1_enable = false;
LOG_D("INPUT", "CH1: DISABLED!");
}
if (in2_enable && bsp_opto_read(BSP_OPTO_CH_IN2) == BSP_OPTO_STATE_INACTIVE)
{
in2_enable = false;
LOG_D("INPUT", "CH2: DISABLED!");
}
if (rs_enable && bsp_opto_read(BSP_OPTO_CH_RS) == BSP_OPTO_STATE_INACTIVE)
{
rs_enable = false;
LOG_D("INPUT", "RS: DISABLED!");
}
bsp_delay(DELAY_MS);
}
}

View file

@ -223,12 +223,12 @@ setup-m5-udev:
fi
# =============================================================================
# ГРУППА: flash — USB Serial Downloader (SDP через ROM-загрузчик)
# ГРУППА: flash-sdp — USB Serial Downloader (SDP через ROM-загрузчик)
# Требует: плата в режиме Serial Downloader (BOOT_MODE = 01)
# =============================================================================
[doc('Прошить образ через USB SDP: flash <firmware_test|bootloader|app> <debug|release>')]
[group('flash')]
[group('flash-sdp')]
flash project type="release":
#!/usr/bin/env bash
set -euo pipefail
@ -241,7 +241,7 @@ flash project type="release":
--firmware "{{ project }}" --build-type "${BUILD_TYPE}"
[doc('Прошить образ в RAM через USB SDP: flash-ram <firmware_test|bootloader|app> <debug|release>')]
[group('flash')]
[group('flash-sdp')]
flash-ram project type="debug":
#!/usr/bin/env bash
set -euo pipefail
@ -254,18 +254,18 @@ flash-ram project type="debug":
--firmware "{{ project }}" --build-type "${BUILD_TYPE}" --ram-only
[doc('Прошить firmware_test Debug через USB SDP')]
[group('flash')]
[group('flash-sdp')]
flash-test-debug:
@just host::flash firmware_test debug
[doc('Прошить firmware_test Release через USB SDP')]
[group('flash')]
[group('flash-sdp')]
flash-test-release:
@just host::flash firmware_test release
[confirm("Flash bootloader + app (Release)?")]
[doc('Прошить bootloader + tft_app Release через USB SDP (производство)')]
[group('flash')]
[group('flash-sdp')]
flash-production:
@just host::flash bootloader release
@just host::flash app release
@ -314,6 +314,15 @@ flash-swd-app-release:
uv run --directory {{ HIL_DIR }} python {{ _flash_swd }} \
--firmware app --build-type Release
[confirm("Стереть всю Flash (W25Q128)? Все прошивки будут удалены.")]
[doc('Стереть всю Flash (W25Q128) через SWD (chip erase). Нужен power cycle после')]
[group('flash-swd')]
flash-swd-erase:
uv run --directory {{ HIL_DIR }} pyocd erase \
--target {{ env('PYOCD_TARGET', 'mimxrt1050_quadspi') }} \
--frequency {{ env('PYOCD_FREQUENCY', '4000000') }} \
--chip
# =============================================================================
# ГРУППА: hil — запуск HIL-тестов на реальном железе
# Сборка target-прошивок — в devcontainer: just build::build-hil
@ -321,30 +330,23 @@ flash-swd-app-release:
_hil_build := BUILD_DIR / "target-debug"
[doc('Загрузить test_host_uart.elf на таргет без запуска тестов (для ручной отладки)')]
[group('hil')]
hil-load:
HIL_BUILD_DIR={{ _hil_build }} \
uv run --directory {{ HIL_DIR }} python load_and_run.py \
{{ _hil_build }}/tests/target/host_uart/test_host_uart.elf
[doc('Запустить все HIL-тесты')]
[group('hil')]
hil-run:
HIL_BUILD_DIR={{ _hil_build }} \
uv run --directory {{ HIL_DIR }} pytest -v
[doc('Запустить только smoke HIL-тесты (быстро, для CI)')]
[doc('Запустить HIL-тест UART')]
[group('hil')]
hil-smoke:
hil-uart:
HIL_BUILD_DIR={{ _hil_build }} \
uv run --directory {{ HIL_DIR }} pytest -v -m smoke
uv run --directory {{ HIL_DIR }} pytest test_uart.py -v
[doc('Запустить HIL-тесты без перезагрузки ELF (прошивка уже запущена)')]
[doc('Запустить HIL-тест оптоизолированных входов')]
[group('hil')]
hil-run-fast:
hil-opto:
HIL_BUILD_DIR={{ _hil_build }} \
uv run --directory {{ HIL_DIR }} pytest -v --no-load
uv run --directory {{ HIL_DIR }} pytest test_opto.py -v
# =============================================================================
# ГРУППА: debug — GDB-сервер для отладки из VSCode (devcontainer)
@ -381,7 +383,7 @@ debug-list-targets:
# =============================================================================
M5_PORT := env('HIL_M5_PORT', '/dev/ttyACM1')
_m5_agent := justfile_directory() / 'tools/hil/m5_agent.py'
_m5_agent := justfile_directory() / 'tools/hil/m5/agent.py'
[doc('Показать подключённые M5Stack устройства (USB CDC)')]
[group('m5')]
@ -421,31 +423,6 @@ m5-scan:
m5-repl:
uv run --directory {{ HIL_DIR }} mpremote connect {{ M5_PORT }} repl
[doc('Скопировать агент на M5Stack (tools/hil/m5_agent.py → /main.py)')]
[group('m5')]
m5-deploy:
#!/usr/bin/env bash
set -euo pipefail
echo " 📤 Копируем агент на M5Stack..."
uv run --directory "{{ HIL_DIR }}" mpremote connect "{{ M5_PORT }}" \
cp "{{ _m5_agent }}" :main.py
echo " ✅ m5_agent.py → /main.py"
echo " ✅ M5Stack перезапущен"
[doc('Показать /main.py на M5Stack (что сейчас установлено)')]
[group('m5')]
m5-show:
uv run --directory "{{ HIL_DIR }}" mpremote connect "{{ M5_PORT }}" \
run - <<< "$(printf 'with open(\"/main.py\") as f:\n print(f.read())\n')"
[doc('Запустить интерактивный CLI для ручного тестирования агента')]
[group('m5')]
m5-cli:
uv run --directory {{ HIL_DIR }} python m5_host_cli.py --port {{ M5_PORT }}
[doc('Проверить версию MicroPython на M5Stack')]
[group('m5')]
m5-version:
@ -463,6 +440,27 @@ m5-start:
sleep 2
just host::m5-cli
[doc('Управление питанием таргета через M5 RLY1: m5-power <on|off>')]
[group('m5')]
m5-power state:
uv run --directory {{ HIL_DIR }} python m5/power.py \
--port {{ M5_PORT }} --state {{ state }}
[doc('Скопировать агент на M5Stack (tools/hil/m5_agent.py → /main.py)')]
[group('m5')]
m5-deploy:
#!/usr/bin/env bash
set -euo pipefail
echo " 📤 Копируем агент на M5Stack..."
uv run --directory "{{ HIL_DIR }}" mpremote connect "{{ M5_PORT }}" \
cp "{{ _m5_agent }}" :main.py
echo " ✅ m5_agent.py → /main.py"
echo " ✅ M5Stack перезапущен"
[doc('Запустить интерактивный CLI для ручного тестирования агента')]
[group('m5')]
m5-cli:
uv run --directory {{ HIL_DIR }} python m5/cli.py --port {{ M5_PORT }}
# =============================================================================
# ГРУППА: util

View file

@ -1 +1,2 @@
add_subdirectory(host_uart)
add_subdirectory(hil_opto)

View file

@ -0,0 +1,25 @@
# =============================================================================
# tests/target/hil_opto/CMakeLists.txt
# =============================================================================
set(TARGET_NAME test_hil_opto)
add_executable(${TARGET_NAME} main.c ${BSP_GENERATED}/clock_config.c
${BSP_STARTUP_FILE} ${BSP_SYSCALLS_FILE})
target_link_options(
${TARGET_NAME}
PRIVATE
-T${CMAKE_SOURCE_DIR}/cmake/linker/MIMXRT1052xxxxx_ram.ld
-Wl,--gc-sections
-Wl,--print-memory-usage
-Wl,-Map=${CMAKE_CURRENT_BINARY_DIR}/${TARGET_NAME}.map)
target_link_libraries(${TARGET_NAME} PRIVATE bsp_boot_ram bsp_board bsp_led
bsp_tick bsp_uart_host bsp_opto)
add_custom_command(
TARGET ${TARGET_NAME}
POST_BUILD
COMMAND ${CMAKE_SIZE} $<TARGET_FILE:${TARGET_NAME}>
COMMENT "Size: ${TARGET_NAME}")

View file

@ -0,0 +1,176 @@
/**
* @file tests/target/hil_opto/main.c
* @brief HIL target CLI для тестирования bsp_opto (оптоизолированные входы).
*
* Протокол: текстовые команды через LPUART1 (MCU-Link VCOM), \r\n-terminated.
*
* Команды:
* PING -> PONG
* OPTO_READ <1|2|3> -> ACTIVE / INACTIVE
* OPTO_EVENTS -> <число> (счётчик callback-событий)
* OPTO_LAST_EVENT -> <ch> <state> (последний канал + состояние)
* OPTO_RESET_EVENTS -> OK (сбросить счётчики)
*
* Стенд:
* M5StampPLC: RLY2->EXT_IN1(ch1), RLY3->EXT_IN2(ch2), RLY4->RS_RX(ch3)
*
* ВАЖНО: bsp_opto_process() вызывается в каждой итерации main loop.
* CLI_RX_TIMEOUT=10ms для быстрого цикла обработки debounce.
*/
#include "board.h"
#include "bsp/led.h"
#include "bsp/opto.h"
#include "bsp/tick.h"
#include "bsp/uart_host.h"
#include "fsl_gpio.h"
#include <stdio.h>
#include <string.h>
/* -------------------------------------------------------------------------- */
#define CLI_BAUD_RATE 115200U
#define CLI_LINE_MAX 128U
#define CLI_RX_TIMEOUT 10U /* мс — короткий, чтобы bsp_opto_process() крутился часто */
/* -------------------------------------------------------------------------- */
/* Event tracking (обновляется из callback, читается из CLI) */
/* -------------------------------------------------------------------------- */
static volatile uint32_t s_event_count;
static volatile bsp_opto_ch_t s_last_ch;
static volatile bsp_opto_state_t s_last_state;
static void opto_callback(bsp_opto_ch_t ch, bsp_opto_state_t state)
{
s_event_count++;
s_last_ch = ch;
s_last_state = state;
}
/* -------------------------------------------------------------------------- */
/* CLI */
/* -------------------------------------------------------------------------- */
static size_t cli_read_line(uint8_t *p_buf, size_t max_len)
{
size_t pos = 0U;
while (pos < (max_len - 1U))
{
int32_t byte = bsp_uart_host_read_byte(CLI_RX_TIMEOUT);
if (byte < 0)
{
break;
}
if ((char) byte == '\r')
{
continue;
}
if ((char) byte == '\n')
{
break;
}
p_buf[pos++] = (uint8_t) byte;
}
p_buf[pos] = '\0';
return pos;
}
static void cli_process_line(const char *p_line)
{
char resp[64];
/* PING */
if (strncmp(p_line, "PING", 4U) == 0)
{
bsp_uart_host_write_str("PONG\r\n");
}
/* OPTO_READ <1|2|3> */
else if (strncmp(p_line, "OPTO_READ ", 10U) == 0)
{
unsigned ch_num = 0U;
if (sscanf(p_line + 10U, "%u", &ch_num) != 1 || ch_num < 1U || ch_num > 3U)
{
bsp_uart_host_write_str("ERR_ARG\r\n");
return;
}
/* ch_num 1..3 -> enum 0..2 */
bsp_opto_state_t st = bsp_opto_read((bsp_opto_ch_t) (ch_num - 1U));
bsp_uart_host_write_str(st == BSP_OPTO_STATE_ACTIVE ? "ACTIVE\r\n" : "INACTIVE\r\n");
}
/* OPTO_EVENTS */
else if (strncmp(p_line, "OPTO_EVENTS", 11U) == 0)
{
snprintf(resp, sizeof(resp), "%lu\r\n", (unsigned long) s_event_count);
bsp_uart_host_write_str(resp);
}
/* OPTO_LAST_EVENT */
else if (strncmp(p_line, "OPTO_LAST_EVENT", 15U) == 0)
{
const char *state_str = (s_last_state == BSP_OPTO_STATE_ACTIVE) ? "ACTIVE" : "INACTIVE";
snprintf(resp, sizeof(resp), "%u %s\r\n", (unsigned) (s_last_ch + 1U), state_str);
bsp_uart_host_write_str(resp);
}
/* OPTO_RESET_EVENTS */
else if (strncmp(p_line, "OPTO_RESET_EVENTS", 17U) == 0)
{
s_event_count = 0U;
s_last_ch = BSP_OPTO_CH_IN1;
s_last_state = BSP_OPTO_STATE_INACTIVE;
bsp_uart_host_write_str("OK\r\n");
}
else if (p_line[0] != '\0')
{
bsp_uart_host_write_str("ERR_UNKNOWN\r\n");
}
}
/* -------------------------------------------------------------------------- */
/* main */
/* -------------------------------------------------------------------------- */
int main(void)
{
board_hw_init();
bsp_tick_init();
bsp_led_init();
bsp_uart_host_init(CLI_BAUD_RATE);
/* --- Opto init --- */
bsp_opto_config_t opto_cfg = {
.callbacks = { opto_callback, opto_callback, opto_callback },
.edges = { BSP_OPTO_EDGE_RISING, BSP_OPTO_EDGE_RISING, BSP_OPTO_EDGE_RISING },
.rs_as_gpio = true,
.debounce_ms = 10U, /* instant — для быстрого тестирования */
};
bsp_opto_init(&opto_cfg);
bsp_led_on(LED_HEARTBEAT);
/* Шлём READY пока хост не подключится */
while (bsp_uart_host_rx_available() == 0U)
{
bsp_uart_host_write_str("READY\r\n");
bsp_led_toggle(LED_APP);
bsp_delay(200U);
}
bsp_led_off(LED_APP);
static uint8_t s_line_buf[CLI_LINE_MAX];
for (;;)
{
bsp_opto_process(); /* <-- ОБЯЗАТЕЛЬНО перед CLI */
size_t len = cli_read_line(s_line_buf, sizeof(s_line_buf));
if (len > 0U)
{
cli_process_line((const char *) s_line_buf);
}
}
return 0;
}

View file

@ -6,6 +6,7 @@ conftest.py — pytest-фикстуры для HIL-тестов MIMXRT1052.
from __future__ import annotations
import json
import logging
import time
from pathlib import Path
@ -19,14 +20,18 @@ from pyocd_utils import flexram_init, load_elf, open_target, run_from_vectors
log = logging.getLogger(__name__)
# Задержка после включения питания таргета (мс стабилизации + POR)
_POWER_ON_SETTLE_S = 1.0
# ---------------------------------------------------------------------------
# CLI-опции pytest (перекрывают .env и os.environ)
# ---------------------------------------------------------------------------
def pytest_addoption(parser: pytest.Parser) -> None:
parser.addoption("--elf", default=None, help="Путь к .elf файлу")
parser.addoption("--vcom", default=cfg.VCOM_PORT, help="VCOM-порт MCU-Link")
parser.addoption("--no-load", action="store_true", help="ELF уже запущен")
parser.addoption("--elf", default=None, help="Путь к .elf файлу")
parser.addoption("--vcom", default=cfg.VCOM_PORT, help="VCOM-порт MCU-Link")
parser.addoption("--m5-port", default=None, help="M5StampPLC serial port")
parser.addoption("--no-load", action="store_true", help="ELF уже запущен")
# ---------------------------------------------------------------------------
@ -49,26 +54,10 @@ def _load_elf(request: pytest.FixtureRequest, default_elf: Path) -> None:
# ---------------------------------------------------------------------------
# Фикстуры загрузки (scope=module — один раз на файл с тестами)
# Общая логика UART: открыть порт, дождаться READY
# ---------------------------------------------------------------------------
@pytest.fixture(scope="module")
def loaded_host_uart(request: pytest.FixtureRequest) -> None:
_load_elf(
request,
Path(cfg.BUILD_DIR) / "tests/target/host_uart/test_host_uart.elf",
)
# ---------------------------------------------------------------------------
# Фикстура UART — открывает порт и ждёт "READY\r\n" от прошивки
# ---------------------------------------------------------------------------
@pytest.fixture(scope="module")
def uart(
request: pytest.FixtureRequest,
loaded_host_uart,
) -> Generator[serial.Serial, None, None]:
def _open_uart_and_wait_ready(request: pytest.FixtureRequest) -> serial.Serial:
"""Открыть VCOM, дождаться READY от прошивки."""
port = request.config.getoption("--vcom")
log.info("UART %s @ %d baud", port, cfg.VCOM_BAUD)
@ -97,10 +86,160 @@ def uart(
)
ser.reset_input_buffer()
return ser
# ---------------------------------------------------------------------------
# Фикстуры загрузки (scope=module — один раз на файл с тестами)
# ---------------------------------------------------------------------------
@pytest.fixture(scope="module")
def loaded_host_uart(request: pytest.FixtureRequest) -> None:
_load_elf(
request,
Path(cfg.BUILD_DIR) / "tests/target/host_uart/test_host_uart.elf",
)
@pytest.fixture(scope="module")
def loaded_hil_opto(request: pytest.FixtureRequest, m5: M5Agent) -> None:
"""Загрузить test_hil_opto.elf. Зависит от m5 — таргет должен быть запитан."""
_load_elf(
request,
Path(cfg.BUILD_DIR) / "tests/target/hil_opto/test_hil_opto.elf",
)
# ---------------------------------------------------------------------------
# Фикстура UART — открывает порт и ждёт "READY\r\n" от прошивки
# ---------------------------------------------------------------------------
@pytest.fixture(scope="module")
def uart(
request: pytest.FixtureRequest,
loaded_host_uart,
) -> Generator[serial.Serial, None, None]:
ser = _open_uart_and_wait_ready(request)
yield ser
ser.close()
@pytest.fixture(scope="module")
def uart_opto(
request: pytest.FixtureRequest,
loaded_hil_opto,
) -> Generator[serial.Serial, None, None]:
ser = _open_uart_and_wait_ready(request)
yield ser
ser.close()
# ---------------------------------------------------------------------------
# M5StampPLC — драйвер для pytest (JSON-lines протокол)
# ---------------------------------------------------------------------------
class M5Agent:
"""Драйвер M5StampPLC для pytest (JSON-lines протокол через USB CDC)."""
def __init__(self, ser: serial.Serial) -> None:
self._ser = ser
def cmd(self, command: str, **kwargs) -> dict:
"""Отправить JSON-команду, вернуть parsed-ответ. RuntimeError при ok=false."""
payload = {"cmd": command, **kwargs}
self._ser.reset_input_buffer()
self._ser.write((json.dumps(payload) + "\r\n").encode("ascii"))
deadline = time.monotonic() + cfg.M5_TIMEOUT
while time.monotonic() < deadline:
raw = self._ser.readline()
if not raw:
continue
text = raw.decode("ascii", errors="replace").strip()
if not text or text == "READY" or not text.startswith("{"):
continue
try:
resp = json.loads(text)
except json.JSONDecodeError:
continue
if not resp.get("ok"):
raise RuntimeError(
f"M5 error: {resp.get('err', '?')} (cmd={command})"
)
return resp
raise TimeoutError(f"M5: нет ответа на '{command}'")
def ping(self) -> None:
self.cmd("ping")
def power(self, state: bool) -> None:
"""Включить/выключить питание таргета (RLY1)."""
self.cmd("power", state=state)
def opto_set(self, ch: int, state: bool) -> None:
self.cmd("opto_set", ch=ch, state=state)
def opto_all_off(self) -> None:
self.cmd("opto_all_off")
# ---------------------------------------------------------------------------
# Фикстура M5StampPLC
# ---------------------------------------------------------------------------
@pytest.fixture(scope="module")
def m5(request: pytest.FixtureRequest) -> Generator[M5Agent, None, None]:
port = request.config.getoption("--m5-port") or cfg.M5_PORT
log.info("M5 %s @ %d baud", port, cfg.M5_BAUD)
ser = serial.Serial(
port=port,
baudrate=cfg.M5_BAUD,
timeout=cfg.M5_TIMEOUT,
write_timeout=1.0,
)
# Ждём READY; fallback на ping если агент уже работает
deadline = time.monotonic() + cfg.READY_TIMEOUT
ready = False
while time.monotonic() < deadline:
line = ser.readline().decode("ascii", errors="replace").strip()
if line == "READY":
ready = True
log.info("M5 READY получен")
break
agent = M5Agent(ser)
if not ready:
try:
agent.ping()
log.info("M5 уже работает (READY пропущен, ping OK)")
except Exception:
ser.close()
pytest.fail(
"M5 agent не отвечает — проверьте HIL_M5_PORT и "
"что m5/agent.py развёрнут (just host::m5-deploy)."
)
# Включаем питание таргета и ждём стабилизации
agent.power(True)
log.info("Питание таргета включено, ждём %.1f с", _POWER_ON_SETTLE_S)
time.sleep(_POWER_ON_SETTLE_S)
agent.opto_all_off() # безопасное начальное состояние
yield agent
# Teardown: выключить всё
try:
agent.opto_all_off()
except Exception:
pass
try:
agent.power(False)
log.info("Питание таргета выключено")
except Exception:
pass
ser.close()
# ---------------------------------------------------------------------------
# Утилита для тестов
# ---------------------------------------------------------------------------

View file

@ -24,3 +24,8 @@ VCOM_PORT: str = _get("HIL_VCOM_PORT", "/dev/ttyACM0")
VCOM_BAUD: int = int(_get("HIL_VCOM_BAUD", "115200"))
READY_TIMEOUT: float = float(_get("HIL_READY_TIMEOUT", "5.0"))
PYOCD_FREQUENCY: int = int(_get("HIL_PYOCD_FREQUENCY", "1000000"))
# M5Stack StamPLC (промежуточная платформа для HIL)
M5_PORT: str = _get("HIL_M5_PORT", "/dev/ttyACM1")
M5_BAUD: int = int(_get("HIL_M5_BAUD", "115200"))
M5_TIMEOUT: float = float(_get("HIL_M5_TIMEOUT", "3.0"))

View file

@ -1,5 +1,5 @@
"""
m5_agent.py MicroPython агент для M5Stack StamPLC.
m5/agent.py MicroPython агент для M5Stack StamPLC.
Размещение: /main.py на устройстве.
Деплой:

View file

@ -105,7 +105,7 @@ def _wait_ready(ser: serial.Serial) -> None:
print(_ok("OK"))
return
print(_err("TIMEOUT"))
print(_err(" Агент не ответил. Проверьте порт и что m5_agent.py загружен на m5stamPLC."))
print(_err(" Агент не ответил. Проверьте порт и что m5/agent.py загружен (just host::m5-deploy)."))
sys.exit(1)

View file

@ -0,0 +1,3 @@
# Загрузчики с microPython
> На HIL-стенд m5stamPLC необходимо предварительно загрузить интерпретор `microPython`.

54
tools/hil/m5/power.py Normal file
View file

@ -0,0 +1,54 @@
"""
m5/power.py управление питанием таргета через M5 RLY1.
Использование:
python m5/power.py --port /dev/ttyACM1 --state on
python m5/power.py --port /dev/ttyACM1 --state off
"""
from __future__ import annotations
import argparse
import json
import sys
import time
import serial
def main() -> None:
parser = argparse.ArgumentParser(description="M5 target power control")
parser.add_argument("--port", required=True, help="M5 serial port")
parser.add_argument("--state", required=True, choices=["on", "off"])
args = parser.parse_args()
state = args.state == "on"
cmd = json.dumps({"cmd": "power", "state": state})
ser = serial.Serial(args.port, 115200, timeout=3)
time.sleep(0.1)
ser.reset_input_buffer()
ser.write(cmd.encode("ascii") + b"\r\n")
deadline = time.monotonic() + 3
while time.monotonic() < deadline:
line = ser.readline().decode("ascii", errors="replace").strip()
if not line.startswith("{"):
continue
resp = json.loads(line)
if resp.get("ok"):
label = "включено" if state else "выключено"
print(f" Питание таргета {label}")
ser.close()
return
print(f" M5 error: {resp.get('err', '?')}")
ser.close()
sys.exit(1)
ser.close()
print(" Нет ответа от M5")
sys.exit(1)
if __name__ == "__main__":
main()

View file

@ -15,9 +15,10 @@ testpaths = ["."]
python_files = ["test_*.py"]
markers = [
"smoke: быстрые sanity-тесты",
"gpio: тесты GPIO",
"uart: тесты UART CLI",
"opto: тесты оптоизолированных входов",
"m5: тесты с участием M5StampPLC",
"slow: тесты с большими таймаутами",
]

113
tools/hil/test_opto.py Normal file
View file

@ -0,0 +1,113 @@
"""
test_opto.py HIL тест bsp_opto через M5StampPLC.
Стенд:
M5StampPLC реле оптопары таргета (PS2801-4, active-LOW):
RLY2 EXT_IN1 (ch1, BSP_OPTO_CH_IN1)
RLY3 EXT_IN2 (ch2, BSP_OPTO_CH_IN2)
RLY4 RS_RX (ch3, BSP_OPTO_CH_RS)
Схема фикстур:
conftest.loaded_hil_opto pyocd_utils: FLEXRAM + load ELF + run
conftest.uart_opto pyserial: открыть VCOM, ждать READY
conftest.m5 M5Agent: JSON-lines к M5StampPLC
тесты uart_cmd() + m5.opto_set() assert
Запуск:
just host::hil-opto
uv run pytest test_opto.py -v
uv run pytest test_opto.py -v --no-load --m5-port /dev/ttyACM1
"""
import time
import pytest
from conftest import uart_cmd
# Задержка после переключения реле перед чтением состояния (с).
# Включает: механическое переключение реле (~5ms), время отклика
# оптопары (~50µs), ISR + process() цикл, UART round-trip (~2ms).
RELAY_SETTLE_S = 0.05
# ── Связь ──────────────────────────────────────────────────────────────────
class TestOptoConnectivity:
"""Проверка каналов связи с таргетом и M5."""
@pytest.fixture(autouse=True)
def _setup(self, uart_opto, m5):
self.ser = uart_opto
self.m5 = m5
def test_target_ping(self):
"""PING → PONG: канал host↔target работает."""
assert uart_cmd(self.ser, "PING") == "PONG"
def test_m5_ping(self):
"""M5 agent отвечает на ping."""
self.m5.ping()
# ── Состояние по умолчанию ─────────────────────────────────────────────────
class TestOptoReadDefault:
"""Все каналы INACTIVE без внешнего воздействия."""
@pytest.fixture(autouse=True)
def _setup(self, uart_opto, m5):
self.ser = uart_opto
self.m5 = m5
self.m5.opto_all_off()
time.sleep(RELAY_SETTLE_S)
def test_ch1_default_inactive(self):
assert uart_cmd(self.ser, "OPTO_READ 1") == "INACTIVE"
def test_ch2_default_inactive(self):
assert uart_cmd(self.ser, "OPTO_READ 2") == "INACTIVE"
def test_ch3_default_inactive(self):
"""RS_RX (ch3) — сконфигурирован как GPIO, должен быть INACTIVE."""
assert uart_cmd(self.ser, "OPTO_READ 3") == "INACTIVE"
# ── Активация / деактивация каналов ────────────────────────────────────────
class TestOptoActivateDeactivate:
"""Активация/деактивация каждого канала по отдельности через M5."""
@pytest.fixture(autouse=True)
def _setup(self, uart_opto, m5):
self.ser = uart_opto
self.m5 = m5
self.m5.opto_all_off()
time.sleep(RELAY_SETTLE_S)
@pytest.mark.parametrize("ch", [1, 2, 3])
def test_activate_single_channel(self, ch):
"""M5 opto ON → таргет читает ACTIVE."""
self.m5.opto_set(ch, True)
time.sleep(RELAY_SETTLE_S)
assert uart_cmd(self.ser, f"OPTO_READ {ch}") == "ACTIVE"
@pytest.mark.parametrize("ch", [1, 2, 3])
def test_deactivate_single_channel(self, ch):
"""ON → OFF → таргет читает INACTIVE."""
self.m5.opto_set(ch, True)
time.sleep(RELAY_SETTLE_S)
self.m5.opto_set(ch, False)
time.sleep(RELAY_SETTLE_S)
assert uart_cmd(self.ser, f"OPTO_READ {ch}") == "INACTIVE"
@pytest.mark.parametrize("ch", [1, 2, 3])
def test_isolation(self, ch):
"""Активация одного канала не влияет на остальные."""
others = [c for c in [1, 2, 3] if c != ch]
self.m5.opto_set(ch, True)
time.sleep(RELAY_SETTLE_S)
for other in others:
assert uart_cmd(self.ser, f"OPTO_READ {other}") == "INACTIVE", \
f"ch{other} должен быть INACTIVE когда активен только ch{ch}"

View file

@ -7,8 +7,8 @@ test_uart.py — HIL тест bsp_uart_host.
тесты uart_cmd() assert
Запуск:
just host::hil-uart
uv run pytest test_uart.py -v
uv run pytest test_uart.py -v -m smoke
uv run pytest test_uart.py -v --no-load # ELF уже запущен
"""
@ -25,12 +25,10 @@ class TestUartBasic:
self.ser = uart
# ------------------------------------------------------------------
@pytest.mark.smoke
def test_ping(self):
"""PING → PONG: канал работает в обе стороны."""
assert uart_cmd(self.ser, "PING") == "PONG"
@pytest.mark.smoke
def test_ping_repeated(self):
"""Десять PING подряд — нет зависаний, нет потерь."""
for i in range(10):