gitea/aio_second_edition
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化了制动和电机电流自动测试代码

Browse Source
This commit is contained in:
gitea
2025-01-18 12:33:29 +08:00
parent 46ce714462
commit 7f815ac63e
7 changed files with 147 additions and 246 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
code/automatic_test/do_brake.py
View File

@ -1,3 +1,4 @@
import random
import time
import os
import paramiko
@ -111,10 +112,15 @@ def gen_result_file(path, axis, t_end, reach, load, speed, speed_target, rounds,
d_stop.extend(d_item)
idx = d_stop.index(0)
av_estop = sum(d_vel[idx - 20:idx])/20 * clibs.RADIAN
av_estop = abs(sum(d_vel[idx - 20:idx])/20 * clibs.RADIAN)
if av_estop / speed_target < threshold:
w2t(f"[av_estop: {av_estop:.2f} | shouldbe: {speed_target:.2f}] 本次触发 ESTOP 时未采集到指定百分比的最大速度，即将重试！\n", "#8A2BE2")
return False
clibs.count += 1
if clibs.count < 3:
return "retry"
else:
clibs.count = 0
w2t(f"尝试三次后仍无法获取正确数据，本次数据无效，继续执行...\n", "red")
df1 = pandas.DataFrame.from_dict({"hw_joint_vel_feedback": d_vel})
df2 = pandas.DataFrame.from_dict({"device_servo_trq_feedback": d_trq})
@ -122,14 +128,18 @@ def gen_result_file(path, axis, t_end, reach, load, speed, speed_target, rounds,
df = pandas.concat([df1, df2, df3], axis=1)
filename = f"{path}/j{axis}/reach{reach}_load{load}_speed{speed}/reach{reach}_load{load}_speed{speed}_{rounds}.data"
df.to_csv(filename, sep="\t", index=False)
return True
def change_curve_state(hr, stat_1, stat_2):
display_pdo_params = [{"name": name, "channel": chl} for name in ["hw_joint_vel_feedback", "device_servo_trq_feedback"] for chl in range(6)]
display_pdo_params.append({"name": "device_safety_estop", "channel": 0})
hr.execution("diagnosis.open", open=stat_1, display_open=stat_2, overrun=True, turn_area=True, delay_motion=False)
hr.execution("diagnosis.set_params", display_pdo_params=display_pdo_params, frequency=50, version="1.4.1")
def run_rl(path, sub, hr, md, config_file, prj_file, result_dirs, avs, w2t):
count, total, speed_target = 0, 63, 0
prj_name = prj_file.split("/")[-1].split(".")[0]
display_pdo_params = [{"name": name, "channel": chl} for name in ["hw_joint_vel_feedback", "device_servo_trq_feedback"] for chl in range(6)]
display_pdo_params.append({"name": "device_safety_estop", "channel": 0})
wb = openpyxl.load_workbook(config_file, read_only=True)
ws = wb["Target"]
write_diagnosis = float(ws.cell(row=2, column=2).value)
@ -147,8 +157,6 @@ def run_rl(path, sub, hr, md, config_file, prj_file, result_dirs, avs, w2t):
else:
w2t("configs.xlsx 中 Target 页面 B5 单元格填写不正确，检查后重新运行...", "red", "DirectionError")
hr.execution("diagnosis.open", open=True, display_open=True, overrun=True, turn_area=True, delay_motion=False)
hr.execution("diagnosis.set_params", display_pdo_params=display_pdo_params, frequency=50, version="1.4.1")
for condition in result_dirs:
reach = condition.split("_")[0].removeprefix("reach")
load = condition.split("_")[1].removeprefix("load")
@ -226,12 +234,13 @@ def run_rl(path, sub, hr, md, config_file, prj_file, result_dirs, avs, w2t):
if (time.time() - t_start) > 20:
w2t("20s 内未收到机器人的运行信号，需要确认 RL 程序编写正确并正常执行...", "red", "ReadySignalTimeoutError")
# 4. 找出最大速度，传递给RL程序，最后清除相关记录
time.sleep(10) # 消除前 10s 的不稳定数据
time.sleep(5) # 消除前 5s 的不稳定数据
change_curve_state(hr, True, True)
start_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(time.time()))
time.sleep(get_init_speed) # 指定时间后获取实际【正|负】方向的最大速度，可通过configs.xlsx配置
end_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(time.time()))
hr.execution("rl_task.stop", tasks=["brake"])
time.sleep(5) # 确保数据都拿到
change_curve_state(hr, False, False)
# 找出最大速度
@clibs.db_lock
@ -284,12 +293,13 @@ def run_rl(path, sub, hr, md, config_file, prj_file, result_dirs, avs, w2t):
else:
w2t("3s 内未收到机器人的运行信号，需要确认 RL 程序配置正确并正常执行...", "red", "ReadySignalTimeoutError")
time.sleep(10) # 排除从其他位姿到零点位姿，再到轴极限位姿的时间
time.sleep(5+random.randint(1, 5)) # 排除从其他位姿到零点位姿，再到轴极限位姿的时间
def exec_brake():
flag, start, data, record = True, time.time(), None, None
change_curve_state(hr, True, True)
while flag:
time.sleep(0.1)
time.sleep(0.05)
if time.time() - start > 20:
w2t("20s 内未触发急停，需排查......", "red", "BrakeTimeoutError")
@ -306,24 +316,22 @@ def run_rl(path, sub, hr, md, config_file, prj_file, result_dirs, avs, w2t):
continue
speed_moment = clibs.RADIAN * sum(item["value"]) / len(item["value"])
if (pon == "positive" and speed_moment > 0) or (pon == "negative" and speed_moment < 0):
if abs(speed_moment) > speed_target * 0.95:
if abs(speed_moment) > speed_target * 0.95:
if (pon == "positive" and speed_moment > 0) or (pon == "negative" and speed_moment < 0):
clibs.c_ec.setdo_value(io_name, "false")
time.sleep(3)
time.sleep(2)
change_curve_state(hr, False, False)
flag = False
break
return time.time()
t_end = exec_brake()
# 6. 保留数据并处理输出
ret = gen_result_file(path, axis, t_end, reach, load, speed, speed_target, rounds, w2t)
if ret:
if ret != "retry":
clibs.count = 0
break
else:
w2t(f"\n{sub.removeprefix("tool")}%负载的制动性能测试执行完毕，如需采集其他负载，须切换负载类型，并更换其他负载，重新执行。\n", "green")
hr.execution("diagnosis.open", open=False, display_open=False, overrun=True, turn_area=True, delay_motion=False)
hr.execution("diagnosis.set_params", display_pdo_params=[], frequency=50, version="1.4.1")
def main():

