"""Validate AMI.dbc against source layouts, firmware IDs and known CAN payloads. Run: python -m pip install -r scripts/requirements.txt python scripts/verify_dbc.py [--sdc-dbc path/to/SDC_ACQ_DAC.dbc] """ import argparse import json import math from pathlib import Path import re import cantools from export_dbc_docs import can2_dbc_text ROOT = Path(__file__).resolve().parents[1] def verify_bit_timing(): ioc = dict(line.split('=', 1) for line in (ROOT / 'firmware/AMI_Controller.ioc').read_text().splitlines() if '=' in line) source = (ROOT / 'firmware/Core/Src/fdcan.c').read_text() clock_source = (ROOT / 'firmware/Core/Src/main.c').read_text() for setting in ('RCC_HSI_DIV1', 'RCC_PLLM_DIV1', 'RCC_PLLR_DIV4', 'RCC_HCLK_DIV1'): assert setting in clock_source assert re.search(r'PLL\.PLLN\s*=\s*16;', clock_source) assert 'RCC_FDCANCLKSOURCE_PCLK1' in source clock = int(ioc['RCC.FDCANFreq_Value']) assert clock == 16000000 * 16 // 4 == int(ioc['RCC.APBFreq_Value']) result = {} for port, expected_rate, expected_prescaler in ((1, 500000, 8), (2, 250000, 16)): def value(field): found = re.search(rf'hfdcan{port}\.Init\.{field}\s*=\s*(\w+);', source) assert found, field return found.group(1) assert value('ClockDivider') == 'FDCAN_CLOCK_DIV1' assert value('FrameFormat') == 'FDCAN_FRAME_CLASSIC' prescaler, seg1, seg2, sjw = (int(value(field)) for field in ('NominalPrescaler', 'NominalTimeSeg1', 'NominalTimeSeg2', 'NominalSyncJumpWidth')) assert (prescaler, seg1, seg2, sjw) == (expected_prescaler, 13, 2, 1) rate = clock // prescaler // (1 + seg1 + seg2) near((1 + seg1) / (1 + seg1 + seg2), 0.875) assert rate == expected_rate == int(ioc[f'FDCAN{port}.CalculateBaudRateNominal']) for field in ('NominalPrescaler', 'NominalTimeSeg1', 'NominalTimeSeg2', 'DataPrescaler', 'DataTimeSeg1', 'DataTimeSeg2'): assert int(value(field)) == int(ioc[f'FDCAN{port}.{field}']) near(float(ioc[f'FDCAN{port}.CalculateTimeBitNominal']), 1e9 / rate) near(float(ioc[f'FDCAN{port}.CalculateTimeQuantumNominal']), 1e9 * prescaler / clock) result[f'CAN{port}'] = rate return result def near(actual, expected): assert math.isclose(actual, expected, rel_tol=0, abs_tol=1e-8), (actual, expected) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--sdc-dbc', type=Path) args = parser.parse_args() db = cantools.database.load_file(ROOT / 'AMI.dbc', strict=True) db2 = cantools.database.load_file(ROOT / 'AMI_CAN2_250k.dbc', strict=True) assert (ROOT / 'AMI_CAN2_250k.dbc').read_text(encoding='ascii') == can2_dbc_text() assert db2.dbc.attributes['Baudrate'].value == 250000 assert db.dbc.attributes['AMIPort'].value == 'CAN1' assert db2.dbc.attributes['AMIPort'].value == 'CAN2' bitrates = verify_bit_timing() header = (ROOT / 'firmware/Core/Inc/ami_protocol.h').read_text(encoding='utf-8') identifiers = {int(value, 16) for value in re.findall( r'^#define\s+AMI_CAN_ID_\w+\s+(0x[\dA-Fa-f]+)', header, re.MULTILINE)} expected = {0x300:(False,8), 0x6A0:(False,8), 0x6A1:(False,2), 0x6A2:(False,8), 0x6A4:(False,4), 0x6A6:(False,8), 0x140:(False,8), 0x156:(False,8), 0x186040F3:(True,8), 0x186140F3:(True,8)} actual = {m.frame_id:(m.is_extended_frame,m.length) for m in db.messages} assert actual == expected, actual assert set(actual) == identifiers, (actual, identifiers) assert len(db.messages) == 10 assert sum(len(m.signals) for m in db.messages) == 33 for m in db.messages: if m.frame_id == 0x300: assert m.senders == ['AMI'] and m.receivers == {'VCU'} else: assert m.receivers == {'AMI'} assert m.senders == (['SDC_ACQ_DAC'] if 0x6A0 <= m.frame_id <= 0x6A6 else ['BMS']) assert db.dbc.attributes['Baudrate'].value == 500000 for ident, cycle in [(0x6A0,100), (0x6A1,500), (0x6A2,20)]: assert db.get_message_by_frame_id(ident).cycle_time == cycle for ident in [0x300,0x6A4,0x6A6,0x140,0x156,0x186040F3,0x186140F3]: assert db.get_message_by_frame_id(ident).cycle_time in (None,0) checks = [] def decode(ident, data): checks.append({'id':hex(ident),'data':data}) return db.decode_message(ident, bytes.fromhex(data), decode_choices=False) # Canonical request at 1000 ms, first sequence, task 2; all six task enums. task = decode(0x300,'01 02 00 01 E8 03 00 00') assert task == {'Protocol_Version':1,'Task_ID':2,'Request_Sequence':0, 'Request_Source':1,'MCU_Uptime':1000} request = db.get_message_by_frame_id(0x300) request2 = db2.get_message_by_frame_id(0x300) assert request.encode(task) == bytes.fromhex('01 02 00 01 E8 03 00 00') for task_id in range(1,7): for seq in (0,255): data=dict(task, Task_ID=task_id, Request_Sequence=seq, MCU_Uptime=0xffffffff) payload=request.encode(data) assert payload == bytes([1,task_id,seq,1,255,255,255,255]) assert request.decode(payload,decode_choices=False) == data assert request2.encode(data) == payload assert request2.decode(payload,decode_choices=False) == data assert set(request.get_signal_by_name('Task_ID').choices) == set(range(1,7)) for raw, volts in [('08 20',8.2),('0C 30',12.3),('88 31',12.68),('A4 38',14.5),('FF FF',65.535)]: near(decode(0x6A1,raw)['Input_Voltage'],volts) assert db.encode_message(0x6A1,{'Input_Voltage':volts}) == bytes.fromhex(raw) heartbeat=decode(0x6A0,'A1 01 34 12 E8 03 00 00') assert heartbeat['Board_Identifier']==0xA1 and heartbeat['Protocol_Version']==1 assert heartbeat['Heartbeat_Sequence']==0x1234 near(heartbeat['Uptime'],1.0) freq=decode(0x6A2,'32 00 64 00 96 00 C8 00') assert list(freq.values()) == [50,100,150,200] gpio=decode(0x6A6,'AA 55 FF 7F E8 03 00 00') assert gpio['SDC_Input_State']==0x55AA and gpio['SDC_Changed_Mask']==0x7fff near(gpio['Timestamp'],1.0) ack=decode(0x6A4,'01 01 C4 09') assert ack['DAC_Set_Result']==1 and ack['DAC_Select']==1 near(ack['DAC_Output_Voltage'],2.5) for ident in (0x140,0x186040F3): for raw, voltage, current in [ ('0F A0 14 03 00 00 00 05',400,12.3), ('10 73 10 D1 00 00 00 05',421.1,-69.5), ('02 21 13 88 64 48 01 05',54.5,0), ('00 00 00 00 00 00 00 00',0,-500), ('FF FF FF FF 00 00 00 00',6553.5,6053.5)]: values=decode(ident,raw) near(values['Pack_Voltage'],voltage) near(values['Pack_Current'],current) # Only bytes 0..3 are consumed; remaining bytes encode to zero. encoded=db.encode_message(ident,values) assert encoded[:4] == bytes.fromhex(raw)[:4] assert encoded[4:] == bytes(4) assert all(not sig.is_signed and sig.byte_order=='big_endian' for sig in db.get_message_by_frame_id(ident).signals) payload=bytes.fromhex('42 FF 43 47 37 0A 50 41') temperatures=decode(0x156,payload.hex()) assert list(temperatures.values()) == [26,215,27,31,15,-30,40,25] valid=[temperature for byte,temperature in zip(payload,temperatures.values()) if byte != 0xff] assert max(valid)==40 assert str(db.decode_message(0x156,payload)['Temperature_2']) == 'Invalid' all_invalid=db.decode_message(0x156,bytes([255])*8) assert all(str(value)=='Invalid' for value in all_invalid.values()) negative=decode(0x156,'00 01 0A FF FF FF FF FF') assert max(value for index,value in enumerate(negative.values()) if index < 3) == -30 for payload,maximum in [('10 04 0C E4 3F 17 00 7B',23), ('10 04 0C E4 64 17 00 55',60)]: assert decode(0x186140F3,payload)['Maximum_Cell_Temperature'] == maximum assert str(db.decode_message(0x186140F3,bytes.fromhex('00 00 00 00 FF 00 00 00'))[ 'Maximum_Cell_Temperature']) == 'Invalid' assert len(db.get_message_by_frame_id(0x186140F3).signals)==1 # Mandatory signal-overlap and frame-length checks are exercised by strict load. try: db.decode_message(0x6A1,bytes([0])) except cantools.database.DecodeError: pass else: raise AssertionError('Truncated ADC payload unexpectedly decoded') source_check=False if args.sdc_dbc: source=cantools.database.load_file(args.sdc_dbc,strict=True) for ident in (0x6A0,0x6A1,0x6A2,0x6A4,0x6A6): original=source.get_message_by_frame_id(ident) current=db.get_message_by_frame_id(ident) assert original.length==current.length and original.is_extended_frame==current.is_extended_frame for a,b in zip(original.signals,current.signals,strict=True): fields=('name','start','length','byte_order','is_signed','scale','offset','unit','minimum','maximum') assert all(getattr(a,key)==getattr(b,key) for key in fields),(ident,a,b) source_check=True for check in checks: ident = int(check['id'], 16) payload = bytes.fromhex(check['data']) for decode_choices in (False, True): assert db.decode_message(ident, payload, decode_choices=decode_choices) == \ db2.decode_message(ident, payload, decode_choices=decode_choices) print(json.dumps({'result':'passed','messages':len(db.messages),'signals':33, 'known_payload_checks':len(checks),'all_task_ids_and_counter_edges':True, 'ports_bitrate':bitrates,'both_dbc_layouts_and_payloads_match':True, 'firmware_and_ioc_bit_timing_match':True, 'firmware_ids_match':True,'sdc_source_layout_match':source_check, 'cantools_version':cantools.__version__},indent=2)) if __name__=='__main__': main()