79 lines
5.1 KiB
Python
79 lines
5.1 KiB
Python
exec(open('tmp/accept_as_hardware.py',encoding='utf-8').read().split('with can.Bus(')[0])
|
|
import numpy as np,pyautogui as pg
|
|
from pywinauto import Desktop
|
|
W=461358
|
|
w=Desktop(backend='win32').window(handle=W);u=Desktop(backend='uia').window(handle=W)
|
|
w.set_focus()
|
|
u.child_window(title='开始',control_type='Hyperlink').invoke();time.sleep(.4)
|
|
canvas=w.child_window(handle=330296).rectangle()
|
|
stop=threading.Event();payload=None;errors=[];checks=[];tx=[];frames=[]
|
|
with can.Bus(interface='pcan',channel='PCAN_USBBUS1',bitrate=500000) as bus:
|
|
def sender():
|
|
try:
|
|
while not stop.is_set():
|
|
msg=payload
|
|
if msg:bus.send(msg,timeout=.15);tx.append(dict(id=hex(msg.arbitration_id),data=msg.data.hex(' ')))
|
|
stop.wait(.1)
|
|
except Exception as e:errors.append(repr(e))
|
|
def receiver():
|
|
try:
|
|
while not stop.is_set():
|
|
m=bus.recv(.1)
|
|
if m:frames.append(dict(id=hex(m.arbitration_id),data=m.data.hex(' '),timestamp=m.timestamp))
|
|
except Exception as e:errors.append(repr(e))
|
|
ts=[threading.Thread(target=fn) for fn in (sender,receiver)]
|
|
for t in ts:t.start()
|
|
def feedback(state,flags=15,version=2):
|
|
global payload
|
|
payload=can.Message(arbitration_id=0x183,is_extended_id=False,data=bytes([2,1,1,flags,0,state,0,version]))
|
|
def shotcheck(name,color,target=None):
|
|
time.sleep(1.3)
|
|
if target is not None:
|
|
snap=memory('hmi-'+name);assert snap['states']==target,(name,snap,target)
|
|
else:snap=None
|
|
im=pg.screenshot(region=(canvas.left,canvas.top,1200,720));im.save(str(O/('can-uart-hmi-'+name+'.png')))
|
|
a=np.asarray(im)[int(379*1.5):int(432*1.5),int(330*1.5):int(504*1.5),:3].astype(np.int16)
|
|
r,g,b=a[:,:,0],a[:,:,1],a[:,:,2]
|
|
masks={'gray':(r>75)&(r<180)&(abs(r-g)<8)&(abs(r-b)<8),
|
|
'white':(r>210)&(g>210)&(b>210),'yellow':(r>180)&(g>180)&(b<80),
|
|
'blue':(b>180)&(g<80)&(r<80),'red':(r>180)&(g<80)&(b<80),
|
|
'green':(g>180)&(r<80)&(b<80)}
|
|
count=int(masks[color].sum());assert count>30,(name,color,count)
|
|
assert bus.status()==0 and not errors
|
|
checks.append(dict(name=name,color=color,pixels=count,ram=snap,passed=True));print(name,'display passed',flush=True)
|
|
try:
|
|
for state,label,color in [(0,'off','white'),(1,'ready','yellow'),(2,'driving','yellow'),(3,'finished','blue'),(4,'emergency','red'),(5,'manual','green')]:
|
|
feedback(state);shotcheck(label,color,[2,1,1,state])
|
|
feedback(1,7);shotcheck('as-invalid','gray',[2,1,1,6])
|
|
feedback(1);shotcheck('before-legacy','yellow',[2,1,1,1])
|
|
feedback(0,7,1);shotcheck('legacy-unknown','gray',[2,1,1,6])
|
|
feedback(1);shotcheck('before-timeout','yellow',[2,1,1,1])
|
|
payload=can.Message(arbitration_id=0x6a0,is_extended_id=False,data=bytes.fromhex('A1 01 00 00 00 00 00 00'))
|
|
shotcheck('timeout-other-traffic','gray',[7,0,0,6])
|
|
feedback(1);shotcheck('recovery','yellow',[2,1,1,1])
|
|
# Actual simulated touch; all displayed feedback still comes through physical UART.
|
|
def touch(x,y):
|
|
pg.moveTo(canvas.left+x*1.5,canvas.top+y*1.5);pg.mouseDown();time.sleep(.15);pg.mouseUp();time.sleep(.2)
|
|
props=json.loads((R/'docs/evidence/2026-10-07/tjc8048-migration/properties-verified.json').read_text(encoding='utf-8'))
|
|
def at(page,component):
|
|
p=next(p for k,p in props[page].items() if k.startswith(component+'('))
|
|
touch(int(p['x'])+int(p['w'])/2,int(p['y'])+int(p['h'])/2)
|
|
at('dash','debug_entry');feedback(4);time.sleep(.8);at('debug','b0')
|
|
shotcheck('cross-page-latest-emergency','red',[2,1,1,4])
|
|
feedback(1);shotcheck('before-touch-request','yellow',[2,1,1,1])
|
|
before=len(frames);at('dash','bt4');at('CONFIRM','b0')
|
|
shotcheck('touch-request-keeps-as-ready','yellow',[2,1,1,1])
|
|
request=next((m for m in frames[before:] if m['id']=='0x300' and bytes.fromhex(m['data'])[1]==5),None)
|
|
assert request,('touch -> physical UART RX -> CAN request absent',frames[before:]);checks[-1]['request']=request
|
|
payload=None;shotcheck('final-unknown','gray',[7,0,0,6])
|
|
finally:
|
|
payload=None;stop.set()
|
|
for t in ts:t.join(timeout=2)
|
|
logs={str(c.handle):c.window_text() for c in w.descendants() if 'RichEdit' in c.class_name()}
|
|
(O/'can-uart-hmi-serial-panel.json').write_text(json.dumps(logs,ensure_ascii=False,indent=2),encoding='utf-8')
|
|
result=dict(result='passed' if len(checks)==16 and not errors else 'partial',scope='physical CAN1 -> MCU RAM -> USART2/COM6 -> official HMI simulator; touch UART RX -> real CAN1 request; no physical screen download',checks=checks,errors=errors)
|
|
(O/'can-uart-hmi-results.json').write_text(json.dumps(result,ensure_ascii=False,indent=2),encoding='utf-8')
|
|
(O/'can-uart-hmi-can.json').write_text(json.dumps(dict(sent=tx,received=frames),indent=2),encoding='utf-8')
|
|
u.child_window(title='停止',control_type='Hyperlink').invoke()
|
|
print('CAN/UART/HMI checks completed:',len(checks),flush=True)
|