rt-4d/tools/agent.py

#!/usr/bin/env python3
"""Клиент интерактивного агента RT-4D (прошивка rt4d_agent.bin).

Шлёт рации `0xAA <len> <сырой FM100B-кадр>`, агент блитит кадр модулю, ответы
модуля возвращаются в порт как `F1 F2 <len> <кадр>`. Работает через powershell.exe (WSL).

    python3 tools/agent.py send 25 01 01        # cmd=25 b=01 sub=01, payload пуст -> версия
    python3 tools/agent.py send 25 01 01 --data 00
    python3 tools/agent.py raw 68 25 01 01 00 00 00 00 10   # готовый кадр целиком
    python3 tools/agent.py listen 10            # просто слушать порт N секунд
"""
import argparse, subprocess, sys, struct


def cksum(frame: bytes) -> int:
    s = 0; i = 0
    while len(frame) - i > 1:
        s += (frame[i] << 8) | frame[i + 1]; i += 2
    if len(frame) - i:
        s += frame[i] << 8
    while s >> 16:
        s = (s & 0xFFFF) + (s >> 16)
    return (~s) & 0xFFFF


def build_frame(cmd, b, sub, payload: bytes) -> bytes:
    f = bytearray([0x68, cmd, b, sub, 0xFF, 0xFF,
                   (len(payload) >> 8) & 0xFF, len(payload) & 0xFF])
    f += payload; f.append(0x10)
    ck = cksum(bytes(f)); f[4], f[5] = (ck >> 8) & 0xFF, ck & 0xFF
    return bytes(f)


def find_port():
    out = subprocess.run(["powershell.exe", "-NoProfile", "-Command",
                          "[System.IO.Ports.SerialPort]::GetPortNames() -join ','"],
                         capture_output=True, text=True).stdout
    ports = [p for p in out.strip().replace("\r", "").split(",") if p]
    if not ports: sys.exit("нет COM-порта")
    return ports[0]


PS = r'''
$p = New-Object System.IO.Ports.SerialPort "{port}",115200,"None",8,"one"
$p.ReadTimeout = 200
try {{ $p.Open() }} catch {{ Write-Output "ERR"; exit 1 }}
$p.DiscardInBuffer()
$tx = [byte[]]@({txlist})
if ($tx.Length -gt 0) {{ $p.Write($tx,0,$tx.Length) }}
$sb = New-Object System.Text.StringBuilder
$sw = [Diagnostics.Stopwatch]::StartNew()
$last = 0
while ($sw.Elapsed.TotalSeconds -lt {secs}) {{
  try {{ [void]$sb.Append("{{0:x2}}" -f $p.ReadByte()); $last=$sw.ElapsedMilliseconds }} catch {{}}
  if ($sb.Length -gt 0 -and ($sw.ElapsedMilliseconds - $last) -gt {idle}) {{ break }}
}}
$p.Close()
Write-Output $sb.ToString()
'''


def xfer(port, tx: bytes, secs=3, idle=400):
    txlist = ",".join(str(x) for x in tx)
    r = subprocess.run(["powershell.exe", "-NoProfile", "-Command",
                        PS.format(port=port, txlist=txlist, secs=secs, idle=idle)],
                       capture_output=True, text=True, encoding="utf-8", errors="replace")
    hexs = "".join(l.strip() for l in r.stdout.splitlines() if l.strip() and l.strip() != "ERR")
    return bytes.fromhex(hexs) if hexs else b""


def parse_frames(raw: bytes):
    out, i = [], 0
    while i < len(raw) - 2:
        if raw[i] == 0xF1 and raw[i + 1] == 0xF2:
            n = raw[i + 2]
            if 9 <= n <= 220 and i + 3 + n <= len(raw):
                out.append(raw[i + 3:i + 3 + n]); i += 3 + n; continue
        i += 1
    return out


def show(raw):
    print(f"принято {len(raw)} байт")
    frames = parse_frames(raw)
    if not frames:
        if raw: print("сырые байты:", raw.hex(' '))
        else: print("(тишина)")
        return
    for f in frames:
        pl = f[8:-1] if len(f) > 9 else b""
        txt = "".join(chr(c) if 32 <= c < 127 else "." for c in pl)
        print(f"  cmd 0x{f[1]:02X}  len={len(pl)}  {f.hex(' ')}" + (f"   «{txt}»" if pl else ""))


def main():
    ap = argparse.ArgumentParser()
    sub = ap.add_subparsers(dest="op", required=True)
    s = sub.add_parser("send"); s.add_argument("cmd"); s.add_argument("b"); s.add_argument("sub")
    s.add_argument("--data", default=""); s.add_argument("--secs", type=float, default=3)
    r = sub.add_parser("raw"); r.add_argument("bytes", nargs="+"); r.add_argument("--secs", type=float, default=3)
    l = sub.add_parser("listen"); l.add_argument("secs", type=float, nargs="?", default=10)
    ap.add_argument("--port")
    a = ap.parse_args()
    port = a.port or find_port()

    if a.op == "listen":
        show(xfer(port, b"", secs=a.secs, idle=99999))
        return
    if a.op == "raw":
        frame = bytes(int(x, 16) for x in a.bytes)
    else:
        payload = bytes.fromhex(a.data) if a.data else b""
        frame = build_frame(int(a.cmd, 16), int(a.b, 16), int(a.sub, 16), payload)
    tx = bytes([0xAA, len(frame)]) + frame
    print(f"TX кадр модулю: {frame.hex(' ')}")
    show(xfer(port, tx, secs=a.secs))


if __name__ == "__main__":
    main()