diff --git a/.gitignore b/.gitignore index 2679cdf..9f04e9f 100644 --- a/.gitignore +++ b/.gitignore @@ -62,6 +62,7 @@ Testing/ # Файлы настройки среды разработки .project compile_commands.json +.pytest_cache/ .cache/ .DS_Store .venv/ diff --git a/docs/MT48LCxxM4,8,16A2.pdf b/docs/board/MT48LCxxM4,8,16A2.pdf similarity index 100% rename from docs/MT48LCxxM4,8,16A2.pdf rename to docs/board/MT48LCxxM4,8,16A2.pdf diff --git a/docs/schematic.pdf b/docs/board/schematic.pdf similarity index 100% rename from docs/schematic.pdf rename to docs/board/schematic.pdf diff --git a/docs/m5stamPLC/K141_sch_StamPLC_V10_CPU.pdf b/docs/m5stamPLC/K141_sch_StamPLC_V10_CPU.pdf new file mode 100644 index 0000000..3621e44 Binary files /dev/null and b/docs/m5stamPLC/K141_sch_StamPLC_V10_CPU.pdf differ diff --git a/docs/m5stamPLC/K141_sch_StamPLC_V10_IO.pdf b/docs/m5stamPLC/K141_sch_StamPLC_V10_IO.pdf new file mode 100644 index 0000000..4810f5e Binary files /dev/null and b/docs/m5stamPLC/K141_sch_StamPLC_V10_IO.pdf differ diff --git a/docs/manufacturing_user's_guide.pdf b/docs/mimxrt1052/manufacturing_user's_guide.pdf similarity index 100% rename from docs/manufacturing_user's_guide.pdf rename to docs/mimxrt1052/manufacturing_user's_guide.pdf diff --git a/docs/mcu_rm.pdf b/docs/mimxrt1052/mcu_rm.pdf similarity index 100% rename from docs/mcu_rm.pdf rename to docs/mimxrt1052/mcu_rm.pdf diff --git a/just/host.just b/just/host.just index 1803fa8..91881e1 100644 --- a/just/host.just +++ b/just/host.just @@ -195,6 +195,7 @@ bootstrap: 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})" fi @@ -205,6 +206,22 @@ bootstrap: echo " Next: open project in VSCode -> 'Reopen in Container'" echo "" +[doc('Добавить пользователя в группу dialout (для /dev/ttyACM* — M5Stack, Linux)')] +[group('setup')] +setup-m5-udev: + #!/usr/bin/env bash + set -euo pipefail + case "$(uname -s)" in + Linux*) ;; + *) echo " ℹ️ dialout group — Linux only, skipping."; exit 0 ;; + esac + if groups "$USER" | grep -q dialout; then + echo " ✅ User '${USER}' already in dialout" + else + sudo usermod -a -G dialout "$USER" + echo " ✅ Added '${USER}' to dialout. Re-login required." + fi + # ============================================================================= # ГРУППА: flash — USB Serial Downloader (SDP через ROM-загрузчик) # Требует: плата в режиме Serial Downloader (BOOT_MODE = 01) @@ -360,21 +377,92 @@ debug-list-targets: uv run --directory {{ _hil_dir }} pyocd list --targets | grep -i mimx # ============================================================================= -# ГРУППА: pipeline +# ГРУППА: m5 — работа с M5Stack StamPLC # ============================================================================= -[doc('Входной контроль: прошить firmware_test Release + запустить HIL-тесты')] -[group('pipeline')] -incoming: flash-test-release +M5_PORT := env('HIL_M5_PORT', '/dev/ttyACM1') +_m5_agent := justfile_directory() / 'tools/hil/m5_agent.py' + +[doc('Показать подключённые M5Stack устройства (USB CDC)')] +[group('m5')] +m5-scan: #!/usr/bin/env bash set -euo pipefail - echo " ▶ Running HIL tests..." - just host::hil-run -[doc('Производственная прошивка: bootloader + tft_app Release')] -[group('pipeline')] -production: flash-production - @echo " ✅ Production firmware flashed" + echo "" + echo " Сканируем устройства через mpremote..." + echo "" + + devices=$(uv run --directory "{{ HIL_DIR }}" mpremote devs 2>&1) + + if [[ -z "$devices" ]]; then + echo " ⚠️ Устройства не найдены" + echo " Проверьте USB-кабель и права доступа к порту" + echo " Linux: sudo usermod -a -G dialout \$USER && newgrp dialout" + echo " macOS: никаких дополнительных прав не требуется" + exit 1 + fi + + echo "$devices" + echo "" + + # Подсветить известный VID:PID для Espressif (M5StamPLC JTAG/serial) + if echo "$devices" | grep -qi "303a:1001"; then + echo " ✅ M5StamPLC (Espressif USB JTAG/serial) обнаружен" + elif echo "$devices" | grep -qi "303a:"; then + echo " ✅ Espressif устройство обнаружено" + else + echo " ⚠️ M5StamPLC не найден среди устройств" + fi + echo "" + +[doc('Открыть MicroPython REPL на M5Stack (Ctrl+X для выхода)')] +[group('m5')] +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: + uv run --directory "{{ HIL_DIR }}" mpremote connect "{{ M5_PORT }}" \ + exec 'import sys; print("MicroPython:", sys.version); import machine; print("Freq:", machine.freq(), "Hz")' + +[doc('Развернуть агент + запустить CLI одной командой (быстрый старт)')] +[group('m5')] +m5-start: + #!/usr/bin/env bash + set -euo pipefail + + just host::m5-deploy + uv run --directory "{{ HIL_DIR }}" mpremote connect "{{ M5_PORT }}" reset + sleep 2 + just host::m5-cli + # ============================================================================= # ГРУППА: util @@ -408,3 +496,20 @@ uart-monitor: uv run --directory {{ HIL_DIR }} python3 -m serial.tools.miniterm \ {{ env('HIL_VCOM_PORT') }} {{ env('HIL_VCOM_BAUD', '115200') }} \ --raw + +# ============================================================================= +# ГРУППА: pipeline +# ============================================================================= + +[doc('Входной контроль: прошить firmware_test Release + запустить HIL-тесты')] +[group('pipeline')] +incoming: flash-test-release + #!/usr/bin/env bash + set -euo pipefail + echo " ▶ Running HIL tests..." + just host::hil-run + +[doc('Производственная прошивка: bootloader + tft_app Release')] +[group('pipeline')] +production: flash-production + @echo " ✅ Production firmware flashed" \ No newline at end of file diff --git a/tools/hil/README.md b/tools/hil/README.md index faa0a48..f455d8c 100644 --- a/tools/hil/README.md +++ b/tools/hil/README.md @@ -26,6 +26,11 @@ tools/hil/ ## Концепция HIL-тестов +```bash +espflash write-bin 0 ESP32_GENERIC_S3-20251209-v1.27.0.bin # microPython + +``` + Каждый HIL-тест — это пара: ``` diff --git a/tools/hil/m5_agent.py b/tools/hil/m5_agent.py new file mode 100644 index 0000000..43ad85e --- /dev/null +++ b/tools/hil/m5_agent.py @@ -0,0 +1,435 @@ +""" +m5_agent.py — MicroPython агент для M5Stack StamPLC. +Размещение: /main.py на устройстве. + +Деплой: + just host::m5-deploy + +Ручное тестирование: + just host::m5-cli + +───────────────────────────────────────────────────── +Аппаратура (из схемы StamPLC v1.0): + + I2C экспандер AW9523B addr=0x59 SDA=G13 SCL=G15 + P0_0..P0_3 → RLY_DRV1..4 (выходы: реле через ULN2003A) + P1_0..P1_7 → SYS_IN1..8 (входы: оптопары EL3H4, active-low) + + RST AW9523B → G3_PHY_RST (GPIO 3) + + CAN трансивер SIT1044 TX=G42 RX=G43 (TWAI ESP32-S3) + + Раскладка реле (конфигурация стенда): + RLY1 → питание таргета (VIN) + RLY2 → EXT_IN1 таргета (BSP_OPTO_CH_IN1) + RLY3 → EXT_IN2 таргета (BSP_OPTO_CH_IN2) + RLY4 → RS_RX таргета (BSP_OPTO_CH_RS) + +───────────────────────────────────────────────────── +Протокол: JSON-lines через USB CDC (115200, нет flow control). + + Хост → агент: {"cmd": "<имя>", ...параметры...}\r\n + Агент → хост: {"ok": true, ...данные...}\r\n + {"ok": false, "err": "<текст>"}\r\n + +Команды: + ping + info + aw_debug читать сырые регистры AW9523 + relay_set ch:int(1-4) state:bool + relay_get ch:int(1-4) + relay_all_off + power state:bool алиас relay_set ch=1 + opto_set ch:int(1-3) state:bool + opto_all_off + input_read ch:int(1-8) читает SYS_IN на самом стенде + can_send id:int list[int] ext:bool=false + can_recv timeout_ms:int=500 +""" + +import json +import sys +import time + +from machine import I2C, Pin + +# --------------------------------------------------------------------------- +# Конфигурация +# --------------------------------------------------------------------------- + +_CFG = { + "i2c_sda": 13, + "i2c_scl": 15, + "i2c_freq": 400_000, + "aw_addr": 0x59, + "can_tx": 42, + "can_rx": 43, + "can_baud": 500_000, +} + +_RST_PIN = 3 # G3_PHY_RST → нога сброса AW9523B + +_OPTO_TO_RELAY = {1: 2, 2: 3, 3: 4} # оптоканал таргета → реле стенда +_IN_PIN_LIST = [4, 5, 6, 7, 12, 13, 14, 15] # нумерация пинов AW9523 +# --------------------------------------------------------------------------- +# AW9523B — регистры +# --------------------------------------------------------------------------- + +_R_IN0 = 0x00 # читать входы P0 +_R_IN1 = 0x01 # читать входы P1 +_R_OUT0 = 0x02 # писать выходы P0 +_R_OUT1 = 0x03 # писать выходы P1 +_R_DIR0 = 0x04 # направление P0: 0=выход, 1=вход +_R_DIR1 = 0x05 # направление P1: 0=выход, 1=вход +_R_GCFG = 0x11 # global config: бит4=P0 режим (0=push-pull) +_R_MODE0 = 0x12 # режим P0: бит=1 → GPIO, бит=0 → LED +_R_MODE1 = 0x13 # режим P1: бит=1 → GPIO, бит=0 → LED + +# --------------------------------------------------------------------------- +# AW9523B — класс +# --------------------------------------------------------------------------- + +class AW9523: + """ + Драйвер AW9523B для StamPLC. + + Распределение портов (по схеме): + P0[0..3] → RLY_DRV1..4 (выходы, управляем реле) + P1[0..7] → SYS_IN1..8 (входы, active-low оптопары) + """ + + def __init__(self, i2c: I2C, addr: int = 0x59) -> None: + self._i2c = i2c + self._addr = addr + self._out0 = 0x00 + + # 1. Проверяем chip ID (должен быть 0x23) + chip_id = self._r(0x10) + if chip_id != 0x23: + raise Exception("AW9523 not found, chip_id=0x%02x" % chip_id) + + # 2. Все пины — входы по умолчанию + self._w(_R_DIR0, 0xFF) + self._w(_R_DIR1, 0xFF) + + # 3. GPIO-режим (не LED) для всех пинов + self._w(_R_MODE0, 0xFF) + self._w(_R_MODE1, 0xFF) + + # 4. PUSH-PULL для Port 0: бит 4 в GCFG = 1 + # openDrainPort0(false) в C++ → bitOn(GCR, 0x10) + self._w(_R_GCFG, 0x10) + + # 5. Отключить прерывания + self._w(0x06, 0xFF) + self._w(0x07, 0xFF) + + # 6. P0_0..P0_3 — выходы (бит=0), P0_4..P0_7 — входы (бит=1) + self._w(_R_DIR0, 0xF0) + # P1 весь — входы + self._w(_R_DIR1, 0xFF) + + # 7. Все выходы выключены + self._out0 = 0x00 + self._w(_R_OUT0, self._out0) + + # ── низкоуровневое ──────────────────────────────────────────────────── + + def _w(self, reg: int, val: int) -> None: + self._i2c.writeto_mem(self._addr, reg, bytes([val])) + + def _r(self, reg: int) -> int: + return self._i2c.readfrom_mem(self._addr, reg, 1)[0] + + # ── реле (P0) ───────────────────────────────────────────────────────── + + def relay_set(self, ch: int, state: bool) -> None: + """ch: 1-4 → P0_0..P0_3.""" + mask = 1 << (ch - 1) + self._out0 = (self._out0 | mask) if state else (self._out0 & ~mask) + self._w(_R_OUT0, self._out0) + + def relay_get(self, ch: int) -> bool: + mask = 1 << (ch - 1) + return bool(self._out0 & mask) + + def relay_all_off(self) -> None: + self._out0 = 0x00 + self._w(_R_OUT0, 0x00) + + # ── входы стенда (P1) ───────────────────────────────────────────────── + + def input_read(self, ch: int) -> bool: + """ + ch: 1-8. + Нумерация пинов AW9523: 0-7=Port0, 8-15=Port1. + SYS_IN1..4 на Port0 биты 4-7, SYS_IN5..8 на Port1 биты 4-7. + Полярность: active HIGH (без инверсии) — как в C++ digitalRead. + """ + pin = _IN_PIN_LIST[ch - 1] + if pin < 8: + val = self._r(_R_IN0) + else: + val = self._r(_R_IN1) + pin -= 8 + return bool(val & (1 << pin)) + + # ── диагностика ─────────────────────────────────────────────────────── + + def debug_regs(self) -> dict: + return { + "in0": self._r(_R_IN0), # P0 как вход (для отладки) + "in1": self._r(_R_IN1), # P1 — наши реальные входы + "out0": self._out0, # теневой регистр реле + "dir0": self._r(_R_DIR0), + "dir1": self._r(_R_DIR1), + } + + +# --------------------------------------------------------------------------- +# CAN (TWAI ESP32-S3) (для поддержки CAN, используем кастомный загрузчик) +# см. https://github.com/straga/micropython-esp32-twai +# --------------------------------------------------------------------------- + +def _make_can(): + try: + import CAN + can = CAN( + 0, + extframe=False, + tx=42, # STAMPLC_PIN_CAN_TX + rx=43, # STAMPLC_PIN_CAN_RX + mode=CAN.NORMAL, + bitrate=_CFG["can_baud"], + auto_restart=False, + ) + return can, True + except Exception as e: + print("[CAN] ошибка инициализации:", e) + return None, False + + + +# --------------------------------------------------------------------------- +# Глобальное состояние +# --------------------------------------------------------------------------- + +_i2c = None +_aw = None +_can = None +_can_ok = False + + +# --------------------------------------------------------------------------- +# Диспетчер команд +# --------------------------------------------------------------------------- + +def _dispatch(cmd: dict) -> dict: + c = cmd.get("cmd", "") + + if c == "ping": + return {"ok": True, "pong": True} + + if c == "info": + return { + "ok": True, + "fw": "agent/1.1", + "python": sys.version[:40], + "can_ok": _can_ok, + "aw_ok": _aw is not None, + "relays": [_aw.relay_get(i) for i in range(1, 5)] if _aw else [], + } + + if c == "relay_set": + if not _aw: + return {"ok": False, "err": "AW not available"} + ch = int(cmd["ch"]) + state = bool(cmd["state"]) + if not (1 <= ch <= 4): + return {"ok": False, "err": "ch out of range 1..4"} + _aw.relay_set(ch, state) + return {"ok": True, "ch": ch, "state": state} + + if c == "relay_get": + if not _aw: + return {"ok": False, "err": "AW not available"} + ch = int(cmd["ch"]) + if not (1 <= ch <= 4): + return {"ok": False, "err": "ch out of range 1..4"} + return {"ok": True, "ch": ch, "state": _aw.relay_get(ch)} + + if c == "relay_all_off": + if not _aw: + return {"ok": False, "err": "AW not available"} + _aw.relay_all_off() + return {"ok": True} + + if c == "power": + if not _aw: + return {"ok": False, "err": "AW not available"} + state = bool(cmd["state"]) + _aw.relay_set(1, state) + return {"ok": True, "state": state} + + if c == "opto_set": + if not _aw: + return {"ok": False, "err": "AW not available"} + ch = int(cmd["ch"]) + state = bool(cmd["state"]) + relay = _OPTO_TO_RELAY.get(ch) + if relay is None: + return { + "ok": False, + "err": "opto ch %d unknown, valid: 1,2,3" % ch, + } + _aw.relay_set(relay, state) + return {"ok": True, "opto_ch": ch, "relay": relay, "state": state} + + if c == "opto_all_off": + if not _aw: + return {"ok": False, "err": "AW not available"} + for relay in _OPTO_TO_RELAY.values(): + _aw.relay_set(relay, False) + return {"ok": True} + + if c == "input_read": + if not _aw: + return {"ok": False, "err": "AW not available"} + ch = int(cmd["ch"]) + if not (1 <= ch <= 8): + return {"ok": False, "err": "ch out of range 1..8"} + return {"ok": True, "ch": ch, "state": _aw.input_read(ch)} + + if c == "can_send": + if not _can_ok: + return {"ok": False, "err": "CAN not available"} + ext = bool(cmd.get("ext", False)) + data = bytes(cmd.get("data", [])) + _can.send(list(cmd.get("data", [])), int(cmd["id"]), extframe=ext) + return {"ok": True} + + if c == "can_recv": + if not _can_ok: + return {"ok": False, "err": "CAN not available"} + timeout_ms = int(cmd.get("timeout_ms", 500)) + msg = _can.recv(timeout=timeout_ms) + if msg is None: + return {"ok": False, "err": "timeout"} + frame_id, _, ext, data = msg + return {"ok": True, "id": frame_id, "ext": ext, "data": list(data)} + + if c == "can_loopback": + if not _can_ok: + return {"ok": False, "err": "CAN not available"} + try: + import CAN as _CAN_MOD + + # Временно переинициализируем в LOOPBACK режиме + _can.deinit() + lb = _CAN_MOD( + 0, + extframe=False, + tx=_CFG["can_tx"], + rx=_CFG["can_rx"], + mode=_CAN_MOD.LOOPBACK, + bitrate=_CFG["can_baud"], + auto_restart=False, + ) + + test_id = int(cmd.get("id", 0x123)) + test_data = list(cmd.get("data", [0x01, 0x02, 0x03, 0x04])) + timeout = int(cmd.get("timeout_ms", 500)) + + lb.send(test_data, test_id) + + deadline = time.ticks_ms() + timeout + msg = None + while time.ticks_diff(deadline, time.ticks_ms()) > 0: + if lb.any(): + msg = lb.recv() + break + time.sleep_ms(5) + + lb.deinit() + + if msg is None: + return {"ok": False, "err": "loopback timeout — фрейм не вернулся"} + + recv_id, _, ext, recv_data = msg + matched = (recv_id == test_id) and (list(recv_data[:len(test_data)]) == test_data) + return { + "ok": True, + "matched": matched, + "sent_id": test_id, + "recv_id": recv_id, + "sent_data": test_data, + "recv_data": list(recv_data), + } + + except Exception as e: + return {"ok": False, "err": "loopback error: %s" % e} + + return {"ok": False, "err": "unknown cmd: %r" % c} + + +# --------------------------------------------------------------------------- +# main +# --------------------------------------------------------------------------- + +def main() -> None: + global _i2c, _aw, _can, _can_ok + + # Снять AW9523B из аппаратного сброса (G3_PHY_RST = 1) + try: + Pin(_RST_PIN, Pin.OUT, value=1) + time.sleep_ms(10) + except Exception: + pass + + # Инициализация AW9523B + try: + _i2c = I2C( + 0, + sda=Pin(_CFG["i2c_sda"]), + scl=Pin(_CFG["i2c_scl"]), + freq=_CFG["i2c_freq"], + ) + _aw = AW9523(_i2c, _CFG["aw_addr"]) + except Exception as exc: + sys.stdout.write( + json.dumps({"ok": False, "err": "AW init: %s" % exc}) + "\r\n" + ) + + # Инициализация CAN + _can, _can_ok = _make_can() + + # Безопасный старт — все реле выключены + if _aw: + _aw.relay_all_off() + + sys.stdout.write("READY\r\n") + + # Основной цикл + while True: + try: + line = sys.stdin.readline() + except Exception: + continue + + if not line: + continue + line = line.strip() + if not line: + continue + + try: + resp = _dispatch(json.loads(line)) + except ValueError as exc: + resp = {"ok": False, "err": "json: %s" % exc} + except Exception as exc: + resp = {"ok": False, "err": str(exc)} + + sys.stdout.write(json.dumps(resp) + "\r\n") + + +main() diff --git a/tools/hil/m5_host_cli.py b/tools/hil/m5_host_cli.py new file mode 100644 index 0000000..71ecefb --- /dev/null +++ b/tools/hil/m5_host_cli.py @@ -0,0 +1,372 @@ +#!/usr/bin/env python3 +""" +m5_cli.py — интерактивный CLI для ручного тестирования агента M5Stack. + +Запуск: + just host::m5-cli + # или напрямую: + uv run --directory tools/hil python m5_cli.py --port /dev/ttyACM1 + +Использование: + > ping + > info + > power on + > power off + > relay 2 on + > relay 2 off + > relay_all_off + > opto 1 on # активировать EXT_IN1 таргета + > opto 2 off + > opto_all_off + > input 3 # прочитать SYS_IN3 стенда + > can_send 0x123 01 02 03 04 + > can_recv 1000 # ждать фрейм 1000мс + > help + > quit +""" + +from __future__ import annotations + +import argparse +import json +import sys +import time + +import serial + +# --------------------------------------------------------------------------- +# Конфигурация +# --------------------------------------------------------------------------- + +_DEFAULT_PORT = "/dev/ttyACM1" +_BAUD = 115_200 +_TIMEOUT = 3.0 +_READY_WAIT = 5.0 + +# --------------------------------------------------------------------------- +# Цвета +# --------------------------------------------------------------------------- + +_GREEN = "\033[92m" +_RED = "\033[91m" +_YELLOW = "\033[93m" +_CYAN = "\033[96m" +_GRAY = "\033[90m" +_BOLD = "\033[1m" +_RESET = "\033[0m" + +def _ok(s: str) -> str: return f"{_GREEN}{s}{_RESET}" +def _err(s: str) -> str: return f"{_RED}{s}{_RESET}" +def _info(s: str) -> str: return f"{_CYAN}{s}{_RESET}" +def _hint(s: str) -> str: return f"{_GRAY}{s}{_RESET}" + +# --------------------------------------------------------------------------- +# Транспорт +# --------------------------------------------------------------------------- + +def _send(ser: serial.Serial, cmd: dict) -> dict: + """Отправить команду и дождаться первого валидного JSON-ответа.""" + line = json.dumps(cmd) + "\r\n" + ser.reset_input_buffer() + ser.write(line.encode("ascii")) + + deadline = time.monotonic() + _TIMEOUT + while time.monotonic() < deadline: + raw = ser.readline() + if not raw: + continue + + text = raw.decode("ascii", errors="replace").strip() + if not text: + continue + + # Игнорируем служебный стартовый маркер + if text == "READY": + continue + + # Игнорируем всё, что не выглядит как JSON-объект + if not text.startswith("{"): + continue + + try: + return json.loads(text) + except json.JSONDecodeError: + continue + + raise TimeoutError("Нет корректного JSON-ответа от агента") + + +def _wait_ready(ser: serial.Serial) -> None: + print(_info(" Ожидаем READY от агента..."), end=" ", flush=True) + deadline = time.monotonic() + _READY_WAIT + while time.monotonic() < deadline: + line = ser.readline().decode("ascii", errors="replace").strip() + if line == "READY": + print(_ok("OK")) + return + print(_err("TIMEOUT")) + print(_err(" Агент не ответил. Проверьте порт и что m5_agent.py загружен на m5stamPLC.")) + sys.exit(1) + + +# --------------------------------------------------------------------------- +# Парсер команд +# --------------------------------------------------------------------------- + +def _parse_and_run(ser: serial.Serial, line: str) -> bool: + """ + Разобрать строку, выполнить команду. + Вернуть False если нужно выйти. + """ + parts = line.strip().split() + if not parts: + return True + cmd = parts[0].lower() + + # ── quit / exit ────────────────────────────────────────────────────────── + if cmd in ("quit", "exit", "q"): + return False + + # ── help ───────────────────────────────────────────────────────────────── + if cmd == "help": + print(_hint(""" + Команды: + ping проверить связь с агентом + info версия, статус CAN, состояние реле + power on|off питание таргета (RLY1) + relay <1-4> on|off прямое управление реле + relay_all_off выключить все реле + opto <1-3> on|off активировать оптовход таргета + opto_all_off деактивировать все оптовходы + input <1-8> прочитать входной канал стенда + can_send [байты...] отправить CAN фрейм (id в hex или dec) + can_recv [timeout_ms] принять CAN фрейм + raw отправить произвольный JSON напрямую + help эта справка + quit / exit / q выход + """)) + return True + + # ── ping ───────────────────────────────────────────────────────────────── + if cmd == "ping": + _run(ser, {"cmd": "ping"}) + return True + + # ── info ───────────────────────────────────────────────────────────────── + if cmd == "info": + resp = _run(ser, {"cmd": "info"}) + if resp and resp.get("ok"): + print(f" fw: {resp.get('fw')}") + print(f" python: {resp.get('python', '?')[:40]}") + print(f" can: {'✅ доступен' if resp.get('can_ok') else '❌ недоступен'}") + relays = resp.get("relays", []) + for i, state in enumerate(relays, 1): + mark = "●" if state else "○" + print(f" RLY{i}: {mark} {'ON ' if state else 'off'}") + return True + + # ── power ───────────────────────────────────────────────────────────────── + if cmd == "power": + if len(parts) < 2 or parts[1].lower() not in ("on", "off"): + print(_err(" Использование: power on|off")) + return True + state = parts[1].lower() == "on" + _run(ser, {"cmd": "power", "state": state}) + return True + + # ── relay ───────────────────────────────────────────────────────────────── + if cmd == "relay": + if len(parts) < 3: + print(_err(" Использование: relay <1-4> on|off")) + return True + try: + ch = int(parts[1]) + except ValueError: + print(_err(" ch должен быть числом 1-4")) + return True + if parts[2].lower() not in ("on", "off"): + print(_err(" Состояние: on или off")) + return True + state = parts[2].lower() == "on" + _run(ser, {"cmd": "relay_set", "ch": ch, "state": state}) + return True + + # ── relay_all_off ───────────────────────────────────────────────────────── + if cmd == "relay_all_off": + _run(ser, {"cmd": "relay_all_off"}) + return True + + # ── opto ────────────────────────────────────────────────────────────────── + if cmd == "opto": + if len(parts) < 3: + print(_err(" Использование: opto <1-3> on|off")) + return True + try: + ch = int(parts[1]) + except ValueError: + print(_err(" ch должен быть числом 1-3")) + return True + if parts[2].lower() not in ("on", "off"): + print(_err(" Состояние: on или off")) + return True + state = parts[2].lower() == "on" + _run(ser, {"cmd": "opto_set", "ch": ch, "state": state}) + return True + + # ── opto_all_off ────────────────────────────────────────────────────────── + if cmd == "opto_all_off": + _run(ser, {"cmd": "opto_all_off"}) + return True + + # ── input ───────────────────────────────────────────────────────────────── + if cmd == "input": + if len(parts) < 2: + print(_err(" Использование: input <1-8>")) + return True + try: + ch = int(parts[1]) + except ValueError: + print(_err(" ch должен быть числом 1-8")) + return True + resp = _run(ser, {"cmd": "input_read", "ch": ch}) + if resp and resp.get("ok"): + state = resp.get("state", False) + mark = _ok("ACTIVE") if state else _hint("inactive") + print(f" SYS_IN{ch}: {mark}") + return True + + # ── can_send ────────────────────────────────────────────────────────────── + if cmd == "can_send": + if len(parts) < 2: + print(_err(" Использование: can_send [байт байт ...]")) + return True + try: + frame_id = int(parts[1], 0) # 0x123 или 291 + data = [int(b, 0) for b in parts[2:]] + except ValueError as exc: + print(_err(f" Ошибка парсинга: {exc}")) + return True + _run(ser, {"cmd": "can_send", "id": frame_id, "data": data}) + return True + + # ── can_recv ────────────────────────────────────────────────────────────── + if cmd == "can_recv": + timeout_ms = int(parts[1]) if len(parts) > 1 else 500 + resp = _run(ser, {"cmd": "can_recv", "timeout_ms": timeout_ms}) + if resp and resp.get("ok"): + frame_id = resp.get("id", 0) + data = resp.get("data", []) + ext = resp.get("ext", False) + data_hex = " ".join(f"{b:02X}" for b in data) + print(f" id=0x{frame_id:X} ext={ext} data=[{data_hex}]") + return True + + # ── raw ─────────────────────────────────────────────────────────────────── + if cmd == "raw": + raw_json = line.strip()[4:].strip() + if not raw_json: + print(_err(" Использование: raw {\"cmd\": \"...\"}")) + return True + try: + payload = json.loads(raw_json) + except json.JSONDecodeError as exc: + print(_err(f" JSON ошибка: {exc}")) + return True + _run(ser, payload) + return True + + print(_err(f" Неизвестная команда: {cmd!r} (введите help)")) + return True + + +def _run(ser: serial.Serial, cmd: dict) -> dict | None: + """Выполнить команду и красиво распечатать результат.""" + print(_hint(f" → {json.dumps(cmd)}")) + try: + resp = _send(ser, cmd) + except TimeoutError as exc: + print(_err(f" ✗ {exc}")) + return None + except Exception as exc: # noqa: BLE001 + print(_err(f" ✗ {exc}")) + return None + + if resp.get("ok"): + # Убираем ok=true из вывода — это шум + display = {k: v for k, v in resp.items() if k != "ok"} + if display: + print(_ok(f" ✓ {json.dumps(display, ensure_ascii=False)}")) + else: + print(_ok(" ✓ ok")) + else: + print(_err(f" ✗ err: {resp.get('err', '?')}")) + + return resp + + +# --------------------------------------------------------------------------- +# main +# --------------------------------------------------------------------------- + +def main() -> None: + parser = argparse.ArgumentParser(description="M5Stack StamPLC CLI") + parser.add_argument("--port", default=_DEFAULT_PORT, help="COM-порт") + parser.add_argument("--baud", type=int, default=_BAUD) + args = parser.parse_args() + + print() + print(_bold(" M5Stack StamPLC CLI")) + print(_hint(f" Порт: {args.port} @ {args.baud} baud")) + print(_hint(" Введите 'help' для списка команд, 'quit' для выхода")) + print() + + try: + ser = serial.Serial( + port=args.port, + baudrate=args.baud, + timeout=_TIMEOUT, + write_timeout=1.0, + ) + except serial.SerialException as exc: + print(_err(f" Не удалось открыть порт {args.port}: {exc}")) + print(_hint(" Проверьте: ls /dev/tty* или just host::m5-scan")) + sys.exit(1) + + _wait_ready(ser) + print() + + # Автоматически запрашиваем info при старте + _parse_and_run(ser, "info") + print() + + try: + while True: + try: + line = input(_bold(" m5-hil-agent> ")) + except EOFError: + break + except KeyboardInterrupt: + print() + break + if not _parse_and_run(ser, line): + break + finally: + # При выходе — безопасно выключить все реле + print() + print(_hint(" Выключаем все реле...")) + try: + _send(ser, {"cmd": "relay_all_off"}) + print(_ok(" ✓ Все реле выключены")) + except Exception: + pass + ser.close() + print(_hint(" Порт закрыт. До свидания!")) + print() + + +def _bold(s: str) -> str: + return f"{_BOLD}{s}{_RESET}" + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/tools/hil/microPython/esp32s3_bl-v1.25.0_twai.bin b/tools/hil/microPython/esp32s3_bl-v1.25.0_twai.bin new file mode 100644 index 0000000..2c748db Binary files /dev/null and b/tools/hil/microPython/esp32s3_bl-v1.25.0_twai.bin differ diff --git a/tools/hil/microPython/esp32s3_bl-v1.27.0.bin b/tools/hil/microPython/esp32s3_bl-v1.27.0.bin new file mode 100644 index 0000000..0124e92 Binary files /dev/null and b/tools/hil/microPython/esp32s3_bl-v1.27.0.bin differ diff --git a/tools/hil/pyproject.toml b/tools/hil/pyproject.toml index 7c78ed4..2b72e48 100644 --- a/tools/hil/pyproject.toml +++ b/tools/hil/pyproject.toml @@ -7,6 +7,7 @@ requires-python = ">=3.10" dependencies = [ "pyocd>=0.43.1", "pyserial>=3.5", + "mpremote>=1.23", "pytest>=9.0.2", ] [tool.pytest.ini_options] diff --git a/tools/hil/uv.lock b/tools/hil/uv.lock index 69fcf2c..2974906 100644 --- a/tools/hil/uv.lock +++ b/tools/hil/uv.lock @@ -190,6 +190,7 @@ name = "hil" version = "0.1.0" source = { virtual = "." } dependencies = [ + { name = "mpremote" }, { name = "pyocd" }, { name = "pyserial" }, { name = "pytest" }, @@ -197,6 +198,7 @@ dependencies = [ [package.metadata] requires-dist = [ + { name = "mpremote", specifier = ">=1.23" }, { name = "pyocd", specifier = ">=0.43.1" }, { name = "pyserial", specifier = ">=3.5" }, { name = "pytest", specifier = ">=9.0.2" }, @@ -318,6 +320,19 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/15/14/2c85379880d475f12ee74a27b02a2ffe435d863f8045fe80e5c246c30f23/libusb_package-1.0.26.3-pp310-pypy310_pp73-win_amd64.whl", hash = "sha256:fadaad1181784713948f9cbb7ad1cab8f2b307e784e2e162ed80ba5d2f745901", size = 90602, upload-time = "2025-04-01T12:58:16.828Z" }, ] +[[package]] +name = "mpremote" +version = "1.27.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "platformdirs" }, + { name = "pyserial" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/d2/1c/be7d96275379eca70af111a1d4711c77b1afa78f1782582f3ad082e2d228/mpremote-1.27.0.tar.gz", hash = "sha256:6bb75774648091dad6833af4f86c5bf6505f8d7aec211380f9e6996c01d23cb5", size = 31420, upload-time = "2025-12-09T14:51:55.587Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/58/bf/cb9a7f38015c0fec0295b2cd014b830561f224d264f2f303c2ec15d8ef2f/mpremote-1.27.0-py3-none-any.whl", hash = "sha256:11d134c69b21b487dae3d03eed54c8ccbf84c916c8732a3e069a97cae47be3d4", size = 36094, upload-time = "2025-12-09T14:51:53.759Z" }, +] + [[package]] name = "natsort" version = "8.4.0" @@ -336,6 +351,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/b7/b9/c538f279a4e237a006a2c98387d081e9eb060d203d8ed34467cc0f0b9b53/packaging-26.0-py3-none-any.whl", hash = "sha256:b36f1fef9334a5588b4166f8bcd26a14e521f2b55e6b9de3aaa80d3ff7a37529", size = 74366, upload-time = "2026-01-21T20:50:37.788Z" }, ] +[[package]] +name = "platformdirs" +version = "4.9.4" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/19/56/8d4c30c8a1d07013911a8fdbd8f89440ef9f08d07a1b50ab8ca8be5a20f9/platformdirs-4.9.4.tar.gz", hash = "sha256:1ec356301b7dc906d83f371c8f487070e99d3ccf9e501686456394622a01a934", size = 28737, upload-time = "2026-03-05T18:34:13.271Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/63/d7/97f7e3a6abb67d8080dd406fd4df842c2be0efaf712d1c899c32a075027c/platformdirs-4.9.4-py3-none-any.whl", hash = "sha256:68a9a4619a666ea6439f2ff250c12a853cd1cbd5158d258bd824a7df6be2f868", size = 21216, upload-time = "2026-03-05T18:34:12.172Z" }, +] + [[package]] name = "pluggy" version = "1.6.0"