# 12 HIL с m5stamPLC в процессе

.gitignore

@@ -62,6 +62,7 @@ Testing/
 # Файлы настройки среды разработки
 .project
 compile_commands.json
+.pytest_cache/
 .cache/
 .DS_Store
 .venv/

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-тесты')]
+M5_PORT := env('HIL_M5_PORT', '/dev/ttyACM1')
-[group('pipeline')]
+_m5_agent := justfile_directory() / 'tools/hil/m5_agent.py'
-incoming: flash-test-release
+
+[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')]
+    echo ""
-[group('pipeline')]
+    echo "  Сканируем устройства через mpremote..."
-production: flash-production
+    echo ""
-    @echo "  ✅  Production firmware flashed"
+
+    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"

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-тест — это пара:
 
 ```

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()

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 <id> [байты...]   отправить CAN фрейм (id в hex или dec)
+    can_recv [timeout_ms]      принять CAN фрейм
+    raw <json>                 отправить произвольный 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 <id_hex> [байт байт ...]"))
+            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()

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]

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"