rt-4d/tools/sniff.py

#!/usr/bin/env python3
"""Захват кадров DMR-модуля с рации, прошитой снифером.

    python3 tools/sniff.py [секунды] [--port COM4] [--raw]

Рация должна быть прошита stock-fw/rt4d_sniffer.bin и включена обычным образом.
Формат потока: F1 F2 <len> <кадр len байт>, кадр — обычный `68 cmd b sub CK CK len len … 10`.

Под WSL порт открывается через powershell.exe.
"""
import argparse, subprocess, sys
from collections import Counter

# что уже известно про команды модуля (из re/dmr.md и разбора образа FM100B)
KNOWN = {
    0x05: "конфигурация канала", 0x06: "установка вызова", 0x07: "данные канала",
    0x09: "разное", 0x0A: "передача/PTT", 0x25: "версия модуля",
    0x3E: "обновление прошивки модуля", 0x42: "параметр", 0x49: "свой DMR ID",
    0x51: "вокодер", 0x59: "активность приёма", 0x62: "параметр",
    0x66: "девиация", 0x67: "компандер", 0x81: "блок SMS", 0x82: "отправка SMS",
    0x84: "запрос контакта",
}

PS = r'''
$p = New-Object System.IO.Ports.SerialPort "{port}",115200,"None",8,"one"
$p.ReadTimeout = 100
try {{ $p.Open() }} catch {{ Write-Output "ERR:$($_.Exception.Message)"; exit 1 }}
$sw = [Diagnostics.Stopwatch]::StartNew()
$sb = New-Object System.Text.StringBuilder
while ($sw.Elapsed.TotalSeconds -lt {secs}) {{
  try {{
    $b = $p.ReadByte()
    [void]$sb.Append("{{0:x2}}" -f $b)
    if ($sb.Length -ge 512) {{ Write-Output $sb.ToString(); [void]$sb.Clear() }}
  }} catch {{
    if ($sb.Length -gt 0) {{ Write-Output $sb.ToString(); [void]$sb.Clear() }}
  }}
}}
if ($sb.Length -gt 0) {{ Write-Output $sb.ToString() }}
$p.Close()
'''


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]


def parse(stream: bytes):
    """выделить кадры по маркеру F1 F2 <len>"""
    out, i = [], 0
    while i < len(stream) - 3:
        if stream[i] == 0xF1 and stream[i + 1] == 0xF2:
            n = stream[i + 2]
            if 9 <= n <= 220 and i + 3 + n <= len(stream):
                out.append(stream[i + 3:i + 3 + n])
                i += 3 + n
                continue
        i += 1
    return out


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("secs", nargs="?", type=int, default=30)
    ap.add_argument("--port")
    ap.add_argument("--raw", action="store_true", help="печатать каждый кадр")
    a = ap.parse_args()
    port = a.port or find_port()
    print(f"[слушаю {port} {a.secs} с — начинайте передачу]", flush=True)

    r = subprocess.run(["powershell.exe", "-NoProfile", "-Command",
                        PS.format(port=port, secs=a.secs)],
                       capture_output=True, text=True, encoding="utf-8", errors="replace")
    hexs = "".join(l.strip() for l in r.stdout.splitlines()
                   if l.strip() and not l.startswith("ERR:"))
    if "ERR:" in r.stdout:
        sys.exit(r.stdout.strip())
    raw = bytes.fromhex(hexs) if hexs else b""
    frames = parse(raw)
    print(f"принято {len(raw)} байт, кадров {len(frames)}\n")

    cnt = Counter(f[1] for f in frames if len(f) > 1)
    print("команды (код — сколько раз — что известно):")
    for cmd, n in cnt.most_common():
        print(f"  0x{cmd:02X}  x{n:<5d} {KNOWN.get(cmd,'— НЕИЗВЕСТНО —')}")

    print("\nпервые кадры каждого кода:")
    seen = set()
    for f in frames:
        if len(f) < 2 or f[1] in seen:
            continue
        seen.add(f[1])
        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"  0x{f[1]:02X}: {f.hex(' ')}")
        if pl:
            print(f"        payload {len(pl)} Б  «{txt}»")
    if a.raw:
        print("\nвсе кадры:")
        for f in frames:
            print("  " + f.hex(" "))


if __name__ == "__main__":
    main()