"""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 ROOT = Path(__file__).resolve().parents[1] 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) 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) 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 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 print(json.dumps({'result':'passed','messages':len(db.messages),'signals':33, 'known_payload_checks':len(checks),'all_task_ids_and_counter_edges':True, 'firmware_ids_match':True,'sdc_source_layout_match':source_check, 'cantools_version':cantools.__version__},indent=2)) if __name__=='__main__': main()