387 lines
No EOL
13 KiB
Python
387 lines
No EOL
13 KiB
Python
"""
|
||
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 → RS_RX таргета (BSP_OPTO_CH_RS)
|
||
RLY3 → EXT_IN1 таргета (BSP_OPTO_CH_IN1)
|
||
RLY4 → EXT_IN2 таргета (BSP_OPTO_CH_IN2)
|
||
|
||
─────────────────────────────────────────────────────
|
||
Протокол: 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": 125_000,
|
||
}
|
||
|
||
_RST_PIN = 3 # G3_PHY_RST → нога сброса AW9523B
|
||
|
||
_OPTO_TO_RELAY = {1: 3, 2: 4, 3: 2} # оптоканал таргета → реле стенда
|
||
# ch1 (BSP_OPTO_CH_IN1 / EXT_IN1) → RLY3
|
||
# ch2 (BSP_OPTO_CH_IN2 / EXT_IN2) → RLY4
|
||
# ch3 (BSP_OPTO_CH_RS / RS_RX ) → RLY2
|
||
_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.2",
|
||
"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))
|
||
_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"}
|
||
# Кортеж от биндинга: (identifier, extd, rtr, data)
|
||
frame_id, extd, rtr, data = msg # ← правильная распаковка
|
||
return {"ok": True, "id": frame_id, "ext": bool(extd), "data": list(data)}
|
||
|
||
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() |