STM32F103, STM32F407 · уровень средний · 46 мин чтения

19е. Файлы DBC: как расшифровать чужую шину CAN

Раздел 19 научил STM32 принимать кадры CAN, разделы 19г и 19д — спрашивать машину по стандарту: OBD-II и диагностика UDS. Но большая часть того, что ходит по шине автомобиля, не отвечает ни на какие запросы: блоки просто шлют свои кадры по кругу — скорость, угол руля, обороты, педали. Что лежит в каком бите, описывает файл DBC. Здесь — как его читать, где в нём ловушки, как расшифровать шину, для которой файла нет, и как из DBC получить код C для STM32.

Опоры: настоящие файлы из проекта opendbc (opendbc/dbc/vw_mqb.dbc и opendbc/dbc/generator/toyota/_toyota_2017.dbc), кадры и алгоритмы контрольных сумм оттуда же, motohawk.dbc и код C, созданный cantools, из тестов cantools. Всё проверено прогоном на ПК: свой разборщик на Python, сгенерированный код C собран gcc в контейнере stm-sys:1 и arm-none-eabi-gcc для Cortex-M3 (STM32F103) в stm-arm:1. Лицензия: opendbc — MIT (Comma.ai), cantools — MIT (Erik Moqvist); из can-utils и python-obd (обе GPL-2.0) взяты только формат журнала и формулы, без кода.

Как это называется в документации

Оригинал Перевод

DBC (CAN database) текстовый файл-описание кадров и сигналов шины

BO_ (message) сообщение: идентификатор, длина, отправитель

SG_ (signal) сигнал: где лежит в кадре и как пересчитать в физическую величину

byte order: Intel / Motorola порядок байтов: младший первым (@1) / старший первым (@0)

factor, offset множитель и смещение: физ = сырое × множитель + смещение

multiplexor (M), multiplexed (mN) мультиплексор и сигналы его ветви N

VAL_, CM_, BA_ таблица значений, комментарий, атрибут

rolling counter, checksum / CRC счётчик кадров и контрольная сумма в самом кадре

Оригинал	Перевод
DBC (CAN database)	текстовый файл-описание кадров и сигналов шины
BO_ (message)	сообщение: идентификатор, длина, отправитель
SG_ (signal)	сигнал: где лежит в кадре и как пересчитать в физическую величину
byte order: Intel / Motorola	порядок байтов: младший первым (@1) / старший первым (@0)
factor, offset	множитель и смещение: физ = сырое × множитель + смещение
multiplexor (M), multiplexed (mN)	мультиплексор и сигналы его ветви N
VAL_, CM_, BA_	таблица значений, комментарий, атрибут
rolling counter, checksum / CRC	счётчик кадров и контрольная сумма в самом кадре

19е.1 Что такое DBC

DBC — обычный текстовый файл. Его понимают, например, cantools и opendbc, на которые опирается этот раздел. Главные строки — две.

BO_ объявляет сообщение. Строка из vw_mqb.dbc:

BO_ 253 ESP_21: 8 Gateway_MQB

253 — идентификатор (0xFD) в десятичном виде, ESP_21 — имя, 8 — длина данных, Gateway_MQB — кто шлёт. У расширенного 29-битного идентификатора в файле взведён старший бит числа: 0x80000000 + идентификатор.

SG_ под ним описывает сигнал:

 SG_ ESP_v_Signal : 32|16@1+ (0.01,0) [0.00|655.32] "Unit_KiloMeterPerHour"  XXX

По порядку: имя; старт 32 и длина 16 бит; @1 — Intel, младший байт первым (@0 — Motorola, старший первым); + — без знака (- — со знаком, дополнение до двух); (0.01,0) — множитель и смещение; [0.00|655.32] — пределы; единица; получатели. Физическое значение = сырое × 0,01 + 0.

Остальные строки — справочные. VAL_ даёт имена значениям (VAL_ 173 GE_Fahrstufe 5 "P" 6 "R" 7 "N" 8 "D" … — рычаг коробки VW), CM_ — комментарий, BA_DEF_/BA_ — атрибуты. Самый полезный атрибут — GenMsgCycleTime, период отправки в миллисекундах; по нему приёмник понимает, что блок пропал. В ford_lincoln_base_pt.dbc он задан 156 сообщениям, а в файлах VW и Toyota из opendbc его нет вовсе.

Что в выбранных файлах, посчитал свой разборщик:

Файл	Сообщений	Сигналов	Intel	Motorola	Со знаком	Таблиц значений
`vw_mqb.dbc`	113	1348	1345	3	0	133
`_toyota_2017.dbc`	50	310	5	305	23	51

Два завода — два порядка байтов: VW почти целиком Intel, Toyota — Motorola. Это первое, на чём спотыкаются.

19е.2 Intel и Motorola: «пила» номеров битов

Номер бита в DBC везде один: бит b — это разряд b % 8 байта b / 8. Бит 0 — младший разряд байта 0, бит 15 — старший разряд байта 1. Разница Intel и Motorola — в том, что значит «старт» и куда идти дальше.

Intel (@1). Старт — младший бит сигнала, дальше номера растут: 32, 33 … 39, 40 … 47. Младший байт значения лежит раньше. Настоящий кадр ESP_21 из тестов opendbc: 66 D0 1F 80 45 05 00 00. Байт 4 = 0x45, байт 5 = 0x05, сырое 0x0545 = 1349, скорость 13,49 км/ч.

Motorola (@0). Старт — старший бит сигнала. Дальше вниз по разрядам байта: 3, 2, 1, 0. Байт кончился — прыжок на старший разряд следующего байта: 15, 14 … 8. Номера идут пилой: 3, 2, 1, 0, 15, 14, 13, 12, 11, 10, 9, 8. Так лежит угол руля Toyota STEER_ANGLE : 3|12@0- (1.5,0) в сообщении 0x25 STEER_ANGLE_SENSOR.

Разбор руками. Надо передать −45°. Сырое = −45 / 1,5 = −30. В 12 битах дополнения до двух это 0xFE2. Старшие 4 бита (0xF) ложатся в биты 3…0 байта 0, младшие 8 (0xE2) — в байт 1. Кадр: 0F E2 00 00 00 00 00 00.

Ошибка порядка не даёт сбоя, она даёт правдоподобное неверное число: те же байты, прочитанные как Intel 3|12, дают сырое −959 и −1438,5°. Второй пример ловушки — сигнал Temperature : 0|12@0- из motohawk.dbc: начинается с бита 0 и занимает биты 0, 15, 14 … 8, 23, 22, 21 — три байта ради 12 бит.

Сгенерированный код (о генераторе — в 19е.5) делает то же сдвигами по кускам:

