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

minor modifications

Browse Source
This commit is contained in:
gitea
2024-07-18 14:59:20 +08:00
parent dd0873637f
commit 1cefe4a16b
5 changed files with 37 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@ -4,21 +4,21 @@ from os import scandir
from paramiko import SSHClient, AutoAddPolicy
from json import loads
from time import sleep, time, strftime, localtime
import pandas as pd
from pandas import DataFrame
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'
durable_data_current_max_xlsx = f'{dirname(__file__)}/../../assets/templates/durable/durable_data_current_max.xlsx'
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": "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},
@ -35,7 +35,7 @@ durable_data_current = {
'axis5': [0 for _ in range(25)],
'axis6': [0 for _ in range(25)],
}
durable_data_velocity = {
durable_data_current_max = {
'time': list(range(1, 26)),
'axis1': [0 for _ in range(25)],
'axis2': [0 for _ in range(25)],
@ -44,7 +44,7 @@ durable_data_velocity = {
'axis5': [0 for _ in range(25)],
'axis6': [0 for _ in range(25)],
}
data_all = [durable_data_current, durable_data_velocity]
data_all = [durable_data_current, durable_data_current_max]
def traversal_files(path, w2t):
@ -214,44 +214,33 @@ def get_durable_data(path, data, scenario_time, wait_time, rcs, hr, w2t):
# for _ in _data_list:
# f_obj.write(f"{_}\n")
_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': [],
}
_d2d_vel = {
'hw_joint_vel_feedback_0': [], 'hw_joint_vel_feedback_1': [], 'hw_joint_vel_feedback_2': [],
'hw_joint_vel_feedback_3': [], 'hw_joint_vel_feedback_4': [], 'hw_joint_vel_feedback_5': [],
}
_d2d_trq = {0: [], 1: [], 2: [], 3: [], 4: [], 5: []}
_d2d_trq_max = {0: [], 1: [], 2: [], 3: [], 4: [], 5: []}
for line in _data_list:
for item in line['data']:
for i in range(6):
item['value'].reverse()
if item.get('channel', None) == i and item.get('name', None) == 'device_servo_trq_feedback':
_d2d_trq[f'device_servo_trq_feedback_{i}'].extend(item['value'])
elif item.get('channel', None) == i and item.get('name', None) == 'hw_joint_vel_feedback':
_d2d_vel[f'hw_joint_vel_feedback_{i}'].extend(item['value'])
_d2d_trq[i].extend(item['value'])
if len(_d2d_trq['device_servo_trq_feedback_0']) / 1000 > scenario_time + 1:
# ========= for trq =========
_df_1 = pd.DataFrame(_d2d_trq)
if len(_d2d_trq[0]) / 1000 > scenario_time + 1:
_df = DataFrame(_d2d_trq)
for i in range(6):
_ = 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}']))
_ = sqrt(100*_df[i].apply(lambda x: (rcs[i]*x/10000)**2).sum()/len(_df[i]))
del data[0][f"axis{i + 1}"][0]
data[0][f"axis{i + 1}"].append(_)
_df_1 = pd.DataFrame(data[0])
# ========= for vel =========
_df_2 = pd.DataFrame(_d2d_vel)
for i in range(6):
_ = 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}']))
_ = rcs[i] * _df[i].abs().max() / 1000
del data[1][f"axis{i + 1}"][0]
data[1][f"axis{i + 1}"].append(_)
_df_2 = pd.DataFrame(data[1])
_df_1 = DataFrame(data[0])
_df_2 = DataFrame(data[1])
while True:
if not hr.durable_lock:
hr.durable_lock = 1
_df_1.to_excel(durable_data_current_xlsx, index=False)
_df_2.to_excel(durable_data_velocity_xlsx, index=False)
_df_2.to_excel(durable_data_current_max_xlsx, index=False)
hr.durable_lock = 0
break
else: