gitea/scripts
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

v0.1.0(2024/05/29)

Browse Source 
1. 修改为customtkinter图形化界面
2. 支持工业机器人制动数据处理（理论上支持，测试数据有问题，待验证）
3. 删除configs.xlsx配置表格，直接在界面配置，新增layout.xlsx文件，存储customtkinter布局
4. 电流尚未支持
This commit is contained in:
gitea 2024-05-29 19:15:44 +08:00
parent 1cee89cd0a
commit 05f461f8c1
6 changed files with 471 additions and 144 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

335
rokae/aio/aio.py
View File

@ -1,21 +1,322 @@
import openpyxl import os.path
from threading import Thread
import tkinter.messagebox
import customtkinter
import brake import brake
import current import current
import time
try: customtkinter.set_appearance_mode("System") # Modes: "System" (standard), "Dark", "Light"
wb_conf = openpyxl.load_workbook('./configs.xlsx', read_only=True) customtkinter.set_default_color_theme("blue") # Themes: "blue" (standard), "green", "dark-blue"
ws_conf = wb_conf['attention'] customtkinter.set_widget_scaling(1.1) # widget dimensions and text size
except Exception as Err: customtkinter.set_window_scaling(1.1) # window geometry dimensions
msg = "无法在当前路径下找到或打开【configs.xlsx】文件，请确认！"
brake.warn_pause_exit(msg, 1, 2)
func_name = ws_conf['B2'].value
if func_name not in wb_conf.sheetnames[1:]: class App(customtkinter.CTk):
msg = f"主功能选择错误，程序没有{func_name}的功能，请确认后重新输入...."
wb_conf.close() def __init__(self):
brake.warn_pause_exit(msg, 1, 1) super().__init__()
else: self.my_font = customtkinter.CTkFont(family="Consolas", size=16, weight="bold")
func_dict = {'brake': brake.main, self.w_param = 96
'current': current.main, # =====================================================================
} # configure window
func_dict[func_name]() self.title("AIO - All in one automatic toolbox")
# self.iconbitmap('./icon.ico')
screen_width = self.winfo_screenwidth()
screen_height = self.winfo_screenheight()
width = screen_width - 200
height = screen_height - 200
self.geometry(f"{width}x{height}+{100}+{100}")
self.protocol("WM_DELETE_WINDOW", self.func_end_call_back)
self.grid_rowconfigure(3, weight=1)
self.grid_columnconfigure((1, 2, 3, 4, 5, 6, 7, 8, 9), weight=1)
self.minsize(1100, 500)
# =====================================================================
# create frame sidebar(left)
self.frame_func = customtkinter.CTkFrame(self, width=120, corner_radius=0)
self.frame_func.grid(row=0, column=0, rowspan=7, sticky='nsew')
# create AIO logo
self.label_logo = customtkinter.CTkLabel(self.frame_func, text="Rokae AIO", height=60, font=customtkinter.CTkFont(family="Segoe Script Bold", size=24, weight="bold"), text_color="DarkSlateGray")
self.label_logo.grid(row=0, column=0, padx=15, pady=15)
# create start button
self.btn_start = customtkinter.CTkButton(self.frame_func, corner_radius=10, text='开始运行', font=self.my_font, command=lambda: self.thread_it(self.func_start_callback))
self.btn_start.grid(row=1, column=0, sticky='new', padx=10, pady=10, ipadx=5, ipady=5)
# create param check button
self.btn_check = customtkinter.CTkButton(self.frame_func, corner_radius=10, text='检查参数', font=self.my_font, command=lambda: self.thread_it(self.func_check_callback))
self.btn_check.grid(row=2, column=0, sticky='new', padx=10, pady=10, ipadx=5, ipady=5)
# create start button
self.btn_log = customtkinter.CTkButton(self.frame_func, corner_radius=10, text='保存日志', font=self.my_font, command=lambda: self.thread_it(self.func_log_callback))
self.btn_log.grid(row=3, column=0, sticky='new', padx=10, pady=10, ipadx=5, ipady=5)
# create start button
self.btn_end = customtkinter.CTkButton(self.frame_func, corner_radius=10, text='结束运行', font=self.my_font, command=lambda: self.thread_it(self.func_end_call_back))
self.btn_end.grid(row=4, column=0, sticky='new', padx=10, pady=10, ipadx=5, ipady=5)
# create version info
self.label_version = customtkinter.CTkLabel(self.frame_func, justify='left', text="Vers: 0.1.0\nDate: 05/29/2024", font=self.my_font, text_color="DarkCyan")
self.frame_func.rowconfigure(6, weight=1)
self.label_version.grid(row=6, column=0, padx=20, pady=20, sticky='s')
# =====================================================================
self.frame_param = customtkinter.CTkFrame(self, height=240, corner_radius=10)
self.frame_param.grid(row=0, column=1, rowspan=3, columnspan=9, sticky='new', ipadx=20, ipady=10, padx=10, pady=10)
# create option menu - main function
self.menu_main = customtkinter.CTkOptionMenu(self.frame_param, values=["INIT", "brake", "current"], font=self.my_font, command=self.func_main_callback)
self.menu_main.grid(row=0, column=1, sticky='we', padx=(20, 10), pady=(10, 0))
self.menu_main.set("Start Here!")
self.menu_sub = customtkinter.CTkOptionMenu(self.frame_param)
self.label_path = customtkinter.CTkLabel(self.frame_param, width=self.w_param//10, text="Path", font=self.my_font)
self.label_path.grid(row=0, column=2, sticky='e', pady=(10, 5))
self.entry_path = customtkinter.CTkEntry(self.frame_param, width=680, placeholder_text="数据文件夹路径", font=self.my_font)
self.entry_path.grid(row=0, column=3, columnspan=7, padx=(5, 10), pady=(10, 5), sticky='w')
self.entry_path.configure(state='disabled')
self.label_av = customtkinter.CTkLabel(self.frame_param, width=self.w_param//10, text="AV", font=self.my_font)
self.label_av.grid(row=1, column=2, sticky='e', pady=(5, 5))
self.entry_av = customtkinter.CTkEntry(self.frame_param, width=self.w_param, placeholder_text=f"角速度", font=self.my_font)
self.entry_av.grid(row=1, column=3, padx=(5, 20), pady=(5, 5), sticky='w')
self.entry_av.configure(state='disabled')
self.label_rc = customtkinter.CTkLabel(self.frame_param, width=self.w_param//10, text="RC", font=self.my_font)
self.label_rc.grid(row=1, column=4, sticky='e', pady=(5, 5))
self.entry_rc = customtkinter.CTkEntry(self.frame_param, width=self.w_param, placeholder_text=f"额定电流", font=self.my_font)
self.entry_rc.grid(row=1, column=5, padx=(5, 20), pady=(5, 5), sticky='w')
self.entry_rc.configure(state='disabled')
self.label_rpm = customtkinter.CTkLabel(self.frame_param, width=self.w_param//10, text="RPM", font=self.my_font)
self.label_rpm.grid(row=1, column=6, sticky='e', pady=(5, 5))
self.entry_rpm = customtkinter.CTkEntry(self.frame_param, width=self.w_param, placeholder_text=f"额定转速", font=self.my_font)
self.entry_rpm.grid(row=1, column=7, padx=(5, 20), pady=(5, 5), sticky='w')
self.entry_rpm.configure(state='disabled')
self.label_rr = customtkinter.CTkLabel(self.frame_param, width=self.w_param//10, text="RR", font=self.my_font)
self.label_rr.grid(row=1, column=8, sticky='e', pady=(5, 5))
self.entry_rr = customtkinter.CTkEntry(self.frame_param, width=self.w_param, placeholder_text=f"减速比", font=self.my_font)
self.entry_rr.grid(row=1, column=9, padx=(5, 20), pady=(5, 5), sticky='w')
self.entry_rr.configure(state='disabled')
self.label_axis = customtkinter.CTkLabel(self.frame_param, width=self.w_param//10, text="AXIS", font=self.my_font)
self.label_axis.grid(row=2, column=2, sticky='e', pady=(5, 5))
self.option_axis = customtkinter.CTkOptionMenu(self.frame_param, values=["1", "2", "3", "4", "5", "6", "7"], width=self.w_param, font=self.my_font)
self.option_axis.grid(row=2, column=3, padx=(5, 20), pady=(5, 5), sticky='w')
self.option_axis.configure(state='disabled')
self.label_vel = customtkinter.CTkLabel(self.frame_param, width=self.w_param//10, text="Vel", font=self.my_font)
self.label_vel.grid(row=2, column=4, sticky='e', pady=(5, 5))
self.option_vel = customtkinter.CTkOptionMenu(self.frame_param, values=["1", "2", "3", "4", "5", "6", "7"], width=self.w_param, font=self.my_font)
self.option_vel.grid(row=2, column=5, padx=(5, 20), pady=(5, 5), sticky='w')
self.option_vel.configure(state='disabled')
self.label_trq = customtkinter.CTkLabel(self.frame_param, width=self.w_param//10, text="Trq", font=self.my_font)
self.label_trq.grid(row=2, column=6, sticky='e', pady=(5, 5))
self.option_trq = customtkinter.CTkOptionMenu(self.frame_param, values=["1", "2", "3", "4", "5", "6", "7"], width=self.w_param, font=self.my_font)
self.option_trq.grid(row=2, column=7, padx=(5, 20), pady=(5, 5), sticky='w')
self.option_trq.configure(state='disabled')
# ========================================
self.textbox = customtkinter.CTkTextbox(self, height=640, wrap='none', font=customtkinter.CTkFont(family="consolas", size=14), text_color="blue")
self.textbox.grid(row=3, column=1, rowspan=5, columnspan=13, ipadx=10, ipady=10, padx=10, pady=10, sticky='nsew')
self.textbox.configure(state='disabled')
def thread_it(self, func, *args):
""" 将函数打包进线程 """
self.myThread = Thread(target=func, args=args)
self.myThread.daemon = True # 主线程退出就直接让子线程跟随退出,不论是否运行完成。
self.myThread.start()
def initialization(self):
self.label_path.configure(text="Path", text_color="black")
self.label_av.configure(text="AV", text_color="black")
self.label_rc.configure(text="RC", text_color="black")
self.label_rpm.configure(text="RPM", text_color="black")
self.label_rr.configure(text="RR", text_color="black")
self.label_axis.configure(text="AXIS", text_color="black")
self.label_vel.configure(text="Vel", text_color="black")
self.label_trq.configure(text="Trq", text_color="black")
self.entry_path.configure(placeholder_text="数据文件夹路径", state="disabled")
self.entry_av.configure(placeholder_text="角速度", state="disabled")
self.entry_rc.configure(placeholder_text="额定电流", state="disabled")
self.entry_rpm.configure(placeholder_text="额定转速", state="disabled")
self.entry_rr.configure(placeholder_text="减速比", state="disabled")
self.option_axis.configure(state="disabled")
self.option_vel.configure(state="disabled")
self.option_trq.configure(state="disabled")
self.menu_sub.grid_forget()
self.textbox.delete(index1='1.0', index2='end')
self.textbox.configure(state='disabled')
def func_main_callback(self, func_name):
self.initialization()
if func_name == 'brake':
self.menu_sub = customtkinter.CTkOptionMenu(self.frame_param, values=["industrial", "cobot"], font=self.my_font, command=self.func_sub_callback)
self.menu_sub.grid(row=1, column=1, sticky='we', padx=(20, 10), pady=(5, 5))
self.menu_sub.set("--select--")
self.label_path.configure(text="*Path", text_color='red')
self.label_av.configure(text="*AV", text_color='red')
self.label_rr.configure(text="*RR", text_color='red')
self.label_axis.configure(text="*AXIS", text_color='red')
self.label_vel.configure(text="*Vel", text_color='red')
self.label_trq.configure(text="*Trq", text_color='red')
self.entry_path.configure(state="normal")
self.entry_av.configure(state="normal")
self.entry_rr.configure(state="normal")
self.option_axis.configure(state="normal")
self.option_vel.configure(state="normal")
self.option_trq.configure(state="normal")
elif func_name == 'current':
self.menu_sub = customtkinter.CTkOptionMenu(self.frame_param, values=["max", "avg"], font=self.my_font, command=self.func_sub_callback)
self.menu_sub.grid(row=1, column=1, sticky='we', padx=(20, 10), pady=(5, 5))
self.menu_sub.set("--select--")
self.label_path.configure(text="*Path", text_color='red')
self.label_rc.configure(text="*RC", text_color='red')
self.label_trq.configure(text="*Trq", text_color='red')
self.entry_path.configure(state="normal")
self.entry_rc.configure(state="normal")
self.option_trq.configure(state="normal")
else:
self.initialization()
self.menu_main.set("Start Here!")
def func_sub_callback(self, func_name):
if func_name == "industrial":
self.label_rpm.configure(text="*RPM", text_color='red')
self.entry_rpm.configure(state="normal")
elif func_name == "cobot":
self.label_rpm.configure(text="RPM", text_color='black')
self.entry_rpm.configure(state="disabled")
elif func_name == "max":
pass
elif func_name == 'avg':
pass
def write2textbox(self, text, exitcode=-1, wait=0):
if exitcode != -1:
self.textbox.configure(state="normal", text_color='red')
self.textbox.insert(index='end', text=text + '\n')
self.textbox.insert(index='end', text=f"Error code:{exitcode}，需要解决如上问题，再重新运行程序......")
self.textbox.update()
self.textbox.see('end')
self.textbox.configure(state="disabled")
raise Exception("Something wrong, check needed......")
elif wait != 0:
self.textbox.configure(state="normal")
self.textbox.insert(index='end', text=text)
self.textbox.update()
self.textbox.see('end')
self.textbox.configure(state="disabled")
else:
self.textbox.configure(state="normal", text_color='blue')
self.textbox.insert(index='end', text=text + '\n')
self.textbox.update()
self.textbox.see('end')
self.textbox.configure(state="disabled")
def check_param(self):
func_name = self.menu_main.get()
if func_name == 'brake':
path = self.entry_path.get().strip(' ')
av = self.entry_av.get().strip('- ')
rr = self.entry_rr.get().strip('- ')
axis = self.option_axis.get()
vel = self.option_vel.get()
trq = self.option_trq.get()
sub_func = self.menu_sub.get()
c1 = os.path.exists(path)
c2 = av.isdigit()
c3 = rr.isdigit()
c4 = rpm = 1
c5 = sub_func in ['industrial', 'cobot']
if self.menu_sub.get() == 'industrial':
rpm = self.entry_rpm.get().strip('- ')
c4 = rpm.isdigit()
elif self.menu_sub.get() == 'cobot':
pass
else:
pass
flag = 1 if c1 and c2 and c3 and c4 and c5 else 0
return flag, path, int(av), int(rr), int(rpm), int(axis), int(vel), int(trq)
elif func_name == 'current':
path = self.entry_path.get()
rc = self.entry_rc.get()
vel = self.option_vel.get()
trq = self.option_trq.get()
sub_func = self.menu_sub.get()
c1 = os.path.exists(path)
c2 = sub_func in ['max', 'avg']
try:
_ = float(rc)
c3 = True
except Exception as Err:
c3 = False
flag = 2 if c1 and c2 and c3 else 0
return flag, path, float(rc), int(vel), int(trq), sub_func
else:
return 0, 0
def func_start_callback(self):
self.textbox.configure(state='normal')
self.textbox.delete(index1='1.0', index2='end')
self.textbox.configure(state='disabled')
flag, *args = self.check_param()
func_dict = {1: brake.main, 2: current.main}
if flag == 1:
func_dict[flag](path=args[0], av=args[1], rr=args[2], rpm=args[3], axis=args[4], vel=args[5], trq=args[6], w2t=self.write2textbox)
elif flag == 2:
func_dict[flag](path=args[0], rc=args[1], sub_func= args[2])
else:
tkinter.messagebox.showerror(title="参数错误", message="请检查对应参数是否填写正确！", )
def func_check_callback(self):
flag, *args = self.check_param()
if flag:
tkinter.messagebox.showinfo(title="参数正确", message="所有参数形式上填写无误，请确定后开始运行！")
else:
tkinter.messagebox.showerror(title="参数错误", message="请检查对应参数是否填写正确！", )
def func_log_callback(self):
content = self.textbox.get(index1='1.0', index2='end')
if len(content) > 1:
try:
now = time.strftime('%Y%m%d%H%M%S', time.localtime(time.time()))
log_name = f"{now}_aio.log"
with open(f'{log_name}', 'w', encoding='utf-8') as objlog:
objlog.write(content)
tkinter.messagebox.showinfo(title="保存成功", message=f'{log_name}已被保存存至↓↓↓\n{os.getcwd()}')
except Exception as Err:
print(Err)
tkinter.messagebox.showerror(title="保存失败", message="未能保存本次日志，或未能完整保存，请准备好相关数据，联系fanmingfu@rokae.com查看详细信息！", )
else:
tkinter.messagebox.showwarning(title="未能保存", message="日志数据为空，不可保存！")
def func_end_call_back(self):
if tkinter.messagebox.askyesno(title="关闭程序", message="相关数据可能未保存，正在运行程序时有概率会损坏数据文件，确定要终止程序运行吗？"):
self.destroy()
else:
pass
pass
if __name__ == "__main__":
aio = App()
aio.mainloop()

240
rokae/aio/brake.py
View File

@ -7,13 +7,14 @@ from threading import Thread
import pandas import pandas
def traversal_files(path): def traversal_files(path, w2t):
# 功能：以列表的形式分别返回指定路径下的文件和文件夹，不包含子目录 # 功能：以列表的形式分别返回指定路径下的文件和文件夹，不包含子目录
# 参数：路径 # 参数：路径
# 返回值：路径下的文件夹列表 路径下的文件列表 # 返回值：路径下的文件夹列表 路径下的文件列表
if not os.path.exists(path): if not os.path.exists(path):
msg = f'数据文件夹{path}不存在，请确认后重试......' msg = f'数据文件夹{path}不存在，请确认后重试......'
warn_pause_exit(msg, 1, 11) w2t(msg, 1)
else: else:
dirs = [] dirs = []
files = [] files = []
@ -26,14 +27,14 @@ def traversal_files(path):
return dirs, files return dirs, files
def get_threshold_step(excel_file, AXIS): def get_threshold_step(excel_file, axis):
# 功能：负载和速度100%，且是j2的时候，做特殊处理 # 功能：负载和速度100%，且是j2的时候，做特殊处理
# 参数：新生成的excel，轴号 # 参数：新生成的excel，轴号
# 返回值：速度差阈值，处理步长 # 返回值：速度差阈值，处理步长
conditions = sorted(excel_file.split('\\')[-2].split('_')) conditions = sorted(excel_file.split('\\')[-2].split('_'))
# 只有负载和速度是100%时，才会启用更敏感的step # 只有负载和速度是100%时，才会启用更敏感的step
flg = 1 if conditions[0][-3:] == '100' and conditions[2][-3:] == '100' else 0 flg = 1 if conditions[0][-3:] == '100' and conditions[2][-3:] == '100' else 0
if flg == 1 and AXIS == 2: if flg == 1 and axis == 2:
threshold = 30 threshold = 30
step = 5 step = 5
else: else:
@ -43,24 +44,24 @@ def get_threshold_step(excel_file, AXIS):
return threshold, step return threshold, step
def find_row_start(data_file, df, conditions, AV, RR, AXIS): def find_row_start(data_file, df, conditions, av, rr, axis, vel, trq, w2t, rpm):
# 功能：查找数据文件中有效数据的行号，也即最后一个速度下降的点位 # 功能：查找数据文件中有效数据的行号，也即最后一个速度下降的点位
# 参数：如上 # 参数：如上
# 返回值：速度下降点位，最后的数据点位 # 返回值：速度下降点位，最后的数据点位
ratio = float(conditions[1].removeprefix('speed'))/100 ratio = float(conditions[1].removeprefix('speed'))/100
speed_max = AV * ratio * RR / 6 speed_max = av * ratio * rr / 6
row_max = row_start = df.index[-1] row_max = row_start = df.index[-1]
threshold, step = get_threshold_step(data_file, AXIS) threshold, step = get_threshold_step(data_file, axis)
while row_start > step+1: while row_start > step+1:
speed = df.iloc[row_start, 0] speed = df.iloc[row_start, vel-1] * rpm
if int(speed) < 1: if int(speed) < 1:
row_start -= 50 row_start -= 50
continue continue
_ = [] _ = []
for i in range(row_start, row_start-step+1, -1): for i in range(row_start, row_start-step+1, -1):
_.append(df.iloc[i, 0]) _.append(df.iloc[i, vel-1] * rpm)
speed_avg = abs(sum(_))/len(_) speed_avg = abs(sum(_))/len(_)
if abs(speed_avg-speed_max) < threshold: if abs(speed_avg-speed_max) < threshold:
@ -69,8 +70,8 @@ def find_row_start(data_file, df, conditions, AV, RR, AXIS):
else: else:
row_start -= step row_start -= step
else: else:
msg = f"可能是{data_file},这个文件数据采集有问题，比如采集的时机不对，也有可能是程序步长设定问题，请检查......" msg = f"可能是{data_file}这个文件数据采集有问题，比如未采集理论速度值，也有可能是程序步长设定问题，请检查......"
warn_pause_exit(msg, 1, 9) w2t(msg, 2)
return row_max, row_start return row_max, row_start
@ -87,7 +88,7 @@ def find_result_sheet_name(conditions, count):
return result_sheet_name return result_sheet_name
def copy_data_to_result(df, ws_result, row_max, row_start): def copy_data_to_result(df, ws_result, row_max, row_start, vel, trq, rpm):
# 功能：将数据文件中有效数据拷贝至结果文件对应的 sheet # 功能：将数据文件中有效数据拷贝至结果文件对应的 sheet
# 参数：如上 # 参数：如上
# 返回值：- # 返回值：-
@ -98,11 +99,12 @@ def copy_data_to_result(df, ws_result, row_max, row_start):
# 将合适的数据复制到结果文件 # 将合适的数据复制到结果文件
row_max = row_start + 399 if row_max-row_start > 400 else row_max row_max = row_start + 399 if row_max-row_start > 400 else row_max
rc = 1 if rpm == 1 else 1000
data = [] data = []
for i in range(row_start, row_max+1): for i in range(row_start, row_max+1):
data.append(df.iloc[i, 0]) data.append(df.iloc[i, vel-1] * rpm)
data.append(df.iloc[i, 1]) data.append(df.iloc[i, trq-1] * rc)
i = 0 i = 0
for row in ws_result.iter_rows(min_row=2, min_col=1, max_row=row_max - row_start + 2, max_col=2): for row in ws_result.iter_rows(min_row=2, min_col=1, max_row=row_max - row_start + 2, max_col=2):
@ -111,90 +113,136 @@ def copy_data_to_result(df, ws_result, row_max, row_start):
i = i + 1 i = i + 1
def single_file_process(data_file, wb_result, count, AV, RR, RC, AXIS): def single_file_process(data_file, wb_result, count, av, rr, axis, vel, trq, w2t, rpm):
# 功能：完成单个数据文件的处理 # 功能：完成单个数据文件的处理
# 参数：如上 # 参数：如上
# 返回值：- # 返回值：-
df = pandas.read_csv(data_file, sep='\t') if data_file.endswith('.data'):
sep = '\t'
df = pandas.read_csv(data_file, sep=sep)
elif data_file.endswith('.csv'):
sep = ','
df = pandas.read_csv(data_file, sep=sep, encoding='gbk', header=8)
conditions = sorted(data_file.split('\\')[-2].split('_')[1:]) conditions = sorted(data_file.split('\\')[-2].split('_')[1:])
result_sheet_name = find_result_sheet_name(conditions, count) result_sheet_name = find_result_sheet_name(conditions, count)
ws_result = wb_result[result_sheet_name] ws_result = wb_result[result_sheet_name]
row_max, row_start = find_row_start(data_file, df, conditions, AV, RR, AXIS) row_max, row_start = find_row_start(data_file, df, conditions, av, rr, axis, vel, trq, w2t, rpm)
copy_data_to_result(df, ws_result, row_max, row_start) copy_data_to_result(df, ws_result, row_max, row_start, vel, trq, rpm)
ws_result["C2"] = int(2) ws_result["C2"] = int(2)
ws_result["G2"] = int(10+4) ws_result["G2"] = int(10+4)
def now_doing_msg(docs, flag): def now_doing_msg(docs, flag, w2t):
# 功能：输出正在处理的文件或目录 # 功能：输出正在处理的文件或目录
# 参数：文件或目录，start 或 done 标识 # 参数：文件或目录，start 或 done 标识
# 返回值：- # 返回值：-
now = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())) now = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
file_type = 'file' if os.path.isfile(docs) else 'dir' file_type = 'file' if os.path.isfile(docs) else 'dir'
if flag == 'start' and file_type == 'dir': if flag == 'start' and file_type == 'dir':
print(f"[{now}] 正在处理目录【{docs}】中的数据......") w2t(f"[{now}] 正在处理目录 {docs} 中的数据......")
elif flag == 'start' and file_type == 'file': elif flag == 'start' and file_type == 'file':
print(f"[{now}] 正在处理文件【{docs}】中的数据......") w2t(f"[{now}] 正在处理文件 {docs} 中的数据......")
elif flag == 'done' and file_type == 'dir': elif flag == 'done' and file_type == 'dir':
print(f"[{now}] 目录【{docs}】数据文件已处理完毕......") w2t(f"[{now}] 目录 {docs} 数据文件已处理完毕......")
elif flag == 'done' and file_type == 'file': elif flag == 'done' and file_type == 'file':
print(f"[{now}] 文件【{docs}】数据文件已处理完毕......") w2t(f"[{now}] 文件 {docs} 数据已处理完毕......")
def data_process(result_file, raw_data_dirs, AV, RR, RC, AXIS): class GetThreadResult(Thread):
def __init__(self, func, args=()):
super(GetThreadResult, self).__init__()
self.func = func
self.args = args
self.result = 0
def run(self):
time.sleep(1)
self.result = self.func(*self.args)
def get_result(self):
Thread.join(self) # 等待线程执行完毕
try:
return self.result
except Exception as Err:
return None
def w2t_local(msg, exitcode, wait, w2t):
while True:
global stop
if stop == 0 and wait != 0:
time.sleep(1)
w2t(msg, exitcode, wait)
else:
break
def data_process(result_file, raw_data_dirs, av, rr, axis, vel, trq, w2t, rpm):
# 功能：完成一个结果文件的数据处理 # 功能：完成一个结果文件的数据处理
# 参数：结果文件，数据目录，以及预读取的参数 # 参数：结果文件，数据目录，以及预读取的参数
# 返回值：- # 返回值：-
file_name = result_file.split('\\')[-1]
w2t(f"打开文件 {file_name} 需要 1min 左右，请耐心等待", -1, 1)
global stop
stop = 0
t_excel = GetThreadResult(openpyxl.load_workbook, args=(result_file, ))
t_wait = Thread(target=w2t_local, args=('.', -1, 1, w2t))
t_excel.start()
t_wait.start()
t_excel.join()
wb_result = t_excel.get_result()
stop = 1
time.sleep(1.1)
w2t('', -1, 0)
prefix = result_file.split('\\')[-1].split('_')[0] prefix = result_file.split('\\')[-1].split('_')[0]
wb_result = openpyxl.load_workbook(result_file)
for raw_data_dir in raw_data_dirs: for raw_data_dir in raw_data_dirs:
if raw_data_dir.split('\\')[-1].split('_')[0] == prefix: if raw_data_dir.split('\\')[-1].split('_')[0] == prefix:
now_doing_msg(raw_data_dir, 'start') now_doing_msg(raw_data_dir, 'start', w2t)
_, data_files = traversal_files(raw_data_dir) _, data_files = traversal_files(raw_data_dir, w2t)
# 数据文件串行处理模式--------------------------------- # 数据文件串行处理模式---------------------------------
# count = 1 # count = 1
# for data_file in data_files: # for data_file in data_files:
# now_doing_msg(data_file, 'start') # now_doing_msg(data_file, 'start', w2t)
# single_file_process(data_file, wb_result, count, AV, RR, RC, AXIS) # single_file_process(data_file, wb_result, count, av, rr, axis, vel, trq, w2t, rpm)
# count += 1 # count += 1
# now_doing_msg(data_file, 'done') # now_doing_msg(data_file, 'done', w2t)
# --------------------------------------------------- # ---------------------------------------------------
# 数据文件并行处理模式--------------------------------- # 数据文件并行处理模式---------------------------------
threads = [Thread(target=single_file_process, args=(data_files[0], wb_result, 1, AV, RR, RC, AXIS)), threads = [Thread(target=single_file_process, args=(data_files[0], wb_result, 1, av, rr, axis, vel, trq, w2t, rpm)),
Thread(target=single_file_process, args=(data_files[1], wb_result, 2, AV, RR, RC, AXIS)), Thread(target=single_file_process, args=(data_files[1], wb_result, 2, av, rr, axis, vel, trq, w2t, rpm)),
Thread(target=single_file_process, args=(data_files[2], wb_result, 3, AV, RR, RC, AXIS))] Thread(target=single_file_process, args=(data_files[2], wb_result, 3, av, rr, axis, vel, trq, w2t, rpm))]
[t.start() for t in threads] [t.start() for t in threads]
[t.join() for t in threads] [t.join() for t in threads]
now_doing_msg(raw_data_dir, 'done') now_doing_msg(raw_data_dir, 'done', w2t)
# --------------------------------------------------- # ---------------------------------------------------
now_doing_msg(result_file, 'done') now_doing_msg(result_file, 'done', w2t)
print(f"保存文件需要1-2min，请耐心等待......") w2t(f"保存文件 {file_name} 需要 1min 左右，请耐心等待", -1, 1)
wb_result.save(result_file) stop = 0
wb_result.close() t_excel = Thread(target=wb_result.save, args=(result_file, ))
t_wait = Thread(target=w2t_local, args=('.', -1, 1, w2t))
t_excel.start()
t_wait.start()
t_excel.join()
stop = 1
time.sleep(1.1)
w2t('\n', -1, 0)
def warn_pause_exit(msg, pause_num, exit_num): def check_files(raw_data_dirs, result_files, w2t):
# 功能：打印告警信息，并推出程序
# 参数：告警信息，暂停的次数，退出的值
# 返回值：-
print(msg + '\n')
for i in range(pause_num):
_ = input("Press ENTER to continue......\n")
sys.exit(exit_num)
def check_files(raw_data_dirs, result_files):
# 功能：检查数据文件以及结果文件的合规性 # 功能：检查数据文件以及结果文件的合规性
# 参数：数据文件夹，结果文件 # 参数：数据文件夹，结果文件
# 返回值：- # 返回值：-
if len(result_files) != 3: if len(result_files) != 3:
msg = "结果文件数目错误，结果文件有且只有三个，请确认！" msg = "结果文件数目错误，结果文件有且只有三个，请确认！"
for result_file in result_files: for result_file in result_files:
print(result_file) w2t(result_file)
warn_pause_exit(msg, 1, 3) w2t(msg, 3)
prefix = [] prefix = []
for result_file in result_files: for result_file in result_files:
@ -203,11 +251,11 @@ def check_files(raw_data_dirs, result_files):
wd = result_files[0].split('\\') wd = result_files[0].split('\\')
del wd[-1] del wd[-1]
wd = '\\'.join(wd) wd = '\\'.join(wd)
msg = f"请关闭所有相关数据文件，并检查工作目录【{wd}】下，有且只允许有类似如下三个文件：\n" \ msg = f"""请关闭所有相关数据文件，并检查工作目录 {wd} 下，有且只允许有类似如下三个文件：
f"1. load33_自研_制动性能测试.xlsx\n" \ 1. load33_自研_制动性能测试.xlsx
f"2. load66_自研_制动性能测试.xlsx\n" \ 2. load66_自研_制动性能测试.xlsx
f"3. load100_自研_制动性能测试.xlsx" 3. load100_自研_制动性能测试.xlsx"""
warn_pause_exit(msg, 1, 8) w2t(msg, 4)
for raw_data_dir in raw_data_dirs: for raw_data_dir in raw_data_dirs:
components = raw_data_dir.split('\\')[-1].split('_') components = raw_data_dir.split('\\')[-1].split('_')
@ -215,61 +263,33 @@ def check_files(raw_data_dirs, result_files):
if components[0] not in ['load33', 'load66', 'load100'] or \ if components[0] not in ['load33', 'load66', 'load100'] or \
components[1] not in ['speed33', 'speed66', 'speed100'] or \ components[1] not in ['speed33', 'speed66', 'speed100'] or \
components[2] not in ['reach33', 'reach66', 'reach100']: components[2] not in ['reach33', 'reach66', 'reach100']:
msg = f"报错信息：数据目录【{raw_data_dir}】命名不合规，请参考如下形式\n" \ msg = f"报错信息：数据目录 {raw_data_dir} 命名不合规，请参考如下形式\n" \
f"命名规则：\n 1. loadAA_speedBB_reachCC\n 2. loadAA_reachBB_speedCC\n" \ f"命名规则：\n 1. loadAA_speedBB_reachCC\n 2. loadAA_reachBB_speedCC\n" \
f"规则解释：AA/BB/CC 指的是负载/速度/臂展的比例\n" \ f"规则解释：AA/BB/CC 指的是负载/速度/臂展的比例\n" \
f"load66_speed100_reach33：66% 负载，100% 速度以及 33% 臂展情况下的测试结果文件夹" f"load66_speed100_reach33：66% 负载，100% 速度以及 33% 臂展情况下的测试结果文件夹"
warn_pause_exit(msg, 1, 7) w2t(msg, 5)
# 直接删掉 excel 文件 _, raw_data_files = traversal_files(raw_data_dir, w2t)
_, raw_data_files = traversal_files(raw_data_dir)
for raw_data_file in raw_data_files:
if raw_data_file.endswith(".xlsx"):
os.remove(raw_data_file)
_, raw_data_files = traversal_files(raw_data_dir)
if len(raw_data_files) != 3: if len(raw_data_files) != 3:
msg = f"数据目录【{raw_data_dir}】下数据文件个数错误，每个数据目录下有且只能有三个以 .data 为后缀的数据文件" msg = f"数据目录 {raw_data_dir} 下数据文件个数错误，每个数据目录下有且只能有三个以 .data 为后缀的数据文件"
warn_pause_exit(msg, 1, 6) w2t(msg, 6)
for raw_data_file in raw_data_files: for raw_data_file in raw_data_files:
if not raw_data_file.split('\\')[-1].endswith('.data'): if not (raw_data_file.split('\\')[-1].endswith('.data') or raw_data_file.split('\\')[-1].endswith('.csv')):
msg = f"数据文件【{raw_data_file}】后缀错误，每个数据目录下有且只能有三个以 .data 为后缀的数据文件" msg = f"数据文件 {raw_data_file} 后缀错误，每个数据目录下有且只能有三个以 .data/csv 为后缀的数据文件"
warn_pause_exit(msg, 1, 5) w2t(msg, 7)
print("数据目录合规性检查结束，未发现问题......") w2t("数据目录合规性检查结束，未发现问题......")
def initialization(): def execution(path, av, rr, rpm, axis,vel, trq, w2t):
# 功能：初始化，记录开始时间，读取预定义参数
# 参数：-
# 返回值：结果文件，数据文件夹，以及预定义参数
time_start = time.time()
try:
# read init configurations from config file
wb_conf = openpyxl.load_workbook('./configs.xlsx', read_only=True)
ws_conf = wb_conf['brake']
except Exception as Err:
msg = "无法在当前路径下找到或打开【configs.xlsx】文件，请确认！"
warn_pause_exit(msg, 1, 2)
DATA_DIR = ws_conf.cell(row=2, column=2).value
AXIS = int(ws_conf.cell(row=3, column=2).value)
AV = int(ws_conf.cell(row=4, column=2).value.split(',')[AXIS-1].strip())
RR = int(ws_conf.cell(row=5, column=2).value.split(',')[AXIS-1].strip())
RC = float(ws_conf.cell(row=6, column=2).value.split(',')[AXIS-1].strip())
wb_conf.close()
raw_data_dirs, result_files = traversal_files(DATA_DIR)
check_files(raw_data_dirs, result_files)
return raw_data_dirs, result_files, time_start, AV, RR, RC, AXIS
def execution(args):
# 功能：执行处理所有数据文件 # 功能：执行处理所有数据文件
# 参数：initialization函数的返回值 # 参数：initialization函数的返回值
# 返回值：- # 返回值：-
raw_data_dirs, result_files, time_start, AV, RR, RC, AXIS = args time_start = time.time()
raw_data_dirs, result_files = traversal_files(path, w2t)
check_files(raw_data_dirs, result_files, w2t)
prefix = [] prefix = []
for raw_data_dir in raw_data_dirs: for raw_data_dir in raw_data_dirs:
prefix.append(raw_data_dir.split('\\')[-1].split("_")[0]) prefix.append(raw_data_dir.split('\\')[-1].split("_")[0])
@ -280,25 +300,27 @@ def execution(args):
if result_file.split('\\')[-1].split('_')[0] not in set(prefix): if result_file.split('\\')[-1].split('_')[0] not in set(prefix):
continue continue
else: else:
now_doing_msg(result_file, 'start') now_doing_msg(result_file, 'start', w2t)
data_process(result_file, raw_data_dirs, AV, RR, RC, AXIS) data_process(result_file, raw_data_dirs, av, rr, axis, vel, trq, w2t, rpm)
# threads.append(Thread(target=data_process, args=(result_file, raw_data_dirs, AV, RR, RC, AXIS))) # threads.append(Thread(target=data_process, args=(result_file, raw_data_dirs, AV, RR, RC, AXIS)))
# [t.start() for t in threads] # [t.start() for t in threads]
# [t.join() for t in threads] # [t.join() for t in threads]
print("----------------------------------------------------------") w2t("----------------------------------------------------------")
print("全部处理完毕") w2t("全部处理完毕")
except Exception as Err: except Exception as Err:
print("程序运行错误，请检查配置文件是否准确设定，以及数据文件组织是否正确，也有可能是结果文件损坏，尝试重新复制一份，再运行！") msg = "程序运行错误，请检查配置文件是否准确设定，以及数据文件组织是否正确，也有可能是结果文件损坏，尝试重新复制一份，再运行！"
w2t(msg)
time_end = time.time() time_end = time.time()
time_total = time_end - time_start time_total = time_end - time_start
msg = f"数据处理时间：{time_total // 3600:02.0f} h {time_total % 3600 // 60:02.0f} min {time_total % 60:02.0f} s" msg = f"数据处理时间：{time_total // 3600:02.0f} h {time_total % 3600 // 60:02.0f} min {time_total % 60:02.0f} s"
warn_pause_exit(msg, 1, 0) w2t(msg)
def main(): def main(path, av, rr, rpm, axis,vel, trq, w2t):
execution(initialization()) execution(path, av, rr, rpm, axis,vel, trq, w2t)
if __name__ == "__main__": if __name__ == "__main__":
main() stop = 0
main(path=sys.argv[1], av=sys.argv[2], rr=sys.argv[3], rpm=sys.argv[4], axis=sys.argv[5], vel=sys.argv[6], trq=sys.argv[7], w2t=sys.argv[8])

BIN
rokae/aio/configs.xlsx
View File

Binary file not shown.

8
rokae/aio/file_version_info.txt
View File

@ -6,8 +6,8 @@ VSVersionInfo(
ffi=FixedFileInfo( ffi=FixedFileInfo(
# filevers and prodvers should be always a tuple with four items: (1, 2, 3, 4) # filevers and prodvers should be always a tuple with four items: (1, 2, 3, 4)
# Set not needed items to zero 0. # Set not needed items to zero 0.
filevers=(0, 0, 6, 0), filevers=(0, 1, 0, 0),
prodvers=(0, 0, 6, 0), prodvers=(0, 1, 0, 0),
# Contains a bitmask that specifies the valid bits 'flags'r # Contains a bitmask that specifies the valid bits 'flags'r
mask=0x3f, mask=0x3f,
# Contains a bitmask that specifies the Boolean attributes of the file. # Contains a bitmask that specifies the Boolean attributes of the file.
@ -31,12 +31,12 @@ VSVersionInfo(
'040904b0', '040904b0',
[StringStruct('CompanyName', 'Rokae - https://www.rokae.com/'), [StringStruct('CompanyName', 'Rokae - https://www.rokae.com/'),
StringStruct('FileDescription', 'All in one automatic operating tool'), StringStruct('FileDescription', 'All in one automatic operating tool'),
StringStruct('FileVersion', '0.0.6 (2024-05-20)'), StringStruct('FileVersion', '0.1.0 (2024-05-24)'),
StringStruct('InternalName', 'AIO.exe'), StringStruct('InternalName', 'AIO.exe'),
StringStruct('LegalCopyright', '© 2024-2024 Manford Fan'), StringStruct('LegalCopyright', '© 2024-2024 Manford Fan'),
StringStruct('OriginalFilename', 'AIO.exe'), StringStruct('OriginalFilename', 'AIO.exe'),
StringStruct('ProductName', 'AIO'), StringStruct('ProductName', 'AIO'),
StringStruct('ProductVersion', '0.0.6 (2024-05-23)')]) StringStruct('ProductVersion', '0.1.0 (2024-05-29)')])
]), ]),
VarFileInfo([VarStruct('Translation', [1033, 1200])]) VarFileInfo([VarStruct('Translation', [1033, 1200])])
] ]

BIN
rokae/aio/layout.xlsx Normal file
View File

Binary file not shown.

32
rokae/aio/readme.txt
View File

@ -11,10 +11,13 @@
最好不用虚拟环境 最好不用虚拟环境
pyinstaller.exe -F --version-file file_version_info.txt -i .\icon.ico .\aio.py -p .\brake.py -p .\current.py pyinstaller.exe -F --version-file file_version_info.txt -i .\icon.ico .\aio.py -p .\brake.py -p .\current.py
https://customtkinter.tomschimansky.com/documentation/packaging
pyinstaller --noconfirm --onedir --windowed --add-data "C:/Users/Administrator/AppData/Local/Programs/Python/Python312/Lib/site-packages/customtkinter;customtkinter/" --version-file file_version_info.txt -i .\icon.ico .\aio.py -p .\brake.py
注意事项： 注意事项：
1. 数据文件存储存储规则 1. 数据文件存储存储规则
所谓数据文件，就是我们拍急停的时候，采集到的 .data 文件，正方向拍三次急停，会采集到三个 .data 文件，存储在同一个文件夹内，即每组（三个 .data 文件）文件必须存储在同一个文件夹内，数据文件的命名无要求， 所谓数据文件，就是我们拍急停的时候，采集到的 .data 文件，正方向拍三次急停，会采集到三个 .data 文件，存储在同一个文件夹内，即每组（三个 .data 文件）文件必须存储在同一个文件夹内，数据文件的命名无要求，
2. 文件夹命名规则 2. 文件夹命名规则
虽然对采集到的 .data 文件没有命名要求，但是对于文件夹的命名是有要求的，必须是如下格式： 虽然对采集到的 .data 文件没有命名要求，但是对于文件夹的命名是有要求的，必须是如下格式：
loadXX_speedXX_reachXX 或者 loadXX_reachXX_speedXX loadXX_speedXX_reachXX 或者 loadXX_reachXX_speedXX
@ -26,9 +29,9 @@
load33_自研_制动性能测试.xlsx load33_自研_制动性能测试.xlsx
load66_自研_制动性能测试.xlsx load66_自研_制动性能测试.xlsx
load100_自研_制动性能测试.xlsx load100_自研_制动性能测试.xlsx
！！结果文件可以是没有数据的，也可以是之前有数据的，只要保证第 6 点中的那几个数据准确即可 ！！结果文件可以是没有数据的，也可以是之前有数据的，只要保证第 6 点中的那几个数据准确即可
4. 数据存储的组织结 4. 数据存储的组织结
..../j1/load100_speed33_reach100 ..../j1/load100_speed33_reach100
..../j1/load100_speed66_reach100 ..../j1/load100_speed66_reach100
@ -37,28 +40,28 @@
..../j1/load100_speed33_reach100/2024_05_16_09_18_52.data ..../j1/load100_speed33_reach100/2024_05_16_09_18_52.data
..../j1/load100_speed33_reach100/2024_05_16_09_19_52.data ..../j1/load100_speed33_reach100/2024_05_16_09_19_52.data
..../j1/load100_speed33_reach100/2024_05_16_09_20_52.data ..../j1/load100_speed33_reach100/2024_05_16_09_20_52.data
..../j1/load33_自研_制动性能测试.xlsx ..../j1/load33_自研_制动性能测试.xlsx
..../j1/load66_自研_制动性能测试.xlsx ..../j1/load66_自研_制动性能测试.xlsx
..../j1/load100_自研_制动性能测试.xlsx ..../j1/load100_自研_制动性能测试.xlsx
5. 文件的打开与关闭 5. 文件的打开与关闭
a. 在执行程序之前，需要关闭所有相关 excle 文件 a. 在执行程序之前，需要关闭所有相关 excle 文件
b. 在执行程序之中，不允许打开相关 excle 文件 b. 在执行程序之中，不允许打开相关 excle 文件
c. 在执行程序之后，需要逐个打开结果文件，并保存一次 c. 在执行程序之后，需要逐个打开结果文件，并保存一次
6. 参数一致性检查 6. 参数一致性检查
执行程序前，需要确定 configs.xlsx 中设定的减速比/最大角速度/额定电流的值是正确的 执行程序前，需要确定 configs.xlsx 中设定的减速比/最大角速度/额定电流的值是正确的
7. 数据准确性检查 7. 数据准确性检查
执行完程序之后，需要对结果文件的数据准确性做核对，通过我自己的数据观察，误差基本在10ms以内，也即10个数据点，误差较大的情况可自行调整 执行完程序之后，需要对结果文件的数据准确性做核对，通过我自己的数据观察，误差基本在10ms以内，也即10个数据点，误差较大的情况可自行调整
8. .data 数据顺序 8. .data 数据顺序
.data 文件的第一列和第二列必须分别是速度和电流 .data 文件的第一列和第二列必须分别是速度和电流
9. 其他 9. 其他
程序运行主要的耗时集中在打开，保存和关闭结果文件，第一次打开的时候会比较慢，是因为 excel 在做首次公式的计算，保存关闭之后，再打开会比较快一些，另外，如果在运行出错并重复运行程序的时候无响应，或者出现异常，请打开任务管理器，关闭一切和excel相关的进程，重新运行即可 程序运行主要的耗时集中在打开，保存和关闭结果文件，第一次打开的时候会比较慢，是因为 excel 在做首次公式的计算，保存关闭之后，再打开会比较快一些，另外，如果在运行出错并重复运行程序的时候无响应，或者出现异常，请打开任务管理器，关闭一切和excel相关的进程，重新运行即可
RELEASE CHANGES RELEASE CHANGES
@ -107,7 +110,8 @@ v0.0.6(2024/05/23)
1. 为了调整多功能框架，aio.py文件将会作为入口程序存在，不实现具体功能，功能的实现将由具体的功能脚本实现，aio.py只负责条件调用 1. 为了调整多功能框架，aio.py文件将会作为入口程序存在，不实现具体功能，功能的实现将由具体的功能脚本实现，aio.py只负责条件调用
2. 新增了自动化处理电流数据（电机/过载）的功能 2. 新增了自动化处理电流数据（电机/过载）的功能
v0.0.7(2024/05/27) v0.1.0(2024/05/29)
1. 该版本制动数据处理变动较大，重写了find_row_start & copy_data_to_result，删除了delete_excel_files 1. 修改为customtkinter图形化界面
2. 主要是修改了数据处理的方式，直接使用pandas进行数据处理，跳过了openpyxl来回变换，节省了大量的IO以及时间 2. 支持工业机器人制动数据处理（理论上支持，测试数据有问题，待验证）
3. 删除configs.xlsx配置表格，直接在界面配置
4. 电流尚未支持