feat: add VCU status feedback and map DEBUG safety inputs

2026-10-03 21:52:14 +08:00
parent dbbda5624a
commit 1e7b2e7f96
305 changed files with 32906 additions and 68 deletions
@@ -0,0 +1,89 @@
+import json,os,threading,time
+from pathlib import Path
+import numpy as np
+import pyautogui as pg
+import win32gui,win32con
+from pywinauto import Desktop
+from pywinauto.controls.win32_controls import ComboBoxWrapper
+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
+W=4262948
+w=Desktop(backend='win32').window(handle=W);u=Desktop(backend='uia').window(handle=W);w.set_focus()
+if not u.child_window(title='停止',control_type='Hyperlink').exists():
+    w.child_window(title='用户MCU输入').click();time.sleep(.3)
+    for c in w.descendants():
+        if 'COMBOBOX' in c.class_name() and c.is_visible():
+            combo=ComboBoxWrapper(c.element_info);items=combo.item_texts()
+            if '9600' in items:combo.select('9600')
+            if 'COM4' in items:combo.select('COM4')
+    u.child_window(title='开始',control_type='Hyperlink').invoke();time.sleep(6)
+canvas_control=next(c for c in w.children() if c.rectangle().width()==720 and c.rectangle().height()==408)
+canvas=canvas_control.rectangle()
+def touch(x,y):
+    point=(y<<16)|x
+    win32gui.SendMessage(canvas_control.handle,win32con.WM_LBUTTONDOWN,win32con.MK_LBUTTON,point)
+    time.sleep(.15)
+    win32gui.SendMessage(canvas_control.handle,win32con.WM_LBUTTONUP,0,point)
+ROIS=[(185,275,445,313),(185,328,445,364),(472,349,705,394)]
+checks=[];tx=[];errors=[];stop=threading.Event();payload=None
+def check(name,colors,delay=1.2):
+    time.sleep(delay)
+    shot=pg.screenshot(region=(canvas.left,canvas.top,720,408));shot.save(str(R/f'can-{name}.png'))
+    pixels=np.asarray(shot);counts=[]
+    for (x1,y1,x2,y2),color in zip(ROIS,colors):
+        a=pixels[y1:y2,x1:x2,:3].astype(np.int16);r,g,b=a[:,:,0],a[:,:,1],a[:,:,2]
+        c=[int(((r>75)&(r<180)&(abs(r-g)<8)&(abs(r-b)<8)).sum()),int(((g>180)&(r<80)&(b<80)).sum()),int(((r>180)&(g<80)&(b<80)).sum())]
+        assert c[color]>20 and (color==1 or c[1]==0) and (color==2 or c[2]==0),(name,colors,c)
+        counts.append(c)
+    assert bus.status()==0 and not errors,(name,errors)
+    checks.append({'name':name,'colors':colors,'pixel_counts':counts,'passed':True,'screenshot':f'can-{name}.png'})
+    print(name,'passed',flush=True)
+with can.Bus(interface='pcan',channel='PCAN_USBBUS1',bitrate=500000) as bus:
+    assert bus.reset()
+    def sender():
+        try:
+            while not stop.is_set():
+                msg=payload
+                if msg:
+                    bus.send(msg,timeout=.2);tx.append({'id':hex(msg.arbitration_id),'data':msg.data.hex(' ')})
+                stop.wait(.1)
+        except Exception as exc:errors.append(repr(exc))
+    t=threading.Thread(target=sender);t.start()
+    def feedback(task,locked,normal,flags=7):
+        global payload
+        payload=can.Message(arbitration_id=0x183,is_extended_id=False,data=bytes([task,locked,normal,flags,0,0,0,1]))
+    try:
+        check('initial-unknown',[0,0,0])
+        for task in range(7):
+            feedback(task,1,1);check(f'task-{task}',[2 if task==0 else 1,1,1])
+        feedback(2,0,0);check('unlocked-ebs-fault',[1,2,2])
+        feedback(2,1,1,5);check('lock-invalid',[1,0,1])
+        feedback(2,1,1,3);check('ebs-invalid',[1,1,0])
+        feedback(2,1,1,6);check('task-invalid',[0,1,1])
+        feedback(2,1,1);check('before-EBS-request',[1,1,1])
+        while bus.recv(0):pass
+        # Actual mouse press/release invokes the real task and confirm events.
+        touch(75,292);time.sleep(.3)
+        pg.screenshot(region=(canvas.left,canvas.top,720,408)).save(str(R/'can-confirm-EBS-request.png'))
+        print('CONFIRM entered; locate confirm control from screenshot',flush=True)
+        # CONFIRM buttons are placed in the lower half of the page.
+        touch(180,323)
+        request=None;end=time.monotonic()+1
+        while time.monotonic()<end:
+            m=bus.recv(.1)
+            if m and m.arbitration_id==0x300:
+                request={'id':'0x300','data':m.data.hex(' ')};break
+        assert request and bytes.fromhex(request['data'])[1]==5,request
+        check('after-EBS-request-task-stays2',[1,1,1])
+        feedback(6,0,0);check('VCU-changes-task6',[1,2,2])
+        payload=can.Message(arbitration_id=0x6a0,is_extended_id=False,data=bytes.fromhex('A1 01 00 00 00 00 00 00'))
+        check('VCU-timeout-other-traffic',[0,0,0],1.4)
+        feedback(3,1,1);check('feedback-recovery-task3',[1,1,1])
+        payload=None;check('final-timeout',[0,0,0],1.4)
+    finally:
+        stop.set();t.join()
+counters=[{'id':c.element_info.automation_id,'text':c.window_text()} for c in u.descendants(control_type='Text') if c.element_info.automation_id.startswith('datalabel')]
+(R/'can-display-results.json').write_text(json.dumps({'scope':'Physical CAN1 PCAN -> STM32 -> USART2 COM4 -> official HMI simulator, not physical panel','checks':checks,'tx':tx,'errors':errors,'counters':counters,'request':request,'passed':True},ensure_ascii=False,indent=2),encoding='utf-8')
+print(len(checks),'physical CAN/UART/display checks passed',counters,flush=True)