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-22 16:14:19 +08:00
parent 7f815ac63e
commit 92bdf133f2
48 changed files with 662 additions and 870240 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
code/automatic_test/do_brake.py
View File

@ -10,7 +10,7 @@ from common import clibs, openapi
def initialization(path, sub, data_dirs, data_files, hr, w2t):
def check_files():
msg = "初始路径下不允许有文件夹，且初始路径下只能存在如下五个文件，确认后重新运行！\n"
msg = "初始路径下不允许有文件夹，初始路径下只能存在如下五个文件，且文件为关闭状态，确认后重新运行！\n"
msg += "1. configs.xlsx\n2. reach33/reach66/reach100_xxxx.xlsx\n3. xxxx.zip\n"
if len(data_dirs) != 0 or len(data_files) != 5:
w2t(msg, "red", "InitFileError")
@ -130,16 +130,19 @@ def gen_result_file(path, axis, t_end, reach, load, speed, speed_target, rounds,
df.to_csv(filename, sep="\t", index=False)
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)
def change_curve_state(hr, stat):
if not stat:
display_pdo_params = []
else:
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, display_open=stat, 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]
prj_name = ".".join(prj_file.split("/")[-1].split(".")[:-1])
wb = openpyxl.load_workbook(config_file, read_only=True)
ws = wb["Target"]
write_diagnosis = float(ws.cell(row=2, column=2).value)
@ -171,6 +174,10 @@ def run_rl(path, sub, hr, md, config_file, prj_file, result_dirs, avs, w2t):
continue
for axis in range(1, 4):
if not clibs.running:
w2t("后台数据清零完成，现在可以重新运行之前停止的程序。", "red")
exit()
# for single condition test
if (single_axis != -1 and single_axis != axis) or (axis == 3 and reach != "100"):
continue
@ -231,16 +238,16 @@ def run_rl(path, sub, hr, md, config_file, prj_file, result_dirs, avs, w2t):
break
else:
time.sleep(1)
if (time.time() - t_start) > 20:
w2t("20s 内未收到机器人的运行信号，需要确认 RL 程序编写正确并正常执行...", "red", "ReadySignalTimeoutError")
if (time.time() - t_start) > 3:
w2t("3s 内未收到机器人的运行信号，需要确认 RL 程序编写正确并正常执行...", "red", "ReadySignalTimeoutError")
# 4. 找出最大速度，传递给RL程序，最后清除相关记录
time.sleep(5) # 消除前 5s 的不稳定数据
change_curve_state(hr, True, True)
change_curve_state(hr, 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"])
change_curve_state(hr, False, False)
change_curve_state(hr, False)
# 找出最大速度
@clibs.db_lock
@ -297,7 +304,7 @@ def run_rl(path, sub, hr, md, config_file, prj_file, result_dirs, avs, w2t):
def exec_brake():
flag, start, data, record = True, time.time(), None, None
change_curve_state(hr, True, True)
change_curve_state(hr, True)
while flag:
time.sleep(0.05)
if time.time() - start > 20:
@ -320,7 +327,7 @@ def run_rl(path, sub, hr, md, config_file, prj_file, result_dirs, avs, w2t):
if (pon == "positive" and speed_moment > 0) or (pon == "negative" and speed_moment < 0):
clibs.c_ec.setdo_value(io_name, "false")
time.sleep(2)
change_curve_state(hr, False, False)
change_curve_state(hr, False)
flag = False
return time.time()

62
code/automatic_test/do_current.py
View File

@ -8,7 +8,7 @@ from common import clibs
def initialization(path, sub, data_dirs, data_files, hr, w2t):
def check_files():
msg = "初始路径下不允许有文件夹，且初始路径下只能存在如下两个文件，确认后重新运行！\n"
msg = "初始路径下不允许有文件夹，初始路径下只能存在如下两个文件，且文件为关闭状态，确认后重新运行！\n"
msg += "1. T_电机电流.xlsx\n2. xxxx.zip\n"
if len(data_dirs) != 0 or len(data_files) != 2:
w2t(msg, "red", "InitFileError")
@ -60,7 +60,7 @@ def initialization(path, sub, data_dirs, data_files, hr, w2t):
def single_axis_proc(path, records, number):
text = "single" if number < 6 else "hold"
number = number if number < 6 else number - 6
d_vel, d_trq, d_sensor = [], [], []
d_vel, d_trq, d_sensor, d_trans = [], [], [], []
for record in records:
data = eval(record[0])["data"]
for item in data:
@ -71,33 +71,40 @@ def single_axis_proc(path, records, number):
d_trq.extend(d_item)
elif item.get("channel", None) == number and item.get("name", None) == "hw_sensor_trq_feedback":
d_sensor.extend(d_item)
elif item.get("channel", None) == number and item.get("name", None) == "hw_estimate_trans_trq_res":
d_trans.extend(d_item)
df1 = pandas.DataFrame.from_dict({"hw_joint_vel_feedback": d_vel})
df2 = pandas.DataFrame.from_dict({"device_servo_trq_feedback": d_trq})
df3 = pandas.DataFrame.from_dict({"hw_sensor_trq_feedback": d_sensor})
df = pandas.concat([df1, df2, df3], axis=1)
df4 = pandas.DataFrame.from_dict({"hw_estimate_trans_trq_res": d_trans})
df = pandas.concat([df1, df2, df3, df4], axis=1)
filename = f"{path}/single/j{number + 1}_{text}_{time.time()}.data"
df.to_csv(filename, sep="\t", index=False)
def scenario_proc(path, records, number, scenario_time):
for axis in range(6):
d_vel, d_trq, d_sensor = [], [], []
for record in records:
data = eval(record[0])["data"]
for item in data:
d_item = reversed(item["value"])
d_vel, d_trq, d_sensor, d_trans = [[], [], [], [], [], []], [[], [], [], [], [], []], [[], [], [], [], [], []], [[], [], [], [], [], []]
for record in records:
data = eval(record[0])["data"]
for item in data:
d_item = reversed(item["value"])
for axis in range(6):
if item.get("channel", None) == axis and item.get("name", None) == "hw_joint_vel_feedback":
d_vel.extend(d_item)
d_vel[axis].extend(d_item)
elif item.get("channel", None) == axis and item.get("name", None) == "device_servo_trq_feedback":
d_trq.extend(d_item)
d_trq[axis].extend(d_item)
elif item.get("channel", None) == axis and item.get("name", None) == "hw_sensor_trq_feedback":
d_sensor.extend(d_item)
d_sensor[axis].extend(d_item)
elif item.get("channel", None) == axis and item.get("name", None) == "hw_estimate_trans_trq_res":
d_trans[axis].extend(d_item)
df1 = pandas.DataFrame.from_dict({"hw_joint_vel_feedback": d_vel})
df2 = pandas.DataFrame.from_dict({"device_servo_trq_feedback": d_trq})
df3 = pandas.DataFrame.from_dict({"hw_sensor_trq_feedback": d_sensor})
df = pandas.concat([df1, df2, df3], axis=1)
for axis in range(6):
df1 = pandas.DataFrame.from_dict({"hw_joint_vel_feedback": d_vel[axis]})
df2 = pandas.DataFrame.from_dict({"device_servo_trq_feedback": d_trq[axis]})
df3 = pandas.DataFrame.from_dict({"hw_sensor_trq_feedback": d_sensor[axis]})
df4 = pandas.DataFrame.from_dict({"hw_estimate_trans_trq_res": d_trans[axis]})
df = pandas.concat([df1, df2, df3, df4], axis=1)
filename = f"{path}/s_{number-11}/j{axis+1}_s_{number-11}_{scenario_time}_{time.time()}.data"
df.to_csv(filename, sep="\t", index=False)
@ -121,14 +128,15 @@ 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)
def change_curve_state(hr, stat):
curves = ["hw_joint_vel_feedback", "device_servo_trq_feedback", "hw_sensor_trq_feedback", "hw_estimate_trans_trq_res"]
display_pdo_params = [] if not stat else [{"name": curve, "channel": chl} for curve in curves for chl in range(6)]
hr.execution("diagnosis.open", open=stat, display_open=stat, 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]
prj_name = ".".join(prj_file.split("/")[-1].split(".")[:-1])
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)",
@ -165,6 +173,10 @@ def run_rl(path, prj_file, hr, md, sub, w2t):
md.r_clear_alarm()
for condition in conditions:
if not clibs.running:
w2t("后台数据清零完成，现在可以重新运行之前停止的程序。", "red")
exit()
number = conditions.index(condition)
w2t(f"正在执行{disc[number]}测试......\n")
@ -198,12 +210,12 @@ def run_rl(path, prj_file, hr, md, sub, w2t):
break
else:
time.sleep(1)
if (time.time() - t_start) > 20:
w2t("20s 内未收到机器人的运行信号，需要确认RL程序和工具通信是否正常执行...", "red", "ReadySignalTimeoutError")
if (time.time() - t_start) > 3:
w2t("3s 内未收到机器人的运行信号，需要确认RL程序和工具通信是否正常执行...", "red", "ReadySignalTimeoutError")
# 4. 执行采集
time.sleep(10) # 消除前 10s 的不稳定数据
change_curve_state(hr, True, True)
change_curve_state(hr, 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: # 单轴
@ -227,7 +239,7 @@ def run_rl(path, prj_file, hr, md, sub, w2t):
end_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(time.time()))
hr.execution("rl_task.stop", tasks=["current"])
time.sleep(2) # 确保数据都拿到
change_curve_state(hr, False, False)
change_curve_state(hr, False)
gen_result_file(path, number, start_time, end_time, scenario_time)
else:
if sub == "tool100":
@ -252,7 +264,7 @@ def main():
e_time = time.time()
time_total = e_time - s_time
w2t(f"-" * 90 + "\n", "purple")
w2t(f"处理总时长：{time_total // 3600:02.0f} h {time_total % 3600 // 60:02.0f} m {time_total % 60:02.0f} s", "green")
w2t(f"处理总时长：{time_total // 3600:02.0f} h {time_total % 3600 // 60:02.0f} m {time_total % 60:02.0f} s\n", "green")
if __name__ == "__main__":