cheng
195 lines
10 KiB
Python
195 lines
10 KiB
Python
"""Exercise Qt and backend shutdown with no device transport in the UI."""
|
|
import sys
|
|
import os
|
|
import time
|
|
from station import runtime_client as runtime
|
|
runtime.start(offline=True)
|
|
import station.operator_app as ui
|
|
app=ui.QApplication.instance();app.setFont(ui.QFont('Noto Sans SC',11));app.setStyleSheet(ui._STYLE)
|
|
window=ui.StationApp();window.resize(1440,900);window.show()
|
|
window._apply_theme('dark',persist=False)
|
|
assert app.property('darkTheme') is True
|
|
assert window._theme_selector.currentData()=='dark'
|
|
window._apply_theme('light',persist=False)
|
|
assert app.property('darkTheme') is False
|
|
# A first status message and its expiry must not resize the dashboard.
|
|
app.processEvents()
|
|
content_before=window.centralWidget().geometry()
|
|
window.statusBar().showMessage('刷新相机 · 状态提示 '*30)
|
|
app.processEvents()
|
|
assert window.centralWidget().geometry()==content_before
|
|
window.statusBar().clearMessage();app.processEvents()
|
|
assert window.centralWidget().geometry()==content_before
|
|
assert window._theme_selector.currentData()=='light'
|
|
assert 'rclpy' not in sys.modules
|
|
assert 'serial' not in sys.modules
|
|
assert 'station.operator_console' not in sys.modules
|
|
assert set(window._pages) == {'main','calibration','settings','ros_monitor'}
|
|
# Offline display uses the configured HOME, while real feedback always wins.
|
|
assert window._home_poses[window._arm] != [0.]*7
|
|
assert window._arm_sketch._pose_key[0] == tuple(round(q,3) for q in window._home_poses[window._arm])
|
|
assert not window._arm_sketch._live
|
|
from unittest.mock import patch
|
|
live_snap=dict(window._snap)
|
|
live_snap['arm_live']={window._arm:{'pose_live':True,'q':[.12]*7}}
|
|
with patch.object(ui,'status',return_value=live_snap):
|
|
window._refresh()
|
|
assert window._arm_sketch._pose_key[0]==(.12,)*7
|
|
assert window._arm_sketch._live
|
|
window._refresh()
|
|
# Calibration follows arm/mode and shares the existing camera preview path.
|
|
from unittest.mock import patch
|
|
window._show_page('calibration')
|
|
window._snap['camera_ownership']=[{'id':cam['id'],'owned':True} for cam in ui.camera_view.targets()]
|
|
with patch.object(ui.camera_view, 'watch') as watch:
|
|
window._calibration_mode.setCurrentIndex(0)
|
|
window._sync_camera_watch()
|
|
wrist=window._calibration_camera()
|
|
assert wrist and wrist['arm']==window._arm
|
|
assert wrist['id']+'_color' in watch.call_args.args[0]
|
|
window._calibration_mode.setCurrentIndex(1)
|
|
outside=window._calibration_camera()
|
|
assert outside and not outside.get('arm')
|
|
assert outside['id']+'_color' in watch.call_args.args[0]
|
|
window._calibration_mode.setCurrentIndex(0)
|
|
window._show_page('main')
|
|
# A photo must preserve the current RGB frame; no capture/device operation is needed.
|
|
import tempfile
|
|
from unittest.mock import patch
|
|
sample=bytes([12,84,192])*(32*18)
|
|
selected=next(cam for _page,cam,_label in window._camera_labels if cam.get('arm')==window._arm)
|
|
with tempfile.TemporaryDirectory() as pictures:
|
|
with patch.object(ui.Path,'home',return_value=ui.Path(pictures)), patch.object(ui.camera_view,'previews',return_value={selected['id']+'_color':(sample,32,18)}):
|
|
window._take_photo()
|
|
photos=list(ui.Path(pictures).rglob('*.png'))
|
|
assert len(photos)==1
|
|
image=ui.QImage(str(photos[0]))
|
|
assert (image.width(),image.height())==(32,18)
|
|
assert image.pixelColor(0,0).getRgb()[:3]==(12,84,192)
|
|
window._open_camera(selected['id'])
|
|
assert selected['id'] in window._camera_windows
|
|
window._camera_windows[selected['id']].close()
|
|
assert not window._camera_windows
|
|
# Device controls only submit explicit actions; changing settings never writes to hardware.
|
|
assert not hasattr(window,'_grip_force') and not hasattr(window,'_grip_target')
|
|
assert window._gripper_probe_button.text()=='连接 / 探测'
|
|
assert window._sensor_refresh_button.text()=='刷新传感器'
|
|
assert window._sensor_reset_button.text()=='重置底图'
|
|
assert window._sensor_reset_button.size()==window._sensor_refresh_button.size()
|
|
assert window._gripper_probe_button.size()==window._sensor_refresh_button.size()
|
|
assert window._gripper_probe_button.height()==28
|
|
original_arm=window._arm
|
|
window._set_arm('91')
|
|
assert window._gripper_sketch._cad_static and window._gripper_sketch._gl_count>0
|
|
assert window._gripper_sketch._caption()==''
|
|
window._set_arm(original_arm)
|
|
assert not window._gripper_sketch._cad_static
|
|
assert window._gripper_sketch._caption()==''
|
|
from PySide6.QtCore import QByteArray,QBuffer,QIODevice
|
|
sample_image=ui.QImage(32,32,ui.QImage.Format.Format_RGB888);sample_image.fill(ui.QColor('#1683ff'))
|
|
encoded=QByteArray();buffer=QBuffer(encoded);buffer.open(QIODevice.OpenModeFlag.WriteOnly)
|
|
assert sample_image.save(buffer,'JPG');buffer.close()
|
|
fake_snap=dict(window._snap)
|
|
fake_snap['backend']={**fake_snap.get('backend',{}),'error':''}
|
|
fake_snap['arm_live']={original_arm:{'tcp_pose':[500.,120.,325.,179.2,.5,88.1]}}
|
|
fake_snap['tactile']={'left':{'connected':True,'fps':29.,'fnormal':12000.,'fshearx':20000.,'fsheary':-10000.,
|
|
'force_n':{'fnormal':.477,'fshearx':.65,'fsheary':-.31}}}
|
|
fake_snap['history_n']={'left':[{'t':time.time()-.1,'fnormal':.477,'fshearx':.65,'fsheary':-.31}]}
|
|
with patch.object(ui,'status',return_value=fake_snap),patch.object(ui.runtime.tactile_view,'frame',return_value=bytes(encoded)):
|
|
window._refresh();window._paint_images()
|
|
assert [v.text() for v in window._tcp_values]==['500.0','120.0','325.0','179.2','0.5','88.1']
|
|
assert '(29fps)' in window._flow_slots[0][0].text()
|
|
assert '·' not in window._flow_slots[0][0].text()
|
|
assert '当前法向力 0.48N' in window._flow_slots[0][0].text()
|
|
assert not hasattr(window,'_flow_details')
|
|
assert window._flow_axis_charts['left']['x']._samples[-1]['fshearx']==.65
|
|
assert window._flow_axis_charts['left']['y']._samples[-1]['fsheary']==-.31
|
|
assert window._flow_axis_charts['left']['x']._scale==window._flow_axis_charts['left']['y']._scale==10.
|
|
assert window._flow_axis_charts['left']['x'].MAX_SAMPLES==150
|
|
app.processEvents()
|
|
for triplet in window._flow_triplets.values():
|
|
rectangles=[item.geometry() for item in triplet.items]
|
|
assert all(rect.width()==rect.height() for rect in rectangles),rectangles
|
|
assert len({rect.width() for rect in rectangles})==1,rectangles
|
|
assert len({rect.y() for rect in rectangles})==1,rectangles
|
|
assert not window._flow_slots[0][1]._source.isNull()
|
|
# Recent samples are spread across a shorter window without altering source data.
|
|
chart=window._flow_axis_charts['left']['x']
|
|
history=[{'seq':i,'fshearx':.2,'fsheary':-.2} for i in range(1800)]
|
|
chart.set_samples(history)
|
|
assert len(chart._samples)==150 and chart._samples[0]['seq']==1650
|
|
assert history[-1]['fshearx']==.2
|
|
assert ui._display_force(.299)==ui._display_force(-.299)==0.
|
|
assert ui._display_force(.3)==.3 and ui._display_force(-.3)==-.3
|
|
assert window._flow_slots[0][0].textFormat()==ui.Qt.TextFormat.PlainText
|
|
title=window._flow_slots[0][0]
|
|
triplet=window._flow_triplets['left']
|
|
assert title.geometry().top()>=triplet.geometry().bottom()
|
|
fake_snap['history_n']['left'][-1]['fnormal']=.25
|
|
window._refresh();window._paint_images()
|
|
assert '当前法向力 0N' in title.text()
|
|
|
|
|
|
errors=[];ticks=[]
|
|
duration=float(os.environ.get("ORISYS_GUI_TEST_SECONDS","2.2"))
|
|
started=time.monotonic()
|
|
frames=[bytes([v,40,80])*(320*180) for v in (30,70,120)]
|
|
def tick():
|
|
ticks.append(time.monotonic())
|
|
if duration>10:
|
|
if len(ticks)%100==0:
|
|
window._show_page(("main","calibration")[(len(ticks)//100)%2])
|
|
for key in ui.camera_view._PREVIEW_WANT:
|
|
ui.camera_view._PREVIEW[key]=(frames[len(ticks)%3],320,180)
|
|
ui.camera_view._META[key]=(320,180,20.)
|
|
original=sys.excepthook
|
|
def fail(*args): errors.append(str(args[1]));original(*args)
|
|
sys.excepthook=fail
|
|
beat=ui.QTimer();beat.timeout.connect(tick);beat.start(50)
|
|
# The GUI must not serialize read-only requests behind a long arm operation.
|
|
parallel=[]
|
|
def fake_command(action,aid,*args,**kwargs):
|
|
parallel.append((action,time.monotonic()))
|
|
if action=='stage_track':
|
|
time.sleep(.65)
|
|
parallel.append(('motion_finished',time.monotonic()))
|
|
return {'ok':True,'message':'isolated test'}
|
|
ui.run_action=fake_command
|
|
ui.stop_started=lambda *args:(parallel.append(('stop',time.monotonic())),{'ok':True,'message':'isolated stop'})[1]
|
|
ui.QTimer.singleShot(220,lambda:window._ask('stage_track'))
|
|
ui.QTimer.singleShot(260,lambda:window._ask('handeye_show'))
|
|
ui.QTimer.singleShot(280,lambda:window._ask('stage_track'))
|
|
ui.QTimer.singleShot(300,window._stop_arm)
|
|
ui.QTimer.singleShot(200,lambda:window._start(lambda:(time.sleep(.7),{'ok':True,'message':'test'})[1],'test'))
|
|
ui.QTimer.singleShot(500,lambda:window._show_page('calibration'))
|
|
ui.QTimer.singleShot(1000,lambda:window._show_page('calibration'))
|
|
ui.QTimer.singleShot(1500,lambda:window._show_page('ros_monitor'))
|
|
ui.QTimer.singleShot(1900,lambda:window._show_page('main'))
|
|
ui.QTimer.singleShot(round(duration*1000),window.close)
|
|
ui.QTimer.singleShot(round((duration+15)*1000),app.quit)
|
|
app.exec()
|
|
assert len(ticks)>20,len(ticks)
|
|
assert [name for name,_ in parallel].count('stage_track')==1,parallel
|
|
assert dict(parallel)['handeye_show']<dict(parallel)['motion_finished'],parallel
|
|
assert dict(parallel)['stop']<dict(parallel)['motion_finished'],parallel
|
|
text = window._detail.toPlainText()
|
|
assert '[点击] 检测跟踪' in text,text
|
|
assert '[未重复发送]' in text,text
|
|
assert 'isolated test' in text,text
|
|
# Repeated refreshes cannot duplicate backend lifecycle events.
|
|
window._snap['operation_log']={'session':'log-test','events':[
|
|
{'seq':1,'time':time.time(),'action':'stage_track','arm':'89','task_id':'123456789','message':'收到操作请求'},
|
|
{'seq':2,'time':time.time(),'action':'stage_track','arm':'89','task_id':'123456789','message':'识别阶段'}]}
|
|
window._sync_operation_log();window._sync_operation_log()
|
|
assert window._detail.toPlainText().count('识别阶段')==1
|
|
assert '#12345678' in window._detail.toPlainText()
|
|
window._ran({'ok':False,'operation':'对孔测试','message':'执行失败','detail':'测试用内部错误明细','log':'/tmp/isolated-motion.log'})
|
|
assert '测试用内部错误明细' in window._detail.toPlainText()
|
|
assert '/tmp/isolated-motion.log' in window._detail.toPlainText()
|
|
assert not errors,errors
|
|
assert runtime._runtime is None
|
|
assert not any(job.isRunning() for job in window._jobs.values())
|
|
print('Qt responsive, no device modules, async task and graceful shutdown PASS',
|
|
{'seconds':round(time.monotonic()-started,2),'ticks':len(ticks),
|
|
'max_gap':round(max((b-a for a,b in zip(ticks,ticks[1:])),default=0),3),'errors':errors},flush=True)
|