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

7. [APIs: btn_functions.py]: 重写了告警输出函数，从日志中拿数据

Browse Source 
8. [APIs: aio.py]: 将日志框输出的内容，也保存至日志文件
9. [APIs: do_brake.py]
   - 修改获取初始速度的逻辑，只获取configs文件中配置的时间内的速度
   - 新增 configs 参数 single_brake，可针对特定条件做测试
This commit is contained in:
gitea
2024-08-01 17:11:12 +08:00
parent e713485d00
commit 60726d9d07
6 changed files with 41 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
aio/README.md
View File

@ -590,4 +590,8 @@ v0.2.0.5(2024/07/31)
- 保持电流，只取最后 15s
- 优化 ssh 输入密码的部分
6. [t_change_ui: all the part]: 引入 commons 包，并定制了 logging 输出，后续持续优化
7. [APIs: btn_functions.py]: 重写了告警输出函数，从日志中拿数据
8. [APIs: aio.py]: 将日志框输出的内容，也保存至日志文件
9. [APIs: do_brake.py]
- 修改获取初始速度的逻辑，只获取configs文件中配置的时间内的速度
- 新增 configs 参数 single_brake，可针对特定条件做测试

BIN
aio/assets/configs.xlsx
View File

Binary file not shown.

1
aio/code/aio.py
View File

@ -449,6 +449,7 @@ class App(customtkinter.CTk):
self.textbox.tag_add(color, 'insert', 'end')
self.textbox.tag_config(tagName=color, foreground=color)
tab_name_cur = self.tabview.get()
logger.info(text)
if tab_name == tab_name_cur:
if wait != 0:

15
aio/code/automatic_test/btn_functions.py
View File

@ -38,12 +38,15 @@ def get_state(hr, w2t):
def warning_info(hr, w2t):
for msg in hr.c_msg:
if 'alarm' in msg.lower():
w2t(str(loads(msg)), tab_name=tab_name)
for msg in hr.c_msg_xs:
if 'alarm' in msg.lower():
w2t(str(loads(msg)), tab_name=tab_name)
for postfix in ['', '.2', '.3', '.4', '.5', '.6', '.7', '.8', '.9', '.10']:
log_name = clibs.log_data + 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):

29
aio/code/automatic_test/do_brake.py
View File

@ -61,7 +61,7 @@ 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:]:
for data in curve_data[-240:]: # 保留最后12s的数据
dict_results = data['data']
for item in dict_results:
try:
@ -85,6 +85,7 @@ def gen_result_file(path, curve_data, axis, _reach, _load, _speed, count):
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},
@ -104,6 +105,8 @@ def run_rl(path, loadsel, hr, md, config_file, result_dirs, w2t):
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 = ws.cell(row=5, column=10).value
if ws.cell(row=1, column=1).value == 'positive':
md.write_pon(1)
elif ws.cell(row=1, column=1).value == 'negative':
@ -118,7 +121,19 @@ def run_rl(path, loadsel, hr, md, config_file, result_dirs, w2t):
_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':
@ -130,7 +145,7 @@ def run_rl(path, loadsel, hr, md, config_file, result_dirs, w2t):
_count += 1
this_time = strftime("%Y-%m-%d %H:%M:%S", localtime(time()))
prj_path = 'target/_build/target.prj'
w2t(f"[{this_time} | {_count}/63] 正在执行 {axis} 轴 {condition} 的第 {count} 次制动测试...", 0, 0, 'purple', tab_name)
w2t(f"[{this_time} | {_count}/{_total}] 正在执行 {axis} 轴 {condition} 的第 {count} 次制动测试...", 0, 0, 'purple', tab_name)
# 1. 关闭诊断曲线，触发软急停，并解除，目的是让可能正在运行着的机器停下来，切手动模式并下电
md.trigger_estop()
@ -180,13 +195,17 @@ def run_rl(path, loadsel, hr, md, config_file, result_dirs, w2t):
w2t("20s内未收到机器人的运行信号，需要确认RL程序编写正确并正常执行...", 0, 111, 'red', tab_name)
else:
sleep(1)
# 4. 打开诊断曲线，并执行采集，之后触发软急停，关闭曲线采集，找出最大速度，传递给RL程序，最后清除相关记录
sleep(get_init_speed) # 获取实际最大速度，可通过configs.xlsx配置
# 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':
@ -195,7 +214,7 @@ def run_rl(path, loadsel, hr, md, config_file, result_dirs, w2t):
speed_max = max(_, speed_max)
elif ws.cell(row=1, column=1).value == 'negative':
speed_max = min(_, speed_max)
print(f"speed max = {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:

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

@ -64,7 +64,7 @@ def data_proc_regular(path, filename, channel, scenario_time):
lines = f_obj.readlines()
_d2d_vel = {'hw_joint_vel_feedback': []}
_d2d_trq = {'device_servo_trq_feedback': []}
for line in lines:
for line in lines[-500:]: # 保留最后25s的数据
data = eval(line.strip())['data']
for item in data:
try:
@ -168,7 +168,7 @@ def data_proc_regular(path, filename, channel, scenario_time):
lines = f_obj.readlines()
_d2d_vel = {'hw_joint_vel_feedback': []}
_d2d_trq = {'device_servo_trq_feedback': []}
for line in lines[-300:]:
for line in lines[-300:]: # 保留最后15s的数据
data = eval(line.strip())['data']
for item in data:
try: