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-14 14:14:33 +08:00
parent 137122947c
commit 0e67a2831c
25 changed files with 524877 additions and 203 deletions
Download Patch File Download Diff File Expand all files Collapse all files

67
code/automatic_test/btn_functions.py Normal file
View File

@ -0,0 +1,67 @@
from sys import argv
from commons import clibs
tab_name = clibs.tab_names['at']
logger = clibs.log_prod
def trigger_estop(md, w2t):
md.trigger_estop()
w2t("触发急停成功，可点击机器状态验证。", 0, 0, 'green', tab_name)
def reset_estop(md, w2t):
md.reset_estop()
w2t("恢复急停成功，可点击机器状态验证。", 0, 0, 'green', tab_name)
def get_state(hr, w2t):
# 获取机器状态
_response = clibs.execution('state.get_state', hr, w2t, tab_name)
stat_desc = {'engine': '上电状态', 'operate': '操作模式', 'rc_state': '控制器状态', 'robot_action': '机器人动作', 'safety_mode': '安全模式', 'servo_mode': '伺服工作模式', 'task_space': '工作任务空间'}
for component, state in _response['data'].items():
w2t(f"{stat_desc[component]}: {state}", tab_name=tab_name)
# 获取设备伺服信息
_response = clibs.execution('device.get_params', hr, w2t, tab_name)
dev_desc = {0: '伺服版本', 1: '伺服参数', 2: '安全板固件', 3: '控制器', 4: '通讯总线', 5: '解释器', 6: '运动控制', 8: '力控版本', 9: '末端固件', 10: '机型文件', 11: '环境包'}
dev_vers = {}
for device in _response['data']['devices']:
dev_vers[device['type']] = device['version']
for i in sorted(dev_desc.keys()):
w2t(f"{dev_desc[i]}: {dev_vers[i]}", tab_name=tab_name)
# 设置示教器模式
_response = clibs.execution('state.set_tp_mode', hr, w2t, tab_name, tp_mode='without')
def warning_info(hr, w2t):
for postfix in ['', '.2', '.3', '.4', '.5', '.6', '.7', '.8', '.9', '.10']:
log_name = clibs.log_data_hmi + postfix
try:
with open(log_name, 'r', encoding='utf-8') as f_log:
for line in f_log:
if 'alarm' in line:
w2t(line.strip(), tab_name=tab_name)
except FileNotFoundError:
pass
def main(hr, md, func, w2t):
if hr is None:
w2t("无法连接机器人，检查是否已经使用Robot Assist软件连接机器，重试中...", 0, 49, 'red', tab_name)
# func: get_state/
match func:
case 'trigger_estop':
trigger_estop(md, w2t)
case 'reset_estop':
reset_estop(md, w2t)
case 'get_state':
get_state(hr, w2t)
case 'warning_info':
warning_info(hr, w2t)
if __name__ == '__main__':
main(*argv[1:])

298
code/automatic_test/do_brake.py Normal file
View File

