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

v0.1.7.5(2024/07/03)

Browse Source 
1. [APIs: aio.py]
   - 增加触发急停和恢复急停功能逻辑
2. [APIs: do_current.py]
   - 重新调整运行顺序，增加数据处理的逻辑（惯量负载逻辑暂不实现，等待软件部解决了修改工程之后不生效的问题再考虑）
3. [APIs: btn_functions.py]
   - 增加触发急停和恢复急停的modbus实现，仅适用于自动化测试

v0.1.7.6(2024/07/04)
1. [APIs: aio.py]
   - Automatic Test逻辑中增加选择current时，需要选负载类型的逻辑
2. [APIs: do_current.py]
   - 单轴/场景电机电流的采集已完成
3. [APIs: openapi.py]
   - 增加了modbus读取浮点数的功能
   - 优化了get_from_id的逻辑
4. [autotest.xml]: 新增了scenario_time只写寄存器
This commit is contained in:
gitea
2024-07-04 13:34:04 +08:00
parent aedac4c90c
commit 524af160d8
6 changed files with 287 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files

234
aio/code/automatic_test/do_current.py
View File

@ -1,9 +1,11 @@
from time import sleep
import os
from time import sleep, time
from sys import argv
from os import scandir
from os.path import exists
from paramiko import SSHClient, AutoAddPolicy
from json import loads
import pandas
def traversal_files(path, w2t):
@ -22,7 +24,7 @@ def traversal_files(path, w2t):
return dirs, files
def check_files(data_dirs, data_files, w2t):
def check_files(path, data_dirs, data_files, w2t):
if len(data_dirs) != 0 or len(data_files) != 3:
w2t('初始路径下不允许有文件夹，且初始路径下只能存在如下三个文件，确认后重新运行！', 0, 0, 'red', tab_name='Automatic Test')
w2t(' 1. configs.xlsx\n 2. T_电机电流.xlsx\n 3. xxxx.zip', 0, 1, 'red', tab_name='Automatic Test')
@ -42,6 +44,12 @@ def check_files(data_dirs, data_files, w2t):
if config_file and current_file and prj_file:
w2t("数据目录合规性检查结束，未发现问题......", tab_name='Automatic Test')
os.mkdir(f"{path}\\single")
os.mkdir(f"{path}\\s_1")
os.mkdir(f"{path}\\s_2")
os.mkdir(f"{path}\\s_3")
os.mkdir(f"{path}\\inertia")
os.mkdir(f"{path}\\hold")
return config_file, current_file, prj_file
else:
w2t('初始路径下不允许有文件夹，且初始路径下只能存在如下三个文件，确认后重新运行！', 0, 0, 'red', tab_name='Automatic Test')
@ -56,11 +64,8 @@ def prj_to_xcore(prj_file):
# stdin, stdout, stderr = ssh.exec_command('rm /tmp/target.zip')
# ssh.exec_command('rm /tmp/target.zip')
sftp.put(prj_file, '/tmp/target.zip')
cmd = 'cd /tmp; '
cmd += 'rm -rf target/; '
cmd += 'mkdir target; '
cmd += 'unzip -d target/ -q target.zip; '
cmd += 'rm target.zip; '
cmd = 'cd /tmp; rm -rf target/; mkdir target; unzip -d target/ -q target.zip; '
cmd += 'chmod 777 -R target/; rm target.zip'
ssh.exec_command(cmd)
cmd = 'sudo rm -rf /home/luoshi/bin/controller/projects/target; '
@ -72,8 +77,8 @@ def prj_to_xcore(prj_file):
print(stderr.read().decode()) # 顺便也执行以下stderr
_prj_name = prj_file.split('\\')[-1].removesuffix('.zip')
cmd = f'cd /home/luoshi/bin/controller/; '
cmd += f'sudo mv projects/target/_build/{_prj_name}.prj projects/target/_build/target.prj '
cmd = 'cd /home/luoshi/bin/controller/; '
cmd += f'sudo mv projects/target/_build/{_prj_name}.prj projects/target/_build/target.prj'
stdin, stdout, stderr = ssh.exec_command(cmd, get_pty=True)
stdin.write('luoshi2019' + '\n')
stdin.flush()
@ -98,15 +103,125 @@ def execution(cmd, hr, w2t, **kwargs):
_id = hr.execution(cmd, **kwargs)
_msg = hr.get_from_id(_id)
if not _msg:
w2t(f"无法获取{_id}请求的响应信息", 0, 0, 'red', tab_name='Automatic Test')
w2t(f"无法获取{_id}请求的响应信息", 0, 7, 'red', tab_name='Automatic Test')
else:
_response = loads(_msg)
validate_resp(_id, _response, w2t)
return _response
def run_rl(hr, md, w2t):
# need to run
def data_proc(path, filename, channel):
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': []}
for line in lines:
data = eval(line.strip())['data']
for item in data:
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'])
df1 = pandas.DataFrame.from_dict(_d2d_vel)
df2 = pandas.DataFrame.from_dict(_d2d_trq)
df = pandas.concat([df1, df2], axis=1)
_filename = f'{path}\\single\\j{channel+1}_single.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_vel_1 = {'hw_joint_vel_feedback': []}
_d2d_trq_1 = {'device_servo_trq_feedback': []}
_d2d_vel_2 = {'hw_joint_vel_feedback': []}
_d2d_trq_2 = {'device_servo_trq_feedback': []}
_d2d_vel_3 = {'hw_joint_vel_feedback': []}
_d2d_trq_3 = {'device_servo_trq_feedback': []}
_d2d_vel_4 = {'hw_joint_vel_feedback': []}
_d2d_trq_4 = {'device_servo_trq_feedback': []}
_d2d_vel_5 = {'hw_joint_vel_feedback': []}
_d2d_trq_5 = {'device_servo_trq_feedback': []}
for line in lines:
data = eval(line.strip())['data']
for item in data:
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) == 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) == 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_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) == 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) == 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'])
df_01 = pandas.DataFrame.from_dict(_d2d_vel_0)
df_02 = pandas.DataFrame.from_dict(_d2d_trq_0)
df = pandas.concat([df_01, df_02], axis=1)
_filename = f'{path}\\s_{channel-5}\\j1_s_1.data'
df.to_csv(_filename, sep='\t', index=False)
df_01 = pandas.DataFrame.from_dict(_d2d_vel_1)
df_02 = pandas.DataFrame.from_dict(_d2d_trq_1)
df = pandas.concat([df_01, df_02], axis=1)
_filename = f'{path}\\s_{channel-5}\\j2_s_1.data'
df.to_csv(_filename, sep='\t', index=False)
df_01 = pandas.DataFrame.from_dict(_d2d_vel_2)
df_02 = pandas.DataFrame.from_dict(_d2d_trq_2)
df = pandas.concat([df_01, df_02], axis=1)
_filename = f'{path}\\s_{channel-5}\\j3_s_1.data'
df.to_csv(_filename, sep='\t', index=False)
df_01 = pandas.DataFrame.from_dict(_d2d_vel_3)
df_02 = pandas.DataFrame.from_dict(_d2d_trq_3)
df = pandas.concat([df_01, df_02], axis=1)
_filename = f'{path}\\s_{channel-5}\\j4_s_1.data'
df.to_csv(_filename, sep='\t', index=False)
df_01 = pandas.DataFrame.from_dict(_d2d_vel_4)
df_02 = pandas.DataFrame.from_dict(_d2d_trq_4)
df = pandas.concat([df_01, df_02], axis=1)
_filename = f'{path}\\s_{channel-5}\\j5_s_1.data'
df.to_csv(_filename, sep='\t', index=False)
df_01 = pandas.DataFrame.from_dict(_d2d_vel_5)
df_02 = pandas.DataFrame.from_dict(_d2d_trq_5)
df = pandas.concat([df_01, df_02], axis=1)
_filename = f'{path}\\s_{channel-5}\\j6_s_1.data'
df.to_csv(_filename, sep='\t', index=False)
def gen_result_file(path, loadsel, disc, number):
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(path, filename, number)
elif loadsel == 'inertia':
data_proc(path, filename, 'all')
def regular_load(path, hr, md, loadsel, w2t):
conditions = [
" scenario(0, j1_p, j1_n, p_speed, p_tool)",
" scenario(0, j2_p, j2_n, p_speed, p_tool)",
@ -118,45 +233,44 @@ def run_rl(hr, md, w2t):
" scenario(2, j5_p, j5_n, p_speed, p_tool)",
" scenario(3, j5_p, j5_n, p_speed, p_tool)",
]
for condition in conditions:
disc = {0: '一轴', 1: '二轴', 2: '三轴', 3: '四轴', 4: '五轴', 5: '六轴', 6: '场景一', 7: '场景二', 8: '场景三'}
disc = {
0: ['一轴', []], 1: ['二轴', []], 2: ['三轴', []], 3: ['四轴', []], 4: ['五轴', []], 5: ['六轴', []],
6: ['场景一', []], 7: ['场景二', []], 8: ['场景三', []]
}
number = conditions.index(condition)
w2t(f"正在执行{disc[number]}测试......", 0, 0, 'purple', 'Automatic Test')
w2t(f"正在执行{disc[number][0]}测试......", 0, 0, 'purple', 'Automatic Test')
# 1. 关闭诊断曲线，触发软急停，并解除，目的是让可能正在运行着的机器停下来
# 1. 关闭诊断曲线，触发软急停，并解除，目的是让可能正在运行着的机器停下来，切手动模式并下电
_response = execution('diagnosis.open', hr, w2t, open=False, display_open=False)
md.trigger_estop()
md.reset_estop()
_response = execution('state.switch_manual', hr, w2t)
_response = execution('state.switch_motor_off', hr, w2t)
# 3. 修改未要执行的场景
# 2. 修改未要执行的场景
ssh = SSHClient()
ssh.set_missing_host_key_policy(AutoAddPolicy())
ssh.connect('192.168.0.160', 22, username='luoshi', password='luoshi2019')
cmd = f'cd /home/luoshi/bin/controller/; '
cmd += f'sudo sed -i "/scenario/d" projects/target/current/main.mod; '
cmd += f'sudo sed -i "/DONOTDELETE/i {condition}" projects/target/current/main.mod; '
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')
stdin.flush()
print(stdout.read().decode()) # 必须得输出一下stdout，才能正确执行sudo
print(stderr.read().decode()) # 顺便也执行以下stderr
# 4. reload工程后，pp2main
# 3. reload工程后，pp2main，并且自动模式和上电
prj_path = 'target/_build/target.prj'
_response = execution('overview.reload', hr, w2t, prj_path=prj_path, tasks=['current'])
_response = execution('overview.get_cur_prj', hr, w2t)
_response = execution('rl_task.pp_to_main', hr, w2t, tasks=['current'])
print(f"set pp2main of prj: {_response}")
# 5. 切换自动并上电
_response = execution('state.switch_auto', hr, w2t)
_response = execution('state.switch_motor_on', hr, w2t)
# 6. 开始运行程序，单轴运行15s
# 4. 开始运行程序，单轴运行15s
_response = execution('rl_task.run', hr, w2t, tasks=['current'])
print(f"run prj: {_response}")
for i in range(3):
if md.read_ready_to_go() == 1:
md.write_act(True)
@ -165,10 +279,10 @@ def run_rl(hr, md, w2t):
break
else:
sleep(1)
continue
else:
w2t("未收到机器人的运行信号，需要确认RL程序编写正确并正常执行...", 0, 111, 'red', 'Automatic Test')
# 7. 打开诊断曲线，并执行采集
# 5. 打开诊断曲线，并执行采集
_response = execution('diagnosis.open', hr, w2t, open=True, display_open=True)
display_pdo_params = [
{"name": "hw_joint_vel_feedback", "channel": 0},
@ -186,32 +300,62 @@ def run_rl(hr, md, w2t):
{"name": "device_safety_estop", "channel": 0},
]
_response = execution('diagnosis.set_params', hr, w2t, display_pdo_params=display_pdo_params)
sleep(8)
# 8. 关闭诊断曲线，停止程序运行，下电并且换成手动模式
if number < 6:
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', 'Automatic Test')
break
else:
if (time()-_t_start)//60 > 3:
w2t(f"未收到场景{number-5}的周期时间，需要确认RL程序编写正确并正常执行...", 0, 111, 'red', 'Automatic Test')
else:
sleep(5)
sleep(1) # 一定要延迟一秒再读一次scenario time寄存器，因为一开始读取的数值不准确
sleep(float(md.read_scenario_time())*0.2) # 再运行周期的20%即可
# 6. 关闭诊断曲线，停止程序运行，下电并且换成手动模式
_response = execution('diagnosis.open', hr, w2t, open=False, display_open=False)
_response = execution('rl_task.stop', hr, w2t, tasks=['current'])
_response = execution('state.switch_motor_off', hr, w2t)
_response = execution('state.switch_manual', hr, w2t)
sleep(1)
# 9. 处理数据并输出文件
# for _msg in hr.c_msg:
# if 'diagnosis.result' in _msg:
# print(_msg)
# _msg = json.loads(_msg)
# if 'channel' in _msg and 'name' in _msg:
# if int(_msg['channel']) == 0 and _msg['name'] == 'device_servo_trq_feedback':
# print(f"diagnosis.result: {_msg}")
# count += 1
# if count * 50 > 5 * 1000:
# break
sleep(2) # 保证所有数据均已返回
# 7. 保留数据并处理输出
for _msg in hr.c_msg:
if 'diagnosis.result' in _msg:
disc[number][1].insert(0, loads(_msg))
else:
_index = 210
for _msg in hr.c_msg:
if 'diagnosis.result' in _msg:
_index = hr.c_msg.index(_msg)
break
del hr.c_msg[_index-10:]
hr.c_msg_xs.clear()
gen_result_file(path, loadsel, disc, number)
else:
w2t("满载情况下，单轴和场景电机电流采集完毕。", 0, 0, 'green', 'Automatic Test')
def main(path, hr, md, w2t):
def inertia_load(path, hr, md, loadsel, w2t):
pass
def run_rl(path, hr, md, loadsel, w2t):
if loadsel == 'tool100':
regular_load(path, hr, md, loadsel, w2t)
elif loadsel == 'inertia':
inertia_load(path, hr, md, loadsel, w2t)
def main(path, hr, md, loadsel, w2t):
data_dirs, data_files = traversal_files(path, w2t)
config_file, current_file, prj_file = check_files(data_dirs, data_files, w2t)
config_file, current_file, prj_file = check_files(path, data_dirs, data_files, w2t)
prj_to_xcore(prj_file)
run_rl(hr, md, w2t)
run_rl(path, hr, md, loadsel, w2t)
if __name__ == '__main__':