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

v0.1.7.2(2024/06/30)

Browse Source 
1. 初步完成NB4h_R580_3BH7.zip工程的设计
2. 重新研究了解包操作，重新实现了一版
3. 修改openapi.pi中excution为execution函数
4. 增减了解包原理性文档
This commit is contained in:
gitea
2024-06-30 20:29:49 +08:00
parent ac71522cc2
commit fc3d5482f8
8 changed files with 1236 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@ -7,9 +7,6 @@ from json import loads
def traversal_files(path, w2t):
# 功能：以列表的形式分别返回指定路径下的文件和文件夹，不包含子目录
# 参数：路径
# 返回值：路径下的文件夹列表 路径下的文件列表
if not exists(path):
msg = f'数据文件夹{path}不存在，请确认后重试......'
w2t(msg, 0, 1, 'red', tab_name='Automatic Test')
@ -26,33 +23,29 @@ def traversal_files(path, w2t):
def check_files(data_dirs, data_files, w2t):
if len(data_dirs) != 0 or len(data_files) != 5:
w2t('初始路径下不允许有文件夹，且初始路径下只能存在如下五个文件，确认后重新运行！', 0, 0, 'red', tab_name='Automatic Test')
w2t(' 1. configs.xlsx\n 2. reach33/reach66/reach100_xxxx.xlsx\n 3. xxxx.zip', 0, 1, 'red', tab_name='Automatic Test')
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')
config_file = reach33 = reach66 = reach100 = prj_file = None
config_file = current_file = 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 == 'T_电机电流.xlsx':
current_file = data_file
elif filename.endswith('.zip'):
prj_file = data_file
else:
w2t('初始路径下不允许有文件夹，且初始路径下只能存在如下五个文件，确认后重新运行！', 0, 0, 'red', tab_name='Automatic Test')
w2t(' 1. configs.xlsx\n 2. reach33/reach66/reach100_xxxx.xlsx\n 3. xxxx.zip', 0, 2, 'red', tab_name='Automatic Test')
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')
if config_file and reach33 and reach66 and reach100 and prj_file:
if config_file and current_file and prj_file:
w2t("数据目录合规性检查结束，未发现问题......", tab_name='Automatic Test')
return config_file, reach33, reach66, reach100, prj_file
return config_file, current_file, prj_file
else:
w2t('初始路径下不允许有文件夹，且初始路径下只能存在如下五个文件，确认后重新运行！', 0, 0, 'red', tab_name='Automatic Test')
w2t(' 1. configs.xlsx\n 2. reach33/reach66/reach100_xxxx.xlsx\n 3. xxxx.zip', 0, 1, 'red', tab_name='Automatic Test')
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')
def prj_to_xcore(prj_file):
@ -105,7 +98,7 @@ def validate_resp(_id, response, w2t):
def execution(cmd, hr, w2t, **kwargs):
_id = hr.excution(cmd, **kwargs)
_id = hr.execution(cmd, **kwargs)
_msg = hr.get_from_id(_id)
if not _msg:
w2t(f"无法获取{_id}请求的响应信息", 0, 6, 'red', tab_name='Automatic Test')
@ -116,38 +109,54 @@ def execution(cmd, hr, w2t, **kwargs):
def run_rl(hr, w2t):
# prj_path = 'target/_build/target.prj'
# _response = execution('overview.reload', hr, w2t, prj_path=prj_path, tasks=['Durable_Test_Com', 'Mechanical_Test_Com'])
# print(f"reload prj: {_response}")
# # _response = execution('overview.get_cur_prj', hr, w2t)
# # print(f"get cur prj name: {_response}")
# _response = execution('rl_task.pp_to_main', hr, w2t, tasks=['Mechanical_Test_Com'])
# print(f"set pp2main of prj: {_response}")
#
# _response = execution('state.switch_auto', hr, w2t)
# _response = execution('state.switch_motor_on', hr, w2t)
#
# _response = execution('rl_task.run', hr, w2t, tasks=['Mechanical_Test_Com'])
# print(f"run prj: {_response}")
# sleep(10)
#
# _response = execution('state.switch_motor_off', hr, w2t)
# _response = execution('state.switch_manual', hr, w2t)
prj_path = 'target/_build/target.prj'
_response = execution('overview.reload', hr, w2t, prj_path=prj_path, tasks=['brake', 'current'])
print(f"reload prj: {_response}")
_response = execution('overview.get_cur_prj', hr, w2t)
print(f"get cur prj name: {_response}")
_response = execution('rl_task.pp_to_main', hr, w2t, tasks=['brake'])
print(f"set pp2main of prj: {_response}")
_response = execution('state.switch_auto', hr, w2t)
_response = execution('state.switch_motor_on', hr, w2t)
_response = execution('diagnosis.open', hr, w2t, open=True, display_open=True)
print(f"打开诊断: {_response}")
# _response = execution('diagnosis.get_params', hr, w2t)
# print(f"显示诊断状态: {_response}")
display_pdo_params = [
{"name": "device_servo_trq_feedback", "channel": 0}
{"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},
{"name": "device_safety_estop", "channel": 0},
{"name": "device_safety_estop", "channel": 0},
{"name": "device_safety_estop", "channel": 0},
{"name": "device_safety_estop", "channel": 0},
{"name": "device_safety_estop", "channel": 0},
]
_response = execution('diagnosis.set_params', hr, w2t, display_pdo_params=display_pdo_params)
print(f"执行采样: {_response}")
sleep(5)
_response = execution('rl_task.run', hr, w2t, tasks=['brake'])
print(f"run prj: {_response}")
sleep(10)
_response = execution('diagnosis.open', hr, w2t, open=False, display_open=False)
print(f"关闭诊断: {_response}")
# sleep(1)
_response = execution('state.switch_motor_off', hr, w2t)
_response = execution('state.switch_manual', hr, w2t)
for _msg in hr.c_msg:
if 'diagnosis.result' in _msg:
print(_msg)
@ -160,14 +169,11 @@ def run_rl(hr, w2t):
# break
else:
sleep(1)
def main(path, hr, w2t):
def main(path, hr, loadsel, w2t):
data_dirs, data_files = traversal_files(path, w2t)
config_file, reach33, reach66, reach100, prj_file = check_files(data_dirs, data_files, w2t)
# prj_to_xcore(prj_file)
config_file, current_file, prj_file = check_files(data_dirs, data_files, w2t)
prj_to_xcore(prj_file)
run_rl(hr, w2t)