215 lines
9.1 KiB
Python
215 lines
9.1 KiB
Python
"""Execute compiled STM32 firmware in Unicorn; HAL I/O is mocked, no hardware used.
|
|
|
|
Requires scripts/requirements-test.txt and a built Debug or Release ELF.
|
|
Tests CAN RX -> display state -> ASCII UART DMA commands using actual ARM code.
|
|
"""
|
|
import argparse
|
|
from decimal import Decimal, ROUND_HALF_UP
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import struct
|
|
|
|
from elftools.elf.elffile import ELFFile
|
|
from unicorn import Uc, UC_ARCH_ARM, UC_MODE_THUMB, UC_MODE_MCLASS, UC_HOOK_CODE
|
|
from unicorn.arm_const import (
|
|
UC_ARM_REG_R0, UC_ARM_REG_R1, UC_ARM_REG_R2, UC_ARM_REG_R3,
|
|
UC_ARM_REG_SP, UC_ARM_REG_LR, UC_ARM_REG_PC, UC_ARM_REG_XPSR,
|
|
)
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
REGS = (UC_ARM_REG_R0, UC_ARM_REG_R1, UC_ARM_REG_R2, UC_ARM_REG_R3)
|
|
|
|
|
|
class Firmware:
|
|
def __init__(self, path):
|
|
self.uc = Uc(UC_ARCH_ARM, UC_MODE_THUMB | UC_MODE_MCLASS)
|
|
self.uc.mem_map(0x08000000, 0x20000)
|
|
self.uc.mem_map(0x20000000, 0x24000)
|
|
self.uc.mem_map(0x40000000, 0x30000)
|
|
self.uc.mem_map(0x50000000, 0x10000)
|
|
with path.open('rb') as file:
|
|
elf = ELFFile(file)
|
|
self.symbols = {s.name: s['st_value'] for s in elf.get_section_by_name('.symtab').iter_symbols()}
|
|
for segment in elf.iter_segments():
|
|
if segment['p_type'] == 'PT_LOAD' and segment['p_filesz']:
|
|
self.uc.mem_write(segment['p_vaddr'], segment.data())
|
|
self.now = 800
|
|
self.rx = []
|
|
self.commands = []
|
|
self.hooks = {self.symbols[name] & ~1: name for name in (
|
|
'HAL_GetTick', 'HAL_GPIO_WritePin', 'HAL_GPIO_TogglePin',
|
|
'HAL_FDCAN_GetRxFifoFillLevel', 'HAL_FDCAN_GetRxMessage',
|
|
'HAL_FDCAN_GetTxFifoFreeLevel', 'HAL_UART_Transmit_DMA',
|
|
)}
|
|
self.uc.hook_add(UC_HOOK_CODE, self.hook)
|
|
for port, instance in ((1,0x40006400), (2,0x40006800)):
|
|
self.write_u32(f'hfdcan{port}', instance)
|
|
|
|
def hook(self, uc, address, size, data):
|
|
name = self.hooks.get(address)
|
|
if name is None:
|
|
return
|
|
result = 0
|
|
if name == 'HAL_GetTick':
|
|
result = self.now
|
|
elif name == 'HAL_FDCAN_GetRxFifoFillLevel':
|
|
result = len(self.rx)
|
|
elif name == 'HAL_FDCAN_GetRxMessage':
|
|
header, payload = self.rx.pop(0)
|
|
uc.mem_write(uc.reg_read(UC_ARM_REG_R2), header)
|
|
uc.mem_write(uc.reg_read(UC_ARM_REG_R3), payload)
|
|
elif name == 'HAL_UART_Transmit_DMA':
|
|
command = bytes(uc.mem_read(uc.reg_read(UC_ARM_REG_R1), uc.reg_read(UC_ARM_REG_R2)))
|
|
assert command.endswith(b'\xff\xff\xff'), command
|
|
self.commands.append(command[:-3].decode('ascii'))
|
|
# TX free level is zero: heartbeat never touches hardware in this test.
|
|
uc.reg_write(UC_ARM_REG_R0, result)
|
|
uc.reg_write(UC_ARM_REG_PC, uc.reg_read(UC_ARM_REG_LR))
|
|
|
|
def call(self, name, *args):
|
|
for register, value in zip(REGS, args):
|
|
self.uc.reg_write(register, value)
|
|
self.uc.reg_write(UC_ARM_REG_SP, 0x20023000)
|
|
self.uc.reg_write(UC_ARM_REG_LR, 0x0801FF01)
|
|
self.uc.reg_write(UC_ARM_REG_XPSR, 1 << 24)
|
|
self.uc.emu_start(self.symbols[name] | 1, 0x0801FF00, count=200000)
|
|
assert self.uc.reg_read(UC_ARM_REG_PC) == 0x0801FF00, f'{name} exceeded instruction budget'
|
|
return self.uc.reg_read(UC_ARM_REG_R0)
|
|
|
|
def read_u32(self, name, offset=0):
|
|
return struct.unpack('<I',self.uc.mem_read(self.symbols[name]+offset,4))[0]
|
|
|
|
def write_u32(self, name, value):
|
|
self.uc.mem_write(self.symbols[name],struct.pack('<I',value))
|
|
|
|
def valid(self):
|
|
return bool(self.uc.mem_read(self.symbols['display_state']+32,1)[0] & 1)
|
|
|
|
def speed(self):
|
|
return self.read_u32('display_state')
|
|
|
|
def receive(self, payload, port=1, ident=0x6A2, id_type=0, remote=0, fd=0, dlc=None):
|
|
dlc = len(payload) if dlc is None else dlc
|
|
header = struct.pack('<10I',ident,id_type,remote,dlc,0,0,fd,0,0,0)
|
|
self.rx.append((header,payload))
|
|
self.call('HAL_FDCAN_RxFifo0Callback',self.symbols[f'hfdcan{port}'],1)
|
|
assert not self.rx
|
|
|
|
def service(self, now=None):
|
|
if now is not None:
|
|
self.now = now & 0xFFFFFFFF
|
|
self.call('AMI_App_Service')
|
|
# Model the completion of the UART DMA transfer, without real UART I/O.
|
|
self.uc.mem_write(self.symbols['screen_tx_busy'],b'\0')
|
|
|
|
def drain(self):
|
|
for _ in range(40):
|
|
self.service()
|
|
|
|
|
|
def verify(path):
|
|
cases = []
|
|
f = Firmware(path)
|
|
f.write_u32('display_dirty',0xFFFFFFFF)
|
|
f.drain()
|
|
assert not f.valid() and f.speed() == 0
|
|
assert 'dash.speed.val=0' in f.commands and 'dash.speed.pco=16904' in f.commands
|
|
cases.append('boot unknown speed: grey zero, UART terminator verified')
|
|
|
|
samples = [(0,0), (10000,11000), (11000,10000), (0,20000),
|
|
(1249,1250), (1250,1250), (1250,1251), (65535,65535), (0,65535)]
|
|
for port in (1,2):
|
|
for left,right in samples:
|
|
f.now = 1000
|
|
f.commands.clear()
|
|
f.receive(struct.pack('<HH',left,right),port=port)
|
|
expected = int(((Decimal(left)+Decimal(right))/2000*Decimal('3.6'))
|
|
.quantize(Decimal('1'),rounding=ROUND_HALF_UP))
|
|
assert f.valid() and f.speed() == expected,(port,left,right,f.speed(),expected)
|
|
raw = struct.unpack('<HH',f.uc.mem_read(f.symbols['sdc_wheel_speed_mm_s'],4))
|
|
assert raw == (left,right)
|
|
assert f.read_u32('can_port_last_rx_ms',4*(port-1)) == f.now
|
|
f.drain()
|
|
# An unchanged display value may legitimately produce no UART command.
|
|
assert f'dash.speed.val={expected}' in f.commands or not any(
|
|
c.startswith('dash.speed.val=') for c in f.commands)
|
|
cases.append(f'CAN{port}: {len(samples)} zero, asymmetric, rounding and saturation vectors')
|
|
|
|
for kwargs,payload in [
|
|
({},bytes.fromhex('32 00 64 00 96 00 C8 00')),
|
|
({},bytes(3)), ({},bytes(5)), ({'dlc':9,'fd':0x00200000},bytes(12)),
|
|
({'id_type':0x40000000},bytes(4)),
|
|
({'remote':0x20000000},bytes(4)), ({'fd':0x00200000},bytes(4)),
|
|
({'ident':0x180},bytes(8)),
|
|
]:
|
|
f = Firmware(path)
|
|
f.now = 1000
|
|
f.receive(struct.pack('<HH',10000,11000))
|
|
snapshot = bytes(f.uc.mem_read(f.symbols['sdc_wheel_speed_mm_s'],4))
|
|
f.now = 1400
|
|
f.receive(payload,**kwargs)
|
|
assert f.speed() == 38 and f.valid()
|
|
assert f.read_u32('wheel_speed_last_rx_ms') == 1000
|
|
assert bytes(f.uc.mem_read(f.symbols['sdc_wheel_speed_mm_s'],4)) == snapshot
|
|
f.service(1500)
|
|
assert f.valid(), '500 ms boundary must still be valid'
|
|
f.service(1501)
|
|
assert not f.valid() and f.speed() == 38
|
|
f.drain()
|
|
assert 'dash.speed.pco=16904' in f.commands
|
|
cases.append('8 invalid DLC/format/legacy frames reject without overwriting or refreshing expiry')
|
|
|
|
f = Firmware(path)
|
|
f.now = 1000
|
|
f.receive(struct.pack('<HH',10000,11000))
|
|
f.drain()
|
|
assert 'dash.speed.val=38' in f.commands and 'dash.speed.pco=65535' in f.commands
|
|
f.now = 1501
|
|
for ident,payload in [(0x6A0,bytes.fromhex('A1 01 00 00 00 00 00 00')),
|
|
(0x6A6,bytes(8)), (0x6A1,bytes.fromhex('0C 30')),
|
|
(0x140,bytes(8)), (0x183,bytes.fromhex('02 01 01 07 00 00 00 01'))]:
|
|
f.receive(payload,ident=ident)
|
|
f.service()
|
|
assert not f.valid() and f.speed() == 38
|
|
f.now = 1600
|
|
f.commands.clear()
|
|
f.receive(struct.pack('<HH',10000,11000),port=2)
|
|
f.drain()
|
|
assert f.valid() and 'dash.speed.val=38' in f.commands and 'dash.speed.pco=65535' in f.commands
|
|
f.receive(bytes(4))
|
|
f.drain()
|
|
assert f.valid() and f.speed() == 0 and 'dash.speed.val=0' in f.commands
|
|
cases.append('unrelated SDC/BMS/VCU traffic cannot renew; same-value recovery and white zero work')
|
|
|
|
for started,boundary,expired in ((0,500,501),(0xFFFFFF00,244,245)):
|
|
f = Firmware(path)
|
|
f.now = started
|
|
f.receive(struct.pack('<HH',10000,11000))
|
|
f.service(boundary)
|
|
assert f.valid()
|
|
f.service(expired)
|
|
assert not f.valid() and f.speed() == 38
|
|
cases.append('tick zero and uint32 wrap: independent timeout boundary works')
|
|
return {'elf':str(path.relative_to(ROOT)), 'sha256':hashlib.sha256(path.read_bytes()).hexdigest(),
|
|
'result':'passed','checks':cases,'evidence':'ARM emulation with mocked HAL, not physical CAN/HMI'}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--elf',type=Path,action='append')
|
|
parser.add_argument('--output',type=Path)
|
|
args = parser.parse_args()
|
|
paths = args.elf or [ROOT/'firmware/build/Debug/AMI_Controller.elf',
|
|
ROOT/'firmware/build/Release/AMI_Controller.elf']
|
|
results = [verify(p.resolve()) for p in paths]
|
|
text = json.dumps(results,ensure_ascii=False,indent=2)
|
|
if args.output:
|
|
args.output.parent.mkdir(parents=True,exist_ok=True)
|
|
args.output.write_text(text+'\n',encoding='utf-8')
|
|
print(text)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|