15
code/automatic_test/do_current.py
View File

@ -121,9 +121,14 @@ def gen_result_file(path, number, start_time, end_time, scenario_time):
p.start()
def change_curve_state(hr, stat_1, stat_2):
display_pdo_params = [{"name": name, "channel": chl} for name in ["hw_joint_vel_feedback", "device_servo_trq_feedback", "hw_sensor_trq_feedback"] for chl in range(6)]
hr.execution("diagnosis.open", open=stat_1, display_open=stat_2, overrun=True, turn_area=True, delay_motion=False)
hr.execution("diagnosis.set_params", display_pdo_params=display_pdo_params, frequency=50, version="1.4.1")
def run_rl(path, prj_file, hr, md, sub, w2t):
prj_name = prj_file.split("/")[-1].split(".")[0]
display_pdo_params = [{"name": name, "channel": chl} for name in ["hw_joint_vel_feedback", "device_servo_trq_feedback", "hw_sensor_trq_feedback"] for chl in range(6)]
c_regular = [
"scenario(0, j1_p, j1_n, p_speed, p_tool, i_tool)",
"scenario(0, j2_p, j2_n, p_speed, p_tool, i_tool)",
@ -155,8 +160,6 @@ def run_rl(path, prj_file, hr, md, sub, w2t):
conditions, disc = c_inertia, disc_inertia
# 打开诊断曲线，触发软急停，并解除，目的是让可能正在运行着的机器停下来
hr.execution("diagnosis.open", open=True, display_open=True, overrun=True, turn_area=True, delay_motion=False)
hr.execution("diagnosis.set_params", display_pdo_params=display_pdo_params, frequency=50, version="1.4.1")
md.r_soft_estop(0)
md.r_soft_estop(1)
md.r_clear_alarm()
@ -200,6 +203,7 @@ def run_rl(path, prj_file, hr, md, sub, w2t):
# 4. 执行采集
time.sleep(10) # 消除前 10s 的不稳定数据
change_curve_state(hr, True, True)
start_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(time.time()))
single_time, stall_time, scenario_time = 40, 10, 0
if number < 6: # 单轴
@ -222,15 +226,14 @@ def run_rl(path, prj_file, hr, md, sub, w2t):
# 5.停止程序运行，保留数据并处理输出
end_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(time.time()))
hr.execution("rl_task.stop", tasks=["current"])
time.sleep(5) # 确保数据都拿到
time.sleep(2) # 确保数据都拿到
change_curve_state(hr, False, False)
gen_result_file(path, number, start_time, end_time, scenario_time)
else:
if sub == "tool100":
w2t("单轴和场景电机电流采集完毕，如需采集惯量负载，须切换负载类型，并更换惯量负载，重新执行。\n", "green")
elif sub == "inertia":
w2t("惯量负载电机电流采集完毕，如需采集单轴/场景/保持电机电流，须切换负载类型，并更换偏置负载，重新执行。\n", "green")
hr.execution("diagnosis.open", open=False, display_open=False, overrun=True, turn_area=True, delay_motion=False)
hr.execution("diagnosis.set_params", display_pdo_params=[], frequency=50, version="1.4.1")
def main():