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

minor fix

Browse Source
This commit is contained in:
gitea
2024-07-18 11:04:31 +08:00
parent 5ab03d23f2
commit dd0873637f
5 changed files with 38 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

61
aio/code/durable_action/factory_test.py
View File

@ -3,12 +3,13 @@ from os.path import exists, dirname
from os import scandir
from paramiko import SSHClient, AutoAddPolicy
from json import loads
from time import sleep, time
from time import sleep, time, strftime, localtime
import pandas as pd
from openpyxl import load_workbook
from math import sqrt
tab_name = 'Durable Action'
count = 0
durable_data_current_xlsx = f'{dirname(__file__)}/../../assets/templates/durable/durable_data_current.xlsx'
durable_data_velocity_xlsx = f'{dirname(__file__)}/../../assets/templates/durable/durable_data_velocity.xlsx'
display_pdo_params = [
@ -170,26 +171,32 @@ def run_rl(path, config_file, hr, md, w2t):
# 6. 关闭诊断曲线，停止程序运行，下电并且换成手动模式
_response = execution('diagnosis.open', hr, w2t, open=False, display_open=False)
_response = execution('diagnosis.set_params', hr, w2t, display_pdo_params=[])
_response = execution('rl_task.stop', hr, w2t, tasks=['current'])
sleep(1) # 保证所有数据均已返回
# 7. 保留数据并处理输出
get_durable_data(path, config_file, data_all, scenario_time, hr, w2t)
_wb = load_workbook(config_file, read_only=True)
_ws = _wb['Target']
wait_time = float(_ws.cell(row=2, column=10).value)
rcs = []
for i in range(6):
rcs.append(float(_ws.cell(row=6, column=i + 2).value))
get_durable_data(path, data_all, scenario_time, wait_time, rcs, hr, w2t)
# 8. 继续运行
_response = execution('rl_task.run', hr, w2t, tasks=['current'])
while True:
# 每3分钟，更新一次数据，打开曲线，获取周期内电流，关闭曲线
sleep(180)
sleep(wait_time)
_response = execution('diagnosis.open', hr, w2t, open=True, display_open=True)
_response = execution('diagnosis.set_params', hr, w2t, display_pdo_params=display_pdo_params)
sleep(scenario_time + 10)
sleep(scenario_time+5)
_response = execution('diagnosis.open', hr, w2t, open=False, display_open=False)
_response = execution('diagnosis.set_params', hr, w2t, display_pdo_params=[])
sleep(2)
# 7. 保留数据并处理输出
get_durable_data(path, config_file, data_all, scenario_time, hr, w2t)
get_durable_data(path, data_all, scenario_time, wait_time, rcs, hr, w2t)
def get_durable_data(path, config_file, data, scenario_time, hr, w2t):
def get_durable_data(path, data, scenario_time, wait_time, rcs, hr, w2t):
_data_list = []
for _msg in hr.c_msg:
if 'diagnosis.result' in _msg:
@ -207,12 +214,6 @@ def get_durable_data(path, config_file, data, scenario_time, hr, w2t):
# for _ in _data_list:
# f_obj.write(f"{_}\n")
_wb = load_workbook(config_file, read_only=True)
_ws = _wb['Target']
rcs = []
for i in range(6):
rcs.append(float(_ws.cell(row=6, column=i + 2).value))
_d2d_trq = {
'device_servo_trq_feedback_0': [], 'device_servo_trq_feedback_1': [], 'device_servo_trq_feedback_2': [],
'device_servo_trq_feedback_3': [], 'device_servo_trq_feedback_4': [], 'device_servo_trq_feedback_5': [],
@ -231,35 +232,35 @@ def get_durable_data(path, config_file, data, scenario_time, hr, w2t):
_d2d_vel[f'hw_joint_vel_feedback_{i}'].extend(item['value'])
if len(_d2d_trq['device_servo_trq_feedback_0']) / 1000 > scenario_time + 1:
_df = pd.DataFrame(_d2d_trq)
# ========= for trq =========
_df_1 = pd.DataFrame(_d2d_trq)
for i in range(6):
_ = sqrt(100*_df[f'device_servo_trq_feedback_{i}'].apply(lambda x: (rcs[i]*x/10000)**2).sum()/len(_df[f'device_servo_trq_feedback_{i}']))
_ = sqrt(100*_df_1[f'device_servo_trq_feedback_{i}'].apply(lambda x: (rcs[i]*x/10000)**2).sum()/len(_df[f'device_servo_trq_feedback_{i}']))
del data[0][f"axis{i + 1}"][0]
data[0][f"axis{i + 1}"].append(_)
_df = pd.DataFrame(data[0])
while True:
if not hr.durable_lock:
hr.durable_lock = 1
_df.to_excel(durable_data_current_xlsx, index=False)
hr.durable_lock = 0
break
else:
sleep(1)
_df = pd.DataFrame(_d2d_vel)
_df_1 = pd.DataFrame(data[0])
# ========= for vel =========
_df_2 = pd.DataFrame(_d2d_vel)
for i in range(6):
_ = sqrt(100*_df[f'hw_joint_vel_feedback_{i}'].apply(lambda x: (rcs[i]*x/10000)**2).sum()/len(_df[f'hw_joint_vel_feedback_{i}']))
_ = sqrt(100*_df_2[f'hw_joint_vel_feedback_{i}'].apply(lambda x: (rcs[i]*x/10000)**2).sum()/len(_df[f'hw_joint_vel_feedback_{i}']))
del data[1][f"axis{i + 1}"][0]
data[1][f"axis{i + 1}"].append(_)
_df = pd.DataFrame(data[1])
_df_2 = pd.DataFrame(data[1])
while True:
if not hr.durable_lock:
hr.durable_lock = 1
_df.to_excel(durable_data_velocity_xlsx, index=False)
_df_1.to_excel(durable_data_current_xlsx, index=False)
_df_2.to_excel(durable_data_velocity_xlsx, index=False)
hr.durable_lock = 0
break
else:
sleep(1)
global count
count += 1
this_time = strftime("%Y-%m-%d %H:%M:%S", localtime(time()))
next_time = strftime("%Y-%m-%d %H:%M:%S", localtime(time()+wait_time+10+scenario_time)).split()[-1]
w2t(f"[{this_time}] 当前次数：{count:09d} | 预计下次数据更新时间：{next_time}", 0, 0, '#008B8B', tab_name)
break
else:
with open(f'{path}\\device_servo_trq_feedback_0.txt', 'w', encoding='utf-8') as f_obj: