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-16 09:38:46 +08:00
parent 855448664c
commit 2413d6d305
40 changed files with 870678 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files

145
code/automatic_test/do_current.py
View File

@@ -1,4 +1,5 @@
import os
import multiprocessing
import time
import paramiko
import pandas
@@ -7,9 +8,10 @@ from common import clibs
def initialization(path, sub, data_dirs, data_files, hr, w2t):
def check_files():
msg = "初始路径下不允许有文件夹，且初始路径下只能存在如下两个文件，确认后重新运行！\n"
msg += "1. T_电机电流.xlsx\n2. xxxx.zip\n"
if len(data_dirs) != 0 or len(data_files) != 2:
w2t("初始路径下不允许有文件夹，且初始路径下只能存在如下两个文件，确认后重新运行！\n", "red")
w2t("1. T_电机电流.xlsx\n2. xxxx.zip\n", "red", "ConfigFileError")
w2t(msg, "red", "InitFileError")
prj_file, count = None, 0
for data_file in data_files:
@@ -20,12 +22,11 @@ def initialization(path, sub, data_dirs, data_files, hr, w2t):
count += 1
prj_file = data_file
else:
w2t(f"{data_file} 不合规：初始路径下只能存在如下两个文件，确认后重新运行！\n", "red")
w2t("1. T_电机电流.xlsx\n2. xxxx.zip\n", "red", "ConfigFileError")
w2t(msg, "red", "InitFileError")
if count != 2:
w2t("初始路径下不允许有文件夹，且初始路径下只能存在如下两个文件，确认后重新运行！\n", "red")
w2t("1. T_电机电流.xlsx\n2. xxxx.zip\n", "red", "ConfigFileError")
w2t(msg, "red", "InitFileError")
w2t("数据目录合规性检查结束，未发现问题......\n")
if sub == "tool100" or sub == "inertia":
os.mkdir(f"{path}/single")
@@ -50,10 +51,10 @@ def initialization(path, sub, data_dirs, data_files, hr, w2t):
local_file = path + f"/{robot_type}.cfg"
clibs.c_pd.pull_file_from_server(server_file, local_file)
prj_file = check_files()
_prj_file = check_files()
get_configs()
return prj_file
return _prj_file
def single_axis_proc(path, records, number):
@@ -61,58 +62,63 @@ def single_axis_proc(path, records, number):
number = number if number < 6 else number - 6
d_vel, d_trq, d_sensor = [], [], []
for record in records:
print(f"record = {record}")
data = eval(record)["data"]
data = eval(record[0])["data"]
for item in data:
d_item = reversed(item["value"])
if item.get("channel", None) == number and item.get("name", None) == "hw_joint_vel_feedback":
d_vel.extend(item["value"])
d_vel.extend(d_item)
elif item.get("channel", None) == number and item.get("name", None) == "device_servo_trq_feedback":
d_trq.extend(item["value"])
d_trq.extend(d_item)
elif item.get("channel", None) == number and item.get("name", None) == "hw_sensor_trq_feedback":
d_sensor.extend(item["value"])
d_sensor.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)
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"])
if item.get("channel", None) == axis and item.get("name", None) == "hw_joint_vel_feedback":
d_vel.extend(d_item)
elif item.get("channel", None) == axis and item.get("name", None) == "device_servo_trq_feedback":
d_trq.extend(d_item)
elif item.get("channel", None) == axis and item.get("name", None) == "hw_sensor_trq_feedback":
d_sensor.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)
filename = f"{path}/single/j{number + 1}_{text}_{time.time()}.data"
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)
def scenario_proc(path, records, number, scenario_time):
d_vel, d_trq, d_sensor = [], [], []
for record in records:
print(f"record = {record}")
data = eval(record)["data"]
for axis in range(6):
for item in data:
if item.get("channel", None) == axis and item.get("name", None) == "hw_joint_vel_feedback":
d_vel.extend(item["value"])
elif item.get("channel", None) == axis and item.get("name", None) == "device_servo_trq_feedback":
d_trq.extend(item["value"])
elif item.get("channel", None) == axis and item.get("name", None) == "hw_sensor_trq_feedback":
d_sensor.extend(item["value"])
def gen_result_file(path, number, start_time, end_time, scenario_time):
@clibs.db_lock
def get_records(s_time, e_time):
clibs.cursor.execute(f"select content from logs where time between '{s_time}' and '{e_time}' and content like '%diagnosis.result%' order by id asc")
_ = clibs.cursor.fetchall()
return _
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)
filename = f"{path}/s_{number-11}/j{axis}_s_{number-11}_{scenario_time}_{time.time()}.data"
df.to_csv(filename, sep="\t", index=False)
@clibs.db_lock
def gen_result_file(path, number, scenario_time):
if number < 12: # 35s/15s == 700/300
len_records = 700 if number < 6 else 300
clibs.cursor.execute(f"select content from logs where content like 'diagnosis.result' limit {len_records}")
records = clibs.cursor.fetchall()
single_axis_proc(path, records, number)
elif number < 15: # scenario time
len_records = int(scenario_time * 20) + 1
clibs.cursor.execute(f"select content from logs where content like 'diagnosis.result' limit {len_records}")
records = clibs.cursor.fetchall()
scenario_proc(path, records, number, scenario_time)
if number < 12:
records = get_records(start_time, end_time)
p = multiprocessing.Process(target=single_axis_proc, args=(path, records, number))
p.daemon = True
p.start()
elif number < 15:
records = get_records(start_time, end_time)
p = multiprocessing.Process(target=scenario_proc, args=(path, records, number, scenario_time))
p.daemon = True
p.start()
def run_rl(path, prj_file, hr, md, sub, w2t):
@@ -151,8 +157,9 @@ def run_rl(path, prj_file, hr, md, sub, w2t):
# 打开诊断曲线，触发软急停，并解除，目的是让可能正在运行着的机器停下来
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.trigger_estop()
md.reset_estop()
md.r_soft_estop(0)
md.r_soft_estop(1)
md.r_clear_alarm()
for condition in conditions:
number = conditions.index(condition)
@@ -178,7 +185,7 @@ def run_rl(path, prj_file, hr, md, sub, w2t):
hr.execution("state.switch_auto")
hr.execution("state.switch_motor_on")
# 3. 开始运行程序，单轴运行35s
# 3. 开始运行程序
hr.execution("rl_task.set_run_params", loop_mode=True, override=1.0)
hr.execution("rl_task.run", tasks=["current"])
t_start = time.time()
@@ -188,34 +195,35 @@ def run_rl(path, prj_file, hr, md, sub, w2t):
break
else:
time.sleep(1)
if (time.time() - t_start) // 20 > 1:
w2t("20s 内未收到机器人的运行信号，需要确认RL程序编写正确并正常执行...", "red", "ReadySignalTimeoutError")
if (time.time() - t_start) > 20:
w2t("20s 内未收到机器人的运行信号，需要确认RL程序和工具通信是否正常执行...", "red", "ReadySignalTimeoutError")
# 4. 打开诊断曲线，并执行采集
time.sleep(10) # 保证程序已经运行起来，其实主要是为了保持电流的采集而设定
scenario_time = 0
# 4. 执行采集
time.sleep(10) # 消除前 10s 的不稳定数据
start_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(time.time()))
single_time, stall_time, scenario_time = 30, 10, 0
if number < 6: # 单轴
time.sleep(35)
time.sleep(single_time)
elif number < 12: # 堵转
time.sleep(15)
time.sleep(stall_time)
else: # 场景
t_start = time.time()
while True:
scenario_time = md.read_scenario_time()
if float(scenario_time) > 1:
w2t(f"场景{number-5}的周期时间：{scenario_time}\n")
scenario_time = float(f"{float(md.read_scenario_time()):.2f}")
if float(scenario_time) != 0:
w2t(f"场景{number - 11}的周期时间：{scenario_time}\n")
break
else:
time.sleep(5)
if (time.time()-t_start)//60 > 3:
w2t(f"未收到场景{number-5}的周期时间，需要确认RL程序编写正确并正常执行...\n", "red", "GetScenarioTimeError")
time.sleep(1) # 一定要延迟一秒再读一次scenario time寄存器，因为一开始读取的数值不准确
scenario_time = md.read_scenario_time()
time.sleep(float(scenario_time)*0.2) # 再运行周期的20%即可
time.sleep(1)
if (time.time()-t_start) > 180:
w2t(f"180s 内未收到场景{number - 11}的周期时间，需要确认RL程序和工具通信交互是否正常执行...\n", "red", "GetScenarioTimeError")
time.sleep(20)
# 5.停止程序运行，保留数据并处理输出
clibs.execution("rl_task.stop", tasks=["current"])
gen_result_file(path, number, scenario_time)
end_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(time.time()))
hr.execution("rl_task.stop", tasks=["current"])
time.sleep(5) # 确保数据都拿到
gen_result_file(path, number, start_time, end_time, scenario_time)
else:
if sub == "tool100":
w2t("单轴和场景电机电流采集完毕，如需采集惯量负载，须切换负载类型，并更换惯量负载，重新执行。\n", "green")
@@ -231,7 +239,6 @@ def main():
w2t = clibs.w2t
hr = clibs.c_hr
md = clibs.c_md
insert_logdb = clibs.insert_logdb
data_dirs, data_files = clibs.traversal_files(path, w2t)
prj_file = initialization(path, sub, data_dirs, data_files, hr, w2t)