212 lines
9.8 KiB
Python
212 lines
9.8 KiB
Python
"""Exercise real Qt widgets against the ROS host with its fake adapter."""
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import textwrap
|
|
|
|
|
|
def test_demo_gui_teach_scan_pause_resume_cancel(tmp_path):
|
|
program = r'''
|
|
import json,time,sys
|
|
from pathlib import Path
|
|
import rclpy
|
|
from PyQt5.QtCore import Qt
|
|
from PyQt5.QtTest import QTest
|
|
from PyQt5.QtWidgets import QApplication,QLabel
|
|
from linkerhand_range_calibration.profiles import load_profile,load_station
|
|
from linkerhand_range_calibration.runtime import CalibrationRuntime
|
|
from linkerhand_range_calibration.ui.window import CalibrationWindow
|
|
|
|
rclpy.init();app=QApplication([])
|
|
station=load_station();station['output_root']=str(Path(sys.argv[1])/'output')
|
|
station['teaching_root']=str(Path(sys.argv[1])/'teaching')
|
|
station['scan'].update(rate=1000,stable_frames=4,settle_seconds=.01)
|
|
node=CalibrationRuntime(load_profile(),station,demo=True)
|
|
window=CalibrationWindow(node);window.show()
|
|
def until(condition,timeout=6):
|
|
end=time.monotonic()+timeout
|
|
while time.monotonic()<end:
|
|
rclpy.spin_once(node,timeout_sec=.015);app.processEvents();window.refresh()
|
|
if condition():return
|
|
raise AssertionError(str(node.snapshot()))
|
|
until(lambda:node.snapshot().get('can_save',False))
|
|
assert not node.adapter.sent and not node.adapter.settings_sent
|
|
assert len(node.previews())==3
|
|
assert len(window.joint_controls)==20
|
|
clearance_choices=[window.save_kind.itemData(i)[1] for i in range(window.save_kind.count())
|
|
if window.save_kind.itemData(i)[0]=='clearances']
|
|
assert clearance_choices==[]
|
|
first,second=window.joint_controls[:2]
|
|
assert first.slider.minimum()==0 and first.slider.maximum()==255
|
|
window.sync_targets();window.fill.click()
|
|
node._tick()
|
|
assert not node.adapter.sent and not node.adapter.settings_sent
|
|
|
|
# Many drag changes before one runtime tick keep only the latest target.
|
|
for value in range(20,101):first.slider.setValue(value)
|
|
assert not node.adapter.sent
|
|
node._tick();window.refresh()
|
|
assert node.engine.state=='TEACHING' and node.adapter.target[0]==100
|
|
assert first.isEnabled() and not window.save.isEnabled()
|
|
assert len(node.adapter.settings_sent)==1
|
|
second.slider.setValue(45)
|
|
first.slider.setValue(30)
|
|
node._tick()
|
|
assert node.adapter.target[:2]==(30,45)
|
|
assert first.editor.value()==30 and second.editor.value()==45
|
|
assert len(node.adapter.settings_sent)==1
|
|
until(lambda:node.engine.state=='IDLE')
|
|
assert node.adapter.target[:2]==(30,45)
|
|
assert node.adapter.target[2:]==(0,)*18
|
|
|
|
# Numeric typing waits for confirmation, and then drives the same slider path.
|
|
first.editor.setFocus();first.editor.selectAll()
|
|
QTest.keyClicks(first.editor,'35')
|
|
node._tick()
|
|
assert node.adapter.target[0]==30 and not node.pending_jog
|
|
QTest.keyClick(first.editor,Qt.Key_Return)
|
|
until(lambda:node.engine.state=='IDLE' and node.adapter.target[0]==35)
|
|
assert first.slider.value()==35
|
|
assert len(node.adapter.settings_sent)==1 # idle between gestures does not resend settings
|
|
|
|
# Finishing one joint must not overwrite another joint's unconfirmed numeric input.
|
|
second.slider.setValue(80);node._tick();window.refresh()
|
|
first.editor.setFocus();first.editor.selectAll()
|
|
QTest.keyClicks(first.editor,'36')
|
|
until(lambda:node.engine.state=='IDLE')
|
|
assert first.editor.text()=='36' and node.adapter.target[0]==35
|
|
QTest.keyClick(first.editor,Qt.Key_Escape)
|
|
window.sync_targets()
|
|
|
|
# Cancel discards a pending gesture and rejects late events from that UI epoch.
|
|
old_epoch=node.snapshot()['manual_epoch']
|
|
before=len(node.adapter.sent)
|
|
first.slider.setValue(200)
|
|
node.submit('cancel')
|
|
node.submit('jog',targets={node.profile.joints[0].name:240},epoch=old_epoch)
|
|
node._tick();window.refresh()
|
|
assert node.engine.state=='CANCELLED' and len(node.adapter.sent)==before
|
|
assert first.slider.value()==35
|
|
|
|
# Pause holds the last sent target, and manual controls remain locked until cancel.
|
|
first.slider.setValue(200);node._tick();window.refresh()
|
|
assert node.engine.state=='TEACHING'
|
|
window.pause.click();until(lambda:node.engine.state=='PAUSED')
|
|
assert not first.isEnabled() and not window.resume.isEnabled()
|
|
before=len(node.adapter.sent)
|
|
node.submit('jog',targets={node.profile.joints[0].name:255},epoch=node.snapshot()['manual_epoch'])
|
|
node._tick()
|
|
assert len(node.adapter.sent)==before
|
|
window.cancel.click();until(lambda:node.engine.state=='CANCELLED')
|
|
|
|
window.start_one.click()
|
|
until(lambda:node.engine.state=='SCANNING')
|
|
assert not first.isEnabled() and not window.fill.isEnabled()
|
|
window.pause.click();until(lambda:node.engine.state=='PAUSED')
|
|
window.resume.click();until(lambda:node.engine.state=='PREPARING')
|
|
window.cancel.click();until(lambda:node.engine.state=='CANCELLED')
|
|
document=json.loads(node.session.result_path.read_text())
|
|
assert len(document['joints'])==20 and all(v=={'min':None,'max':None} for v in document['joints'].values())
|
|
held=node.engine.command
|
|
first.slider.setValue(10)
|
|
until(lambda:node.engine.state=='IDLE' and node.adapter.target[0]==10)
|
|
assert node.adapter.target[1:]==held[1:]
|
|
until(lambda:node.snapshot()['can_save'])
|
|
window.save.click();until(lambda:node.teaching_path.exists())
|
|
assert node.teaching.data['baseline']['target'][0]==10
|
|
|
|
# Saving a nonzero baseline keeps the fixed thumb clearance at zero.
|
|
index_control=window.joint_controls[2]
|
|
index_control.slider.setValue(37)
|
|
until(lambda:node.snapshot()['can_save'] and node.adapter.target[2]==37)
|
|
window.save.click()
|
|
until(lambda:node.teaching.data['baseline']['target'][2]==37)
|
|
saved=json.loads(json.dumps(node.teaching.data))
|
|
assert saved['baseline']['target'][0]==10
|
|
assert not saved['clearances']
|
|
tasks={task.name:task for task in node.profile.tasks}
|
|
assert node.teaching.prepare(tasks['thumb_cmc_roll'])[2]==0
|
|
assert node.teaching.prepare(tasks['four_finger_roll'])[2]==37
|
|
|
|
# A command received during a tick must not expose its epoch before it is applied.
|
|
original_tick=node.engine.tick
|
|
def cancel_during_tick(now,stamp):
|
|
node.submit('cancel')
|
|
original_tick(now,stamp)
|
|
node.engine.tick=cancel_during_tick
|
|
node._tick()
|
|
node.engine.tick=original_tick
|
|
assert node.snapshot()['manual_epoch']<node.manual_epoch
|
|
node.submit('jog',targets={node.profile.joints[0].name:255},epoch=node.snapshot()['manual_epoch'])
|
|
node._tick()
|
|
assert node.engine.state=='CANCELLED' and not node.pending_jog
|
|
|
|
# SDK diagnostics remain in the journal and never add a banner or lock controls.
|
|
node.adapter.warning='thumb_cmc_roll：执行器层判定堵转；thumb_cmc_yaw：执行器过温'
|
|
node._tick();window.refresh()
|
|
assert window.start_one.isEnabled() and all(control.isEnabled() for control in window.joint_controls)
|
|
assert not any('自动标定不可用' in label.text() or '堵转' in label.text() or '过温' in label.text()
|
|
for label in window.findChildren(QLabel))
|
|
window.start_one.click();until(lambda:node.engine.state=='SCANNING')
|
|
events=[json.loads(line) for line in (node.session.directory/'samples.jsonl').read_text().splitlines()]
|
|
assert any(event['kind']=='diagnostic_warning' and '过温' in event['reason'] for event in events)
|
|
window.pause.click();until(lambda:node.engine.state=='PAUSED')
|
|
assert window.resume.isEnabled()
|
|
window.resume.click();until(lambda:node.engine.state=='PREPARING')
|
|
window.cancel.click();until(lambda:node.engine.state=='CANCELLED')
|
|
|
|
# Manual teaching uses the same readiness check, including the joint named in a diagnostic.
|
|
node._tick();window.refresh()
|
|
assert first.isEnabled() and index_control.isEnabled() and second.isEnabled()
|
|
assert window.start_one.isEnabled()
|
|
settings_with_warning=len(node.adapter.settings_sent)
|
|
second.slider.setValue(75)
|
|
index_control.slider.setValue(45)
|
|
until(lambda:node.snapshot()['can_save'] and node.adapter.target[2]==45 and node.adapter.target[1]==75)
|
|
assert node.engine.state=='IDLE'
|
|
assert window.save.isEnabled() and len(node.adapter.settings_sent)==settings_with_warning
|
|
window.save.click()
|
|
until(lambda:node.teaching.data['baseline']['target'][2]==45)
|
|
node.adapter.warning=None
|
|
node._tick();window.refresh()
|
|
assert second.isEnabled() and window.start_one.isEnabled()
|
|
|
|
# After a connection fault, the next deliberate gesture reapplies settings once.
|
|
settings_before=len(node.adapter.settings_sent)
|
|
node.adapter.error='模拟反馈断流'
|
|
node._tick();window.refresh()
|
|
assert not first.isEnabled() and node.motion_settings_key is None
|
|
node.adapter.error=None
|
|
node._tick();window.refresh()
|
|
first.slider.setValue(11)
|
|
node._tick()
|
|
assert node.adapter.target[0]==11
|
|
assert len(node.adapter.settings_sent)==settings_before+1
|
|
until(lambda:node.engine.state=='IDLE')
|
|
|
|
# A single thumb-roll task restores the thumb before releasing the index clearance.
|
|
window.start_one.click();until(lambda:node.engine.state=='RESTORING')
|
|
assert not any(control.isEnabled() for control in window.joint_controls)
|
|
assert not window.save.isEnabled() and window.cancel.isEnabled()
|
|
assert '测后恢复基础姿态' in window.status.text()
|
|
assert node.engine.restoring_joint=='thumb_cmc_roll'
|
|
assert 'thumb_cmc_roll 恢复基础姿态' in window.status.text()
|
|
assert node.engine.command[2]==0
|
|
until(lambda:node.engine.restoring_joint=='index_mcp_roll')
|
|
assert 'index_mcp_roll 恢复基础姿态' in window.status.text()
|
|
assert node.engine.command[0]==node.teaching.data['baseline']['target'][0]
|
|
assert not any(control.isEnabled() for control in window.joint_controls)
|
|
until(lambda:node.engine.state=='COMPLETED')
|
|
assert first.slider.value()==node.teaching.data['baseline']['target'][0]
|
|
assert index_control.slider.value()==node.teaching.data['baseline']['target'][2]
|
|
assert all(control.isEnabled() for control in window.joint_controls)
|
|
window.close();node.close_session();node.destroy_node();rclpy.shutdown()
|
|
print('GUI workflow passed')
|
|
'''
|
|
env = dict(os.environ, ROS_DOMAIN_ID='199', ROS_LOCALHOST_ONLY='1', QT_QPA_PLATFORM='offscreen')
|
|
env['PYTHONPATH'] = str(Path(__file__).resolve().parents[1])+os.pathsep+env.get('PYTHONPATH','')
|
|
result = subprocess.run([sys.executable,'-c',textwrap.dedent(program),str(tmp_path)],env=env,
|
|
capture_output=True,text=True,timeout=25)
|
|
assert result.returncode==0, result.stdout+result.stderr
|