DMR-данные (SMS) на TYT MD-UV380 под OpenGD77
Восстановлено из транскрипта сессии после потери рабочего дерева на сервере сборки. Что внутри: - четыре новых модуля прошивки (приём, сборка исходящего, меню Messages); - правки существующих файлов в виде скриптов, с порядком применения; - инструменты стенда: снятие экрана рации по USB, дамп кадров, эталонная сборка пакета; - NOTES.md: что проверено в эфире, что не доделано, где грабли. Работало в эфире в августе 2026: данные ходят в обе стороны в режиме хотспота, рация принимает SMS и показывает на экране. Отправка собрана и сверена с эталоном сервера побайтово, но в эфире не проверялась. Claude-Session: https://claude.ai/code/session_01GJtvG5WwqYNe1b7s9LLtgy
Этот коммит содержится в:
@@ -0,0 +1,112 @@
|
||||
#!/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()
|
||||
@@ -0,0 +1,115 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Grab the radio's screen over USB and save it as a PNG.
|
||||
|
||||
Uses the CPS read command (area 6 = display buffer), which the firmware answers only
|
||||
while the radio is NOT in modem mode - and, importantly, such requests do not switch it
|
||||
into modem mode either. So the screen can be watched while the radio does its normal job.
|
||||
|
||||
Display is 160x128, RGB565, 2 bytes per pixel.
|
||||
"""
|
||||
import os, sys, termios, time, zlib, struct, argparse
|
||||
|
||||
WIDTH, HEIGHT = 160, 128
|
||||
CPS_ACCESS_DISPLAY_BUFFER = 6
|
||||
CHUNK = 1024
|
||||
|
||||
|
||||
def open_port(path):
|
||||
fd = os.open(path, os.O_RDWR | os.O_NOCTTY)
|
||||
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] = 20
|
||||
termios.tcsetattr(fd, termios.TCSANOW, a)
|
||||
termios.tcflush(fd, termios.TCIOFLUSH)
|
||||
return fd
|
||||
|
||||
|
||||
def read_exact(fd, n, timeout=3.0):
|
||||
buf = b""
|
||||
deadline = time.time() + timeout
|
||||
while len(buf) < n and time.time() < deadline:
|
||||
try:
|
||||
part = os.read(fd, n - len(buf))
|
||||
except BlockingIOError:
|
||||
part = b""
|
||||
if part:
|
||||
buf += part
|
||||
else:
|
||||
time.sleep(0.003)
|
||||
return buf
|
||||
|
||||
|
||||
def read_block(fd, area, addr, length):
|
||||
req = bytes([ord('R'), area,
|
||||
(addr >> 24) & 0xFF, (addr >> 16) & 0xFF, (addr >> 8) & 0xFF, addr & 0xFF,
|
||||
(length >> 8) & 0xFF, length & 0xFF])
|
||||
termios.tcflush(fd, termios.TCIFLUSH)
|
||||
os.write(fd, req)
|
||||
head = read_exact(fd, 3)
|
||||
if len(head) < 3 or head[0:1] != b'R':
|
||||
return None
|
||||
got = (head[1] << 8) | head[2]
|
||||
data = read_exact(fd, got)
|
||||
return data if len(data) == got else None
|
||||
|
||||
|
||||
def write_png(path, rgb_rows):
|
||||
"""Minimal PNG writer (stdlib only)."""
|
||||
raw = b"".join(b"\x00" + row for row in rgb_rows)
|
||||
|
||||
def chunk(tag, data):
|
||||
c = struct.pack(">I", len(data)) + tag + data
|
||||
return c + struct.pack(">I", zlib.crc32(tag + data) & 0xFFFFFFFF)
|
||||
|
||||
png = b"\x89PNG\r\n\x1a\n"
|
||||
png += chunk(b"IHDR", struct.pack(">IIBBBBB", WIDTH, HEIGHT, 8, 2, 0, 0, 0))
|
||||
png += chunk(b"IDAT", zlib.compress(raw, 6))
|
||||
png += chunk(b"IEND", b"")
|
||||
open(path, "wb").write(png)
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--port", default="/dev/ttyACM0")
|
||||
ap.add_argument("--out", default="/home/viktor/screen.png")
|
||||
args = ap.parse_args()
|
||||
|
||||
fd = open_port(args.port)
|
||||
try:
|
||||
total = WIDTH * HEIGHT * 2
|
||||
data = b""
|
||||
while len(data) < total:
|
||||
n = min(CHUNK, total - len(data))
|
||||
part = read_block(fd, CPS_ACCESS_DISPLAY_BUFFER, len(data), n)
|
||||
if part is None:
|
||||
print("read failed at offset %d - is the radio in modem mode?" % len(data))
|
||||
return 2
|
||||
data += part
|
||||
finally:
|
||||
os.close(fd)
|
||||
|
||||
rows = []
|
||||
for y in range(HEIGHT):
|
||||
row = bytearray()
|
||||
for x in range(WIDTH):
|
||||
i = (y * WIDTH + x) * 2
|
||||
# RGB565, little endian as stored by the display driver
|
||||
v = data[i] | (data[i + 1] << 8)
|
||||
r = ((v >> 11) & 0x1F) << 3
|
||||
g = ((v >> 5) & 0x3F) << 2
|
||||
b = (v & 0x1F) << 3
|
||||
row += bytes((r, g, b))
|
||||
rows.append(bytes(row))
|
||||
|
||||
write_png(args.out, rows)
|
||||
nonblack = sum(1 for i in range(0, len(data), 2) if data[i] or data[i + 1])
|
||||
print("saved %s (%d of %d pixels non-black)" % (args.out, nonblack, WIDTH * HEIGHT))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,117 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Reference implementation of outgoing SMS packet assembly, mirrored from the spec.
|
||||
|
||||
Written on the PC first so the tricky parts (CRC-32 with its byte-pair swap, the IP and
|
||||
UDP checksums, the block accounting) can be checked against a known-good packet before
|
||||
any of it goes into firmware. The firmware version will be a literal translation of this.
|
||||
"""
|
||||
|
||||
|
||||
def crc32_dmr(data: bytes) -> bytes:
|
||||
"""ETSI/MOTOTRBO CRC-32: swap bytes within each 16-bit pair, then MSB-first
|
||||
CRC32 (poly 0x04C11DB7, init 0, no final xor). Returned little endian."""
|
||||
swapped = bytearray(data)
|
||||
for i in range(0, len(swapped) - 1, 2):
|
||||
swapped[i], swapped[i + 1] = swapped[i + 1], swapped[i]
|
||||
|
||||
crc = 0
|
||||
for b in swapped:
|
||||
crc ^= b << 24
|
||||
for _ in range(8):
|
||||
crc = ((crc << 1) ^ 0x04C11DB7) & 0xFFFFFFFF if (crc & 0x80000000) else ((crc << 1) & 0xFFFFFFFF)
|
||||
return crc.to_bytes(4, "little")
|
||||
|
||||
|
||||
def ones_complement(data: bytes) -> int:
|
||||
if len(data) % 2:
|
||||
data += b"\x00"
|
||||
total = 0
|
||||
for i in range(0, len(data), 2):
|
||||
total += (data[i] << 8) | data[i + 1]
|
||||
total = (total & 0xFFFF) + (total >> 16)
|
||||
return (~total) & 0xFFFF
|
||||
|
||||
|
||||
def crc16_ccitt(data: bytes) -> int:
|
||||
crc = 0x0000
|
||||
for b in data:
|
||||
crc ^= b << 8
|
||||
for _ in range(8):
|
||||
crc = ((crc << 1) ^ 0x1021) & 0xFFFF if (crc & 0x8000) else ((crc << 1) & 0xFFFF)
|
||||
return crc
|
||||
|
||||
|
||||
def build_tms(text: str, flag: int = 0xA0, seq: int = 0x81) -> bytes:
|
||||
"""[len 2B BE][flag][00][seq][04 0D 00 0A 00][text UTF-16LE]"""
|
||||
body = bytes([flag, 0x00, seq]) + bytes([0x04, 0x0D, 0x00, 0x0A, 0x00]) + text.encode("utf-16-le")
|
||||
tms_len = 8 + 2 * len(text)
|
||||
assert tms_len == len(body), "length mismatch: %d vs %d" % (tms_len, len(body))
|
||||
return tms_len.to_bytes(2, "big") + body
|
||||
|
||||
|
||||
def build_packet(src_id: int, dst_id: int, is_group: bool, text: str,
|
||||
flag: int = 0xA0, seq: int = 0x81, ip_id: int = 0x0001) -> bytes:
|
||||
tms = build_tms(text, flag, seq)
|
||||
|
||||
udp_len = 8 + len(tms)
|
||||
src_ip = (0x0C << 24) | src_id
|
||||
dst_ip = ((0xE1 if is_group else 0x0C) << 24) | dst_id
|
||||
ttl = 0x01 if is_group else 0x40
|
||||
|
||||
udp = bytearray()
|
||||
udp += (0x0FA7).to_bytes(2, "big") # source port 4007
|
||||
udp += (0x0FA7).to_bytes(2, "big") # destination port 4007
|
||||
udp += udp_len.to_bytes(2, "big")
|
||||
udp += b"\x00\x00" # checksum placeholder
|
||||
udp += tms
|
||||
|
||||
pseudo = src_ip.to_bytes(4, "big") + dst_ip.to_bytes(4, "big") + b"\x00\x11" + udp_len.to_bytes(2, "big")
|
||||
udp_csum = ones_complement(pseudo + bytes(udp))
|
||||
udp[6:8] = udp_csum.to_bytes(2, "big")
|
||||
|
||||
ip_len = 20 + udp_len
|
||||
ip = bytearray()
|
||||
ip += b"\x45\x00"
|
||||
ip += ip_len.to_bytes(2, "big")
|
||||
ip += ip_id.to_bytes(2, "big")
|
||||
ip += b"\x00\x00"
|
||||
ip += bytes([ttl, 0x11])
|
||||
ip += b"\x00\x00" # header checksum placeholder
|
||||
ip += src_ip.to_bytes(4, "big")
|
||||
ip += dst_ip.to_bytes(4, "big")
|
||||
ip[10:12] = ones_complement(bytes(ip)).to_bytes(2, "big")
|
||||
|
||||
packet = bytes(ip) + bytes(udp)
|
||||
|
||||
# pad to a multiple of 12, then append the CRC-32 as its own tail
|
||||
pad = (-len(packet)) % 12
|
||||
packet += b"\x00" * pad
|
||||
packet += crc32_dmr(packet)
|
||||
return packet, pad
|
||||
|
||||
|
||||
def build_data_header(src_id: int, dst_id: int, is_group: bool, blocks: int, pad: int) -> bytes:
|
||||
h = bytearray(12)
|
||||
h[0] = (0x80 if is_group else 0x00) | 0x02 # DPF=2 unconfirmed
|
||||
h[1] = 0x40 | (pad & 0x0F) # SAP=4 (IP), pad octets
|
||||
h[2:5] = dst_id.to_bytes(3, "big")
|
||||
h[5:8] = src_id.to_bytes(3, "big")
|
||||
h[8] = 0x80 | (blocks & 0x7F)
|
||||
h[9] = 0x10
|
||||
crc = crc16_ccitt(bytes(h[0:10])) ^ 0xCCCC
|
||||
h[10] = (crc >> 8) & 0xFF
|
||||
h[11] = crc & 0xFF
|
||||
return bytes(h)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
text = "PRIVET"
|
||||
packet, pad = build_packet(777777, 6, True, text)
|
||||
blocks = len(packet) // 12
|
||||
hdr = build_data_header(777777, 6, True, blocks, pad)
|
||||
|
||||
print("text : %r" % text)
|
||||
print("packet len : %d bytes, pad %d, blocks %d" % (len(packet), pad, blocks))
|
||||
print("packet hex : %s" % packet.hex(" "))
|
||||
print("last 4 bytes: %s (reference ends with 92 ba 1c 62)" % packet[-4:].hex(" "))
|
||||
print("data header : %s" % hdr.hex(" "))
|
||||
Ссылка в новой задаче
Block a user
© 2026 StackWorks