Surprisingly I made it! 完成了制动性能测试程序
This commit is contained in:
parent
9fa42fb3e1
commit
fa5a9f0f89
Binary file not shown.
@ -1,4 +1,4 @@
|
||||
import random
|
||||
from random import randint
|
||||
from time import sleep, time
|
||||
from sys import argv
|
||||
from os import scandir, mkdir
|
||||
@ -178,33 +178,33 @@ def run_rl(path, loadsel, hr, md, config_file, prj_file, result_dirs, w2t):
|
||||
continue
|
||||
for count in range(1, 4):
|
||||
_count += 1
|
||||
w2t(f"[{_count}/64-{count}] 正在执行{axis}轴{condition}的制动测试......", 0, 0, 'purple', 'Automatic Test')
|
||||
w2t(f"[{_count}/63-{count}] 正在执行{axis}轴{condition}的制动测试......", 0, 0, 'purple', 'Automatic Test')
|
||||
|
||||
# 1. 关闭诊断曲线,触发软急停,并解除,目的是让可能正在运行着的机器停下来,切手动模式并下电
|
||||
md.trigger_estop()
|
||||
md.reset_estop()
|
||||
md.write_act(False)
|
||||
_response = execution('diagnosis.open', hr, w2t, open=False, display_open=False)
|
||||
sleep(2) # 让曲线彻底关闭
|
||||
sleep(1) # 让曲线彻底关闭
|
||||
_response = execution('state.switch_manual', hr, w2t)
|
||||
_response = execution('state.switch_motor_off', hr, w2t)
|
||||
|
||||
# 2. 修改未要执行的场景
|
||||
if count == 1:
|
||||
ssh = SSHClient()
|
||||
ssh.set_missing_host_key_policy(AutoAddPolicy())
|
||||
ssh.connect('192.168.0.160', 22, username='luoshi', password='luoshi2019')
|
||||
_rl_cmd = f"brake_E(j{axis}_{_reach}_p, j{axis}_{_reach}_n, p_speed, p_tool)"
|
||||
_rl_speed = f"VelSet {_speed}"
|
||||
cmd = 'cd /home/luoshi/bin/controller/; '
|
||||
cmd += 'sudo sed -i "/brake_E/d" projects/target/_build/brake/main.mod; '
|
||||
cmd += f'sudo sed -i "/DONOTDELETE/i {_rl_cmd}" projects/target/_build/brake/main.mod; '
|
||||
cmd += f'sudo sed -i "/VelSet/d" projects/target/_build/brake/main.mod; '
|
||||
cmd += f'sudo sed -i "/MoveAbsJ/i {_rl_speed}" projects/target/_build/brake/main.mod; '
|
||||
stdin, stdout, stderr = ssh.exec_command(cmd, get_pty=True)
|
||||
stdin.write('luoshi2019' + '\n')
|
||||
stdin.flush()
|
||||
print(stdout.read().decode()) # 必须得输出一下stdout,才能正确执行sudo
|
||||
print(stderr.read().decode()) # 顺便也执行以下stderr
|
||||
ssh = SSHClient()
|
||||
ssh.set_missing_host_key_policy(AutoAddPolicy())
|
||||
ssh.connect('192.168.0.160', 22, username='luoshi', password='luoshi2019')
|
||||
_rl_cmd = f"brake_E(j{axis}_{_reach}_p, j{axis}_{_reach}_n, p_speed, p_tool)"
|
||||
_rl_speed = f"VelSet {_speed}"
|
||||
cmd = 'cd /home/luoshi/bin/controller/; '
|
||||
cmd += 'sudo sed -i "/brake_E/d" projects/target/_build/brake/main.mod; '
|
||||
cmd += f'sudo sed -i "/DONOTDELETE/i {_rl_cmd}" projects/target/_build/brake/main.mod; '
|
||||
cmd += f'sudo sed -i "/VelSet/d" projects/target/_build/brake/main.mod; '
|
||||
cmd += f'sudo sed -i "/MoveAbsJ/i {_rl_speed}" projects/target/_build/brake/main.mod; '
|
||||
stdin, stdout, stderr = ssh.exec_command(cmd, get_pty=True)
|
||||
stdin.write('luoshi2019' + '\n')
|
||||
stdin.flush()
|
||||
print(stdout.read().decode()) # 必须得输出一下stdout,才能正确执行sudo
|
||||
print(stderr.read().decode()) # 顺便也执行以下stderr
|
||||
|
||||
# 3. reload工程后,pp2main,并且自动模式和上电,最后运行程序
|
||||
prj_path = 'target/_build/target.prj'
|
||||
@ -216,8 +216,6 @@ def run_rl(path, loadsel, hr, md, config_file, prj_file, result_dirs, w2t):
|
||||
for i in range(3):
|
||||
if md.read_ready_to_go() == 1:
|
||||
md.write_act(True)
|
||||
sleep(1)
|
||||
md.write_act(False)
|
||||
break
|
||||
else:
|
||||
sleep(1)
|
||||
@ -226,7 +224,6 @@ def run_rl(path, loadsel, hr, md, config_file, prj_file, result_dirs, w2t):
|
||||
|
||||
# 4. 第一次打开诊断曲线,并执行采集8s,之后触发软急停,关闭曲线采集,找出最大速度,传递给RL程序,最后清除相关记录
|
||||
if count == 1:
|
||||
sleep(4) # 排除从零点位置到目标位置的干扰
|
||||
_response = execution('diagnosis.open', hr, w2t, open=True, display_open=True)
|
||||
_response = execution('diagnosis.set_params', hr, w2t, display_pdo_params=display_pdo_params)
|
||||
sleep(8) # 前八秒获取实际最大速度
|
||||
@ -244,10 +241,8 @@ def run_rl(path, loadsel, hr, md, config_file, prj_file, result_dirs, w2t):
|
||||
wb = load_workbook(config_file, read_only=True)
|
||||
ws = wb['Target']
|
||||
speed_target = float(ws.cell(row=3, column=axis+1).value) * float(_speed) / 100
|
||||
print(f"speed target = {speed_target}")
|
||||
print(f"speed max = {speed_max}")
|
||||
if speed_max < speed_target*0.95 or speed_max > speed_target*1.05:
|
||||
w2t(f"Axis: {axis}-{count} | Speed: {speed_max*RADIAN} | Shouldbe: {speed_target}", 0, 0, 'indigo')
|
||||
w2t(f"Axis: {axis}-{count} | Speed: {speed_max} | Shouldbe: {speed_target}", 0, 0, 'indigo', 'Automatic Test')
|
||||
|
||||
md.write_speed_max(speed_max)
|
||||
sleep(1)
|
||||
@ -260,27 +255,24 @@ def run_rl(path, loadsel, hr, md, config_file, prj_file, result_dirs, w2t):
|
||||
break
|
||||
|
||||
# 5. 清除软急停,重新运行程序,并打开曲线发送继续运动信号,当速度达到最大值时,通过DO触发急停
|
||||
if count == 1:
|
||||
md.reset_estop()
|
||||
_response = execution('overview.reload', hr, w2t, prj_path=prj_path, tasks=['brake', 'stop0_related'])
|
||||
_response = execution('rl_task.pp_to_main', hr, w2t, tasks=['brake', 'stop0_related'])
|
||||
_response = execution('state.switch_auto', hr, w2t)
|
||||
_response = execution('state.switch_motor_on', hr, w2t)
|
||||
_response = execution('rl_task.run', hr, w2t, tasks=['brake', 'stop0_related'])
|
||||
for i in range(3):
|
||||
if md.read_ready_to_go() == 1:
|
||||
md.write_act(True)
|
||||
sleep(1)
|
||||
md.write_act(False)
|
||||
break
|
||||
else:
|
||||
sleep(1)
|
||||
md.reset_estop()
|
||||
_response = execution('overview.reload', hr, w2t, prj_path=prj_path, tasks=['brake', 'stop0_related'])
|
||||
_response = execution('rl_task.pp_to_main', hr, w2t, tasks=['brake', 'stop0_related'])
|
||||
_response = execution('state.switch_auto', hr, w2t)
|
||||
_response = execution('state.switch_motor_on', hr, w2t)
|
||||
_response = execution('rl_task.run', hr, w2t, tasks=['brake', 'stop0_related'])
|
||||
for i in range(3):
|
||||
if md.read_ready_to_go() == 1:
|
||||
md.write_act(True)
|
||||
break
|
||||
else:
|
||||
w2t("未收到机器人的运行信号,需要确认RL程序编写正确并正常执行...", 0, 111, 'red', 'Automatic Test')
|
||||
sleep(1)
|
||||
else:
|
||||
w2t("未收到机器人的运行信号,需要确认RL程序编写正确并正常执行...", 0, 111, 'red', 'Automatic Test')
|
||||
|
||||
_response = execution('diagnosis.open', hr, w2t, open=True, display_open=True)
|
||||
_response = execution('diagnosis.set_params', hr, w2t, display_pdo_params=display_pdo_params)
|
||||
sleep(random.randint(2, 8))
|
||||
# sleep(randint(3, 6))
|
||||
md.write_probe(True)
|
||||
_t_start = time()
|
||||
while True:
|
||||
@ -321,7 +313,7 @@ def main(path, hr, md, loadsel, w2t):
|
||||
run_rl(path, loadsel, hr, md, config_file, prj_file, result_dirs, w2t)
|
||||
_e_time = time()
|
||||
time_total = _e_time - _s_time
|
||||
w2t(f"处理总时长:{time_total // 3600:02.0f} h {time_total % 3600 // 60:02.0f} m {time_total % 60:02.0f} s")
|
||||
w2t(f"处理总时长:{time_total // 3600:02.0f} h {time_total % 3600 // 60:02.0f} m {time_total % 60:02.0f} s", 0, 0, 'green', 'Automatic Test')
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
Reference in New Issue
Block a user