opengd77-uv380-sms/tools/dmr_dump.py

#!/usr/bin/env python3
"""Listen on the radio's MMDVM port and dump every received DMR burst verbatim.

Prints the MMDVM control byte and the raw 33-byte burst as hex, so the frames can be
decoded and compared byte for byte elsewhere. Read-only towards the air: the only thing
sent to the radio is a periodic GET_STATUS, which is required to put/keep it in modem
mode (the firmware leaves modem mode if the host goes quiet, and only enters it when it
sees an MMDVM frame).
"""
import os, sys, termios, time, argparse

MMDVM_FRAME_START = 0xE0
MMDVM_GET_STATUS  = 0x01
MMDVM_DMR_DATA1   = 0x18
MMDVM_DMR_DATA2   = 0x1A

DATA_TYPES = {
    0x0: "PI header", 0x1: "voice LC header", 0x2: "terminator with LC",
    0x3: "CSBK", 0x4: "MBC header", 0x5: "MBC continuation",
    0x6: "data header", 0x7: "rate 1/2 data", 0x8: "rate 3/4 data",
    0x9: "idle", 0xA: "rate 1 data",
}


def open_port(path):
    fd = os.open(path, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK)
    a = termios.tcgetattr(fd)
    a[0] = 0
    a[1] = 0
    a[2] = termios.CS8 | termios.CREAD | termios.CLOCAL
    a[3] = 0
    a[6][termios.VMIN] = 0
    a[6][termios.VTIME] = 0
    termios.tcsetattr(fd, termios.TCSANOW, a)
    termios.tcflush(fd, termios.TCIOFLUSH)
    return fd


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--port", default="/dev/ttyACM0")
    ap.add_argument("--secs", type=int, default=600)
    args = ap.parse_args()

    fd = open_port(args.port)
    buf = b""
    deadline = time.time() + args.secs
    next_poll = 0.0
    counts = {}
    total = 0

    print("dumping raw DMR bursts for %d s (GET_STATUS keepalive every 3 s)" % args.secs, flush=True)
    try:
        while time.time() < deadline:
            now = time.time()
            if now >= next_poll:
                # keepalive: also what puts the radio into modem mode in the first place
                try:
                    os.write(fd, bytes([MMDVM_FRAME_START, 3, MMDVM_GET_STATUS]))
                except OSError:
                    pass
                next_poll = now + 3.0

            try:
                chunk = os.read(fd, 4096)
            except (BlockingIOError, OSError):
                chunk = b""
            if chunk:
                buf += chunk
            else:
                time.sleep(0.002)

            # parse MMDVM frames: E0 <len> <type> <payload...>
            while len(buf) >= 3:
                if buf[0] != MMDVM_FRAME_START:
                    buf = buf[1:]
                    continue
                length = buf[1]
                if length < 3:
                    buf = buf[1:]
                    continue
                if len(buf) < length:
                    break
                frame, buf = buf[:length], buf[length:]
                ftype = frame[2]

                if ftype in (MMDVM_DMR_DATA1, MMDVM_DMR_DATA2):
                    total += 1
                    control = frame[3] if len(frame) > 3 else 0
                    burst = frame[4:37]
                    slot = 1 if ftype == MMDVM_DMR_DATA1 else 2
                    if control & 0x40:                      # data sync burst
                        dt = control & 0x0F
                        label = "data/%s" % DATA_TYPES.get(dt, "type %d" % dt)
                    elif control & 0x20:
                        label = "voice sync"
                    else:
                        label = "voice seq %d" % (control & 0x0F)
                    counts[label] = counts.get(label, 0) + 1
                    print("%s slot%d ctrl=%02X %-28s %s" % (
                        time.strftime("%H:%M:%S", time.gmtime()), slot, control, label, burst.hex()),
                        flush=True)
    finally:
        os.close(fd)

    print("\n--- summary: %d DMR bursts ---" % total, flush=True)
    for k in sorted(counts, key=lambda x: -counts[x]):
        print("  %-28s %d" % (k, counts[k]), flush=True)


if __name__ == "__main__":
    main()