@ -0,0 +1,298 @@
from time import sleep, time, strftime, localtime
from sys import argv
from os import mkdir
from paramiko import SSHClient, AutoAddPolicy
from json import loads
from openpyxl import load_workbook
from pandas import DataFrame, concat
from commons import clibs
tab_name = clibs.tab_names['at']
logger = clibs.log_prod
def check_files(path, loadsel, data_dirs, data_files, w2t):
if len(data_dirs) != 0 or len(data_files) != 5:
w2t('初始路径下不允许有文件夹，且初始路径下只能存在如下五个文件，确认后重新运行！', 0, 0, 'red', tab_name)
w2t(' 1. configs.xlsx\n 2. reach33/reach66/reach100_xxxx.xlsx\n 3. xxxx.zip', 0, 1, 'red', tab_name)
config_file = reach33 = reach66 = reach100 = prj_file = None
for data_file in data_files:
filename = data_file.split('\\')[-1]
if filename == 'configs.xlsx':
config_file = data_file
elif filename.startswith('reach33_') and filename.endswith('.xlsx'):
reach33 = data_file
elif filename.startswith('reach66_') and filename.endswith('.xlsx'):
reach66 = data_file
elif filename.startswith('reach100_') and filename.endswith('.xlsx'):
reach100 = data_file
elif filename.endswith('.zip'):
prj_file = data_file
else:
w2t('初始路径下不允许有文件夹，且初始路径下只能存在如下五个文件，确认后重新运行！', 0, 0, 'red', tab_name)
w2t(' 1. configs.xlsx\n 2. reach33/reach66/reach100_xxxx.xlsx\n 3. xxxx.zip', 0, 2, 'red', tab_name)
if config_file and reach33 and reach66 and reach100 and prj_file:
result_dirs = []
mkdir(f"{path}\\j1")
mkdir(f"{path}\\j2")
mkdir(f"{path}\\j3")
for _reach in ['reach33', 'reach66', 'reach100']:
for _load in [f'load{loadsel.removeprefix("tool")}']:
for _speed in ['speed33', 'speed66', 'speed100']:
dir_name = '_'.join([_reach, _load, _speed])
result_dirs.append(dir_name)
mkdir(f"{path}\\j1\\{dir_name}")
mkdir(f"{path}\\j2\\{dir_name}")
if _reach == 'reach100':
mkdir(f"{path}\\j3\\{dir_name}")
w2t("数据目录合规性检查结束，未发现问题......", 0, 0, 'blue', tab_name)
return config_file, reach33, reach66, reach100, prj_file, result_dirs
else:
w2t('初始路径下不允许有文件夹，且初始路径下只能存在如下五个文件，确认后重新运行！', 0, 0, 'red', tab_name)
w2t(' 1. configs.xlsx\n 2. reach33/reach66/reach100_xxxx.xlsx\n 3. xxxx.zip', 0, 1, 'red', tab_name)
def gen_result_file(path, curve_data, axis, _reach, _load, _speed, count):
_d2d_vel = {'hw_joint_vel_feedback': []}
_d2d_trq = {'device_servo_trq_feedback': []}
_d2d_stop = {'device_safety_estop': []}
for data in curve_data[-240:]: # 保留最后12s的数据
dict_results = data['data']
for item in dict_results:
try:
item['value'].reverse()
except KeyError:
continue
if item.get('channel', None) == axis-1 and item.get('name', None) == 'hw_joint_vel_feedback':
_d2d_vel['hw_joint_vel_feedback'].extend(item['value'])
elif item.get('channel', None) == axis-1 and item.get('name', None) == 'device_servo_trq_feedback':
_d2d_trq['device_servo_trq_feedback'].extend(item['value'])
elif item.get('channel', None) == 0 and item.get('name', None) == 'device_safety_estop':
_d2d_stop['device_safety_estop'].extend(item['value'])
df1 = DataFrame.from_dict(_d2d_vel)
df2 = DataFrame.from_dict(_d2d_trq)
df3 = DataFrame.from_dict(_d2d_stop)
df = concat([df1, df2, df3], axis=1)
_filename = f"{path}\\j{axis}\\reach{_reach}_load{_load}_speed{_speed}\\reach{_reach}_load{_load}_speed{_speed}_{count}.data"
df.to_csv(_filename, sep='\t', index=False)
def run_rl(path, loadsel, hr, md, config_file, result_dirs, w2t):
_count = 0
_total = 63
display_pdo_params = [
{"name": "hw_joint_vel_feedback", "channel": 0},
{"name": "hw_joint_vel_feedback", "channel": 1},
{"name": "hw_joint_vel_feedback", "channel": 2},
{"name": "hw_joint_vel_feedback", "channel": 3},
{"name": "hw_joint_vel_feedback", "channel": 4},
{"name": "hw_joint_vel_feedback", "channel": 5},
{"name": "device_servo_trq_feedback", "channel": 0},
{"name": "device_servo_trq_feedback", "channel": 1},
{"name": "device_servo_trq_feedback", "channel": 2},
{"name": "device_servo_trq_feedback", "channel": 3},
{"name": "device_servo_trq_feedback", "channel": 4},
{"name": "device_servo_trq_feedback", "channel": 5},
{"name": "device_safety_estop", "channel": 0},
]
wb = load_workbook(config_file, read_only=True)
ws = wb['Target']
write_diagnosis = float(ws.cell(row=3, column=10).value)
get_init_speed = float(ws.cell(row=4, column=10).value)
single_brake = str(ws.cell(row=5, column=10).value)
logger.info(f"write_diagnosis = {write_diagnosis}, get_init_speed = {get_init_speed}, single_brake = {single_brake}")
if ws.cell(row=1, column=1).value == 'positive':
md.write_pon(1)
elif ws.cell(row=1, column=1).value == 'negative':
md.write_pon(0)
else:
w2t("configs.xlsx中Target页面A1单元格填写不正确，检查后重新运行...", 0, 111, 'red', tab_name)
clibs.execution('diagnosis.open', hr, w2t, tab_name, open=True, display_open=True)
clibs.execution('diagnosis.set_params', hr, w2t, tab_name, display_pdo_params=display_pdo_params)
for condition in result_dirs:
_reach = condition.split('_')[0].removeprefix('reach')
_load = condition.split('_')[1].removeprefix('load')
_speed = condition.split('_')[2].removeprefix('speed')
# for single condition test
_single_axis = 0
if single_brake != '0':
_total = 3
_single_axis = int(single_brake.split('-')[0])
if _reach != single_brake.split('-')[1] or _load != single_brake.split('-')[2] or _speed != single_brake.split('-')[3]:
continue
for axis in range(1, 4):
# for single condition test
if _single_axis != 0 and _single_axis != axis:
continue
md.write_axis(axis)
speed_max = 0
if axis == 3 and _reach != '100':
continue
w2t(f"-"*90, 0, 0, 'purple', tab_name)
for count in range(1, 4):
_count += 1
this_time = strftime("%Y-%m-%d %H:%M:%S", localtime(time()))
prj_path = 'target/_build/target.prj'
w2t(f"[{this_time} | {_count}/{_total}] 正在执行 {axis} 轴 {condition} 的第 {count} 次制动测试...", 0, 0, 'purple', tab_name)
# 1. 关闭诊断曲线，触发软急停，并解除，目的是让可能正在运行着的机器停下来，切手动模式并下电
md.trigger_estop()
md.reset_estop()
md.clear_alarm()
md.write_act(0)
sleep(write_diagnosis) # 软急停超差后，等待写诊断时间，可通过configs.xlsx配置
while count == 1:
# 2. 修改要执行的场景
ssh = SSHClient()
ssh.set_missing_host_key_policy(AutoAddPolicy())
ssh.connect(hostname=clibs.ip_addr, port=22, username='luoshi', password='luoshi2019')
if ws.cell(row=1, column=1).value == 'positive':
_rl_cmd = f"brake_E(j{axis}_{_reach}_p, j{axis}_{_reach}_n, p_speed, p_tool)"
elif ws.cell(row=1, column=1).value == 'negative':
_rl_cmd = f"brake_E(j{axis}_{_reach}_n, j{axis}_{_reach}_p, p_speed, p_tool)"
else:
w2t("configs.xlsx中Target页面A1单元格填写不正确，检查后重新运行...", 0, 111, 'red', tab_name)
_rl_speed = f"VelSet {_speed}"
_rl_tool = f"tool p_tool = tool{loadsel.removeprefix('tool')}"
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 += '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; '
cmd += 'sudo sed -i "/tool p_tool/d" projects/target/_build/brake/main.mod; '
cmd += f'sudo sed -i "/VelSet/i {_rl_tool}" projects/target/_build/brake/main.mod; '
stdin, stdout, stderr = ssh.exec_command(cmd, get_pty=True)
stdin.write('luoshi2019' + '\n')
stdout.read().decode() # 需要read一下才能正常执行
stderr.read().decode()
# 3. reload工程后，pp2main，并且自动模式和上电，最后运行程序
clibs.execution('overview.reload', hr, w2t, tab_name, prj_path=prj_path, tasks=['brake', 'stop0_related'])
clibs.execution('rl_task.pp_to_main', hr, w2t, tab_name, tasks=['brake', 'stop0_related'])
clibs.execution('state.switch_auto', hr, w2t, tab_name)
clibs.execution('state.switch_motor_on', hr, w2t, tab_name)
clibs.execution('rl_task.set_run_params', hr, w2t, tab_name, loop_mode=True, override=1.0)
clibs.execution('rl_task.run', hr, w2t, tab_name, tasks=['brake', 'stop0_related'])
_t_start = time()
while True:
if md.read_ready_to_go() == 1:
md.write_act(True)
break
else:
if (time() - _t_start) // 20 > 1:
w2t("20s内未收到机器人的运行信号，需要确认RL程序编写正确并正常执行...", 0, 111, 'red', tab_name)
else:
sleep(1)
# 4. 找出最大速度，传递给RL程序，最后清除相关记录
sleep(get_init_speed+5) # 冗余5s，指定时间后获取实际【正|负】方向的最大速度，可通过configs.xlsx配置
clibs.execution('rl_task.stop', hr, w2t, tab_name, tasks=['brake'])
# 找出最大速度
_c_msg = hr.c_msg.copy()
_number = 0
for _msg in _c_msg:
if _number > get_init_speed*20: # 最开始回零点的时候，二轴速度可以达到最大值，实际摆动时，某一方向可能达不到
break
if 'diagnosis.result' in _msg:
_number += 1
dict_results = loads(_msg)['data']
for item in dict_results:
if item.get('channel', None) == axis-1 and item.get('name', None) == 'hw_joint_vel_feedback':
_ = clibs.RADIAN * sum(item['value']) / len(item['value'])
if ws.cell(row=1, column=1).value == 'positive':
speed_max = max(_, speed_max)
elif ws.cell(row=1, column=1).value == 'negative':
speed_max = min(_, speed_max)
logger.info(f"speed max = {speed_max}")
speed_max = abs(speed_max)
speed_target = float(ws.cell(row=3, column=axis+1).value) * float(_speed) / 100
if speed_max < speed_target*0.95 or speed_max > speed_target*1.05:
w2t(f"Axis: {axis}-{count} | Speed: {speed_max} | Shouldbe: {speed_target}", 0, 0, 'indigo', tab_name)
md.write_speed_max(speed_max)
hr.c_msg_xs.clear()
if len(hr.c_msg) > 270:
del hr.c_msg[270:]
if speed_max < 10:
md.clear_alarm()
w2t("未获取到正确的速度，即将重新获取...", 0, 0, 'red', tab_name)
continue
else:
break
# 5. 清除软急停，重新运行程序，并打开曲线发送继续运动信号，当速度达到最大值时，通过DO触发急停
md.reset_estop() # 其实没必要
md.clear_alarm()
# clibs.execution('overview.reload', hr, w2t, tab_name, prj_path=prj_path, tasks=['brake', 'stop0_related'])
clibs.execution('rl_task.pp_to_main', hr, w2t, tab_name, tasks=['brake', 'stop0_related'])
clibs.execution('state.switch_auto', hr, w2t, tab_name)
clibs.execution('state.switch_motor_on', hr, w2t, tab_name)
clibs.execution('rl_task.run', hr, w2t, tab_name, tasks=['brake', 'stop0_related'])
for i in range(3):
if md.read_ready_to_go() == 1:
md.write_act(1)
break
else:
sleep(1)
else:
w2t("未收到机器人的运行信号，需要确认RL程序编写正确并正常执行...", 0, 111, 'red', tab_name)
sleep(10) # 排除从其他位姿到零点位姿，再到轴极限位姿的时间
md.write_probe(1)
_t_start = time()
while True:
if md.read_brake_done() == 1:
sleep(4) # 保证速度归零
md.write_probe(0)
break
else:
if (time() - _t_start) > 30:
w2t(f"30s内未触发急停，该条数据无效，需要确认RL/Python程序编写正确并正常执行，或者判别是否是机器本体问题，比如正负方向速度是否一致...", 0, 0, 'red', tab_name)
md.write_probe(0)
break
else:
sleep(1)
# 6. 保留数据并处理输出
curve_data = []
_c_msg = hr.c_msg.copy()
for _msg in _c_msg:
if 'diagnosis.result' in _msg:
curve_data.insert(0, loads(_msg))
else:
hr.c_msg_xs.clear()
if len(hr.c_msg) > 270:
del hr.c_msg[270:]
gen_result_file(path, curve_data, axis, _reach, _load, _speed, count)
else:
w2t(f"\n{loadsel.removeprefix('tool')}%负载的制动性能测试执行完毕，如需采集其他负载，须切换负载类型，并更换其他负载，重新执行。", 0, 0, 'green', tab_name)
def main(path, hr, md, loadsel, w2t):
_s_time = time()
data_dirs, data_files = clibs.traversal_files(path, w2t)
config_file, reach33, reach66, reach100, prj_file, result_dirs = check_files(path, loadsel, data_dirs, data_files, w2t)
clibs.prj_to_xcore(prj_file)
run_rl(path, loadsel, hr, md, config_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", 0, 0, 'green', tab_name)
if __name__ == '__main__':
main(*argv[1:])

405
code/automatic_test/do_current.py Normal file
View File

@ -0,0 +1,405 @@
import json
from os import mkdir
import time
from time import sleep, time
from sys import argv
from paramiko import SSHClient, AutoAddPolicy
from json import loads
from pandas import DataFrame, concat
from common import clibs
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)]
def initialization(path, sub, data_dirs, data_files, hr, w2t):
def check_files():
if len(data_dirs) != 0 or len(data_files) != 2:
w2t("初始路径下不允许有文件夹，且初始路径下只能存在如下两个文件，确认后重新运行！\n", "red")
w2t("1. T_电机电流.xlsx\n2. xxxx.zip\n", "red", "ConfigFileError")
prj_file, count = None, 0
for data_file in data_files:
filename = data_file.split("/")[-1]
if filename == "T_电机电流.xlsx":
count += 1
elif filename.endswith(".zip"):
count += 1
prj_file = data_file
else:
w2t(f"{data_file} 不合规：初始路径下只能存在如下两个文件，确认后重新运行！\n", "red")
w2t("1. T_电机电流.xlsx\n2. xxxx.zip\n", "red", "ConfigFileError")
if count != 2:
w2t("初始路径下不允许有文件夹，且初始路径下只能存在如下两个文件，确认后重新运行！\n", "red")
w2t("1. T_电机电流.xlsx\n2. xxxx.zip\n", "red", "ConfigFileError")
w2t("数据目录合规性检查结束，未发现问题......\n")
if sub == "tool100" or sub == "inertia":
mkdir(f"{path}/single")
mkdir(f"{path}/s_1")
mkdir(f"{path}/s_2")
mkdir(f"{path}/s_3")
elif sub == "inertia":
mkdir(f"{path}/inertia")
else:
w2t("负载选择错误，电机电流测试只能选择 tool100/inertia 规格！\n", "red", "LoadSelectError")
return prj_file
def get_configs():
robot_type = None
msg_id, state = hr.execution("controller.get_params")
records = hr.get_from_id(msg_id, state)
for record in records:
if "请求发送成功" not in record[0]:
robot_type = eval(record[0])["data"]["robot_type"]
server_file = f"/home/luoshi/bin/controller/robot_cfg/{robot_type}/{robot_type}.cfg"
local_file = path + f"/{robot_type}.cfg"
clibs.c_pd.pull_file_from_server(server_file, local_file)
return local_file
prj_file = check_files()
config_file = get_configs()
return config_file, prj_file
def data_proc_regular(path, filename, channel, scenario_time):
if channel in list(range(6)):
with open(filename, 'r', encoding='utf-8') as f_obj:
lines = f_obj.readlines()
_d2d_vel = {'hw_joint_vel_feedback': []}
_d2d_trq = {'device_servo_trq_feedback': []}
_d2d_sensor = {'hw_sensor_trq_feedback': []}
for line in lines[-500:]: # 保留最后25s的数据
data = eval(line.strip())['data']
for item in data:
try:
item['value'].reverse()
except KeyError:
continue
if item.get('channel', None) == channel and item.get('name', None) == 'hw_joint_vel_feedback':
_d2d_vel['hw_joint_vel_feedback'].extend(item['value'])
elif item.get('channel', None) == channel and item.get('name', None) == 'device_servo_trq_feedback':
_d2d_trq['device_servo_trq_feedback'].extend(item['value'])
elif item.get('channel', None) == channel and item.get('name', None) == 'hw_sensor_trq_feedback':
_d2d_sensor['hw_sensor_trq_feedback'].extend(item['value'])
df1 = DataFrame.from_dict(_d2d_vel)
df2 = DataFrame.from_dict(_d2d_trq)
df3 = DataFrame.from_dict(_d2d_sensor)
df = concat([df1, df2, df3], axis=1)
_filename = f'{path}\\single\\j{channel+1}_single_{time()}.data'
df.to_csv(_filename, sep='\t', index=False)
elif channel in list(range(6, 9)):
with open(filename, 'r', encoding='utf-8') as f_obj:
lines = f_obj.readlines()
_d2d_vel_0 = {'hw_joint_vel_feedback': []}
_d2d_trq_0 = {'device_servo_trq_feedback': []}
_d2d_sensor_0 = {'hw_sensor_trq_feedback': []}
_d2d_vel_1 = {'hw_joint_vel_feedback': []}
_d2d_trq_1 = {'device_servo_trq_feedback': []}
_d2d_sensor_1 = {'hw_sensor_trq_feedback': []}
_d2d_vel_2 = {'hw_joint_vel_feedback': []}
_d2d_trq_2 = {'device_servo_trq_feedback': []}
_d2d_sensor_2 = {'hw_sensor_trq_feedback': []}
_d2d_vel_3 = {'hw_joint_vel_feedback': []}
_d2d_trq_3 = {'device_servo_trq_feedback': []}
_d2d_sensor_3 = {'hw_sensor_trq_feedback': []}
_d2d_vel_4 = {'hw_joint_vel_feedback': []}
_d2d_trq_4 = {'device_servo_trq_feedback': []}
_d2d_sensor_4 = {'hw_sensor_trq_feedback': []}
_d2d_vel_5 = {'hw_joint_vel_feedback': []}
_d2d_trq_5 = {'device_servo_trq_feedback': []}
_d2d_sensor_5 = {'hw_sensor_trq_feedback': []}
for line in lines:
data = eval(line.strip())['data']
for item in data:
try:
item['value'].reverse()
except KeyError:
continue
if item.get('channel', None) == 0 and item.get('name', None) == 'hw_joint_vel_feedback':
_d2d_vel_0['hw_joint_vel_feedback'].extend(item['value'])
elif item.get('channel', None) == 0 and item.get('name', None) == 'device_servo_trq_feedback':
_d2d_trq_0['device_servo_trq_feedback'].extend(item['value'])
elif item.get('channel', None) == 0 and item.get('name', None) == 'hw_sensor_trq_feedback':
_d2d_sensor_0['hw_sensor_trq_feedback'].extend(item['value'])
elif item.get('channel', None) == 1 and item.get('name', None) == 'hw_joint_vel_feedback':
_d2d_vel_1['hw_joint_vel_feedback'].extend(item['value'])
elif item.get('channel', None) == 1 and item.get('name', None) == 'device_servo_trq_feedback':
_d2d_trq_1['device_servo_trq_feedback'].extend(item['value'])
elif item.get('channel', None) == 1 and item.get('name', None) == 'hw_sensor_trq_feedback':
_d2d_sensor_1['hw_sensor_trq_feedback'].extend(item['value'])
elif item.get('channel', None) == 2 and item.get('name', None) == 'hw_joint_vel_feedback':
_d2d_vel_2['hw_joint_vel_feedback'].extend(item['value'])
elif item.get('channel', None) == 2 and item.get('name', None) == 'device_servo_trq_feedback':
_d2d_trq_2['device_servo_trq_feedback'].extend(item['value'])
elif item.get('channel', None) == 3 and item.get('name', None) == 'hw_sensor_trq_feedback':
_d2d_sensor_2['hw_sensor_trq_feedback'].extend(item['value'])
elif item.get('channel', None) == 3 and item.get('name', None) == 'hw_joint_vel_feedback':
_d2d_vel_3['hw_joint_vel_feedback'].extend(item['value'])
elif item.get('channel', None) == 3 and item.get('name', None) == 'device_servo_trq_feedback':
_d2d_trq_3['device_servo_trq_feedback'].extend(item['value'])
elif item.get('channel', None) == 3 and item.get('name', None) == 'hw_sensor_trq_feedback':
_d2d_sensor_3['hw_sensor_trq_feedback'].extend(item['value'])
elif item.get('channel', None) == 4 and item.get('name', None) == 'hw_joint_vel_feedback':
_d2d_vel_4['hw_joint_vel_feedback'].extend(item['value'])
elif item.get('channel', None) == 4 and item.get('name', None) == 'device_servo_trq_feedback':
_d2d_trq_4['device_servo_trq_feedback'].extend(item['value'])
elif item.get('channel', None) == 4 and item.get('name', None) == 'hw_sensor_trq_feedback':
_d2d_sensor_4['hw_sensor_trq_feedback'].extend(item['value'])
elif item.get('channel', None) == 5 and item.get('name', None) == 'hw_joint_vel_feedback':
_d2d_vel_5['hw_joint_vel_feedback'].extend(item['value'])
elif item.get('channel', None) == 5 and item.get('name', None) == 'device_servo_trq_feedback':
_d2d_trq_5['device_servo_trq_feedback'].extend(item['value'])
elif item.get('channel', None) == 5 and item.get('name', None) == 'hw_sensor_trq_feedback':
_d2d_sensor_5['hw_sensor_trq_feedback'].extend(item['value'])
df_01 = DataFrame.from_dict(_d2d_vel_0)
df_02 = DataFrame.from_dict(_d2d_trq_0)
df_03 = DataFrame.from_dict(_d2d_sensor_0)
df = concat([df_01, df_02, df_03], axis=1)
_filename = f'{path}\\s_{channel-5}\\j1_s_{channel-5}_{scenario_time}_{time()}.data'
df.to_csv(_filename, sep='\t', index=False)
df_01 = DataFrame.from_dict(_d2d_vel_1)
df_02 = DataFrame.from_dict(_d2d_trq_1)
df_03 = DataFrame.from_dict(_d2d_sensor_1)
df = concat([df_01, df_02, df_03], axis=1)
_filename = f'{path}\\s_{channel-5}\\j2_s_{channel-5}_{scenario_time}_{time()}.data'
df.to_csv(_filename, sep='\t', index=False)
df_01 = DataFrame.from_dict(_d2d_vel_2)
df_02 = DataFrame.from_dict(_d2d_trq_2)
df_03 = DataFrame.from_dict(_d2d_sensor_2)
df = concat([df_01, df_02, df_03], axis=1)
_filename = f'{path}\\s_{channel-5}\\j3_s_{channel-5}_{scenario_time}_{time()}.data'
df.to_csv(_filename, sep='\t', index=False)
df_01 = DataFrame.from_dict(_d2d_vel_3)
df_02 = DataFrame.from_dict(_d2d_trq_3)
df_03 = DataFrame.from_dict(_d2d_sensor_3)
df = concat([df_01, df_02, df_03], axis=1)
_filename = f'{path}\\s_{channel-5}\\j4_s_{channel-5}_{scenario_time}_{time()}.data'
df.to_csv(_filename, sep='\t', index=False)
df_01 = DataFrame.from_dict(_d2d_vel_4)
df_02 = DataFrame.from_dict(_d2d_trq_4)
df_03 = DataFrame.from_dict(_d2d_sensor_4)
df = concat([df_01, df_02, df_03], axis=1)
_filename = f'{path}\\s_{channel-5}\\j5_s_{channel-5}_{scenario_time}_{time()}.data'
df.to_csv(_filename, sep='\t', index=False)
df_01 = DataFrame.from_dict(_d2d_vel_5)
df_02 = DataFrame.from_dict(_d2d_trq_5)
df_03 = DataFrame.from_dict(_d2d_sensor_5)
df = concat([df_01, df_02, df_03], axis=1)
_filename = f'{path}\\s_{channel-5}\\j6_s_{channel-5}_{scenario_time}_{time()}.data'
df.to_csv(_filename, sep='\t', index=False)
elif channel in list(range(9, 15)):
with open(filename, 'r', encoding='utf-8') as f_obj:
lines = f_obj.readlines()
_d2d_vel = {'hw_joint_vel_feedback': []}
_d2d_trq = {'device_servo_trq_feedback': []}
_d2d_sensor = {'hw_sensor_trq_feedback': []}
for line in lines[-300:]: # 保留最后15s的数据
data = eval(line.strip())['data']
for item in data:
try:
item['value'].reverse()
except KeyError:
continue
if item.get('channel', None) == channel-9 and item.get('name', None) == 'hw_joint_vel_feedback':
_d2d_vel['hw_joint_vel_feedback'].extend(item['value'])
elif item.get('channel', None) == channel-9 and item.get('name', None) == 'device_servo_trq_feedback':
_d2d_trq['device_servo_trq_feedback'].extend(item['value'])
elif item.get('channel', None) == channel-9 and item.get('name', None) == 'hw_sensor_trq_feedback':
_d2d_sensor['hw_sensor_trq_feedback'].extend(item['value'])
df1 = DataFrame.from_dict(_d2d_vel)
df2 = DataFrame.from_dict(_d2d_trq)
df3 = DataFrame.from_dict(_d2d_sensor)
df = concat([df1, df2, df3], axis=1)
_filename = f'{path}\\single\\j{channel-8}_hold_{time()}.data'
df.to_csv(_filename, sep='\t', index=False)
def data_proc_inertia(path, filename, channel):
with open(filename, 'r', encoding='utf-8') as f_obj:
lines = f_obj.readlines()
_d2d_vel = {'hw_joint_vel_feedback': []}
_d2d_trq = {'device_servo_trq_feedback': []}
_d2d_sensor = {'hw_sensor_trq_feedback': []}
for line in lines:
data = eval(line.strip())['data']
for item in data:
try:
item['value'].reverse()
except KeyError:
continue
if item.get('channel', None) == channel+3 and item.get('name', None) == 'hw_joint_vel_feedback':
_d2d_vel['hw_joint_vel_feedback'].extend(item['value'])
elif item.get('channel', None) == channel+3 and item.get('name', None) == 'device_servo_trq_feedback':
_d2d_trq['device_servo_trq_feedback'].extend(item['value'])
elif item.get('channel', None) == channel+3 and item.get('name', None) == 'hw_sensor_trq_feedback':
_d2d_trq['hw_sensor_trq_feedback'].extend(item['value'])
df1 = DataFrame.from_dict(_d2d_vel)
df2 = DataFrame.from_dict(_d2d_trq)
df3 = DataFrame.from_dict(_d2d_sensor)
df = concat([df1, df2, df3], axis=1)
_filename = f'{path}\\inertia\\j{channel+4}_inertia_{time()}.data'
df.to_csv(_filename, sep='\t', index=False)
def gen_result_file(path, loadsel, disc, number, scenario_time):
filename = path + f'\\data.txt'
with open(filename, 'w', encoding='utf-8') as f_obj:
for line in disc[number][1]:
f_obj.write(str(line)+'\n')
if loadsel == 'tool100':
data_proc_regular(path, filename, number, scenario_time)
elif loadsel == 'inertia':
data_proc_inertia(path, filename, number)
def run_rl(path, hr, md, sub, w2t):
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)",
"scenario(0, j3_p, j3_n, p_speed, p_tool, i_tool)",
"scenario(0, j4_p, j4_n, p_speed, p_tool, i_tool)",
"scenario(0, j5_p, j5_n, p_speed, p_tool, i_tool)",
"scenario(0, j6_p, j6_n, p_speed, p_tool, i_tool)",
"scenario(1, j6_p, j6_n, p_speed, p_tool, i_tool)",
"scenario(2, j6_p, j6_n, p_speed, p_tool, i_tool)",
"scenario(3, j6_p, j6_n, p_speed, p_tool, i_tool)",
"scenario(4, j1_hold, j1_hold, p_speed, p_tool, i_tool)",
"scenario(4, j2_hold, j2_hold, p_speed, p_tool, i_tool)",
"scenario(4, j3_hold, j3_hold, p_speed, p_tool, i_tool)",
"scenario(4, j4_hold, j4_hold, p_speed, p_tool, i_tool)",
"scenario(4, j5_hold, j5_hold, p_speed, p_tool, i_tool)",
"scenario(4, j6_hold, j6_hold, p_speed, p_tool, i_tool)",
]
c_inertia = [
"scenario(5, j4_p_inertia, j4_n_inertia, p_speed, p_tool, i_tool)",
"scenario(5, j5_p_inertia, j5_n_inertia, p_speed, p_tool, i_tool)",
"scenario(5, j6_p_inertia, j6_n_inertia, p_speed, p_tool, i_tool)",
]
disc_regular = {
0: ['一轴', []], 1: ['二轴', []], 2: ['三轴', []], 3: ['四轴', []], 4: ['五轴', []], 5: ['六轴', []],
6: ['场景一', []], 7: ['场景二', []], 8: ['场景三', []], 9: ['一轴保持', []], 10: ['二轴保持', []],
11: ['三轴保持', []], 12: ['四轴保持', []], 13: ['五轴保持', []], 14: ['六轴保持', []]
}
disc_inertia = {0: ['四轴惯量', []], 1: ['五轴惯量', []], 2: ['六轴惯量', []]}
if sub == 'tool100':
conditions = c_regular
disc = disc_regular
elif sub == 'inertia':
conditions = c_inertia
disc = disc_inertia
# preparation 触发软急停，并解除，目的是让可能正在运行着的机器停下来
hr.execution('diagnosis.open', hr, w2t, tab_name, open=True, display_open=True)
hr.execution('diagnosis.set_params', hr, w2t, tab_name, display_pdo_params=display_pdo_params)
# clibs.execution('diagnosis.save', hr, w2t, tab_name, save=True) # 这条命令有问题
md.trigger_estop()
md.reset_estop()
for condition in conditions:
number = conditions.index(condition)
w2t(f"正在执行{disc[number][0]}测试......", 0, 0, 'purple', tab_name)
# 1. 将act重置为False，并修改未要执行的场景
md.write_act(False)
ssh = SSHClient()
ssh.set_missing_host_key_policy(AutoAddPolicy())
ssh.connect(clibs.ip_addr, 22, username='luoshi', password='luoshi2019')
cmd = 'cd /home/luoshi/bin/controller/; '
cmd += 'sudo sed -i "/scenario/d" projects/target/_build/current/main.mod; '
cmd += f'sudo sed -i "/DONOTDELETE/i {condition}" projects/target/_build/current/main.mod'
stdin, stdout, stderr = ssh.exec_command(cmd, get_pty=True)
stdin.write('luoshi2019' + '\n')
stdout.read().decode() # 需要read一下才能正常执行
stderr.read().decode()
# 2. reload工程后，pp2main，并且自动模式和上电
prj_path = 'target/_build/target.prj'
clibs.execution('overview.reload', hr, w2t, tab_name, prj_path=prj_path, tasks=['current'])
clibs.execution('rl_task.pp_to_main', hr, w2t, tab_name, tasks=['current'])
clibs.execution('state.switch_auto', hr, w2t, tab_name)
clibs.execution('state.switch_motor_on', hr, w2t, tab_name)
# 3. 开始运行程序，单轴运行35s
clibs.execution('rl_task.set_run_params', hr, w2t, tab_name, loop_mode=True, override=1.0)
clibs.execution('rl_task.run', hr, w2t, tab_name, tasks=['current'])
_t_start = time()
while True:
if md.read_ready_to_go() == 1:
md.write_act(True)
break
else:
if (time() - _t_start) // 20 > 1:
w2t("20s内未收到机器人的运行信号，需要确认RL程序编写正确并正常执行...", 0, 111, 'red', tab_name)
else:
sleep(1)
# 4. 打开诊断曲线，并执行采集
sleep(10) # 保证程序已经运行起来，其实主要是为了保持电流的采集而设定
scenario_time = 0
if number < 6:
sleep(35)
elif number > 8:
sleep(15)
else:
_t_start = time()
while True:
scenario_time = md.read_scenario_time()
if float(scenario_time) > 1:
w2t(f"场景{number-5}的周期时间：{scenario_time}", 0, 0, 'green', tab_name)
break
else:
if (time()-_t_start)//60 > 3:
w2t(f"未收到场景{number-5}的周期时间，需要确认RL程序编写正确并正常执行...", 0, 111, 'red', tab_name)
else:
sleep(5)
sleep(1) # 一定要延迟一秒再读一次scenario time寄存器，因为一开始读取的数值不准确
scenario_time = md.read_scenario_time()
sleep(float(scenario_time)*0.2) # 再运行周期的20%即可
# 5.停止程序运行，保留数据并处理输出
clibs.execution('rl_task.stop', hr, w2t, tab_name, tasks=['current'])
_c_msg = hr.c_msg.copy()
for _msg in _c_msg:
if 'diagnosis.result' in _msg:
disc[number][1].insert(0, loads(_msg))
else:
hr.c_msg_xs.clear()
if len(hr.c_msg) > 270:
del hr.c_msg[270:]
gen_result_file(path, loadsel, disc, number, scenario_time)
else:
if loadsel == 'tool100':
w2t("单轴和场景电机电流采集完毕，如需采集惯量负载，须切换负载类型，并更换惯量负载，重新执行。", 0, 0, 'green', tab_name)
elif loadsel == 'inertia':
w2t("惯量负载电机电流采集完毕，如需采集单轴/场景/保持电机电流，须切换负载类型，并更换偏置负载，重新执行。", 0, 0, 'green', tab_name)
def main():
path = clibs.data_at["_path"]
sub = clibs.data_at["_sub"]
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)
config_file, prj_file = initialization(path, sub, data_dirs, data_files, hr, w2t)
clibs.c_pd.push_prj_to_server(prj_file)
clibs.c_hr.execution("diagnosis.open", open=False, display_open=False, overrun=True, turn_area=True, delay_motion=False)
# run_rl(path, hr, md, sub, w2t)
if __name__ == '__main__':
main()