148 lines
7.0 KiB
Python
148 lines
7.0 KiB
Python
"""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()
|