DMR-чип опознан по схеме: HF6853 = AUCTUS A6 (RDA8809), не SCT3258

Опознание БЕЗ разбора рации, по схеме RT4DDLT01 (Radtel v2.1):
DMR-модуль «FM100B» = один чип U700 HF6853 — одночиповый DMR-SoC
семейства Auctus A6 (внутри RDA8809): CPU + ВЧ-трансивер + DSP +
AMBE-вокодер + аудио-кодек в одном кристалле, прошивка во внешнем
SPI-flash, кварц 26 МГц. Прежняя версия «Sicomm SCT3258TD» неверна
(SCT3258 — baseband-only без ВЧ; и нативный протокол другой).

Скрытый ATE/CPS-интерфейс полностью реверснут (jhart99/a6tools,
вендорен в research/a6tools). Даёт КОНКРЕТНУЮ починку нашей
TX-проблемы: AT+GETFREQERR / AT+DMOFREQERR=N, offset=-2500+10*N,
±2500 Гц — наши ~1100 Гц внутри диапазона. Плюс DMR_ADJTXSYMDEV
(девиация), FGU_AFC, DMOSETPOWER, CPS chanInfo, дамп прошивки модуля.

- docs/hf6853-auctus-a6.md — идентификация + протокол + план TX-fix
- tools/a6_freqfix.py — заготовка тула (probe/read/write, не запускать вслепую)
- research/a6tools/ — вендоренный реверс (MIT), atcommands.md/cpecommands.md
- docs/dmr-tx-recovery.md, README — путь 0 (AT+DMOFREQERR) как лучший
- research/rt4ddlt01-dmr-module-sheet.png — рендер листа схемы DMR-модуля

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NikMGoqQWWk9wy2ww2vJAr
Этот коммит содержится в:
2026-07-24 03:10:24 +09:00
co-authored by Claude Opus 4.8
родитель 6d49ba3cba
Коммит c10ab3c6f1
20 изменённых файлов: 1373 добавлений и 1 удалений
+26
Просмотреть файл
@@ -0,0 +1,26 @@
from .escaper import escaper
from .escaper import unescaper
from .rdadebug import compute_check
from .rdadebug import rda_debug_frame
from .rdadebug import read_word
from .rdadebug import write_register_int8
from .rdadebug import read_register_int8
from .rdadebug import write_block
from .a6commands import h2p_command
from .a6commands import set_uart_to_host
from .a6commands import set_uart_to_normal
from .a6commands import read_uart_to_host
from .a6commands import ate_command
from .a6commands import cps_command
from .a6commands import reboot_and_freeze
from .serialio import send_uart_setup
from .serialio import fetch_memory_address
from .serialio import send_ate_command
from .serialio import send_cps_command
from .serialio import atecps_resp_read
from .serialio import read_mem_range
from .serialio import get_chan_info
from .serialio import get_freq_err
from .serialio import parse_freq_err_resp
from .serialio import set_freq_err
from .serialio import SerialIO
+168
Просмотреть файл
@@ -0,0 +1,168 @@
from .rdadebug import write_register_int8
from .rdadebug import read_register_int8
from .rdadebug import write_block
from .rdadebug import compute_check
from .eprint import eprint
def h2p_command(msg):
""" Format a frame for an h2p command
The CPS software sends commands to a special debug register
00000005. Writing a value to this register throws an interupt
which is picked up by a function on the device.
0x00 : Command finished, clears semaphore
0xA5 : Process command with RxByHostPortCB
0xEE : Reboot
0xFF : Handle with boot_HstCmdBasicHandler
@param msg: the message to send
@return: the frame to send
"""
return write_register_int8(0x5, msg)
def set_uart_to_normal():
""" Set device uart to host mode
The CPS software sends repeated requests to set internal register
00000003 to 0x80 which has the effect of locking the UART to debug
mode
@return: the frame to send
"""
return write_register_int8(3, 0x00)
def set_uart_to_host():
""" Set device uart to host mode
The CPS software sends repeated requests to set internal register
00000003 to 0x80 which has the effect of locking the UART to debug
mode
@return: the frame to send
"""
return write_register_int8(3, 0x80)
def reboot_and_freeze():
""" Reboot and freeze the processor
This command comes from coolwatcher and resets the processor and
immediately halts it. This is useful for stepping through the
boot process, but also allows some areas of ROM to be read without
crashing
@return: the frame to send
"""
return write_register_int8(0, 0x03)
def read_uart_to_host():
""" make a frame containing a knock command
this function creates a frame that I assume wakes up the device
for further commands.
@return: the frame to send
"""
return read_register_int8(3)
def ate_command(cmd, p_atecps_write):
""" make a frame containing an ATE command
@param cmd: the command to send
@param p_atecps_write: the address of the CPS write register
@return: the frame to send
"""
cmd = bytearray(cmd, 'utf-8') + b'\r'
cmd += bytes(4 - len(cmd) % 4)
return write_block(p_atecps_write, cmd)
def cps_command(cmd, p_atecps_write):
""" make a frame containing an CPS command
@param cmd: the command to send
@param p_atecps_write: the address of the CPS write register
@return: the frame to send
"""
length = (len(cmd) + 4).to_bytes(1, 'big')
check = compute_check(length + cmd)
begin = bytes([0xaa])
end = bytes([0xbb])
msg = begin + length + cmd + check + end
padding = 4 - (len(msg) % 4)
return write_block(p_atecps_write, msg + bytes([0x00]) * padding)
class CPSFrame:
""" Received CPS class
This class decodes CPS frames received from the device.
"""
check_fail = False
length = 0
type = 0
content = bytes([])
def __init__(self, msg):
eprint(msg.hex())
if (msg[-1].to_bytes(1, 'big') != compute_check(msg[1:-2])):
self.check_fail = True
eprint('CPS frame check failed')
return
self.length = msg[1]
self.type = int.from_bytes(msg[2:4], 'big')
self.is_ok = msg[4] == 0x01
self.content = msg[5:-3]
def __repr__(self):
return 'packet length {} type {} is_ok {} content {}'.format(self.length, self.type, self.is_ok, self.content)
class ChanInfoFrame(CPSFrame):
""" Received ChanInfoFrame class
This class decodes ChanInfoFrame frames received from the device.
"\tcpsInst.chanInfo.nChanIndex=%d\n
\tcpsInst.chanInfo.nChanType=%d\n
\tcpsInst.chanInfo.nVox=%d\n
\tcpsInst.chanInfo.nPower=%d\n
\tcpsInst.chanInfo.nRxFreq=%d\n
\tcpsInst.chanInfo.nTxFreq=%d\n
\tcpsInst.chanInfo.nTxContactsIdx=0x%08x\n
\tcpsInst.chanInfo.nColorCode=%d\n
\tcpsInst.chanInfo.nTimeSlot=%d\n
\tcpsInst.chanInfo.bPoliteCall=%d\n"
\tcpsInst.chanInfo.nEmrSys=%d\n
\tcpsInst.chanInfo.nEncry=%d\n
\tcpsInst.chanInfo.nTypeWideNarrow=%d\n
\tcpsInst.chanInfo.nRxCtdcs=%d\n
\tcpsInst.chanInfo.bRxCtdcsInvert=%d\n
\tcpsInst.chanInfo.bTxCtdcsInvert=%d\n
\tcpsInst.chanInfo.nTxCtdcs=%d\n
\tcpsInst.chanInfo.nRxGrpListIdx=%d\n"
"""
def __init__(self, msg):
super().__init__(msg)
self.index = int.from_bytes(self.content[0:2], 'little')
self.chantype = self.content[2]
self.rxFreq = int.from_bytes(self.content[4:8], 'little')
self.txFreq = int.from_bytes(self.content[8:12], 'little')
self.txContactIndex = int.from_bytes(self.content[12:16], 'little')
self.colorCode = self.content[16]
self.timeslot = self.content[17]
self.polite = self.content[18]
self.emrSys = int.from_bytes(self.content[1:2], 'big')
self.encryption = int.from_bytes(self.content[1:2], 'big')
self.widenarrow = int.from_bytes(self.content[1:2], 'big')
self.rxctdcs = int.from_bytes(self.content[1:2], 'big')
self.rxctdcsinvert = int.from_bytes(self.content[1:2], 'big')
self.txctdcsinvert = int.from_bytes(self.content[1:2], 'big')
self.txctdcs = int.from_bytes(self.content[1:2], 'big')
self.rxGroupIdx = int.from_bytes(self.content[1:2], 'big')
self.vox = int.from_bytes(self.content[1:2], 'big')
def __repr__(self):
return 'packet length {} type {} is_ok {} index {} chantype {} rxfreq {} txfreq {}'.format(
self.length, self.type, self.is_ok, self.index, self.chantype, self.rxFreq, self.txFreq)
+14
Просмотреть файл
@@ -0,0 +1,14 @@
import sys
def eprint(*args, **kwargs):
""" print to stderr
This function takes its arguments just as if it were the normal
print function and instead prints to stderr.
@param args: the arguments to print
@param kwargs: the keyword arguments to print
"""
print(*args, file=sys.stderr, **kwargs)
+34
Просмотреть файл
@@ -0,0 +1,34 @@
def escaper(msg):
""" escape message
this function escapes special characters in the message. These
are 0x5c, 0x11 and 0x13 which are '\' and XON and XOFF characters.
@param msg: the message to escape
@return: the escaped message
"""
out = bytes(sum([[0x5c, 0xFF ^ x ] if x in [0x11, 0x13, 0x5c] else [x] for x in msg], []))
return out
def unescaper(msg):
""" unescape message
this function undoes any escape sequences in a received message
@param msg: the message to unescape
@return: the unescaped message
"""
out = []
escape = False
for x in msg:
if x == 0x5c:
escape = True
continue
if escape:
x = 0x5c ^ x ^ 0xa3
escape = False
out.append(x)
return bytes(out)
+121
Просмотреть файл
@@ -0,0 +1,121 @@
import functools
import operator
from .escaper import escaper
from .escaper import unescaper
from .eprint import eprint
__author__ = "jhart99"
__license__ = "MIT"
def compute_check(msg):
""" Compute the check value for a message
AUCTUS messages use a check byte which is simply the XOR of all
the values of the message.
"""
if len(msg) == 0:
return int(0).to_bytes(1, 'little')
return functools.reduce(operator.xor, msg).to_bytes(1, 'little')
def rda_debug_frame(flow, cmd, message):
""" Format a raw message into a frame
AUCTUS frames are of the form AD 00 XX FF ...message... YY
where XX is the length of the message and YY is the check byte
Additionally certain bytes in the message are escaped.
"""
header = int(0xad).to_bytes(1, 'big')
msg = flow + cmd + message
msglen = len(msg).to_bytes(2, 'big')
check = compute_check(msg)
return escaper(header + msglen + msg + check)
def read_word(addr, seq = 1):
""" make a frame to read a word at a memory address
this function creates a frame to read the memory from the device
suitable for serial transmission.
"""
flow = bytes([0xff])
command = bytes([0x02])
if isinstance(addr, int):
addr = addr.to_bytes(4, 'little')
msg = addr + seq.to_bytes(1, 'big')
return rda_debug_frame(flow, command, msg)
def write_register_int8(addr, msg):
""" write to a byte to an internal register
this function creates a frame to do some device magic and these
frames are used in the preamble and finalizer commands.
"""
flow = bytes([0xff])
command = bytes([0x84])
msg = addr.to_bytes(4, 'little') + msg.to_bytes(1, 'little')
return rda_debug_frame(flow, command, msg)
def read_register_int8(addr, seq=1):
""" make a frame containing a knock command
this function creates a frame that I assume wakes up the device
for further commands.
"""
flow = bytes([0xff])
command = bytes([0x04])
msg = addr.to_bytes(4, 'little') + seq.to_bytes(1, 'big')
return rda_debug_frame(flow, command, msg)
def write_block(addr, msg):
""" make a frame containing a write command
this function creates a frame to do write a multiple byte content
at a specific memory address. The length need not be a word, but
could be 16 bytes or more.
"""
flow = bytes([0xff])
command = bytes([0x83])
if isinstance(addr, int):
addr = addr.to_bytes(4, 'little')
msg = addr + msg
return rda_debug_frame(flow, command, msg)
class RdaFrame:
""" Received Frame class
This class decodes possible received Frames.
"""
ack = False
check_fail = False
seq = 0
length = 0
content = bytes([])
def __init__(self, msg):
msg = unescaper(msg)
if len(msg) <= 4:
if(msg == b'\x11\x13'):
self.ack = True
else:
# impossibly short frame something is wrong.
self.check_fail = True
return
if (msg[-1].to_bytes(1, 'big') != compute_check(msg[3:-1])):
self.check_fail = True
return
self.seq = msg[4]
self.length = msg[2]
self.content = msg[5:-1]
def __repr__(self):
return 'packet length {} seq {} content {} ack {} check {}'.format(self.length, self.seq, self.content, self.ack, self.check_fail)
+365
Просмотреть файл
@@ -0,0 +1,365 @@
import serial
import time
import sys
import re
from .eprint import eprint
from .a6commands import CPSFrame, h2p_command
from .a6commands import ChanInfoFrame, h2p_command
from .a6commands import ate_command
from .a6commands import cps_command
from .a6commands import read_uart_to_host
from .rdadebug import RdaFrame
from .rdadebug import read_word
class Singleton(object):
def __new__(cls, *args, **kwargs):
""" Singleton class
@param args: arguments
@param kwargs: keyword arguments
@return: object
"""
it = cls.__dict__.get("__it__")
if it is not None:
return it
cls.__it__ = it = object.__new__(cls)
it.init(*args, **kwargs)
return it
def init(self, *args, **kwargs):
"""
"""
pass
class SerialIO(Singleton):
def init(self, port, baudrate=921600, verbosity=0, timeout=0.1):
""" Initialize the serial port
@param port: serial port
@param baudrate: baud rate
@param verbosity: verbosity level
"""
self.port = port
self.sio = serial.Serial(port, baudrate,
serial.EIGHTBITS, serial.PARITY_NONE, serial.STOPBITS_ONE,
xonxoff=True, rtscts=False, timeout=timeout)
self.verbosity = verbosity
self._ate_cps_addr = 0
self._ate_cps_resp_addr = 0
self._ate_cps_resp_length_addr = 0
self._uart_resp_addr = 0
self.sio.flush()
if verbosity > 0:
eprint("SerialIO: {} initialized".format(self.port))
def __del__(self):
""" Close the serial port
"""
self.sio.close()
def write(self, msg):
""" Write a message to the serial port
@param msg: message
"""
if self.verbosity > 0:
eprint("write : ", msg.hex())
self.sio.write(msg)
def read(self, nbytes):
""" Read nbytes from the serial port
@param nbytes: number of bytes
@return: message
"""
data = self.sio.read(nbytes)
if self.verbosity > 0:
eprint("read : ", data.hex())
return data
def flush(self):
""" Flush the serial port
"""
self.sio.flush()
@property
def in_waiting(self):
""" return the number of bytes in the serial port
"""
return self.sio.in_waiting
@property
def ate_cps_addr(self):
""" return the address of the ate command
"""
if self._ate_cps_addr == 0:
self._ate_cps_addr = fetch_memory_address(0x81c00270)
self._ate_cps_addr = int.from_bytes(self._ate_cps_addr, byteorder='little')
return self._ate_cps_addr
@property
def ate_cps_resp_addr(self):
""" return the address of the ate command response
"""
if self._ate_cps_resp_addr == 0:
self._ate_cps_resp_addr = fetch_memory_address(0x81c00264)
self._ate_cps_resp_addr = int.from_bytes(self._ate_cps_resp_addr, byteorder='little')
return self._ate_cps_resp_addr
@property
def ate_cps_resp_length_addr(self):
""" return the address of the ate command response
"""
return self.ate_cps_resp_addr - 4
@property
def uart_resp_addr(self):
""" return the address of the ate command response
"""
if self._uart_resp_addr == 0:
self._uart_resp_addr = fetch_memory_address(0x81c0026c)
self._uart_resp_addr = int.from_bytes(self._uart_resp_addr, byteorder='little')
return self._uart_resp_addr
def write_flush_pause(msg, sleep = 0.07):
""" Write out to serial and wait for the radio to process the command
@param msg: bytes to write
@param sleep: time to sleep after writing in ms
"""
uart = SerialIO()
uart.write(msg)
uart.flush()
time.sleep(0.07)
def send_ate_command(msg):
""" Send a command to the ATE/CPS function on the radio
To send a command to the ATE or CPS software on the radio, it has
to be surrounded by these h2p commands which clear the registers
and then throw and interupt which causes the command to be
executed
@param msg: bytes to write
"""
uart = SerialIO()
write_flush_pause(h2p_command(0))
write_flush_pause(ate_command(msg, uart.ate_cps_addr))
write_flush_pause(h2p_command(0xa5))
def send_cps_command(msg):
""" Send a command to the ATE/CPS function on the radio
To send a command to the ATE or CPS software on the radio, it has
to be surrounded by these h2p commands which clear the registers
and then throw and interupt which causes the command to be
executed
@param msg: bytes to write
"""
uart = SerialIO()
write_flush_pause(h2p_command(0))
write_flush_pause(cps_command(msg, uart.ate_cps_addr))
write_flush_pause(h2p_command(0xa5))
def wait_on_read(retries=256, delay=0):
""" Wait until a read happens
This function waits until something is received from the serial or
will abort after a certain number of retries.
@param retries: number of retries before aborting
@param delay: time to sleep between retries
"""
uart = SerialIO()
size = uart.in_waiting
countdown = retries
while size == 0 and countdown > 0:
size = uart.in_waiting
countdown -= 1
if delay: time.sleep(delay)
if countdown == 0:
# nothing received
return b''
if size > 0:
data = uart.read(size)
return data
def send_uart_setup():
""" Replays the initial UART setup sequence
This sequence and timing is from the CPS software capture.
"""
uart = SerialIO()
knock_worked = False
retries = 25
while not knock_worked and retries > 0:
uart.write(read_uart_to_host())
uart.flush()
time.sleep(0.001)
data = wait_on_read()
response = RdaFrame(data)
if response.seq == 1 and response.content == b'\x80':
knock_worked = True
else:
time.sleep(0.25)
retries -= 1
return knock_worked
def fetch_memory_address(addr, seq=1):
""" Attempt to read a memory address and keep trying until it succeeds
"""
uart = SerialIO()
read_ok = False
retval = b''
retries = 25
while not read_ok:
frame = read_word(addr, seq)
uart.write(frame)
uart.flush()
size = uart.in_waiting
i = retries
while size == 0 and i > 0:
time.sleep(0.001)
size = uart.in_waiting
i -= 1
if retries == 0:
continue
data = uart.read(size)
inbound_frame = RdaFrame(data)
read_ok = inbound_frame.seq == seq and not inbound_frame.check_fail
retval = inbound_frame.content
return retval
def atecps_resp_read():
""" Read the response from an ATECPS command
@return: response from ATECPS command
"""
uart = SerialIO()
length = fetch_memory_address(uart.ate_cps_resp_length_addr)
length = int.from_bytes(length, 'little')
response = read_mem_range(uart.ate_cps_resp_addr, uart.ate_cps_resp_addr + length)
return response
def uart_resp_read():
""" Read the response from an ATECPS command
@return: response from ATECPS command
"""
uart = SerialIO()
length = fetch_memory_address(uart.uart_resp_addr)
length = length[1]
response = read_mem_range(uart.uart_resp_addr, uart.uart_resp_addr + length)
return response
def read_mem_range(begin, end):
""" Read a memory range
@param begin: start address
@param end: end address
@return: the data in bytes
"""
addr = begin
datalist = []
while addr < end:
data = fetch_memory_address(addr)
if len(data) == 4:
datalist.append(data)
addr = addr + 4
return b''.join(datalist)
def read_mem_burst(sio, begin, end, offset=0, verbosity=0):
""" Read a limited memory range using a burst read
@param sio: serial object
@param begin: start address
@param end: end address
@param offset: offset of the sequence number
@param verbosity: verbosity level
@return: the data in bytes
"""
if end - begin > 0x100:
raise ValueError('burst read only supports ranges of less than 256 bytes')
# the burst is in words of 4 bytes
burst = (end - begin) / 4
# preallocate the lists
recvflags = [False] * burst
recvdata = [0] * burst
i = 0
data = b''
while sum(recvflags) != burst:
while i < burst:
if not recvflags[i]:
sio.write(read_word(begin + 4 * i, i + offset + 1))
i += 1
size = sio.in_waiting
if size > 0:
data += sio.read(size)
i = 0
return b''.join(recvdata)
def get_chan_info(channel = 0):
""" Get the channel info
@param channel: channel number
@return: the channel info
"""
cmd = bytes([0, 0x12]) + channel.to_bytes(1, 'little')
send_cps_command(cmd)
resp = uart_resp_read()
print(ChanInfoFrame(resp))
# sys.stdout.buffer.write(resp)
def get_freq_err():
""" Get the frequency error from the Radio
@return: frequency error in Hz
"""
send_ate_command("AT+DMOCONNECT")
send_ate_command("AT+GETFREQERR")
resp = atecps_resp_read()
resp = resp.split(b'\x00')
resp = [x.decode('utf-8') for x in resp]
return parse_freq_err_resp(resp[0])
def parse_freq_err_resp(resp):
""" Parse the frequency error response
@param resp: response from ATECPS
@return: frequency error in Hz
"""
pattern = '\[(.+)\]'
freqerr = re.search(pattern, resp)
if freqerr:
return int(freqerr.group(1))
else:
return 0
def set_freq_err(freqerr):
""" Set the frequency error on the Radio
@param freqerr: frequency error parameter which is (-2500 + 10 * freqerr) in Hz
"""
send_ate_command("AT+DMOCONNECT")
send_ate_command("AT+DMOFREQERR={}".format(freqerr))
resp = atecps_resp_read()
resp = resp.split(b'\x00')
resp = [x.decode('utf-8') for x in resp]
for line in resp:
print(line)