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

1. [current: do_current.py] 增加了 hw_sensor_trq_feedback 曲线的采集

Browse Source 
2. [current: current.py] 增加了 hw_sensor_trq_feedback 曲线数据的处理，以及修改了之前数据处理的逻辑
3. [current: clibs.py] 新增可手动修改连接 IP 地址的功能，存储在 assets/templates/ipaddr.txt 中，默认是 192.168.0.160
This commit is contained in:
gitea
2024-12-05 16:14:59 +08:00
parent 5c5168442f
commit 4d297118e0
10 changed files with 83 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
aio/code/aio.py
View File

@ -96,7 +96,7 @@ class App(customtkinter.CTk):
btns_func['log']['btn'].configure(command=lambda: self.thread_it(self.func_log_callback))
btns_func['end']['btn'].configure(command=lambda: self.thread_it(self.func_end_callback))
# 1.3 create version info
self.label_version = customtkinter.CTkLabel(self.frame_func, justify='left', text="Vers: 0.2.0.9\nDate: 10/09/2024", font=self.my_font, text_color="#4F4F4F")
self.label_version = customtkinter.CTkLabel(self.frame_func, justify='left', text="Vers: 0.2.1.0\nDate: 12/05/2024", font=self.my_font, text_color="#4F4F4F")
self.frame_func.rowconfigure(6, weight=1)
self.label_version.grid(row=6, column=0, padx=20, pady=20, sticky='s')
# =====================================================================

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

