Как превратить бинарный CP-лог модема SIMCom A7665E в читаемый
Введение
Отладочные логи мобильных модемов часто выглядят как поток непонятных бинарных данных. Внутри могут находиться сообщения о регистрации в сети, выборе PLMN, параметрах serving cell, APN, IP-адресах и состоянии протоколов LTE. Однако без знания транспортного формата и таблицы сообщений прошивки этот поток практически бесполезен.
В этой статье разберём, как устроено распознавание debug-лога модема SIMCom A7665E на базе ASR «Crane» ASR1603. Цель — пройти весь путь от байтов, поступающих через USB, до человекочитаемой строки:
байтовый поток → бинарная запись → форматное сообщение → декодированные аргументы
Практически задача состоит из двух независимых частей:
Первую проблему можно решить детерминированно. Вторая требует базы форматных строк, соответствующей конкретной сборке прошивки.
1. Исходные условия
В качестве устройства используется модем SIMCom A7665E
Важно не перепутать этот поток с Qualcomm DIAG. CP-log ASR имеет другой бинарный формат, поэтому инструменты, рассчитанные на Qualcomm DIAG, например qc_debug_monitor, его не распознают.
Итоговая архитектура декодера выглядит так:
/dev/ttyUSBx
│
│ AT+ECPLOG=1
▼
поток бинарных данных
│
▼
asr_proto.py
байты → Record
│
▼
asr_mdb.py
(msg_type, fmt_id) + payload → строка
│
▼
asr_debug_monitor.py
захват, фильтрация, цветной вывод, replayТакое разделение ответственности удобно по нескольким причинам:
- транспортный парсер не зависит от базы форматов;
- декодер семантики не зависит от последовательного порта;
- один и тот же протокол можно использовать в CLI, web-мониторе или тестовой системе;
- каждый слой можно тестировать отдельно.
2. Поиск CP-log-порта
Почему нельзя полагаться на ttyUSB0
После перезагрузки модема Linux может назначить интерфейсам другие имена:
до перезагрузки: ttyUSB0, ttyUSB1, ttyUSB2 после перезагрузки: ttyUSB3, ttyUSB4, ttyUSB5
Поэтому поиск порта по имени ненадёжен. Лучше использовать USB VID и номер интерфейса в USB-топологии.
В рассматриваемом устройстве CP-log находится на интерфейсе #2. В pyserial его можно найти по полю location:
from serial.tools import list_ports
MODEM_VID = 0x1e0e
DEBUG_IFACE = 2
def find_debug_port(iface=DEBUG_IFACE, vid=MODEM_VID):
fallback = None
for port in list_ports.comports():
if port.vid != vid:
continue
location = port.location or ""
if location.endswith(f":1.{iface}"):
return port.device
fallback = fallback or port.device
return fallbacklocation.endswith(":1.2")Номер ttyUSBx может измениться, а номер USB-интерфейса в конкретной композиции обычно остаётся постоянным.
Включение потока
CP-log не начинает передаваться автоматически. Сначала на тот же порт необходимо отправить AT-команду:
ENABLE_CMD = b"AT+ECPLOG=1\r\n" DISABLE_CMD = b"AT+ECPLOG=0\r\n"
import serial
import time
def open_device(device, baudrate=115200):
ser = serial.Serial(device, baudrate, timeout=0.2)
ser.reset_input_buffer()
ser.write(ENABLE_CMD)
ser.flush()
time.sleep(0.4)
# Удаляем эхо команды и текстовый ответ OK
ser.reset_input_buffer()
return serЗдесь есть важная особенность: после AT+ECPLOG=1 модем сначала может вернуть обычный текстовый ответ:
AT+ECPLOG=1 OK
Затем начинается бинарный поток. Если не очистить входной буфер, парсер может попытаться интерпретировать ASCII-ответ как начало бинарной записи.
При завершении работы полезно отключить лог:
ser.write(DISABLE_CMD) ser.flush() time.sleep(0.1) ser.reset_input_buffer() ser.close()
3. Формат бинарной записи
Как была найдена граница сообщений
В дампе регулярно повторяется последовательность:
маленькое число → несколько нулей → полезная нагрузка
Это позволяет предположить, что первые два байта являются длиной записи в формате little-endian.
- прочитать первые два байта;
- интерпретировать их как
u16; - перейти вперёд на указанное количество байт;
- проверить, похожи ли следующие два байта на очередную длину;
- повторить процедуру до конца буфера.
Если цепочка проходит через весь дамп без рассинхронизации, гипотеза подтверждается.
Такой формат обладает свойством самосинхронизации: если в поток попал мусор, длина обычно выходит за разумные пределы, и парсер может начать поиск следующей потенциальной записи.
Заголовок
В Python структура описывается так:
import struct
HDR = struct.Struct("<HHHHHHI")
HDR_LEN = HDR.size # 16 байтlength, flags, channel, seq, msg_type, fmt_id, timestamp
Таймстамп
Поле timestamp не является временем в микросекундах или миллисекундах. Экспериментальное сопоставление с известными интервалами показало, что устройство использует частоту 32768 Гц:
TS_HZ = 32768
def timestamp_seconds(timestamp):
return timestamp / TS_HZtimestamp = 2 572 383 time = 2 572 383 / 32 768 ≈ 78.503 с
Это абсолютное время работы устройства до момента включения ECPLOG. В пользовательском интерфейсе удобнее вычесть timestamp первой записи и показывать относительное время:
[+0.000000] ... [+0.003174] ... [+0.006287] ...
4. Потоковый парсер
Парсер должен корректно работать в двух режимах:
Разбор одной записи
MIN_REC = 16
MAX_REC = 16384
def take_record(buf, offset, flush=False):
end = len(buf)
if offset + 2 > end:
return None, offset, True
length = buf[offset] | (buf[offset + 1] << 8)
if length < MIN_REC or length > MAX_REC:
# Некорректная длина: пробуем синхронизацию со следующего байта
return None, offset + 1, False
if offset + length > end:
if flush:
# Финальный обрезанный хвост
return None, offset + 1, False
# Для живого потока ждём следующую порцию данных
return None, offset, True
fields = HDR.unpack_from(buf, offset)
record = {
"length": fields[0],
"flags": fields[1],
"channel": fields[2],
"seq": fields[3],
"msg_type": fields[4],
"fmt_id": fields[5],
"timestamp": fields[6],
"payload": bytes(
buf[offset + HDR_LEN:offset + length]
),
"offset": offset,
}
return record, offset + length, FalseИнкрементальный режим
class StreamParser:
def __init__(self):
self.buffer = bytearray()
def feed(self, chunk):
self.buffer.extend(chunk)
offset = 0
end = len(self.buffer)
while offset < end:
record, next_offset, need_more = take_record(
self.buffer,
offset,
flush=False
)
if need_more:
break
offset = next_offset
if record is not None:
yield record
if offset:
del self.buffer[:offset]
# Защита от бесконечного роста при потере синхронизации
if len(self.buffer) > 1 << 20:
del self.buffer[:len(self.buffer) - (1 << 16)]Если в середине записи пришёл только заголовок и часть payload, она остаётся во внутреннем буфере. Следующий вызов feed() продолжает разбор.
5. Разбор первой записи
Рассмотрим пример записи размером 34 байта:
22 00 | 00 00 | 00 00 | 01 00 | 2b 00 | 96 14 | 5f 40 27 00 | 02 00 63 00 78 08 12 00 01 00 00 00 02 00 00 00 00 00
length flags channel seq 22 00 00 00 00 00 01 00 msg_type fmt_id timestamp 2b 00 96 14 5f 40 27 00 payload 02 00 63 00 78 08 12 00 01 00 00 00 02 00 00 00 00 00
В данном случае msg_type = 0x2b соответствует текстовой записи. Однако payload содержит бинарный префикс, поэтому его нельзя безусловно интерпретировать как обычную C-строку.
6. Типы сообщений
Поле msg_type определяет, как нужно обрабатывать запись:
TYPE_TEXT = 0x2B
TYPE_NAMES = {
0x2B: "TEXT",
0x1D: "TRACE",
0x17: "SIGNAL",
0x1A: "T1A",
0x16: "T16",
0x0E: "T0E",
0x03: "T03",
}TEXT-записи
TEXT-записи уже содержат готовые фрагменты текста. Их можно декодировать без базы форматных строк, извлекая печатные ASCII-последовательности.
def printable_runs(data, min_run=3):
runs = []
current = bytearray()
for byte in data:
if 32 <= byte < 127:
current.append(byte)
else:
if len(current) >= min_run:
runs.append(current.decode("latin-1"))
current.clear()
if len(current) >= min_run:
runs.append(current.decode("latin-1"))
return runsДля текстовых записей достаточно использовать короткий порог:
runs = printable_runs(payload, min_run=3) text = " ".join(runs)
Terse-записи
Terse-записи содержат не сам текст, а:
Их можно представить как удалённый вызов:
diagPrintf("Serving Cell: rsrp = %d", rsrp);fmt_id + бинарное значение rsrp
Сама строка формата находится в базе данных прошивки.
7. Записи file:line без базы
Часть assert-сообщений содержит имя исходного файла и номер строки:
rrcdsutils.c\0\xa2\x0a\x00\x00
имя файла + NUL + номер строки u32
Такой формат можно распознать независимо от версии базы:
import re
SOURCE_RE = re.compile(
r"\.(c|h|cpp|cc|cxx)quot;,
re.IGNORECASE
)
def decode_file_line(payload):
nul = payload.find(b"\x00")
if nul < 4:
return None
raw_name = payload[:nul]
if not all(32 <= byte < 127 for byte in raw_name):
return None
name = raw_name.decode("latin-1")
if not SOURCE_RE.search(name):
return None
name = name.replace("\\", "/").rsplit("/", 1)[-1]
if len(payload) < nul + 5:
return None
line = int.from_bytes(
payload[nul + 1:nul + 5],
"little"
)
return name, linerrcdsutils.c:2722
Строгая проверка необходима: случайный бинарный payload тоже может содержать печатные байты. Условие с расширением исходного файла снижает вероятность ложного распознавания.
8. База форматных строк
Зачем нужна база
Terse-сообщение само по себе неполно. Например, запись может содержать:
msg_type = 0x1d fmt_id = 0x0b3d payload = бинарные аргументы
Чтобы понять, что означают эти байты, нужна запись из базы:
"updates ( %ld , %d ) measUpdateTime: ( 0x%lx ) , ( rsrp = %d )"
В случае ASR используется база cp_MDB.txt формата CATStudio TxtDbHeader.
Секции базы
EnumValsRes— таблица форматных строк;Enums— преобразование числовых значений в имена;Signals— описания структур для%S{...}.
Извлечение секций
В начале файла находится индекс с байтовыми смещениями:
EnumValsRes,1234,987654 Enums,987654,1048576 Signals,1048576,4567890
def parse_sections(raw):
marker = raw.find(b"<end>")
if marker < 0:
return {}
header = raw[:marker].decode(
"latin-1",
"replace"
)
sections = {}
for line in header.splitlines():
parts = line.split(",")
if len(parts) != 3:
continue
name, start, end = parts
if start.isdigit() and end.isdigit():
sections[name] = (
int(start),
int(end)
)
return sectionsТаблица форматных строк
Строка в EnumValsRes выглядит примерно так:
table,id,flag,flag,GROUP,MODULE,NAME,diagPrintf("формат", args);Запятые могут встречаться внутри самой форматной строки, поэтому разделять строку нужно только по первым семи запятым:
parts = line.split(",", 7)Для поддержки всех вариантов вызова используется регулярное выражение:
import re
FORMAT_RE = re.compile(
r'diag(?:Text|Struct)?Printf\s*\('
r'\s*"((?:[^"\\]|\\.)*)"'
)diagPrintf(...) TextPrintf(...) StructPrintf(...)
Это исправляет важную ошибку: если искать только diagPrintf, записи diagTextPrintf и diagStructPrintf останутся без форматной строки и будут ошибочно показаны как hex.
9. Подбор подходящей базы
База форматных строк должна соответствовать сборке прошивки. Даже если две версии используют один и тот же чипсет, идентификаторы сообщений могут отличаться.
Для выбора наиболее подходящей базы удобно использовать измеримые метрики.
Key-match
Первая метрика — доля пар (msg_type, fmt_id), найденных в базе:
distinct_keys = {
(record["msg_type"], record["fmt_id"])
for record in terse_records
}
matched = {
key for key in distinct_keys
if key in mdb_entries
}
key_match = len(matched) / len(distinct_keys)Эта метрика показывает, совпадает ли схема идентификаторов.
Scalar-clean
Вторая метрика проверяет, удалось ли полностью разобрать аргументы скалярных сообщений без маркера <?>.
scalar-clean = число сообщений без underflow / общее число проверяемых сообщений
Она проверяет уже не только идентификаторы, но и ширины аргументов.
10. Главная особенность: int занимает 16 бит
Это наиболее нетривиальная часть реверса.
Симптом
При стандартной интерпретации printf-аргументов каждый %d ожидает 4 байта. Но результаты выглядели неправильно:
rsrp = 104922215
rsrp = <?>
Причина — payload заканчивался раньше, чем ожидал декодер.
Проверка гипотезы
Для каждого сообщения можно вычислить ожидаемый размер аргументов и сравнить его с фактической длиной payload:
def fit_score(records, short_width):
exact = 0
underflow = 0
leftover = 0
for record in records:
fmt = get_format(record)
if not fmt:
continue
required = calculate_argument_size(
fmt,
short_width=short_width,
long_width=4
)
actual = len(record["payload"])
if required == actual:
exact += 1
elif required > actual:
underflow += 1
else:
leftover += 1
return exact, underflow, leftoverПри ширине 2 байта количество underflow уменьшается примерно в пять раз.
Практическое правило
Для ASR-терс-сообщений используется следующее соответствие:
if conversion in "eEfFgG":
width = 8 if length in ("l", "ll", "L") else 4
elif conversion == "p":
width = 4
elif length == "ll":
width = 8
elif length == "hh":
width = 1
elif length == "l":
width = 4
elif length in ("L", "z", "j", "t"):
width = 4
elif length == "h":
width = 2
else:
width = 2Значения %d и %i необходимо интерпретировать как знаковые:
value = int.from_bytes(
payload[pos:pos + width],
"little",
signed=conversion in "di"
)11. Сквозной пример декодирования
msg_type = 0x1d fmt_id = 0x0b3d payload = e40c0000 d100 021c0f00 61e4
"updates ( %ld , %d ) 's measUpdateTime: " "( 0x%lx ) , ( rsrp = %d ) "
updates ( 3300, 209 ) measUpdateTime: ( 0xf1c02 ), ( rsrp = -7071 )
При ошибочной ширине %d = 4 байта декодер начал бы читать следующие аргументы с неправильных смещений и в итоге получил бы <?>.
12. Поддержка %S{...} и структур
Некоторые диагностические сообщения передают не скаляры, а структуры:
%S{EmmPrintDebugInfoInd_MobileId}Описание полей находится в секции Signals.
structs = {
"Example": [
("field_a", "int", 1, ""),
("field_b", "enum", 1, "StateType"),
("field_c", "pointer", 1, ""),
]
}def decode_struct(name, payload, pos=0, depth=0):
if depth > 6:
return f"{{{name}?}}", pos
fields = structs.get(name)
if fields is None:
return f"{{{name}?}}", pos
result = []
for field_name, field_type, count, detail in fields:
values = []
for _ in range(count):
if field_type == "enum":
value = read_u32(payload, pos)
pos += 4
value = enums.get(detail, {}).get(
value,
str(value)
)
elif field_type == "pointer":
value = read_u32(payload, pos)
pos += 4
value = hex(value)
elif field_type == "int":
value = read_u32(payload, pos, signed=True)
pos += 4
elif detail in structs:
value, pos = decode_struct(
detail,
payload,
pos,
depth + 1
)
else:
value = "<?>"
values.append(value)
rendered = (
values[0]
if count == 1
else "[" + ",".join(map(str, values)) + "]"
)
result.append(f"{field_name}={rendered}")
return "{" + ", ".join(result) + "}", posПочему структуры могут декодироваться неверно
Секция Signals из близкой базы может не полностью соответствовать конкретной сборке модема. Например, поля большой EMM-структуры могут иметь другую раскладку или тип.
В результате появляются значения вроде:
mcc = -1727901696
В таких случаях проблема находится не в транспортном парсере, а в несовпадении описания структуры. Исправить её можно только точной базой cp_MDB.txt для конкретной версии прошивки.
13. Приоритет способов декодирования
Одна запись может быть распознана несколькими способами. Поэтому форматтеру нужен строгий порядок выбора:
1. TEXT-запись └─ готовая строка, база не нужна 2. file:line └─ имя исходного файла и номер строки 3. terse-декодирование через MDB └─ форматная строка + бинарные аргументы 4. встроенная ASCII-строка └─ длинный печатный фрагмент внутри payload 5. raw hex └─ неизвестный или неподдерживаемый payload
Формат вывода может выглядеть так:
[+0.000000] [TEXT] +ZDON [+0.003174] [NAS] New Selected PLMN (250,1) [+0.006287] [RRC] Serving Cell rsrp=-110.5 dBm [+0.009460] [ASSERT] rrcdsutils.c:2722
Большие необработанные записи лучше скрывать по умолчанию и включать отдельным флагом:
./asr_debug_monitor.py --replay capture.bin -t
где -t показывает terse-записи, а:
./asr_debug_monitor.py --replay capture.bin -a
показывает все записи, включая объёмные signal-сообщения.
14. Контроль потерь по seq
Поле seq — 16-битный счётчик записей. Оно позволяет заметить потерю данных на USB или переполнение буфера.
Поскольку счётчик циклический, сравнение выполняется по модулю 2162^{16}216:
def sequence_gap(previous, current):
return (current - previous - 1) & 0xFFFFgap = (record.seq - last_seq - 1) & 0xFFFF
if 0 < gap < 0x8000:
dropped += gapОграничение gap < 0x8000 нужно, чтобы отличать нормальное переполнение счётчика от большого обратного скачка.
В интерфейсе можно показывать:
[warning] потеряно записей: 12
Это особенно полезно при высокоскоростном логе, когда USB-порт, пользовательский процесс или внутренний буфер не успевают обработать весь поток.
15. Верификация
Декодер необходимо проверять на двух уровнях.
Автономные тесты
Полезно протестировать отдельно:
- разбор заголовка;
- length-framing;
- повторную синхронизацию после мусора;
- инкрементальный
StreamParser; - TEXT-записи;
file:line;- разбор MDB;
%S{...};%e{...};- 16-битные аргументы.
В исходном проекте такой набор из девяти тестов проходит на синтетической базе и искусственно сформированных записях.
Сквозная проверка
Затем необходимо проверить полный путь:
A7665E → USB CP-log → StreamParser → MDB decoder → человекочитаемый вывод
В корректно декодированных сообщениях должны распознаваться, например:
+ZDON;- выбранная PLMN;
- serving cell;
- RSRP и RSRQ;
- APN;
- IP-адрес;
- события NAS и RRC;
- сообщения IPC между L1 и protocol stack.
16. Ограничения метода
Даже при правильно найденном транспортном формате часть сообщений может оставаться неполностью декодированной.
Особенно важно различать ошибки разных уровней:
неправильная длина записи
→ проблема транспортного парсера
правильная запись, но <?>
→ проблема формата или ширины аргумента
правильный текст, но неверные поля структуры
→ несовпадение Signals с конкретной прошивкойТранспортный формат при этом может быть полностью понятен, даже если семантическая расшифровка остаётся неполной.
17. Запуск декодера
cd /home/sam/work/modems_debug/asr_debug
Живой захват с автоматическим поиском порта:
./asr_debug_monitor.py
./asr_debug_monitor.py \
--replay boot_registration.bin./asr_debug_monitor.py \
--replay boot_registration.bin \
-t./asr_debug_monitor.py \
--replay boot_registration.bin \
-a./asr_debug_monitor.py \
-m /path/to/exact/cp_MDB.txtДля захвата, переживающего ребут модема, нужен внешний wrapper:
- дождаться появления нужного USB-интерфейса;
- открыть порт;
- отправить
AT+ECPLOG=1; - записывать необработанный поток;
- обнаружить исчезновение порта;
- дождаться повторной USB-энумерации;
- открыть новую сессию в отдельном файле.
Так можно захватывать загрузку модема с самого начала, даже если во время ребута меняются имена ttyUSBx.
Заключение
Распознавание бинарного CP-лога ASR состоит из нескольких независимых уровней:
USB-порт → AT+ECPLOG=1 → length-framing → 16-байтный заголовок → msg_type + fmt_id → база форматных строк → декодирование printf-аргументов → структуры, enum и timestamp → читаемый трейс
Самая важная практическая находка — нестандартная упаковка аргументов: обычные %d, %u и %x в terse-сообщениях ASR занимают 2 байта, а %l... — 4 байта. Без этого правила идентификаторы сообщений могут находиться правильно, но значения будут «плыть», а декодер будет регулярно выдавать <?>.
Второй важный вывод — транспорт и семантику нужно разрабатывать отдельно. Формат записи можно восстановить и протестировать независимо от базы. А база, в свою очередь, может заменяться без изменений в протоколе и потоковом парсере.
Для полного покрытия остаётся получить точную cp_MDB.txt под прошивку A7665M5_B01V01. После этого должны исчезнуть ошибки в больших структурах, неизвестные enum и большая часть необработанных бинарных хвостов.