void toy_steer_angle_sensor_unpack(toy_steer_angle_sensor_t *m, const uint8_t *d)
{
    uint32_t v;
    v = (uint32_t)(d[0] & 0x0Fu) << 8 | (uint32_t)d[1];
    if (v & 0x800u) v |= 0xFFFFF000u;   /* знак: дополнение до двух */
    m->steer_angle = (int16_t)v;

19е.3 Мультиплексор: несколько кадров под одним идентификатором

Когда данных больше восьми байт, а новый идентификатор жалко, часть кадра объявляют мультиплексором (M), а сигналы помечают номером ветви (m0, m1 …). Сигнал ветви есть в кадре, только если мультиплексор равен её номеру. В vw_mqb.dbc так передаётся номер кузова VIN в сообщении 0x6B4 VIN_01:

 SG_ VIN_01_MUX M : 0|2@1+ (1,0) [0|3] ""  Airbag_MQB
 SG_ KS_Geheimnis_1 m0 : 8|8@1+ (1,0) [0|255] "" Vector__XXX
 SG_ VIN_4 m1 : 8|8@1+ (1,0) [0|255] ""  Airbag_MQB
 SG_ VIN_11 m2 : 8|8@1+ (1,0) [0|255] ""  Airbag_MQB

Байт 1 — это KS_Geheimnis_1, VIN_4 или VIN_11 в зависимости от двух младших битов байта 0. Семнадцать букв VIN разложены по трём ветвям: в ветви 0 три буквы и четыре байта ключа, в ветвях 1 и 2 — по семь букв. Проверка собрала учебный VIN TESTVIN1234567890 в три кадра 0000000000544553, 015456494E313233, 0234353637383930 и разобрала обратно. В коде C ветви становятся switch по мультиплексору; при упаковке пишется только ветвь, выбранная мультиплексором.

В vw_mqb.dbc два мультиплексора и 24 сигнала ветвей. Расширенное мультиплексирование (SG_MUL_VAL_, ветвь на диапазон значений) встречается реже; наш разборщик его не понимает.

19е.4 Счётчики и контрольные суммы

Многие кадры — в примерах ниже это датчик угла руля и тормозная система — несут два служебных поля. Счётчик растёт на 1 в каждом кадре по кругу: приёмник видит пропуск или «замёрзший» передатчик. Контрольная сумма — от данных кадра: приёмник видит искажение, а подделка без знания алгоритма не проходит. В DBC это обычные сигналы COUNTER и CHECKSUM; как считать сумму, DBC не говорит — алгоритм живёт в коде. В opendbc он выбирается по имени файла (opendbc/can/dbc.py, функция get_checksum_state): для toyota_ — toyota_checksum, для vw_mqb — volkswagen_mqb_meb_checksum.

Toyota (opendbc/car/toyota/toyotacan.py) — простая сумма: длина кадра + байты идентификатора + все байты данных, кроме последнего, по модулю 256. Сумма лежит в последнем байте, например CHECKSUM : 63|8@0+ в сообщении 0x260 STEER_TORQUE_SENSOR.

VW MQB (opendbc/car/volkswagen/mqbcan.py) — CRC-8 с многочленом 0x2F (AUTOSAR E2E Profile 2): байт 0 — CRC, младшая тетрада байта 1 — счётчик 0…15. CRC идёт по байтам 1…7, затем по «секретному» байту, выбранному по счётчику; у каждого сообщения своя таблица из 16 байтов. Без неё правильный CRC не получить, даже зная многочлен. Наш вариант на C (перенесён из opendbc, без таблицы на 256 байт):

uint8_t vw_mqb_crc(uint32_t addr, const uint8_t *d, uint8_t len)
{
    const uint8_t *magic = addr == 0x86u ? magic_86 : addr == 0xFDu ? magic_fd : 0;
    uint8_t crc = 0xFF;
    for (uint8_t i = 1; i < len; i++) crc = crc8_2f(crc, d[i]);
    if (magic) crc = crc8_2f(crc, magic[d[1] & 0x0Fu]);
    return crc ^ 0xFFu;
}

Проверка — на настоящих кадрах из opendbc/can/tests/test_checksums.py: 16 кадров LWI_01 (0x86, угол руля) и 16 кадров ESP_21 (0xFD) со счётчиками 0…15. CRC сошёлся во всех 16 + 16, и на Python, и в коде C.

Ещё одна ловушка там же. Кадр LWI_01 03 03 BE A2 12 00 00 00: LWI_Lenkradwinkel : 16|13@1+ (0.1,0) даёт сырое 702 и 70,2°, но LWI_VZ_Lenkradwinkel (бит 29) равен 1 — это знак. Угол −70,2°. VW хранит знак отдельным битом, а DBC такого не умеет: - в DBC — всегда дополнение до двух. Код, который берёт только сигнал угла, путает лево и право.

19е.5 Код C из DBC: cantools и свой генератор

cantools умеет делать из DBC код C: python -m cantools generate_c_source файл.dbc даёт пару .c/.h со структурой на сообщение и функциями _pack, _unpack, _encode, _decode. Но в контейнере stm-sys:1 он не запускается: ModuleNotFoundError: No module named 'bitstruct' — нет пакетов bitstruct и textparser, а раздел обходится без сторонних пакетов.

Поэтому генератор свой: dbcmini.py в каталоге проверки, 301 строка. Понимает BO_, SG_ (Intel и Motorola, знак, множитель и смещение, мультиплексор M/mN), VAL_, CM_, BA_ сообщений; не понимает SG_MUL_VAL_, сигналы с плавающей точкой (SIG_VALTYPE_) и сигналы длиннее 32 бит. На каждое сообщение он пишет:

#define …_ID, …_DLC, а если есть GenMsgCycleTime — …_CYCLE_MS;
структуру с сырыми значениями в самых узких целых (int16_t для 12 бит со знаком);
…_unpack() и …_pack() — сдвиги и маски по кускам, как в листинге 19е.2; pack сначала обнуляет кадр;
static inline функции пересчёта в физическую величину; тип задаёт DBC_REAL (по умолчанию double, на F103 разумнее float или вовсе целые).

Сравнение с эталоном. В тестах cantools лежит код, созданный самим cantools для motohawk.dbc (tests/files/c_source/motohawk.c). Он и наш код для того же файла собраны вместе gcc и прогнаны на 100 000 случайных кадрах: распаковка совпала в 100 000, упаковка обратно — тоже в 100 000.

Размеры для STM32F103 (arm-none-eabi-gcc 12.2.1, -mcpu=cortex-m3 -Os):

Код	Флеш (text), байт
Toyota: 2 сообщения, pack + unpack	272
VW: LWI_01, ESP_21, VIN_01	916
motohawk, наш код	98
motohawk, cantools (с encode/decode и проверками пределов)	440
суммы Toyota и CRC VW	198

По функциям: vw_mqb_crc — 108 байт, toyota_checksum — 38; распаковка motohawk у нас 48 байт, у cantools 58. ОЗУ не нужно ни байта: data и bss равны нулю.

19е.6 Шина без DBC: разведка

Файла под вашу машину может не быть. Тогда его пишут сами. Порядок работы:

Записать шину. candump -l can0 пишет журнал строками (0000000004.540000) can0 0FD#AB0300009D060400 — время, интерфейс, идентификатор, байты (формат — candump.c из can-utils). Адаптер — хоть собственный сниффер на STM32 (раздел 198е).
Разделить по идентификаторам и посмотреть, какие биты меняются. cansniffer из can-utils подсвечивает меняющиеся байты вживую.
Найти служебные поля. Счётчик виден сразу: младший бит меняется в каждом кадре, следующий — в половине, дальше 1/4, 1/8. Байт CRC похож на шум: каждый бит меняется примерно в половине кадров.
Двигать одно и следить. Нажали педаль, повернули руль — в каком поле пошли изменения? Лучше, чем на глаз: параллельно спрашивать по OBD (раздел 19г) известную величину и искать поле, которое с ней коррелирует.
Подобрать множитель прямой по точкам «сырое — OBD» и записать сигнал в свой DBC.

Шаги 3–5 проверены на модели. Модель — 60 с езды: скорость 0 → 72 → 0 км/ч, руль туда-сюда; кадры ESP_21 и LWI_01 раз в 20 мс собраны по vw_mqb.dbc со счётчиком и настоящим CRC; ответ OBD на PID 0x0D (скорость, байт A = км/ч, как в python-obd) — раз в 100 мс. Всего 6600 кадров, из них 600 ответов OBD. Результат:

биты 8…11 кадра 0xFD меняются в долях 1,0, 0,5, 0,25 и 0,12 кадров — счётчик;
биты байта 0 меняются в долях от 0,48 до 0,50 — это CRC;
39 бит из 64 не меняются никогда;
перебор 237 полей (старт, длина 8/12/16, оба порядка) против скорости OBD нашёл поле 32|16@1 с корреляцией 0,999913 и множителем 0,00999 при смещении −0,48 (OBD отдаёт целые км/ч с отбрасыванием дробной части — отсюда почти −0,5).

Но у полей 31|16@1 и 30|16@1 корреляция та же, 0,999913, а множители 0,005 и 0,0025: биты 30 и 31 в модели не меняются, и сдвинутое поле — то же число, умноженное на 2 или 4. Статистика границу не находит. Решают соседние биты: поле начинается с самого младшего меняющегося бита, а множитель у инженеров круглый — 0,01.

19е.7 Полный код генератора и разбора

Весь код раздела свой: генератор dbcmini.py, созданные им файлы C, контрольные суммы dbc19e_e2e.c и разведка шины из прогона dbc19e.py. Ниже он по порядку. Из dbc19e_vw.c показана одна функция из шести — распаковка VIN_01 с мультиплексором: остальные того же вида, что у Toyota, и генератор пишет их сам. Стенд dbc19e_host.c, который сверяет наш код с Python и с motohawk.c из cantools, не приведён: он только читает команды прогона и печатает ответы. Прошивки с приёмом кадров по bxCAN в этом разделе нет — код проверен на ПК и собран в объектные файлы.

Файлы C пишет dbc19e.py --gen <каталог>. Файлы DBC он берёт из клонов opendbc (github.com/commaai/opendbc) и cantools (github.com/cantools/cantools), которые лежат рядом в каталоге из переменной OSS_DIR. Сборка для Cortex-M3 — те же ключи, что в прогоне:

python dbc19e.py --gen .
for f in dbc19e_toy dbc19e_vw dbc19e_mh dbc19e_e2e; do
  arm-none-eabi-gcc -std=c99 -mcpu=cortex-m3 -mthumb -Os -ffunction-sections -Wall -Wextra -I. -c $f.c -o $f.o
done
arm-none-eabi-size dbc19e_toy.o dbc19e_vw.o dbc19e_mh.o dbc19e_e2e.o

Разбор DBC. Регулярные выражения на строки BO_, SG_, VAL_, CM_ и BA_, классы сигнала и сообщения, функция parse(). Сигналы относятся к последнему BO_ выше; первая строка другого вида обрывает их список.

# -*- coding: utf-8 -*-
"""Раздел 19е. Маленький разборщик DBC (подмножество) и генератор кода C для STM32.

Понимает: BO_, SG_ (Intel @1 и Motorola @0, знак +/-, множитель и смещение,
пределы, единица, получатели, мультиплексор M и ветви mN), VAL_, CM_ (BO_ и SG_),
BA_DEF_ / BA_DEF_DEF_ / BA_ для сообщений (GenMsgCycleTime и т. п.).
Не понимает: SG_MUL_VAL_ (расширенное мультиплексирование), SIG_VALTYPE_ (float),
сигналы длиннее 32 бит. cantools умеет больше, но в stm-sys:1 он не запускается:
нет пакетов bitstruct и textparser, а ставить их нельзя.
"""
import re

RE_BO = re.compile(r"^BO_ (\d+) (\w+) *: *(\d+) (\w+)", re.M)
RE_SG = re.compile(r'^ SG_ (\w+) *(M|m\d+)? *: *(\d+)\|(\d+)@([01])([+-]) *\(([^,]+),([^)]+)\) *'
                   r'\[([^|]*)\|([^\]]*)\] *"([^"]*)" *(.*)$')
RE_VAL = re.compile(r'^VAL_ (\d+) (\w+) (.*?);', re.M)
RE_CM = re.compile(r'^CM_ (BO_|SG_) (\d+) (?:(\w+) )?"(.*?)";', re.M | re.S)
RE_BADEF = re.compile(r'^BA_DEF_ BO_ +"(\w+)" (\w+)', re.M)
RE_BADEFDEF = re.compile(r'^BA_DEF_DEF_ +"(\w+)" +([^;]+);', re.M)
RE_BA = re.compile(r'^BA_ "(\w+)" BO_ (\d+) ([^;]+);', re.M)


class Signal:
    def __init__(self, m):
        (self.name, mux, sb, ln, bo, sg, fa, of, mn, mx, self.unit, rx) = m.groups()
        self.start, self.length = int(sb), int(ln)
        self.intel = bo == "1"                     # @1 — Intel (младший байт первым), @0 — Motorola
        self.signed = sg == "-"
        self.factor, self.offset = float(fa), float(of)
        self.min, self.max = float(mn or 0), float(mx or 0)
        self.receivers = [r for r in re.split(r"[ ,]+", rx.strip()) if r]
        self.is_mux = mux == "M"
        self.mux_id = int(mux[1:]) if mux and mux != "M" else None
        self.choices, self.comment = {}, None


class Message:
    def __init__(self, fid, name, dlc, sender):
        self.frame_id, self.name, self.dlc, self.sender = fid, name, dlc, sender
        self.extended = bool(fid & 0x80000000)
        self.id = fid & 0x1FFFFFFF
        self.signals, self.comment, self.attrs = [], None, {}

    def sig(self, name):
        return next(s for s in self.signals if s.name == name)

    @property
    def mux(self):
        return next((s for s in self.signals if s.is_mux), None)


def parse(text):
    msgs, cur = {}, None
    for line in text.splitlines():
        m = RE_BO.match(line)
        if m:
            fid = int(m.group(1))
            cur = msgs[fid] = Message(fid, m.group(2), int(m.group(3)), m.group(4))
            continue
        m = RE_SG.match(line)
        if m and cur is not None:
            s = Signal(m)
            assert s.length <= 32, s.name
            cur.signals.append(s)
        else:
            cur = None
    for fid, name, body in RE_VAL.findall(text):
        if int(fid) in msgs:
            s = next((x for x in msgs[int(fid)].signals if x.name == name), None)
            if s is None:
                continue
            s.choices = {int(k): v for k, v in re.findall(r'(-?\d+) +"([^"]*)"', body)}
    for kind, fid, name, txt in RE_CM.findall(text):
        if int(fid) not in msgs:
            continue
        if kind == "BO_":
            msgs[int(fid)].comment = txt
        else:
            s = next((x for x in msgs[int(fid)].signals if x.name == name), None)
            if s is not None:
                s.comment = txt
    defaults = {n: v.strip().strip('"') for n, v in RE_BADEFDEF.findall(text)}
    for n, _ in RE_BADEF.findall(text):
        for mm in msgs.values():
            if n in defaults:
                mm.attrs[n] = defaults[n]
    for n, fid, v in RE_BA.findall(text):
        if int(fid) in msgs:
            msgs[int(fid)].attrs[n] = v.strip().strip('"')
    return msgs

Номера битов. bit_positions() — та самая пила из 19е.2: у Motorola после разряда 0 номер прыгает на 15 вверх. decode_raw() нарочно считает иначе, через одно большое число: сверка C с Python ловит ошибки генератора, а не повторяет их.

# ── Нумерация битов ─────────────────────────────────────────────────────
def bit_positions(s):
    """Номера битов сигнала от старшего к младшему (нумерация DBC: бит b = байт b//8, разряд b%8)."""
    if s.intel:
        return [s.start + i for i in range(s.length)][::-1]
    out, b = [], s.start                       # Motorola: старт — старший бит, «пила»
    for _ in range(s.length):
        out.append(b)
        b = b + 15 if b % 8 == 0 else b - 1
    return out


def decode_raw(s, data):
    """Сырое значение другим способом, чем генератор C: через одно большое число."""
    n = len(data)
    if s.intel:
        raw = (int.from_bytes(data, "little") >> s.start) & ((1 << s.length) - 1)
    else:
        msb = (s.start // 8) * 8 + (7 - s.start % 8)       # номер старшего бита при счёте «слева направо»
        raw = (int.from_bytes(data, "big") >> (n * 8 - msb - s.length)) & ((1 << s.length) - 1)
    if s.signed and raw >> (s.length - 1):
        raw -= 1 << s.length
    return raw


def encode_raw(s, raw, data):
    raw &= (1 << s.length) - 1
    for k, b in enumerate(bit_positions(s)):
        bit = (raw >> (s.length - 1 - k)) & 1
        data[b // 8] = (data[b // 8] & ~(1 << (b % 8))) | (bit << (b % 8))
    return data


def phys(s, raw):
    return raw * s.factor + s.offset


def active(msg, data):
    """Сигналы, которые есть в кадре: мультиплексор выбирает свою ветвь."""
    mx = msg.mux
    sel = decode_raw(mx, data) if mx else None
    return [s for s in msg.signals if s.mux_id is None or s.mux_id == sel]


def decode(msg, data):
    return {s.name: decode_raw(s, data) for s in active(msg, data)}


def used_mask(msg, data):
    mask = bytearray(msg.dlc)
    for s in active(msg, data):
        for b in bit_positions(s):
            mask[b // 8] |= 1 << (b % 8)
    return bytes(mask)

Генератор C. segments() режет сигнал на куски по байтам, из кусков gen_c() пишет строки распаковки и упаковки. gen_dump() и py_dump() нужны только стенду на ПК: они печатают одно и то же двумя независимыми путями.

# ── Генератор C ─────────────────────────────────────────────────────────
def segments(s):
    """Куски сигнала по байтам: (байт, сдвиг в байте, ширина, позиция в значении)."""
    pos = bit_positions(s)                    # от старшего к младшему
    L, segs, k = s.length, [], 0
    while k < L:
        byte = pos[k] // 8
        j = k
        while j + 1 < L and pos[j + 1] // 8 == byte and pos[j + 1] == pos[j] - 1:
            j += 1
        lo = pos[j] % 8                       # младший разряд куска в байте
        segs.append((byte, lo, j - k + 1, L - 1 - j))
        k = j + 1
    return segs


def ctype(s):
    for n in (8, 16, 32):
        if s.length <= n:
            return ("int%d_t" if s.signed else "uint%d_t") % n


def utype(s):
    return "u" + ctype(s).lstrip("u")


def cname(x):
    return re.sub(r"[^0-9a-z_]", "_", x.lower())


def fmt(x):
    r = repr(float(x))
    return r if ("." in r or "e" in r) else r + ".0"


def gen_c(prefix, msgs, names, hname):
    """Возвращает (текст .h, текст .c) для сообщений names."""
    H = ["/* Сгенерировано dbcmini.py (раздел 19е) — не править руками. */",
         "#ifndef %s_H" % prefix.upper(), "#define %s_H" % prefix.upper(),
         "#include <stdint.h>", "", "#ifndef DBC_REAL", "#define DBC_REAL double", "#endif", ""]
    C = ["/* Сгенерировано dbcmini.py (раздел 19е) — не править руками. */",
         '#include <string.h>', '#include "%s"' % hname, ""]
    for nm in names:
        m = next(x for x in msgs.values() if x.name == nm)
        t = "%s_%s_t" % (prefix, cname(m.name))
        f = "%s_%s" % (prefix, cname(m.name))
        cyc = m.attrs.get("GenMsgCycleTime")
        H.append("#define %s_ID 0x%Xu%s" % (f.upper(), m.id, "  /* расширенный */" if m.extended else ""))
        H.append("#define %s_DLC %du" % (f.upper(), m.dlc))
        if cyc is not None:
            H.append("#define %s_CYCLE_MS %su" % (f.upper(), cyc))
        H.append("typedef struct {")
        for s in m.signals:
            H.append("    %s %s;%s" % (ctype(s), cname(s.name),
                                        "  /* %s%s */" % ("мультиплексор" if s.is_mux else "ветвь %d" % s.mux_id,
                                                          "") if (s.is_mux or s.mux_id is not None) else ""))
        H.append("} %s;" % t)
        H.append("void %s_unpack(%s *m, const uint8_t *d);" % (f, t))
        H.append("void %s_pack(uint8_t *d, const %s *m);" % (f, t))
        for s in m.signals:
            H.append("static inline DBC_REAL %s_%s_phys(%s raw) { return (DBC_REAL)raw * (DBC_REAL)%s + (DBC_REAL)%s; }"
                     % (f, cname(s.name), ctype(s), fmt(s.factor), fmt(s.offset)))
        H.append("")
        mx = m.mux
        plain = [x for x in m.signals if x.mux_id is None]
        plain.sort(key=lambda x: not x.is_mux)
        branches = sorted({x.mux_id for x in m.signals if x.mux_id is not None})

        def body(emit, ind):
            out = []
            for x in plain:
                out += emit(x, ind)
            if branches:
                out.append(ind + "switch (m->%s) {" % cname(mx.name))
                for k in branches:
                    out.append(ind + "case %d:" % k)
                    for x in m.signals:
                        if x.mux_id == k:
                            out += emit(x, ind + "    ")
                    out.append(ind + "    break;")
                out.append(ind + "default:")
                out.append(ind + "    break;")
                out.append(ind + "}")
            return out

        def emit_unpack(s, ind):
            parts = []
            for byte, lo, w, p in segments(s):
                e = "(d[%d] >> %d)" % (byte, lo) if lo else "d[%d]" % byte
                if lo + w < 8:
                    e = "(%s & 0x%02Xu)" % (e, (1 << w) - 1)
                parts.append("(uint32_t)%s << %d" % (e, p) if p else "(uint32_t)%s" % e)
            out = [ind + "v = " + " | ".join(parts) + ";"]
            if s.signed:
                out.append(ind + "if (v & 0x%Xu) v |= 0x%Xu;   /* знак: дополнение до двух */"
                           % (1 << (s.length - 1), (0xFFFFFFFF << s.length) & 0xFFFFFFFF))
            out.append(ind + "m->%s = (%s)v;" % (cname(s.name), ctype(s)))
            return out

        def emit_pack(s, ind):
            out = [ind + "v = (uint32_t)(%s)m->%s;" % (utype(s), cname(s.name))]
            for byte, lo, w, p in segments(s):
                e = "(v >> %d)" % p if p else "v"
                if w < 8:
                    e = "(%s & 0x%02Xu)" % (e, (1 << w) - 1)
                out.append(ind + "d[%d] |= (uint8_t)%s;" % (byte, "(%s << %d)" % (e, lo) if lo else e))
            return out

        C.append("void %s_unpack(%s *m, const uint8_t *d)" % (f, t))
        C += ["{", "    uint32_t v;"] + body(emit_unpack, "    ") + ["}", ""]
        C.append("void %s_pack(uint8_t *d, const %s *m)" % (f, t))
        C += ["{", "    uint32_t v;", "    memset(d, 0, %d);" % m.dlc] + body(emit_pack, "    ") + ["}", ""]
    H.append("#endif")
    return "\n".join(H) + "\n", "\n".join(C)


def gen_dump(prefix, msgs, names):
    """Только для проверки на ПК: распаковать, напечатать сигналы, упаковать обратно."""
    C = []
    for nm in names:
        m = next(x for x in msgs.values() if x.name == nm)
        f = "%s_%s" % (prefix, cname(m.name))
        C.append("static void %s_dump(const uint8_t *d)" % f)
        C.append("{")
        C.append("    %s_t m; uint8_t o[8]; int i;" % f)
        C.append("    memset(&m, 0, sizeof m);")
        C.append("    %s_unpack(&m, d);" % f)
        C.append('    printf("%s");' % m.name)
        mx = m.mux
        for s in m.signals:
            cond = "1" if s.mux_id is None else "m.%s == %d" % (cname(mx.name), s.mux_id)
            C.append('    if (%s) printf(" %s=%%ld:%%.10g", (long)m.%s, %s_%s_phys(m.%s));'
                     % (cond, s.name, cname(s.name), f, cname(s.name), cname(s.name)))
        C.append("    %s_pack(o, &m);" % f)
        C.append('    printf(" |");')
        C.append('    for (i = 0; i < %d; i++) printf("%%02X", o[i]);' % m.dlc)
        C.append('    printf("\\n");')
        C.append("}")
        C.append("")
    return "\n".join(C) + "\n"


def py_dump(msg, data):
    """То же, что печатает gen_dump, но независимым путём decode_raw/encode_raw."""
    vals = decode(msg, data)
    out = msg.name
    for s in msg.signals:
        if s.name in vals:
            out += " %s=%d:%s" % (s.name, vals[s.name], "%.10g" % phys(s, vals[s.name]))
    o = bytearray(msg.dlc)
    for s in active(msg, data):
        encode_raw(s, vals[s.name], o)
    return out + " |" + o.hex().upper()

Какие сообщения генерировать — начало dbc19e.py. DBC_TOY, DBC_VW и DBC_MH — пути к opendbc/dbc/generator/toyota/_toyota_2017.dbc, opendbc/dbc/vw_mqb.dbc и к tests/files/dbc/motohawk.dbc из cantools:

TOY = dm.parse(read(DBC_TOY))
VW = dm.parse(read(DBC_VW))
MH = dm.parse(read(DBC_MH))
SETS = [("toy", TOY, ["STEER_ANGLE_SENSOR", "STEER_TORQUE_SENSOR"]),
        ("vw", VW, ["LWI_01", "ESP_21", "VIN_01"]),
        ("mh", MH, ["ExampleMessage"])]
BYNAME = {n: next(m for m in db.values() if m.name == n) for _, db, ns in SETS for n in ns}


def generate(outdir):
    for pre, db, names in SETS:
        h, c = dm.gen_c(pre, db, names, "dbc19e_%s.h" % pre)
        for ext, txt in (("h", h), ("c", c)):
            with open(os.path.join(outdir, "dbc19e_%s.%s" % (pre, ext)), "w", encoding="utf-8", newline="\n") as fh:
                fh.write(txt)

Что вышло для Toyota — заголовок и код целиком. Пересчёт в физическую величину — функции static inline в заголовке, их тип задаёт DBC_REAL (19е.5):

/* Сгенерировано dbcmini.py (раздел 19е) — не править руками. */
#ifndef TOY_H
#define TOY_H
#include <stdint.h>

#ifndef DBC_REAL
#define DBC_REAL double
#endif

#define TOY_STEER_ANGLE_SENSOR_ID 0x25u
#define TOY_STEER_ANGLE_SENSOR_DLC 8u
typedef struct {
    int16_t steer_angle;
    int8_t steer_fraction;
    int16_t steer_rate;
} toy_steer_angle_sensor_t;
void toy_steer_angle_sensor_unpack(toy_steer_angle_sensor_t *m, const uint8_t *d);
void toy_steer_angle_sensor_pack(uint8_t *d, const toy_steer_angle_sensor_t *m);
static inline DBC_REAL toy_steer_angle_sensor_steer_angle_phys(int16_t raw) { return (DBC_REAL)raw * (DBC_REAL)1.5 + (DBC_REAL)0.0; }
static inline DBC_REAL toy_steer_angle_sensor_steer_fraction_phys(int8_t raw) { return (DBC_REAL)raw * (DBC_REAL)0.1 + (DBC_REAL)0.0; }
static inline DBC_REAL toy_steer_angle_sensor_steer_rate_phys(int16_t raw) { return (DBC_REAL)raw * (DBC_REAL)1.0 + (DBC_REAL)0.0; }

#define TOY_STEER_TORQUE_SENSOR_ID 0x260u
#define TOY_STEER_TORQUE_SENSOR_DLC 8u
typedef struct {
    int16_t steer_torque_eps;
    int16_t steer_torque_driver;
    int16_t steer_angle;
    uint8_t steer_angle_initializing;
    uint8_t steer_override;
    uint8_t checksum;
} toy_steer_torque_sensor_t;
void toy_steer_torque_sensor_unpack(toy_steer_torque_sensor_t *m, const uint8_t *d);
void toy_steer_torque_sensor_pack(uint8_t *d, const toy_steer_torque_sensor_t *m);
static inline DBC_REAL toy_steer_torque_sensor_steer_torque_eps_phys(int16_t raw) { return (DBC_REAL)raw * (DBC_REAL)1.0 + (DBC_REAL)0.0; }
static inline DBC_REAL toy_steer_torque_sensor_steer_torque_driver_phys(int16_t raw) { return (DBC_REAL)raw * (DBC_REAL)1.0 + (DBC_REAL)0.0; }
static inline DBC_REAL toy_steer_torque_sensor_steer_angle_phys(int16_t raw) { return (DBC_REAL)raw * (DBC_REAL)0.0573 + (DBC_REAL)0.0; }
static inline DBC_REAL toy_steer_torque_sensor_steer_angle_initializing_phys(uint8_t raw) { return (DBC_REAL)raw * (DBC_REAL)1.0 + (DBC_REAL)0.0; }
static inline DBC_REAL toy_steer_torque_sensor_steer_override_phys(uint8_t raw) { return (DBC_REAL)raw * (DBC_REAL)1.0 + (DBC_REAL)0.0; }
static inline DBC_REAL toy_steer_torque_sensor_checksum_phys(uint8_t raw) { return (DBC_REAL)raw * (DBC_REAL)1.0 + (DBC_REAL)0.0; }

#endif

/* Сгенерировано dbcmini.py (раздел 19е) — не править руками. */
#include <string.h>
#include "dbc19e_toy.h"

void toy_steer_angle_sensor_unpack(toy_steer_angle_sensor_t *m, const uint8_t *d)
{
    uint32_t v;
    v = (uint32_t)(d[0] & 0x0Fu) << 8 | (uint32_t)d[1];
    if (v & 0x800u) v |= 0xFFFFF000u;   /* знак: дополнение до двух */
    m->steer_angle = (int16_t)v;
    v = (uint32_t)(d[4] >> 4);
    if (v & 0x8u) v |= 0xFFFFFFF0u;   /* знак: дополнение до двух */
    m->steer_fraction = (int8_t)v;
    v = (uint32_t)(d[4] & 0x0Fu) << 8 | (uint32_t)d[5];
    if (v & 0x800u) v |= 0xFFFFF000u;   /* знак: дополнение до двух */
    m->steer_rate = (int16_t)v;
}

void toy_steer_angle_sensor_pack(uint8_t *d, const toy_steer_angle_sensor_t *m)
{
    uint32_t v;
    memset(d, 0, 8);
    v = (uint32_t)(uint16_t)m->steer_angle;
    d[0] |= (uint8_t)((v >> 8) & 0x0Fu);
    d[1] |= (uint8_t)v;
    v = (uint32_t)(uint8_t)m->steer_fraction;
    d[4] |= (uint8_t)((v & 0x0Fu) << 4);
    v = (uint32_t)(uint16_t)m->steer_rate;
    d[4] |= (uint8_t)((v >> 8) & 0x0Fu);
    d[5] |= (uint8_t)v;
}

void toy_steer_torque_sensor_unpack(toy_steer_torque_sensor_t *m, const uint8_t *d)
{
    uint32_t v;
    v = (uint32_t)d[5] << 8 | (uint32_t)d[6];
    if (v & 0x8000u) v |= 0xFFFF0000u;   /* знак: дополнение до двух */
    m->steer_torque_eps = (int16_t)v;
    v = (uint32_t)d[1] << 8 | (uint32_t)d[2];
    if (v & 0x8000u) v |= 0xFFFF0000u;   /* знак: дополнение до двух */
    m->steer_torque_driver = (int16_t)v;
    v = (uint32_t)d[3] << 8 | (uint32_t)d[4];
    if (v & 0x8000u) v |= 0xFFFF0000u;   /* знак: дополнение до двух */
    m->steer_angle = (int16_t)v;
    v = (uint32_t)((d[0] >> 3) & 0x01u);
    m->steer_angle_initializing = (uint8_t)v;
    v = (uint32_t)(d[0] & 0x01u);
    m->steer_override = (uint8_t)v;
    v = (uint32_t)d[7];
    m->checksum = (uint8_t)v;
}

void toy_steer_torque_sensor_pack(uint8_t *d, const toy_steer_torque_sensor_t *m)
{
    uint32_t v;
    memset(d, 0, 8);
    v = (uint32_t)(uint16_t)m->steer_torque_eps;
    d[5] |= (uint8_t)(v >> 8);
    d[6] |= (uint8_t)v;
    v = (uint32_t)(uint16_t)m->steer_torque_driver;
    d[1] |= (uint8_t)(v >> 8);
    d[2] |= (uint8_t)v;
    v = (uint32_t)(uint16_t)m->steer_angle;
    d[3] |= (uint8_t)(v >> 8);
    d[4] |= (uint8_t)v;
    v = (uint32_t)(uint8_t)m->steer_angle_initializing;
    d[0] |= (uint8_t)((v & 0x01u) << 3);
    v = (uint32_t)(uint8_t)m->steer_override;
    d[0] |= (uint8_t)(v & 0x01u);
    v = (uint32_t)(uint8_t)m->checksum;
    d[7] |= (uint8_t)v;
}

motohawk. Сигнал Temperature, который идёт через три байта (19е.2), стал тремя кусками с разными сдвигами:

/* Сгенерировано dbcmini.py (раздел 19е) — не править руками. */
#include <string.h>
#include "dbc19e_mh.h"

void mh_examplemessage_unpack(mh_examplemessage_t *m, const uint8_t *d)
{
    uint32_t v;
    v = (uint32_t)(d[0] & 0x01u) << 11 | (uint32_t)d[1] << 3 | (uint32_t)(d[2] >> 5);
    if (v & 0x800u) v |= 0xFFFFF000u;   /* знак: дополнение до двух */
    m->temperature = (int16_t)v;
    v = (uint32_t)((d[0] >> 1) & 0x3Fu);
    m->averageradius = (uint8_t)v;
    v = (uint32_t)(d[0] >> 7);
    m->enable = (uint8_t)v;
}

void mh_examplemessage_pack(uint8_t *d, const mh_examplemessage_t *m)
{
    uint32_t v;
    memset(d, 0, 8);
    v = (uint32_t)(uint16_t)m->temperature;
    d[0] |= (uint8_t)((v >> 11) & 0x01u);
    d[1] |= (uint8_t)(v >> 3);
    d[2] |= (uint8_t)((v & 0x07u) << 5);
    v = (uint32_t)(uint8_t)m->averageradius;
    d[0] |= (uint8_t)((v & 0x3Fu) << 1);
    v = (uint32_t)(uint8_t)m->enable;
    d[0] |= (uint8_t)((v & 0x01u) << 7);
}

Мультиплексор VIN_01 — switch по двум младшим битам байта 0 (19е.3):

void vw_vin_01_unpack(vw_vin_01_t *m, const uint8_t *d)
{
    uint32_t v;
    v = (uint32_t)(d[0] & 0x03u);
    m->vin_01_mux = (uint8_t)v;
    switch (m->vin_01_mux) {
    case 0:
        v = (uint32_t)d[1];
        m->ks_geheimnis_1 = (uint8_t)v;
        v = (uint32_t)d[2];
        m->ks_geheimnis_2 = (uint8_t)v;
        v = (uint32_t)d[3];
        m->ks_geheimnis_3 = (uint8_t)v;
        v = (uint32_t)d[4];
        m->ks_geheimnis_4 = (uint8_t)v;
        v = (uint32_t)d[5];
        m->vin_1 = (uint8_t)v;
        v = (uint32_t)d[6];
        m->vin_2 = (uint8_t)v;
        v = (uint32_t)d[7];
        m->vin_3 = (uint8_t)v;
        break;
    case 1:
        v = (uint32_t)d[1];
        m->vin_4 = (uint8_t)v;
        v = (uint32_t)d[2];
        m->vin_5 = (uint8_t)v;
        v = (uint32_t)d[3];
        m->vin_6 = (uint8_t)v;
        v = (uint32_t)d[4];
        m->vin_7 = (uint8_t)v;
        v = (uint32_t)d[5];
        m->vin_8 = (uint8_t)v;
        v = (uint32_t)d[6];
        m->vin_9 = (uint8_t)v;
        v = (uint32_t)d[7];
        m->vin_10 = (uint8_t)v;
        break;
    case 2:
        v = (uint32_t)d[1];
        m->vin_11 = (uint8_t)v;
        v = (uint32_t)d[2];
        m->vin_12 = (uint8_t)v;
        v = (uint32_t)d[3];
        m->vin_13 = (uint8_t)v;
        v = (uint32_t)d[4];
        m->vin_14 = (uint8_t)v;
        v = (uint32_t)d[5];
        m->vin_15 = (uint8_t)v;
        v = (uint32_t)d[6];
        m->vin_16 = (uint8_t)v;
        v = (uint32_t)d[7];
        m->vin_17 = (uint8_t)v;
        break;
    default:
        break;
    }
}

Суммы и счётчик (19е.4): сумма Toyota, CRC-8 0x2F VW MQB с «секретными» байтами и проверка счётчика:

/* Раздел 19е: контрольные суммы и счётчики автомобильных кадров. */
#ifndef DBC19E_E2E_H
#define DBC19E_E2E_H
#include <stdint.h>

uint8_t toyota_checksum(uint32_t addr, const uint8_t *d, uint8_t len);
uint8_t vw_mqb_crc(uint32_t addr, const uint8_t *d, uint8_t len);
int     counter_next_ok(uint8_t prev, uint8_t now, uint8_t bits);

#endif

/* Раздел 19е: контрольные суммы и счётчики автомобильных кадров.
   Алгоритмы пересказаны по opendbc (MIT, Comma.ai): toyota_checksum —
   opendbc/car/toyota/toyotacan.py, vw_mqb_crc — opendbc/car/volkswagen/mqbcan.py
   (CRC-8 с многочленом 0x2F, как AUTOSAR E2E Profile 2). */
#include "dbc19e_e2e.h"

/* Toyota: длина + байты адреса + все байты, кроме последнего (там сама сумма) */
uint8_t toyota_checksum(uint32_t addr, const uint8_t *d, uint8_t len)
{
    uint32_t s = len;
    while (addr) { s += addr & 0xFFu; addr >>= 8; }
    for (uint8_t i = 0; i + 1u < len; i++) s += d[i];
    return (uint8_t)s;
}

static uint8_t crc8_2f(uint8_t crc, uint8_t b)
{
    crc ^= b;
    for (int k = 0; k < 8; k++)
        crc = (crc & 0x80u) ? (uint8_t)((crc << 1) ^ 0x2Fu) : (uint8_t)(crc << 1);
    return crc;
}

/* «Секретный» байт на каждое значение счётчика: свой у каждого кадра */
static const uint8_t magic_86[16] = { 0x86, 0x86, 0x86, 0x86, 0x86, 0x86, 0x86, 0x86,
                                      0x86, 0x86, 0x86, 0x86, 0x86, 0x86, 0x86, 0x86 };
static const uint8_t magic_fd[16] = { 0xB4, 0xEF, 0xF8, 0x49, 0x1E, 0xE5, 0xC2, 0xC0,
                                      0x97, 0x19, 0x3C, 0xC9, 0xF1, 0x98, 0xD6, 0x61 };

/* VW MQB: CRC по байтам 1..len-1, затем по байту magic[счётчик]; счётчик — d[1] & 0x0F */
uint8_t vw_mqb_crc(uint32_t addr, const uint8_t *d, uint8_t len)
{
    const uint8_t *magic = addr == 0x86u ? magic_86 : addr == 0xFDu ? magic_fd : 0;
    uint8_t crc = 0xFF;
    for (uint8_t i = 1; i < len; i++) crc = crc8_2f(crc, d[i]);
    if (magic) crc = crc8_2f(crc, magic[d[1] & 0x0Fu]);
    return crc ^ 0xFFu;
}

/* Счётчик должен вырасти ровно на 1 по модулю 2^bits */
int counter_next_ok(uint8_t prev, uint8_t now, uint8_t bits)
{
    return now == (uint8_t)((prev + 1u) & ((1u << bits) - 1u));
}

Разведка шины (19е.6). Тот же CRC VW на Python — через таблицу на 256 байт; им модель подписывает свои кадры:

CRC8H2F = []
for i in range(256):
    c = i
    for _ in range(8):
        c = ((c << 1) ^ 0x2F) & 0xFF if c & 0x80 else (c << 1) & 0xFF
    CRC8H2F.append(c)
MAGIC = {0x86: [0x86] * 16, 0xFD: [0xB4, 0xEF, 0xF8, 0x49, 0x1E, 0xE5, 0xC2, 0xC0, 0x97, 0x19, 0x3C, 0xC9, 0xF1, 0x98, 0xD6, 0x61]}


def vw_crc(addr, dd):
    c = 0xFF
    for x in dd[1:]:
        c = CRC8H2F[c ^ x]
    c = CRC8H2F[c ^ MAGIC[addr][dd[1] & 0x0F]]
    return c ^ 0xFF

Модель поездки, доли смен по битам и перебор полей против скорости OBD — конец dbc19e.py целиком. esp и lwi — сообщения ESP_21 и LWI_01 из разобранного vw_mqb.dbc, dm — модуль dbcmini:

# ══ 6. Разведка чужой шины: модель ══════════════════════════════════════
# 60 с езды: скорость 0 → 72 → 0 км/ч, руль туда-сюда. Кадры ESP_21 и LWI_01 собираются
# настоящими правилами из vw_mqb.dbc (счётчик + CRC), OBD-ответы 0x7E8 на PID 0x0D — раз в 100 мс.
rnd = random.Random(1999)
log, obd = [], []
cnt = {0xFD: 0, 0x86: 0}
for i in range(3000):                          # шаг 20 мс
    t = i * 0.02
    v = 72.0 * math.sin(math.pi * t / 60.0) + rnd.uniform(-0.05, 0.05)
    ang = 90.0 * math.sin(2 * math.pi * t / 7.0)
    for addr, m, vals in ((0xFD, esp, {"ESP_v_Signal": round(max(v, 0) / 0.01), "ESP_Systemstatus": 1}),
                          (0x86, lwi, {"LWI_Lenkradwinkel": round(abs(ang) / 0.1), "LWI_VZ_Lenkradwinkel": int(ang < 0),
                                       "LWI_Lenkradw_Geschw": rnd.randrange(0, 40)})):
        dd = bytearray(8)
        vals["COUNTER"] = cnt[addr]
        cnt[addr] = (cnt[addr] + 1) & 15
        for n, r in vals.items():
            dm.encode_raw(m.sig(n), r, dd)
        dd[0] = vw_crc(addr, dd)
        log.append((t, addr, bytes(dd)))
    if i % 5 == 0:
        a = int(max(v, 0))                     # PID 0x0D: байт A = км/ч
        log.append((t + 0.001, 0x7E8, bytes([0x03, 0x41, 0x0D, a, 0x55, 0x55, 0x55, 0x55])))
        obd.append((t, a))
print("модель: кадров в журнале %d, из них OBD-ответов %d; строка журнала: (%010d.%06d) can0 %03X#%s" % (
    len(log), len(obd), int(log[500][0]), round(log[500][0] % 1 * 1e6), log[500][1], log[500][2].hex().upper()))
for addr in (0xFD, 0x86):
    fr = [x[2] for x in log if x[1] == addr]
    flips = [sum(((fr[k][b // 8] ^ fr[k + 1][b // 8]) >> (b % 8)) & 1 for k in range(len(fr) - 1)) / (len(fr) - 1) for b in range(64)]
    print("0x%X: доля смен по битам 8..11 (счётчик) %s; байт 0 (CRC) от %.2f до %.2f; никогда не меняются %d бит из 64" % (
        addr, [round(flips[b], 2) for b in range(8, 12)], min(flips[:8]), max(flips[:8]), sum(f == 0 for f in flips)))
fr = [(x[0], x[2]) for x in log if x[1] == 0xFD]
best = []
for start in range(0, 64):
    for ln in (8, 12, 16):
        for intel in (True, False):
            if intel and start + ln > 64:
                continue
            s = dm.Signal(dm.RE_SG.match(' SG_ X : %d|%d@%s+ (1,0) [0|0] "" X' % (start, ln, "1" if intel else "0")))
            if not intel and max(dm.bit_positions(s)) > 63 or not intel and any(b < 0 for b in dm.bit_positions(s)):
                continue
            xs, ys = [], []
            j = 0
            for t, a in obd:
                while j + 1 < len(fr) and fr[j + 1][0] <= t:
                    j += 1
                xs.append(dm.decode_raw(s, fr[j][1]))
                ys.append(a)
            mx, my = sum(xs) / len(xs), sum(ys) / len(ys)
            sxx = sum((x - mx) ** 2 for x in xs)
            syy = sum((y - my) ** 2 for y in ys)
            if sxx == 0:
                continue
            sxy = sum((x - mx) * (y - my) for x, y in zip(xs, ys))
            best.append((sxy / math.sqrt(sxx * syy), start, ln, intel, sxy / sxx, my - sxy / sxx * mx))
best.sort(reverse=True)
r, start, ln, intel, k, b = best[0]
print("разведка 0xFD: лучший кандидат %d|%d@%s, корреляция с OBD %.6f, множитель %.5f, смещение %.2f" % (
    start, ln, "1" if intel else "0", r, k, b))
for c in best[1:3]:
    print("  следующий: %d|%d@%s, корреляция %.6f, множитель %.5f" % (c[1], c[2], "1" if c[3] else "0", c[0], c[4]))
print("кандидатов перебрано:", len(best))

19е.8 Грабли

Угол руля −1438° вместо −45°. Перепутан порядок байтов: Motorola прочитан как Intel. Toyota — @0, VW — @1; смотреть на символ после @ в каждой строке, а не на файл целиком — в vw_mqb.dbc 3 сигнала Motorola среди 1345 Intel.

Сигнал Motorola «съехал» на соседние байты. Старт Motorola — старший бит, а не младший, и номера идут пилой. Разбор руками одного кадра (как в 19е.2) ловит ошибку сразу.

Лево и право перепутаны. Знак хранится отдельным битом (LWI_VZ_… у VW), а сигнал угла объявлен +. DBC знак-модуль не выражает — это надо сделать в коде.

Свой кадр блок игнорирует. Нет счётчика, неверный CRC или не тот «секретный» байт. В VW MQB CRC зависит от таблицы на 16 значений счётчика, своей у каждого сообщения. Алгоритма в DBC нет — он в коде opendbc.

Сигналы ветви читаются мусором. Не проверен мультиплексор: в кадре VIN_01 с мультиплексором 1 байт 1 — буква VIN, а не байт ключа.

Разведка нашла поле с идеальной корреляцией, а масштаб странный. Соседние неизменные биты дают ту же корреляцию со сдвигом ×2, ×4. Брать младший меняющийся бит и круглый множитель.

cantools не ставится. Для генерации нужен Python с bitstruct и textparser. Нет сети — либо свой генератор под нужное подмножество, либо готовые .c/.h, созданные на другой машине.

19е.9 Итоги

DBC — текст: BO_ — сообщение (идентификатор, длина, отправитель), SG_ — сигнал старт|длина@порядок знак (множитель,смещение) [мин|макс] "единица".
@1 — Intel, старт — младший бит; @0 — Motorola, старт — старший бит, номера пилой: 3, 2, 1, 0, 15 … 8. −45° в STEER_ANGLE 3|12@0- — это 0F E2; как Intel те же байты дают −1438,5°.
В vw_mqb.dbc 113 сообщений и 1348 сигналов, почти всё Intel; в _toyota_2017.dbc — 50 и 310, почти всё Motorola.
Мультиплексор M и ветви mN: VIN в VIN_01 идёт тремя кадрами.
Счётчик и CRC — обычные сигналы, но алгоритм не в DBC. CRC-8 0x2F VW MQB с «секретными» байтами сошёлся на 16 + 16 настоящих кадрах.
Свой генератор C сверен с кодом cantools: 100 000 из 100 000 кадров; на F103 pack + unpack двух сообщений Toyota — 272 байта флеша, ОЗУ — 0.
Разведка чужой шины: счётчик — доли смен 1, 1/2, 1/4…, CRC — около 0,5; поле скорости нашлось по корреляции с OBD (0,999913), множитель 0,01 — с учётом соседних битов.

Полные исходники

dbcmini.py — маленький разборщик DBC и генератор кода C.

# -*- coding: utf-8 -*-
"""Раздел 19е. Маленький разборщик DBC (подмножество) и генератор кода C для STM32.

Понимает: BO_, SG_ (Intel @1 и Motorola @0, знак +/-, множитель и смещение,
пределы, единица, получатели, мультиплексор M и ветви mN), VAL_, CM_ (BO_ и SG_),
BA_DEF_ / BA_DEF_DEF_ / BA_ для сообщений (GenMsgCycleTime и т. п.).
Не понимает: SG_MUL_VAL_ (расширенное мультиплексирование), SIG_VALTYPE_ (float),
сигналы длиннее 32 бит. cantools умеет больше, но в stm-sys:1 он не запускается:
нет пакетов bitstruct и textparser, а ставить их нельзя.
"""
import re

RE_BO = re.compile(r"^BO_ (\d+) (\w+) *: *(\d+) (\w+)", re.M)
RE_SG = re.compile(r'^ SG_ (\w+) *(M|m\d+)? *: *(\d+)\|(\d+)@([01])([+-]) *\(([^,]+),([^)]+)\) *'
                   r'\[([^|]*)\|([^\]]*)\] *"([^"]*)" *(.*)$')
RE_VAL = re.compile(r'^VAL_ (\d+) (\w+) (.*?);', re.M)
RE_CM = re.compile(r'^CM_ (BO_|SG_) (\d+) (?:(\w+) )?"(.*?)";', re.M | re.S)
RE_BADEF = re.compile(r'^BA_DEF_ BO_ +"(\w+)" (\w+)', re.M)
RE_BADEFDEF = re.compile(r'^BA_DEF_DEF_ +"(\w+)" +([^;]+);', re.M)
RE_BA = re.compile(r'^BA_ "(\w+)" BO_ (\d+) ([^;]+);', re.M)


class Signal:
    def __init__(self, m):
        (self.name, mux, sb, ln, bo, sg, fa, of, mn, mx, self.unit, rx) = m.groups()
        self.start, self.length = int(sb), int(ln)
        self.intel = bo == "1"                     # @1 — Intel (младший байт первым), @0 — Motorola
        self.signed = sg == "-"
        self.factor, self.offset = float(fa), float(of)
        self.min, self.max = float(mn or 0), float(mx or 0)
        self.receivers = [r for r in re.split(r"[ ,]+", rx.strip()) if r]
        self.is_mux = mux == "M"
        self.mux_id = int(mux[1:]) if mux and mux != "M" else None
        self.choices, self.comment = {}, None


class Message:
    def __init__(self, fid, name, dlc, sender):
        self.frame_id, self.name, self.dlc, self.sender = fid, name, dlc, sender
        self.extended = bool(fid & 0x80000000)
        self.id = fid & 0x1FFFFFFF
        self.signals, self.comment, self.attrs = [], None, {}

    def sig(self, name):
        return next(s for s in self.signals if s.name == name)

    @property
    def mux(self):
        return next((s for s in self.signals if s.is_mux), None)


def parse(text):
    msgs, cur = {}, None
    for line in text.splitlines():
        m = RE_BO.match(line)
        if m:
            fid = int(m.group(1))
            cur = msgs[fid] = Message(fid, m.group(2), int(m.group(3)), m.group(4))
            continue
        m = RE_SG.match(line)
        if m and cur is not None:
            s = Signal(m)
            assert s.length <= 32, s.name
            cur.signals.append(s)
        else:
            cur = None
    for fid, name, body in RE_VAL.findall(text):
        if int(fid) in msgs:
            s = next((x for x in msgs[int(fid)].signals if x.name == name), None)
            if s is None:
                continue
            s.choices = {int(k): v for k, v in re.findall(r'(-?\d+) +"([^"]*)"', body)}
    for kind, fid, name, txt in RE_CM.findall(text):
        if int(fid) not in msgs:
            continue
        if kind == "BO_":
            msgs[int(fid)].comment = txt
        else:
            s = next((x for x in msgs[int(fid)].signals if x.name == name), None)
            if s is not None:
                s.comment = txt
    defaults = {n: v.strip().strip('"') for n, v in RE_BADEFDEF.findall(text)}
    for n, _ in RE_BADEF.findall(text):
        for mm in msgs.values():
            if n in defaults:
                mm.attrs[n] = defaults[n]
    for n, fid, v in RE_BA.findall(text):
        if int(fid) in msgs:
            msgs[int(fid)].attrs[n] = v.strip().strip('"')
    return msgs


# ── Нумерация битов ─────────────────────────────────────────────────────
def bit_positions(s):
    """Номера битов сигнала от старшего к младшему (нумерация DBC: бит b = байт b//8, разряд b%8)."""
    if s.intel:
        return [s.start + i for i in range(s.length)][::-1]
    out, b = [], s.start                       # Motorola: старт — старший бит, «пила»
    for _ in range(s.length):
        out.append(b)
        b = b + 15 if b % 8 == 0 else b - 1
    return out


def decode_raw(s, data):
    """Сырое значение другим способом, чем генератор C: через одно большое число."""
    n = len(data)
    if s.intel:
        raw = (int.from_bytes(data, "little") >> s.start) & ((1 << s.length) - 1)
    else:
        msb = (s.start // 8) * 8 + (7 - s.start % 8)       # номер старшего бита при счёте «слева направо»
        raw = (int.from_bytes(data, "big") >> (n * 8 - msb - s.length)) & ((1 << s.length) - 1)
    if s.signed and raw >> (s.length - 1):
        raw -= 1 << s.length
    return raw


def encode_raw(s, raw, data):
    raw &= (1 << s.length) - 1
    for k, b in enumerate(bit_positions(s)):
        bit = (raw >> (s.length - 1 - k)) & 1
        data[b // 8] = (data[b // 8] & ~(1 << (b % 8))) | (bit << (b % 8))
    return data


def phys(s, raw):
    return raw * s.factor + s.offset


def active(msg, data):
    """Сигналы, которые есть в кадре: мультиплексор выбирает свою ветвь."""
    mx = msg.mux
    sel = decode_raw(mx, data) if mx else None
    return [s for s in msg.signals if s.mux_id is None or s.mux_id == sel]


def decode(msg, data):
    return {s.name: decode_raw(s, data) for s in active(msg, data)}


def used_mask(msg, data):
    mask = bytearray(msg.dlc)
    for s in active(msg, data):
        for b in bit_positions(s):
            mask[b // 8] |= 1 << (b % 8)
    return bytes(mask)


# ── Генератор C ─────────────────────────────────────────────────────────
def segments(s):
    """Куски сигнала по байтам: (байт, сдвиг в байте, ширина, позиция в значении)."""
    pos = bit_positions(s)                    # от старшего к младшему
    L, segs, k = s.length, [], 0
    while k < L:
        byte = pos[k] // 8
        j = k
        while j + 1 < L and pos[j + 1] // 8 == byte and pos[j + 1] == pos[j] - 1:
            j += 1
        lo = pos[j] % 8                       # младший разряд куска в байте
        segs.append((byte, lo, j - k + 1, L - 1 - j))
        k = j + 1
    return segs


def ctype(s):
    for n in (8, 16, 32):
        if s.length <= n:
            return ("int%d_t" if s.signed else "uint%d_t") % n


def utype(s):
    return "u" + ctype(s).lstrip("u")


def cname(x):
    return re.sub(r"[^0-9a-z_]", "_", x.lower())


def fmt(x):
    r = repr(float(x))
    return r if ("." in r or "e" in r) else r + ".0"


def gen_c(prefix, msgs, names, hname):
    """Возвращает (текст .h, текст .c) для сообщений names."""
    H = ["/* Сгенерировано dbcmini.py (раздел 19е) — не править руками. */",
         "#ifndef %s_H" % prefix.upper(), "#define %s_H" % prefix.upper(),
         "#include <stdint.h>", "", "#ifndef DBC_REAL", "#define DBC_REAL double", "#endif", ""]
    C = ["/* Сгенерировано dbcmini.py (раздел 19е) — не править руками. */",
         '#include <string.h>', '#include "%s"' % hname, ""]
    for nm in names:
        m = next(x for x in msgs.values() if x.name == nm)
        t = "%s_%s_t" % (prefix, cname(m.name))
        f = "%s_%s" % (prefix, cname(m.name))
        cyc = m.attrs.get("GenMsgCycleTime")
        H.append("#define %s_ID 0x%Xu%s" % (f.upper(), m.id, "  /* расширенный */" if m.extended else ""))
        H.append("#define %s_DLC %du" % (f.upper(), m.dlc))
        if cyc is not None:
            H.append("#define %s_CYCLE_MS %su" % (f.upper(), cyc))
        H.append("typedef struct {")
        for s in m.signals:
            H.append("    %s %s;%s" % (ctype(s), cname(s.name),
                                        "  /* %s%s */" % ("мультиплексор" if s.is_mux else "ветвь %d" % s.mux_id,
                                                          "") if (s.is_mux or s.mux_id is not None) else ""))
        H.append("} %s;" % t)
        H.append("void %s_unpack(%s *m, const uint8_t *d);" % (f, t))
        H.append("void %s_pack(uint8_t *d, const %s *m);" % (f, t))
        for s in m.signals:
            H.append("static inline DBC_REAL %s_%s_phys(%s raw) { return (DBC_REAL)raw * (DBC_REAL)%s + (DBC_REAL)%s; }"
                     % (f, cname(s.name), ctype(s), fmt(s.factor), fmt(s.offset)))
        H.append("")
        mx = m.mux
        plain = [x for x in m.signals if x.mux_id is None]
        plain.sort(key=lambda x: not x.is_mux)
        branches = sorted({x.mux_id for x in m.signals if x.mux_id is not None})

        def body(emit, ind):
            out = []
            for x in plain:
                out += emit(x, ind)
            if branches:
                out.append(ind + "switch (m->%s) {" % cname(mx.name))
                for k in branches:
                    out.append(ind + "case %d:" % k)
                    for x in m.signals:
                        if x.mux_id == k:
                            out += emit(x, ind + "    ")
                    out.append(ind + "    break;")
                out.append(ind + "default:")
                out.append(ind + "    break;")
                out.append(ind + "}")
            return out

        def emit_unpack(s, ind):
            parts = []
            for byte, lo, w, p in segments(s):
                e = "(d[%d] >> %d)" % (byte, lo) if lo else "d[%d]" % byte
                if lo + w < 8:
                    e = "(%s & 0x%02Xu)" % (e, (1 << w) - 1)
                parts.append("(uint32_t)%s << %d" % (e, p) if p else "(uint32_t)%s" % e)
            out = [ind + "v = " + " | ".join(parts) + ";"]
            if s.signed:
                out.append(ind + "if (v & 0x%Xu) v |= 0x%Xu;   /* знак: дополнение до двух */"
                           % (1 << (s.length - 1), (0xFFFFFFFF << s.length) & 0xFFFFFFFF))
            out.append(ind + "m->%s = (%s)v;" % (cname(s.name), ctype(s)))
            return out

        def emit_pack(s, ind):
            out = [ind + "v = (uint32_t)(%s)m->%s;" % (utype(s), cname(s.name))]
            for byte, lo, w, p in segments(s):
                e = "(v >> %d)" % p if p else "v"
                if w < 8:
                    e = "(%s & 0x%02Xu)" % (e, (1 << w) - 1)
                out.append(ind + "d[%d] |= (uint8_t)%s;" % (byte, "(%s << %d)" % (e, lo) if lo else e))
            return out

        C.append("void %s_unpack(%s *m, const uint8_t *d)" % (f, t))
        C += ["{", "    uint32_t v;"] + body(emit_unpack, "    ") + ["}", ""]
        C.append("void %s_pack(uint8_t *d, const %s *m)" % (f, t))
        C += ["{", "    uint32_t v;", "    memset(d, 0, %d);" % m.dlc] + body(emit_pack, "    ") + ["}", ""]
    H.append("#endif")
    return "\n".join(H) + "\n", "\n".join(C)


def gen_dump(prefix, msgs, names):
    """Только для проверки на ПК: распаковать, напечатать сигналы, упаковать обратно."""
    C = []
    for nm in names:
        m = next(x for x in msgs.values() if x.name == nm)
        f = "%s_%s" % (prefix, cname(m.name))
        C.append("static void %s_dump(const uint8_t *d)" % f)
        C.append("{")
        C.append("    %s_t m; uint8_t o[8]; int i;" % f)
        C.append("    memset(&m, 0, sizeof m);")
        C.append("    %s_unpack(&m, d);" % f)
        C.append('    printf("%s");' % m.name)
        mx = m.mux
        for s in m.signals:
            cond = "1" if s.mux_id is None else "m.%s == %d" % (cname(mx.name), s.mux_id)
            C.append('    if (%s) printf(" %s=%%ld:%%.10g", (long)m.%s, %s_%s_phys(m.%s));'
                     % (cond, s.name, cname(s.name), f, cname(s.name), cname(s.name)))
        C.append("    %s_pack(o, &m);" % f)
        C.append('    printf(" |");')
        C.append('    for (i = 0; i < %d; i++) printf("%%02X", o[i]);' % m.dlc)
        C.append('    printf("\\n");')
        C.append("}")
        C.append("")
    return "\n".join(C) + "\n"


def py_dump(msg, data):
    """То же, что печатает gen_dump, но независимым путём decode_raw/encode_raw."""
    vals = decode(msg, data)
    out = msg.name
    for s in msg.signals:
        if s.name in vals:
            out += " %s=%d:%s" % (s.name, vals[s.name], "%.10g" % phys(s, vals[s.name]))
    o = bytearray(msg.dlc)
    for s in active(msg, data):
        encode_raw(s, vals[s.name], o)
    return out + " |" + o.hex().upper()

dbc19e_mh.h — пример сгенерированного заголовка.

/* Сгенерировано dbcmini.py (раздел 19е) — не править руками. */
#ifndef MH_H
#define MH_H
#include <stdint.h>

#ifndef DBC_REAL
#define DBC_REAL double
#endif

#define MH_EXAMPLEMESSAGE_ID 0x1F0u
#define MH_EXAMPLEMESSAGE_DLC 8u
#define MH_EXAMPLEMESSAGE_CYCLE_MS 0u
typedef struct {
    int16_t temperature;
    uint8_t averageradius;
    uint8_t enable;
} mh_examplemessage_t;
void mh_examplemessage_unpack(mh_examplemessage_t *m, const uint8_t *d);
void mh_examplemessage_pack(uint8_t *d, const mh_examplemessage_t *m);
static inline DBC_REAL mh_examplemessage_temperature_phys(int16_t raw) { return (DBC_REAL)raw * (DBC_REAL)0.01 + (DBC_REAL)250.0; }
static inline DBC_REAL mh_examplemessage_averageradius_phys(uint8_t raw) { return (DBC_REAL)raw * (DBC_REAL)0.1 + (DBC_REAL)0.0; }
static inline DBC_REAL mh_examplemessage_enable_phys(uint8_t raw) { return (DBC_REAL)raw * (DBC_REAL)1.0 + (DBC_REAL)0.0; }

#endif

Дальше: раздел 19ж «USB-CAN адаптер и SocketCAN: candleLight, can-utils и стенд для автодиагностики» — как компьютер читает ту же шину через адаптер на STM32 и can-utils.

Открыть в интерактивном справочнике