@ -21,6 +21,12 @@ display_pdo_params = [
{"name": "device_servo_trq_feedback", "channel": 3},
{"name": "device_servo_trq_feedback", "channel": 4},
{"name": "device_servo_trq_feedback", "channel": 5},
{"name": "hw_sensor_trq_feedback", "channel": 0},
{"name": "hw_sensor_trq_feedback", "channel": 1},
{"name": "hw_sensor_trq_feedback", "channel": 2},
{"name": "hw_sensor_trq_feedback", "channel": 3},
{"name": "hw_sensor_trq_feedback", "channel": 4},
{"name": "hw_sensor_trq_feedback", "channel": 5},
]
@ -63,6 +69,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': []}
_d2d_sensor = {'hw_sensor_trq_feedback': []}
for line in lines[-500:]: # 保留最后25s的数据
data = eval(line.strip())['data']
for item in data:
@ -74,10 +81,13 @@ def data_proc_regular(path, filename, channel, scenario_time):
_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'])
elif item.get('channel', None) == channel and item.get('name', None) == 'hw_sensor_trq_feedback':
_d2d_sensor['hw_sensor_trq_feedback'].extend(item['value'])
df1 = DataFrame.from_dict(_d2d_vel)
df2 = DataFrame.from_dict(_d2d_trq)
df = concat([df1, df2], axis=1)
df3 = DataFrame.from_dict(_d2d_sensor)
df = concat([df1, df2, df3], axis=1)
_filename = f'{path}\\single\\j{channel+1}_single_{time()}.data'
df.to_csv(_filename, sep='\t', index=False)
elif channel in list(range(6, 9)):
@ -85,16 +95,22 @@ def data_proc_regular(path, filename, channel, scenario_time):
lines = f_obj.readlines()
_d2d_vel_0 = {'hw_joint_vel_feedback': []}
_d2d_trq_0 = {'device_servo_trq_feedback': []}
_d2d_sensor_0 = {'hw_sensor_trq_feedback': []}
_d2d_vel_1 = {'hw_joint_vel_feedback': []}
_d2d_trq_1 = {'device_servo_trq_feedback': []}
_d2d_sensor_1 = {'hw_sensor_trq_feedback': []}
_d2d_vel_2 = {'hw_joint_vel_feedback': []}
_d2d_trq_2 = {'device_servo_trq_feedback': []}
_d2d_sensor_2 = {'hw_sensor_trq_feedback': []}
_d2d_vel_3 = {'hw_joint_vel_feedback': []}
_d2d_trq_3 = {'device_servo_trq_feedback': []}
_d2d_sensor_3 = {'hw_sensor_trq_feedback': []}
_d2d_vel_4 = {'hw_joint_vel_feedback': []}
_d2d_trq_4 = {'device_servo_trq_feedback': []}
_d2d_sensor_4 = {'hw_sensor_trq_feedback': []}
_d2d_vel_5 = {'hw_joint_vel_feedback': []}
_d2d_trq_5 = {'device_servo_trq_feedback': []}
_d2d_sensor_5 = {'hw_sensor_trq_feedback': []}
for line in lines:
data = eval(line.strip())['data']
for item in data:
@ -106,60 +122,78 @@ def data_proc_regular(path, filename, channel, scenario_time):
_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) == 0 and item.get('name', None) == 'hw_sensor_trq_feedback':
_d2d_sensor_0['hw_sensor_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) == 1 and item.get('name', None) == 'hw_sensor_trq_feedback':
_d2d_sensor_1['hw_sensor_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_sensor_trq_feedback':
_d2d_sensor_2['hw_sensor_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) == 3 and item.get('name', None) == 'hw_sensor_trq_feedback':
_d2d_sensor_3['hw_sensor_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) == 4 and item.get('name', None) == 'hw_sensor_trq_feedback':
_d2d_sensor_4['hw_sensor_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'])
elif item.get('channel', None) == 5 and item.get('name', None) == 'hw_sensor_trq_feedback':
_d2d_sensor_5['hw_sensor_trq_feedback'].extend(item['value'])
df_01 = DataFrame.from_dict(_d2d_vel_0)
df_02 = DataFrame.from_dict(_d2d_trq_0)
df = concat([df_01, df_02], axis=1)
df_03 = DataFrame.from_dict(_d2d_sensor_0)
df = concat([df_01, df_02, df_03], axis=1)
_filename = f'{path}\\s_{channel-5}\\j1_s_{channel-5}_{scenario_time}_{time()}.data'
df.to_csv(_filename, sep='\t', index=False)
df_01 = DataFrame.from_dict(_d2d_vel_1)
df_02 = DataFrame.from_dict(_d2d_trq_1)
df = concat([df_01, df_02], axis=1)
df_03 = DataFrame.from_dict(_d2d_sensor_1)
df = concat([df_01, df_02, df_03], axis=1)
_filename = f'{path}\\s_{channel-5}\\j2_s_{channel-5}_{scenario_time}_{time()}.data'
df.to_csv(_filename, sep='\t', index=False)
df_01 = DataFrame.from_dict(_d2d_vel_2)
df_02 = DataFrame.from_dict(_d2d_trq_2)
df = concat([df_01, df_02], axis=1)
df_03 = DataFrame.from_dict(_d2d_sensor_2)
df = concat([df_01, df_02, df_03], axis=1)
_filename = f'{path}\\s_{channel-5}\\j3_s_{channel-5}_{scenario_time}_{time()}.data'
df.to_csv(_filename, sep='\t', index=False)
df_01 = DataFrame.from_dict(_d2d_vel_3)
df_02 = DataFrame.from_dict(_d2d_trq_3)
df = concat([df_01, df_02], axis=1)
df_03 = DataFrame.from_dict(_d2d_sensor_3)
df = concat([df_01, df_02, df_03], axis=1)
_filename = f'{path}\\s_{channel-5}\\j4_s_{channel-5}_{scenario_time}_{time()}.data'
df.to_csv(_filename, sep='\t', index=False)
df_01 = DataFrame.from_dict(_d2d_vel_4)
df_02 = DataFrame.from_dict(_d2d_trq_4)
df = concat([df_01, df_02], axis=1)
df_03 = DataFrame.from_dict(_d2d_sensor_4)
df = concat([df_01, df_02, df_03], axis=1)
_filename = f'{path}\\s_{channel-5}\\j5_s_{channel-5}_{scenario_time}_{time()}.data'
df.to_csv(_filename, sep='\t', index=False)
df_01 = DataFrame.from_dict(_d2d_vel_5)
df_02 = DataFrame.from_dict(_d2d_trq_5)
df = concat([df_01, df_02], axis=1)
df_03 = DataFrame.from_dict(_d2d_sensor_5)
df = concat([df_01, df_02, df_03], axis=1)
_filename = f'{path}\\s_{channel-5}\\j6_s_{channel-5}_{scenario_time}_{time()}.data'
df.to_csv(_filename, sep='\t', index=False)
elif channel in list(range(9, 15)):
@ -167,6 +201,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': []}
_d2d_sensor = {'hw_sensor_trq_feedback': []}
for line in lines[-300:]: # 保留最后15s的数据
data = eval(line.strip())['data']
for item in data:
@ -178,10 +213,13 @@ def data_proc_regular(path, filename, channel, scenario_time):
_d2d_vel['hw_joint_vel_feedback'].extend(item['value'])
elif item.get('channel', None) == channel-9 and item.get('name', None) == 'device_servo_trq_feedback':
_d2d_trq['device_servo_trq_feedback'].extend(item['value'])
elif item.get('channel', None) == channel-9 and item.get('name', None) == 'hw_sensor_trq_feedback':
_d2d_sensor['hw_sensor_trq_feedback'].extend(item['value'])
df1 = DataFrame.from_dict(_d2d_vel)
df2 = DataFrame.from_dict(_d2d_trq)
df = concat([df1, df2], axis=1)
df3 = DataFrame.from_dict(_d2d_sensor)
df = concat([df1, df2, df3], axis=1)
_filename = f'{path}\\single\\j{channel-8}_hold_{time()}.data'
df.to_csv(_filename, sep='\t', index=False)
@ -191,6 +229,7 @@ def data_proc_inertia(path, filename, channel):
lines = f_obj.readlines()
_d2d_vel = {'hw_joint_vel_feedback': []}
_d2d_trq = {'device_servo_trq_feedback': []}
_d2d_sensor = {'hw_sensor_trq_feedback': []}
for line in lines:
data = eval(line.strip())['data']
for item in data:
@ -202,10 +241,13 @@ def data_proc_inertia(path, filename, channel):
_d2d_vel['hw_joint_vel_feedback'].extend(item['value'])
elif item.get('channel', None) == channel+3 and item.get('name', None) == 'device_servo_trq_feedback':
_d2d_trq['device_servo_trq_feedback'].extend(item['value'])
elif item.get('channel', None) == channel+3 and item.get('name', None) == 'hw_sensor_trq_feedback':
_d2d_trq['hw_sensor_trq_feedback'].extend(item['value'])
df1 = DataFrame.from_dict(_d2d_vel)
df2 = DataFrame.from_dict(_d2d_trq)
df = concat([df1, df2], axis=1)
df3 = DataFrame.from_dict(_d2d_sensor)
df = concat([df1, df2, df3], axis=1)
_filename = f'{path}\\inertia\\j{channel+4}_inertia_{time()}.data'
df.to_csv(_filename, sep='\t', index=False)

13
aio/code/commons/clibs.py
View File

@ -8,16 +8,15 @@ from logging import getLogger
from logging.config import dictConfig
import concurrent_log_handler
ip_addr = '192.168.0.160' # for product
# ip_addr = '192.168.84.129' # for test
RADIAN = 57.3 # 180 / 3.1415926
MAX_FRAME_SIZE = 1024
TIMEOUT = 5
setdefaulttimeout(TIMEOUT)
tab_names = {'dp': 'Data Process', 'at': 'Automatic Test', 'da': 'Duration Action', 'op': 'openapi'}
# PREFIX = '' # for pyinstaller packaging
PREFIX = '../assets/' # for source code debug
PREFIX = '../assets/' # for source code testing and debug
app_icon = f'{PREFIX}templates/icon.ico'
ip_file = f'{PREFIX}templates/ipaddr.txt'
log_path = f'{PREFIX}templates/logs/'
log_data_hmi = f'{PREFIX}templates/logs/c_msg.log'
log_data_debug = f'{PREFIX}templates/logs/debug.log'
@ -42,6 +41,14 @@ durable_data_current_max = {
'axis5': [0 for _ in range(18)],
'axis6': [0 for _ in range(18)],
}
try:
with open(ip_file, mode="r", encoding="utf-8") as f_ipaddr:
ip_addr = f_ipaddr.read().strip()
except:
ip_addr = '192.168.0.160'
# ip_addr = '192.168.0.160' # for product
# ip_addr = '192.168.84.129' # for test
if not exists(log_path):
mkdir(log_path)

2
aio/code/commons/openapi.py
View File

@ -19,7 +19,7 @@ class ModbusRequest(object):
self.host = clibs.ip_addr
self.port = 502
self.interval = 0.3
self.c = ModbusTcpClient(self.host, self.port)
self.c = ModbusTcpClient(host=self.host, port=self.port)
self.c.connect()
def motor_off(self):

13
aio/code/data_process/current.py
View File

@ -33,7 +33,8 @@ def initialization(path, sub, w2t):
else:
if not (match('j[1-7].*\\.data', filename) or match('j[1-7].*\\.csv', filename)):
msg = f"不合规 {data_file}\n"
msg += f"所有数据文件必须以 jx_ 开头，以 .data/csv 结尾（x取值1-7），配置文件需要命名为\"configs.xlsx\"，结果文件需要命名为\"T_电机电流.xlsx\"，请检查后重新运行。"
msg += f"所有数据文件必须以 jx_ 开头，以 .data/csv 结尾（x取值1-7），配置文件需要命名为\"configs.xlsx\"，结果文件需要命名为\"T_电机电流.xlsx\"，请检查后重新运行。\n"
msg += "使用max/avg功能时，需要有配置文件表格\"configs.xlsx\"；使用cycle功能时，需要有电机电流数据处理\"T_电机电流.xlsx\"和配置文件\"configs.xlsx\"两个表格，确认后重新运行！"
w2t(msg, 0, 6, 'red')
if not ((sub == 'cycle' and count == 2) or (sub != 'cycle' and count == 1)):
@ -186,7 +187,6 @@ def find_point(data_file, pos, flag, df, _row_s, _row_e, w2t, exitcode, threshol
else:
return _row_s, _row_e
else:
# w2t(f"[{pos}] {data_file}数据有误，需要检查，无法找到第{exitcode}个有效点...", 0, exitcode, 'red')
w2t(f"[{pos}] {data_file}数据有误，需要检查，无法找到第{exitcode}个有效点...", 0, 0, 'red')
elif flag == 'gt':
while _row_e > end_point:
@ -198,7 +198,6 @@ def find_point(data_file, pos, flag, df, _row_s, _row_e, w2t, exitcode, threshol
else:
return _row_s, _row_e
else:
# w2t(f"[{pos}] {data_file}数据有误，需要检查，无法找到有效起始点或结束点...", 0, exitcode, 'red')
w2t(f"[{pos}] {data_file}数据有误，需要检查，无法找到有效起始点或结束点...", 0, 0, 'red')
@ -234,6 +233,7 @@ def p_single(wb, single, vel, trq, rpms, rrs, w2t):
df_1 = df[col_names[vel-1]].multiply(rpm*addition)
df_2 = df[col_names[trq-1]].multiply(scale)
# print(df_1.abs().max())
df_origin = df
df = concat([df_1, df_2], axis=1)
_step = 5 if data_file.endswith('.csv') else 50
@ -270,11 +270,12 @@ def p_single(wb, single, vel, trq, rpms, rrs, w2t):
row_start = _row_s + _adjust
data = []
for row in range(row_start, row_end):
data.append(df.iloc[row, 0])
data.append(df.iloc[row, 1])
data.append(df_origin.iloc[row, 0])
data.append(df_origin.iloc[row, 1])
data.append(df_origin.iloc[row, 2])
i = 0
for row in ws.iter_rows(min_row=2, min_col=2, max_row=150000, max_col=3):
for row in ws.iter_rows(min_row=2, min_col=2, max_row=150000, max_col=4):
for cell in row:
try:
_ = f"{data[i]:.2f}"