112 lines
6.9 KiB
Python
112 lines
6.9 KiB
Python
import json,os,re,struct,subprocess,threading,time
|
|
from pathlib import Path
|
|
R=Path(__file__).resolve().parent
|
|
P=R.parents[1]
|
|
DLL=P/'tmp/can1-test-20261002/pcan-basic/x64'
|
|
os.environ['PATH']=str(DLL)+os.pathsep+os.environ['PATH']
|
|
dh=os.add_dll_directory(str(DLL))
|
|
import can,serial
|
|
CLI=r'D:\STM32CubeCLT_1.19.0\STM32CubeProgrammer\bin\STM32_Programmer_CLI.exe'
|
|
NM=r'D:\STM32CubeCLT_1.19.0\GNU-tools-for-STM32\bin\arm-none-eabi-nm.exe'
|
|
nm=subprocess.check_output([NM,'-n',str(P/'firmware/build/Release/AMI_Controller.elf')]).decode().replace('\r','')
|
|
S={n:int(a,16) for a,k,n in re.findall(r'^(\w+)\s+(\w)\s+(\w+)$',nm,re.M)}
|
|
(R/'symbols.json').write_text(json.dumps(S,indent=2))
|
|
names=['vcu_display_state','vcu_status_last_rx_ms','vcu_status_received','display_state','can_rx_error_count','uart_rx_error_count','screen_tx_busy']
|
|
base=min(S[n] for n in names)&~3
|
|
length=((max(S[n]+(56 if n=='display_state' else 4) for n in names)-base+3)//4)*4
|
|
def memory(name):
|
|
p=subprocess.run([CLI,'-c','port=SWD','mode=HOTPLUG','freq=1000','-r32',hex(base),str(length)],capture_output=True,timeout=15)
|
|
(R/f'{name}-ram.log').write_bytes(p.stdout+p.stderr)
|
|
assert p.returncode==0
|
|
mem={}
|
|
for a,words in re.findall(r'0x([0-9A-Fa-f]+)\s*:\s*((?:[0-9A-Fa-f]{8}\s*)+)',p.stdout.decode(errors='replace')):
|
|
a=int(a,16)
|
|
for word in words.split():
|
|
for i,b in enumerate(struct.pack('<I',int(word,16))):mem[a+i]=b
|
|
a+=4
|
|
def val(n,l=4):return int.from_bytes(bytes(mem[S[n]+i] for i in range(l)),'little')
|
|
return {'states':list(bytes(mem[S['vcu_display_state']+i] for i in range(3))),
|
|
'timestamp':val('vcu_status_last_rx_ms'),'received':val('vcu_status_received',1),
|
|
'can_errors':val('can_rx_error_count'),'uart_errors':val('uart_rx_error_count'),
|
|
'tx_busy':val('screen_tx_busy',1),
|
|
'debug':list(bytes(mem[S['display_state']+33+i] for i in range(16))),
|
|
'dash':list(struct.unpack('<8i',bytes(mem[S['display_state']+i] for i in range(32))))}
|
|
checks=[];errors=[];tx=[];uart=bytearray();halt=threading.Event();payload=None
|
|
with serial.Serial('COM4',9600,timeout=.05) as port,can.Bus(interface='pcan',channel='PCAN_USBBUS1',bitrate=500000) as bus:
|
|
assert bus.reset()
|
|
def reader():
|
|
while not halt.is_set():uart.extend(port.read(512))
|
|
def sender():
|
|
try:
|
|
while not halt.is_set():
|
|
msg=payload
|
|
if msg:
|
|
bus.send(msg,timeout=.2)
|
|
tx.append({'id':hex(msg.arbitration_id),'data':msg.data.hex(' '),'extended':msg.is_extended_id,'remote':msg.is_remote_frame})
|
|
halt.wait(.1)
|
|
except Exception as exc:errors.append(repr(exc))
|
|
threads=[threading.Thread(target=reader),threading.Thread(target=sender)]
|
|
for t in threads:t.start()
|
|
def set_payload(data,**kw):
|
|
global payload
|
|
payload=can.Message(arbitration_id=0x183,is_extended_id=False,data=bytes(data),**kw)
|
|
def check(name,target,delay=.2):
|
|
time.sleep(delay);values=memory(name)
|
|
assert values['states']==target,(name,values,target)
|
|
assert values['can_errors']==values['uart_errors']==0 and bus.status()==0 and not errors,(name,values,errors)
|
|
checks.append({'name':name,'expected':target,'snapshot':values,'passed':True})
|
|
print(name,'passed',flush=True)
|
|
return values
|
|
try:
|
|
check('boot_unknown',[7,0,0])
|
|
for task in range(7):
|
|
set_payload([task,task&1,(task+1)&1,7,task,0,0,1])
|
|
check(f'task_{task}',[task,1 if task&1 else 2,1 if (task+1)&1 else 2])
|
|
for flags in range(8):
|
|
set_payload([2,1,1,flags,255,0,0,1])
|
|
check(f'valid_mask_{flags}',[2 if flags&1 else 7,1 if flags&2 else 0,1 if flags&4 else 0])
|
|
set_payload([2,1,1,7,0,0,0,1]);check('counter_wrap',[2,1,1])
|
|
invalid=[]
|
|
for name,byte,value in [('task_range',0,7),('lock_range',1,2),('ebs_range',2,2),('flag_reserved',3,15),('reserved_5',5,1),('reserved_6',6,1),('version',7,2)]:
|
|
data=[3,0,0,7,1,0,0,1];data[byte]=value;invalid.append((name,data,{}))
|
|
invalid.extend([('short_dlc',[3,0,0,7,1,0,0],{}),('extended',[3,0,0,7,1,0,0,1],{'is_extended_id':True}),('remote',[],{'is_remote_frame':True,'dlc':8})])
|
|
for name,data,kw in invalid:
|
|
set_payload([2,1,1,7,0,0,0,1]);check(name+'_baseline',[2,1,1])
|
|
payload=None;time.sleep(.03)
|
|
before=memory(name+'_stopped_baseline')
|
|
payload=can.Message(arbitration_id=0x183,is_extended_id=kw.pop('is_extended_id',False),data=bytes(data),**kw)
|
|
# SWD reads can exceed the 500 ms lease. Repeated invalid traffic
|
|
# must not renew that lease, so verify expiry and unchanged stamp.
|
|
after=check(name+'_rejected_and_expired',[7,0,0],.65)
|
|
assert after['timestamp']==before['timestamp'],(name,before,after)
|
|
payload=can.Message(arbitration_id=0x6A0,is_extended_id=False,data=bytes.fromhex('A1 01 00 00 00 00 00 00'))
|
|
check('timeout_despite_other_can',[7,0,0],.8)
|
|
set_payload([6,0,0,7,1,0,0,1]);check('feedback_recovery',[6,2,2])
|
|
set_payload([2,1,1,7,2,0,0,1]);check('request_baseline',[2,1,1])
|
|
while bus.recv(0):pass
|
|
port.write(bytes.fromhex('AA 55 01 05 0D 0A'));port.flush()
|
|
request=None;end=time.monotonic()+.6
|
|
while time.monotonic()<end:
|
|
m=bus.recv(.1)
|
|
if m and m.arbitration_id==0x300:request={'id':'0x300','data':m.data.hex(' ')};break
|
|
check('task_request_does_not_set_task',[2,1,1])
|
|
print('Physical UART request reception:',request,flush=True)
|
|
bus.send(can.Message(arbitration_id=0x6a6,is_extended_id=False,data=struct.pack('<HHI',2,2,1000)))
|
|
time.sleep(.1);snap=memory('gpio_regression')
|
|
assert snap['debug'][5]==2 and snap['debug'][0]==1
|
|
bus.send(can.Message(arbitration_id=0x182,is_extended_id=False,data=bytes.fromhex('F4 01 C2 01 90 01 07 01')))
|
|
time.sleep(.1);snap=memory('drivetrain_regression')
|
|
assert [snap['dash'][i] for i in (2,4,5)]==[500,45,40]
|
|
checks.append({'name':'GPIO_and_drivetrain_regression','passed':True,'snapshot':snap})
|
|
payload=None;check('final_timeout',[7,0,0],.8)
|
|
# Include at least one complete periodic UART resync in the capture.
|
|
time.sleep(5)
|
|
finally:
|
|
halt.set()
|
|
for t in threads:t.join()
|
|
(R/'uart-capture.bin').write_bytes(uart)
|
|
commands=[x.decode('gb2312',errors='replace') for x in uart.split(b'\xff\xff\xff') if x]
|
|
report={'scope':'Physical PCAN CAN1 500k -> STM32 RAM plus COM4 capture, not panel display proof','checks':checks,'tx':tx,'errors':errors,'uart_bytes':len(uart),'uart_commands':commands,'physical_task_request':request,'mcu_passed':True}
|
|
(R/'mcu-results.json').write_text(json.dumps(report,ensure_ascii=False,indent=2),encoding='utf-8')
|
|
print(json.dumps({'checks':len(checks),'uart_bytes':len(uart),'physical_task_request':request,'mcu_passed':True}),flush=True)
|