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

fix merging

Browse Source
This commit is contained in:
gitea
2024-06-16 14:30:12 +08:00
parent 78a79e4aa0 5ba3c76386
commit 2f2f0d430d
14 changed files with 217 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

370
aio/code/aio.py Normal file
View File

@ -0,0 +1,370 @@
from os.path import exists
from os import getcwd
from threading import Thread
import tkinter.messagebox
import customtkinter
import brake, current, iso, wavelogger
from time import time, strftime, localtime
from urllib.request import urlopen
from socket import setdefaulttimeout
customtkinter.set_appearance_mode("System") # Modes: "System" (standard), "Dark", "Light"
customtkinter.set_default_color_theme("blue") # Themes: "blue" (standard), "green", "dark-blue"
customtkinter.set_widget_scaling(1.1) # widget dimensions and text size
customtkinter.set_window_scaling(1.1) # window geometry dimensions
setdefaulttimeout(10)
# global vars
btns = {
'start': {'btn': '', 'row': 1, 'text': '开始运行'},
'check': {'btn': '', 'row': 2, 'text': '检查参数'},
'log': {'btn': '', 'row': 3, 'text': '保存日志'},
'end': {'btn': '', 'row': 4, 'text': '结束运行'},
}
widgits = {
'path': {'label': '', 'entry': '', 'row': 1, 'col': 2, 'text': '数据文件夹路径'},
'av': {'label': '', 'entry': '', 'row': 2, 'col': 2, 'text': '角速度'},
'rc': {'label': '', 'entry': '', 'row': 2, 'col': 4, 'text': '额定电流'},
'rpm': {'label': '', 'entry': '', 'row': 2, 'col': 6, 'text': '额定转速'},
'rr': {'label': '', 'entry': '', 'row': 2, 'col': 8, 'text': '减速比'},
'dur': {'label': '', 'entry': '', 'row': 2, 'col': 10, 'text': '周期时间'},
'axis': {'label': '', 'optionmenu': '', 'row': 3, 'col': 2, 'text': ''},
'vel': {'label': '', 'optionmenu': '', 'row': 3, 'col': 4, 'text': ''},
'trq': {'label': '', 'optionmenu': '', 'row': 3, 'col': 6, 'text': ''},
'trqh': {'label': '', 'optionmenu': '', 'row': 3, 'col': 8, 'text': ''},
'estop': {'label': '', 'optionmenu': '', 'row': 3, 'col': 10, 'text': ''},
'rc1': {'label': '', 'entry': '', 'row': 4, 'col': 2, 'text': '额定电流'},
'rc2': {'label': '', 'entry': '', 'row': 4, 'col': 4, 'text': '额定电流'},
'rc3': {'label': '', 'entry': '', 'row': 4, 'col': 6, 'text': '额定电流'},
'rc4': {'label': '', 'entry': '', 'row': 4, 'col': 8, 'text': '额定电流'},
'rc5': {'label': '', 'entry': '', 'row': 4, 'col': 10, 'text': '额定电流'},
'rc6': {'label': '', 'entry': '', 'row': 4, 'col': 12, 'text': '额定电流'},
}
class App(customtkinter.CTk):
def __init__(self):
super().__init__()
self.my_font = customtkinter.CTkFont(family="Consolas", size=16, weight="bold")
self.w_param = 84
# =====================================================================
# configure window
self.title("AIO - All in one automatic toolbox")
# self.iconbitmap('./icon.ico')
self.geometry("1200x550+30+30")
self.protocol("WM_DELETE_WINDOW", self.func_end_callback)
self.config(bg='#E9E9E9')
self.grid_rowconfigure(5, weight=1)
self.grid_columnconfigure((1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13), weight=1)
self.minsize(1200, 550)
# =====================================================================
# create frame sidebar(left)
self.frame_func = customtkinter.CTkFrame(self, width=120, corner_radius=0, fg_color='#E9E9E9')
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="#4F4F4F")
self.label_logo.grid(row=0, column=0, padx=15, pady=15)
# create buttons
for func in btns:
btns[func]['btn'] = customtkinter.CTkButton(self.frame_func, corner_radius=10, text=btns[func]['text'], fg_color='#4F4F4F', font=self.my_font)
btns[func]['btn'].grid(row=btns[func]['row'], column=0, sticky='new', padx=10, pady=10, ipadx=5, ipady=5)
btns['start']['btn'].configure(command=lambda: self.thread_it(self.func_start_callback))
btns['check']['btn'].configure(command=lambda: self.thread_it(self.func_check_callback))
btns['log']['btn'].configure(command=lambda: self.thread_it(self.func_log_callback))
btns['end']['btn'].configure(command=lambda: self.thread_it(self.func_end_callback))
# create version info
self.label_version = customtkinter.CTkLabel(self.frame_func, justify='left', text="Vers: 0.1.6.2\nDate: 06/16/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')
# =====================================================================
# create tabviews
self.tabview = customtkinter.CTkTabview(self, width=10000, height=100, anchor='w', fg_color='#E9E9E9', border_width=2, border_color='#CDCDCD')
self.tabview.grid(row=0, column=1, padx=10, pady=5, sticky="nsew")
self.tabview.add("Data Process")
self.tabview.add("Automatic Test")
# create main menu
self.menu_main_dp = customtkinter.CTkOptionMenu(self.tabview.tab('Data Process'), values=["init", "brake", "current", "iso", "wavelogger"], font=self.my_font, text_color='yellow', button_color='red', fg_color='green', command=self.func_main_callback)
self.menu_main_dp.grid(row=1, column=1, sticky='we', padx=5, pady=5)
self.menu_main_dp.set("Start Here!")
# create sub menu
self.menu_sub_dp = customtkinter.CTkOptionMenu(self.tabview.tab('Data Process'))
# =====================================================================
# create widgits
for widgit in widgits:
if widgit == 'path':
widgits[widgit]['label'] = customtkinter.CTkLabel(self.tabview.tab('Data Process'), text=f'{widgit.upper()}', font=self.my_font)
widgits[widgit]['label'].grid(row=widgits[widgit]['row'], column=widgits[widgit]['col'], sticky='e', pady=5)
widgits[widgit]['entry'] = customtkinter.CTkEntry(self.tabview.tab('Data Process'), width=670, placeholder_text=widgits[widgit]['text'], font=self.my_font)
widgits[widgit]['entry'].grid(row=widgits[widgit]['row'], column=widgits[widgit]['col']+1, columnspan=11, padx=(5, 10), pady=5, sticky='we')
widgits[widgit]['entry'].configure(state='disabled')
elif widgit in ['av', 'rc', 'rpm', 'rr', 'dur', 'rc1', 'rc2', 'rc3', 'rc4', 'rc5', 'rc6']:
widgits[widgit]['label'] = customtkinter.CTkLabel(self.tabview.tab('Data Process'), text=f"{widgit.upper()}", font=self.my_font)
widgits[widgit]['label'].grid(row=widgits[widgit]['row'], column=widgits[widgit]['col'], sticky='e', pady=5)
widgits[widgit]['entry'] = customtkinter.CTkEntry(self.tabview.tab('Data Process'), width=self.w_param, placeholder_text=f"{widgits[widgit]['text']}", font=self.my_font)
widgits[widgit]['entry'].grid(row=widgits[widgit]['row'], column=widgits[widgit]['col']+1, padx=(5, 10), pady=5, sticky='w')
widgits[widgit]['entry'].configure(state='disabled')
elif widgit in ['axis', 'vel', 'trq', 'trqh', 'estop']:
widgits[widgit]['label'] = customtkinter.CTkLabel(self.tabview.tab('Data Process'), text=f"{widgit.upper()}", font=self.my_font)
widgits[widgit]['label'].grid(row=widgits[widgit]['row'], column=widgits[widgit]['col'], sticky='e', pady=5)
widgits[widgit]['optionmenu'] = customtkinter.CTkOptionMenu(self.tabview.tab('Data Process'), button_color='#708090', fg_color='#778899', values=["1", "2", "3", "4", "5", "6", "7"], width=self.w_param, font=self.my_font)
widgits[widgit]['optionmenu'].grid(row=widgits[widgit]['row'], column=widgits[widgit]['col']+1, padx=(5, 10), pady=5, sticky='w')
widgits[widgit]['optionmenu'].configure(state='disabled')
# =====================================================================
# create textbox
self.textbox = customtkinter.CTkTextbox(self, wrap='none', font=customtkinter.CTkFont(family="consolas", size=14), text_color="blue", fg_color='#E9E9E9', border_width=2, border_color='#CDCDCD', border_spacing=5)
self.textbox.grid(row=5, column=1, rowspan=2, columnspan=13, ipadx=10, ipady=10, padx=10, pady=(5, 10), sticky='nsew')
self.textbox.configure(state='disabled')
# =====================================================================
# version check
cur_vers = self.label_version.cget("text").replace('\n', ' @ ').replace("Vers: ", '').replace("Date: ", '')
url_vers = 'http://10.2.23.150:10008/vers'
try:
new_vers = urlopen(url_vers).read().decode('utf-8')
if cur_vers.strip() != new_vers.strip():
msg = f"""当前版本：{cur_vers}\n更新版本：{new_vers}\n\n请及时更新 http://10.2.23.150:10003/s/jRfM"""
tkinter.messagebox.showwarning(title="版本更新", message=msg)
except:
tkinter.messagebox.showwarning(title="版本更新", message="连接服务器失败，无法确认当前是否是最新版本......")
def thread_it(self, func, *args):
""" 将函数打包进线程 """
self.myThread = Thread(target=func, args=args)
self.myThread.daemon = True # 主线程退出就直接让子线程跟随退出,不论是否运行完成。
self.myThread.start()
def initialization(self):
for widgit in widgits:
if widgit in ['path', 'av', 'rc', 'rpm', 'rr', 'dur', 'rc1', 'rc2', 'rc3', 'rc4', 'rc5', 'rc6']:
widgits[widgit]['label'].configure(text=f'{widgit.upper()}', text_color='black')
widgits[widgit]['entry'].configure(placeholder_text=widgits[widgit]['text'], state='disabled')
elif widgit in ['axis', 'vel', 'trq', 'trqh', 'estop']:
widgits[widgit]['label'].configure(text=f'{widgit.upper()}', text_color="black")
widgits[widgit]['optionmenu'].configure(state='disabled')
self.menu_sub_dp.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':
for widgit in widgits:
if widgit in ['path', 'av', 'rr']:
widgits[widgit]['label'].configure(text_color='red')
widgits[widgit]['entry'].configure(state='normal')
elif widgit in ['axis', 'vel', 'trq', 'estop']:
widgits[widgit]['label'].configure(text_color="red")
widgits[widgit]['optionmenu'].configure(state='normal')
elif func_name == 'current':
self.menu_sub_dp = customtkinter.CTkOptionMenu(self.tabview.tab('Data Process'), values=["max", "avg", "cycle"], font=self.my_font, button_color='red', fg_color='green', command=self.func_sub_callback)
self.menu_sub_dp.grid(row=2, column=1, sticky='we', padx=5, pady=5)
self.menu_sub_dp.set("--select--")
self.menu_sub_dp.configure(text_color='yellow')
for widgit in widgits:
if widgit in ['path', 'rc', 'rc1', 'rc2', 'rc3', 'rc4', 'rc5', 'rc6']:
color = 'blue' if widgit == 'rc' else 'red'
widgits[widgit]['label'].configure(text_color=color)
widgits[widgit]['entry'].configure(state='normal')
elif widgit in ['trqh',]:
widgits[widgit]['label'].configure(text_color="red")
widgits[widgit]['optionmenu'].configure(state='normal')
elif func_name == 'iso' or func_name == 'wavelogger':
for widgit in widgits:
if widgit in ['path',]:
widgits[widgit]['label'].configure(text_color='red')
widgits[widgit]['entry'].configure(state='normal')
else:
self.initialization()
self.menu_main_dp.set("Start Here!")
def func_sub_callback(self, func_name):
if func_name == "max":
for widgit in widgits:
if widgit in ['rpm', 'dur']:
widgits[widgit]['label'].configure(text_color='black')
widgits[widgit]['entry'].configure(state='disabled')
elif widgit in ['vel', 'trq']:
widgits[widgit]['label'].configure(text_color='black')
widgits[widgit]['optionmenu'].configure(state='disabled')
elif func_name == 'avg':
for widgit in widgits:
if widgit in ['rpm', 'dur']:
widgits[widgit]['label'].configure(text_color='black')
widgits[widgit]['entry'].configure(state='disabled')
elif widgit in ['vel', 'trq']:
widgits[widgit]['label'].configure(text_color='black')
widgits[widgit]['optionmenu'].configure(state='disabled')
elif func_name == 'cycle':
for widgit in widgits:
if widgit in ['rpm', 'dur']:
widgits[widgit]['label'].configure(text_color='blue')
widgits[widgit]['entry'].configure(state='normal')
elif widgit in ['vel', 'trq']:
widgits[widgit]['label'].configure(text_color="red")
widgits[widgit]['optionmenu'].configure(state='normal')
def write2textbox(self, text, wait=0, exitcode=0, color='blue'):
self.textbox.tag_add(color, 'insert', 'end')
self.textbox.tag_config(tagName=color, foreground=color)
if wait != 0:
self.textbox.insert(index='end', text=text, tags=color)
self.textbox.update()
self.textbox.see('end')
elif exitcode != 0:
self.textbox.insert(index='end', text=text + '\n', tags=color)
self.textbox.update()
self.textbox.see('end')
self.textbox.configure(state='disabled')
raise Exception(f"Error code: {exitcode}")
else:
self.textbox.insert(index='end', text=text + '\n', tags=color)
self.textbox.update()
self.textbox.see('end')
def is_float(self, flag, *args):
for item in args:
try:
if flag == 'optional':
item = 0 if item == '' else item
_ = float(item)
elif flag == 'required':
_ = float(item)
except Exception as Err:
tkinter.messagebox.showerror(title="参数错误", message="请检查对应参数是否填写正确！", )
self.write2textbox(f"错误信息：{Err}\n参数数据缺失，或者数据类型错误，更正后重新运行...\n", 0, 3, 'red')
return True
def check_param(self):
func_name = self.menu_main_dp.get()
if func_name == 'brake':
path = widgits['path']['entry'].get().strip()
av = widgits['av']['entry'].get().strip('- ')
rr = widgits['rr']['entry'].get().strip('- ')
axis = widgits['axis']['optionmenu'].get()
vel = widgits['vel']['optionmenu'].get()
trq = widgits['trq']['optionmenu'].get()
estop = widgits['estop']['optionmenu'].get()
c1 = exists(path)
c2 = self.is_float('required', av, rr)
c3 = True if len({vel, trq, estop}) == 3 else False
if c1 and c2 and c3:
return 1, path, float(av), float(rr), int(axis), int(vel), int(trq), int(estop)
else:
return 0, 0
# =======================================================
elif func_name == 'current':
path = widgits['path']['entry'].get().strip()
rc = widgits['rc']['entry'].get().strip('- ')
rpm = widgits['rpm']['entry'].get().strip()
dur = widgits['dur']['entry'].get().strip()
rc1 = widgits['rc1']['entry'].get().strip()
rc2 = widgits['rc2']['entry'].get().strip()
rc3 = widgits['rc3']['entry'].get().strip()
rc4 = widgits['rc4']['entry'].get().strip()
rc5 = widgits['rc5']['entry'].get().strip()
rc6 = widgits['rc6']['entry'].get().strip()
vel = widgits['vel']['optionmenu'].get()
trq = widgits['trq']['optionmenu'].get()
trqh = widgits['trqh']['optionmenu'].get()
sub = self.menu_sub_dp.get()
c1 = exists(path)
c2 = sub in ['max', 'avg', 'cycle']
c3 = self.is_float('optional', rc, rpm)
c4 = self.is_float('required', rc1, rc2, rc3, rc4, rc5, rc6)
c5 = c6 = True
if sub == 'cycle':
c5 = True if len({vel, trq}) == 2 else False
c6 = self.is_float('optional', dur)
if c1 and c2 and c3 and c4 and c5 and c6:
rcs = []
for x in [rc1, rc2, rc3, rc4, rc5, rc6]:
rcs.append(float(x))
rc = 0 if rc == '' else rc
dur = 0 if sub != 'cycle' or dur == '' else dur
rpm = 0 if sub != 'cycle' or rpm == '' else rpm
rcs.append(float(rc))
return 2, path, sub, rcs, int(vel), int(trq), int(trqh), float(dur), float(rpm)
else:
return 0, 0
# =======================================================
elif func_name == 'iso':
path = widgits['path']['entry'].get().strip()
c1 = exists(path)
if c1:
return 3, path
else:
return 0, 0
# =======================================================
elif func_name == 'wavelogger':
path = widgits['path']['entry'].get().strip()
c1 = exists(path)
if c1:
return 4, path
else:
return 0, 0
# =======================================================
else:
return 0, 0
def func_start_callback(self):
self.textbox.configure(state='normal')
self.textbox.delete(index1='1.0', index2='end')
flag, *args = self.check_param()
func_dict = {1: brake.main, 2: current.main, 3: iso.main, 4: wavelogger.main}
if flag == 1:
func_dict[flag](path=args[0], av=args[1], rr=args[2], axis=args[3], vel=args[4], trq=args[5], estop=args[6], w2t=self.write2textbox)
elif flag == 2:
func_dict[flag](path=args[0], sub=args[1], rcs=args[2], vel=args[3], trq=args[4], trqh=args[5], dur=args[6], rpm=args[7], w2t=self.write2textbox)
elif flag == 3:
func_dict[flag](path=args[0], w2t=self.write2textbox)
elif flag == 4:
func_dict[flag](path=args[0], w2t=self.write2textbox)
else:
tkinter.messagebox.showerror(title="参数错误", message="请检查对应参数是否填写正确！", )
self.textbox.configure(state='disabled')
def func_check_callback(self):
self.textbox.configure(state='normal')
self.textbox.delete(index1='1.0', index2='end')
flag, *args = self.check_param()
if flag:
tkinter.messagebox.showinfo(title="参数正确", message="所有参数形式上填写无误，可以开始运行！")
else:
tkinter.messagebox.showerror(title="参数错误", message="需要检查对应参数是否填写正确！", )
self.textbox.configure(state='disabled')
def func_log_callback(self):
content = self.textbox.get(index1='1.0', index2='end')
if len(content) > 1:
try:
now = strftime('%Y%m%d%H%M%S', localtime(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{getcwd()}')
except Exception as Err:
self.write2textbox(Err)
tkinter.messagebox.showerror(title="保存失败", message="未能保存本次日志，或未能完整保存，请准备好相关数据，联系fanmingfu@rokae.com查看详细信息！", )
else:
tkinter.messagebox.showwarning(title="未能保存", message="日志数据为空，不可保存！")
def func_end_callback(self):
if tkinter.messagebox.askyesno(title="关闭程序", message="相关数据可能未保存，正在运行程序时有概率会损坏数据文件，确定要终止程序运行吗？"):
self.destroy()
if __name__ == "__main__":
aio = App()
aio.mainloop()

298
aio/code/brake.py Normal file
View File

@ -0,0 +1,298 @@
# coding: utf-8
from os import scandir
from os.path import isfile, exists
from sys import argv
from openpyxl import load_workbook
from time import time, sleep, strftime, localtime
from threading import Thread
from pandas import read_csv
class GetThreadResult(Thread):
def __init__(self, func, args=()):
super(GetThreadResult, self).__init__()
self.func = func
self.args = args
self.result = 0
def run(self):
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 data_process(result_file, raw_data_dirs, av, rr, axis, vel, trq, w2t, estop):
# 功能：完成一个结果文件的数据处理
# 参数：结果文件，数据目录，以及预读取的参数
# 返回值：-
file_name = result_file.split('\\')[-1]
w2t(f"正在打开文件 {file_name} 需要 1min 左右", 1, 0, 'orange')
global stop
stop = 0
t_excel = GetThreadResult(load_workbook, args=(result_file, ))
t_wait = Thread(target=w2t_local, args=('.', 1, w2t))
t_excel.start()
t_wait.start()
t_excel.join()
wb_result = t_excel.get_result()
stop = 1
sleep(1.1)
w2t('')
prefix = result_file.split('\\')[-1].split('_')[0]
for raw_data_dir in raw_data_dirs:
if raw_data_dir.split('\\')[-1].split('_')[0] == prefix:
now_doing_msg(raw_data_dir, 'start', w2t)
_, data_files = traversal_files(raw_data_dir, w2t)
# 数据文件串行处理模式---------------------------------
# count = 1
# for data_file in data_files:
# now_doing_msg(data_file, 'start', w2t)
# single_file_process(data_file, wb_result, count, av, rr, axis, vel, trq, w2t, estop)
# count += 1
# now_doing_msg(data_file, 'done', w2t)
# ---------------------------------------------------
# 数据文件并行处理模式---------------------------------
threads = [Thread(target=single_file_process, args=(data_files[0], wb_result, 1, av, rr, axis, vel, trq, w2t, estop)),
Thread(target=single_file_process, args=(data_files[1], wb_result, 2, av, rr, axis, vel, trq, w2t, estop)),
Thread(target=single_file_process, args=(data_files[2], wb_result, 3, av, rr, axis, vel, trq, w2t, estop))]
[t.start() for t in threads]
[t.join() for t in threads]
# ---------------------------------------------------
now_doing_msg(raw_data_dir, 'done', w2t)
now_doing_msg(result_file, 'done', w2t)
w2t(f"正在保存文件 {file_name} 需要 1min 左右", 1, 0, 'orange')
stop = 0
t_excel = Thread(target=wb_result.save, args=(result_file, ))
t_wait = Thread(target=w2t_local, args=('.', 1, w2t))
t_excel.start()
t_wait.start()
t_excel.join()
stop = 1
sleep(1.1)
w2t('\n')
def check_files(raw_data_dirs, result_files, w2t):
# 功能：检查数据文件以及结果文件的合规性
# 参数：数据文件夹，结果文件
# 返回值：-
if len(result_files) != 3:
msg = "结果文件数目错误，结果文件有且只有三个，请确认！"
for result_file in result_files:
w2t(result_file)
w2t(msg, 0, 2, 'red')
prefix = []
for result_file in result_files:
prefix.append(result_file.split('\\')[-1].split('_')[0])
if not sorted(prefix) == sorted(['reach33', 'reach66', 'reach100']):
wd = result_files[0].split('\\')
del wd[-1]
wd = '\\'.join(wd)
msg = f"""请关闭所有相关数据文件，并检查工作目录 {wd} 下，有且只允许有类似如下三个文件：
1. reach33_XXX制动性能测试.xlsx
2. reach66_XXX制动性能测试.xlsx
3. reach100_XX制动性能测试.xlsx"""
w2t(msg, 0, 3, 'red')
for raw_data_dir in raw_data_dirs:
components = raw_data_dir.split('\\')[-1].split('_')
sorted(components)
if components[0] not in ['reach33', 'reach66', 'reach100'] or \
components[1] not in ['load33', 'load66', 'load100'] or \
components[2] not in ['speed33', 'speed66', 'speed100']:
msg = f"""报错信息：数据目录 {raw_data_dir} 命名不合规，请参考如下形式：
命名规则：reachAA_loadBB_speedCC
规则解释：AA/BB/CC 指的是臂展/负载/速度的比例，例如reach66_load100_speed33：66%臂展，100%负载以及33%速度情况下的测试结果文件夹"""
w2t(msg, 0, 4, 'red')
_, raw_data_files = traversal_files(raw_data_dir, w2t)
if len(raw_data_files) != 3:
msg = f"数据目录 {raw_data_dir} 下数据文件个数错误，每个数据目录下有且只能有三个以 .data 为后缀的数据文件"
w2t(msg, 0, 5, 'red')
for raw_data_file in raw_data_files:
if not raw_data_file.split('\\')[-1].endswith('.data'):
msg = f"数据文件 {raw_data_file} 后缀错误，每个数据目录下有且只能有三个以 .data 为后缀的数据文件"
w2t(msg, 0, 6, 'red')
w2t("数据目录合规性检查结束，未发现问题......")
def now_doing_msg(docs, flag, w2t):
# 功能：输出正在处理的文件或目录
# 参数：文件或目录，start 或 done 标识
# 返回值：-
now = strftime('%Y-%m-%d %H:%M:%S', localtime(time()))
file_type = 'file' if isfile(docs) else 'dir'
if flag == 'start' and file_type == 'dir':
w2t(f"[{now}] 正在处理目录 {docs} 中的数据......")
elif flag == 'start' and file_type == 'file':
w2t(f"[{now}] 正在处理文件 {docs} 中的数据......")
elif flag == 'done' and file_type == 'dir':
w2t(f"[{now}] 目录 {docs} 数据文件已处理完毕")
elif flag == 'done' and file_type == 'file':
w2t(f"[{now}] 文件 {docs} 数据已处理完毕")
def w2t_local(msg, wait, w2t):
while True:
global stop
if stop == 0 and wait != 0:
sleep(1)
w2t(msg, wait, 0, 'orange')
else:
break
def single_file_process(data_file, wb_result, count, av, rr, axis, vel, trq, w2t, estop):
# 功能：完成单个数据文件的处理
# 参数：如上
# 返回值：-
df = read_csv(data_file, sep='\t')
conditions = sorted(data_file.split('\\')[-2].split('_')) # ['loadxx', 'reachxx', 'speedxx']
result_sheet_name = find_result_sheet_name(conditions, count)
ws_result = wb_result[result_sheet_name]
row_start, row_end = find_row_start(data_file, df, conditions, av, rr, axis, vel, w2t, estop)
copy_data_to_result(df, ws_result, row_start, row_end, vel, trq, estop)
def copy_data_to_result(df, ws_result, row_start, row_end, vel, trq, estop):
# 功能：将数据文件中有效数据拷贝至结果文件对应的 sheet
# 参数：如上
# 返回值：-
# 结果文件数据清零
data = []
for _row in range(row_start, row_end + 1):
data.append(df.iloc[_row, vel-1])
data.append(df.iloc[_row, trq-1])
data.append(df.iloc[_row, estop-1])
i = 0
row_max = 2000 if row_end-row_start < 2000 else row_end-row_start+20
for _row in range(2, row_max):
try:
ws_result.cell(row=_row, column=1).value = data[i]
ws_result.cell(row=_row, column=2).value = data[i+1]
ws_result.cell(row=_row, column=3).value = data[i+2]
i += 3
except:
ws_result.cell(row=_row, column=1).value = None
ws_result.cell(row=_row, column=2).value = None
ws_result.cell(row=_row, column=3).value = None
def find_result_sheet_name(conditions, count):
# 功能：获取结果文件准确的sheet页名称
# 参数：臂展和速度的列表
# 返回值：结果文件对应的sheet name
# 33%负载_33%速度_1 - ['loadxx', 'reachxx', 'speedxx']
load = conditions[0].removeprefix('load')
speed = conditions[2].removeprefix('speed')
result_sheet_name = f"{load}%负载_{speed}%速度_{count}"
return result_sheet_name
def find_row_start(data_file, df, conditions, av, rr, axis, vel, w2t, estop):
# 功能：查找数据文件中有效数据的行号，也即最后一个速度下降的点位
# 参数：如上
# 返回值：速度下降点位，最后的数据点位
ratio = float(conditions[2].removeprefix('speed'))/100
av_max = av * ratio
row_max = df.index[-1]
# threshold = 30 if axis == 2 and conditions[0].removeprefix('load') == '100' else 10
threshold = 0.95
for _row in range(row_max, -1, -1):
if df.iloc[_row, estop-1] != 0:
row_start = _row-20 if _row-20 > 0 else 0
break
else:
w2t(f"数据文件 {data_file} 采集的数据中没有 ESTOP 为非 0 的情况，需要确认", 0, 9, 'red')
for _row in range(row_start, row_max):
speed_row = (df.iloc[_row, vel-1] * 180) / 3.1415926 * rr * 60 / 360
if abs(speed_row) < 1:
row_end = _row+100 if _row+100 <= row_max else row_max
break
else:
w2t(f"数据文件 {data_file} 最后的速度未降为零 ", 0, 10, 'red')
av_estop = abs((df.iloc[row_start-10:row_start+10, vel-1].abs().mean() * 180) / 3.1415926)
if abs(av_estop/av_max) < threshold:
filename = data_file.split('\\')[-1]
w2t(f"[av_estop: {av_estop:.2f} | shouldbe: {av_max:.2f}] 数据文件 {filename} 触发 ESTOP 时未采集到指定百分比的最大速度，需要检查", 0, 0, '#8A2BE2')
return row_start, row_end
def traversal_files(path, w2t):
# 功能：以列表的形式分别返回指定路径下的文件和文件夹，不包含子目录
# 参数：路径
# 返回值：路径下的文件夹列表 路径下的文件列表
if not exists(path):
msg = f'数据文件夹{path}不存在，请确认后重试......'
w2t(msg, 0, 1, 'red')
else:
dirs = []
files = []
for item in scandir(path):
if item.is_dir():
dirs.append(item.path)
elif item.is_file():
files.append(item.path)
return dirs, files
def main(path, av, rr, axis, vel, trq, estop, w2t):
# 功能：执行处理所有数据文件
# 参数：initialization函数的返回值
# 返回值：-
time_start = time()
raw_data_dirs, result_files = traversal_files(path, w2t)
try:
# threads = []
check_files(raw_data_dirs, result_files, w2t)
prefix = []
for raw_data_dir in raw_data_dirs:
prefix.append(raw_data_dir.split('\\')[-1].split("_")[0])
for result_file in result_files:
if result_file.split('\\')[-1].split('_')[0] not in set(prefix):
continue
else:
now_doing_msg(result_file, 'start', w2t)
data_process(result_file, raw_data_dirs, av, rr, axis, vel, trq, w2t, estop)
# threads.append(Thread(target=data_process, args=(result_file, raw_data_dirs, AV, RR, RC, AXIS)))
# [t.start() for t in threads]
# [t.join() for t in threads]
except Exception as Err:
msg = f"出现错误：{Err}\n程序运行错误，请检查配置文件是否准确设定，以及数据文件组织是否正确，也有可能是结果文件损坏，尝试重新复制一份，再运行！"
w2t(msg, 0, 11, 'red')
w2t("----------------------------------------------------------")
w2t("全部处理完毕")
time_end = time()
time_total = time_end - time_start
msg = f"数据处理时间：{time_total // 3600:02.0f} h {time_total % 3600 // 60:02.0f} m {time_total % 60:02.0f} s"
w2t(msg)
if __name__ == "__main__":
stop = 0
main(path=argv[1], av=argv[2], rr=argv[3], axis=argv[4], vel=argv[5], trq=argv[6], estop=argv[7], w2t=argv[8])

371
aio/code/current.py Normal file
View File

@ -0,0 +1,371 @@
from openpyxl import load_workbook
from os import scandir
from os.path import exists
from sys import argv
from pandas import read_csv, concat, set_option
from re import match
from threading import Thread
from time import sleep
from csv import reader
class GetThreadResult(Thread):
def __init__(self, func, args=()):
super(GetThreadResult, self).__init__()
self.func = func
self.args = args
self.result = 0
def run(self):
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, wait, w2t):
while True:
global stop
if stop == 0 and wait != 0:
sleep(1)
w2t(msg, wait, 0, 'orange')
else:
break
def traversal_files(path, w2t):
# 功能：以列表的形式分别返回指定路径下的文件和文件夹，不包含子目录
# 参数：路径
# 返回值：路径下的文件夹列表 路径下的文件列表
if not exists(path):
msg = f'数据文件夹{path}不存在，请确认后重试......'
w2t(msg, 0, 8, 'red')
else:
dirs = []
files = []
for item in scandir(path):
if item.is_dir():
dirs.append(item.path)
elif item.is_file():
files.append(item.path)
return dirs, files
def initialization(path, sub, w2t):
_, data_files = traversal_files(path, w2t)
count = 0
for data_file in data_files:
filename = data_file.split('\\')[-1]
if sub != 'cycle':
if not (match('j[1-7].*\\.data', filename) or match('j[1-7].*\\.csv', filename)):
msg = f"所有文件必须以 jx_ 开头，以 .data/csv 结尾（x取值1-7），请检查后重新运行。"
w2t(msg, 0, 6, 'red')
else:
if filename.endswith('.xlsx'):
count += 1
elif not (match('j[1-7].*\\.data', filename) or match('j[1-7].*\\.csv', filename)):
msg = f"所有文件必须以 jx_ 开头，以 .data/csv 结尾（x取值1-7），请检查后重新运行。"
w2t(msg, 0, 7, 'red')
if sub == 'cycle' and count != 1:
w2t("未找到电机电流数据处理excel表格，确认后重新运行！", 0, 5, 'red')
return data_files
def current_max(data_files, rcs, trqh, w2t):
current = {1: [], 2: [], 3: [], 4: [], 5: [], 6: [], 7: []}
for data_file in data_files:
if data_file.endswith('.data'):
df = read_csv(data_file, sep='\t')
elif data_file.endswith('.csv'):
df = read_csv(data_file, sep=',', encoding='gbk', header=8)
axis = int(data_file.split('\\')[-1].split('_')[0].removeprefix('j'))
rca = rcs[axis-1]
col = df.columns.values[trqh-1]
c_max = df[col].max()
scale = 1 if data_file.endswith('.csv') else 1000
_ = abs(c_max/scale*rca)
current[axis].append(_)
w2t(f"{data_file}: {_:.4f}")
for axis, cur in current.items():
if not cur:
continue
else:
w2t(f"{axis}轴数据：", 1, 0, 'purple')
for value in cur:
w2t(f"{value:.4f} ", 1, 0, 'purple')
w2t('')
w2t("\n【MAX】数据处理完毕......")
return current
def current_avg(data_files, rcs, trqh, w2t):
current = {1: [], 2: [], 3: [], 4: [], 5: [], 6: [], 7: []}
for data_file in data_files:
if data_file.endswith('.data'):
df = read_csv(data_file, sep='\t')
elif data_file.endswith('.csv'):
df = read_csv(data_file, sep=',', encoding='gbk', header=8)
axis = int(data_file.split('\\')[-1].split('_')[0].removeprefix('j'))
rca = rcs[axis-1]
col = df.columns.values[trqh - 1]
c_std = df[col].std()
c_avg = df[col].mean()
scale = 1 if data_file.endswith('.csv') else 1000
_ = (abs(c_avg)+c_std)/scale*rca
current[axis].append(_)
w2t(f"{data_file}: {_:.4f}")
for axis, cur in current.items():
if not cur:
continue
else:
w2t(f"{axis}轴数据：", 1, 0, 'purple')
for value in cur:
w2t(f"{value:.4f} ", 1, 0, 'purple')
w2t('')
w2t("\n【AVG】数据处理完毕......")
return current
def current_cycle(dur, data_files, rcs, vel, trq, trqh, rpm, w2t):
result = None
hold = []
single = []
for data_file in data_files:
filename = data_file.split('\\')[-1]
if data_file.endswith('.xlsx'):
result = data_file
elif match('j[1-7]_hold_.*\\.data', filename) or match('j[1-7]_hold_.*\\.csv', filename):
hold.append(data_file)
else:
single.append(data_file)
w2t(f"正在打开文件 {result}，需要 10s 左右", 1, 0, 'orange')
global stop
stop = 0
t_excel = GetThreadResult(load_workbook, args=(result, ))
t_wait = Thread(target=w2t_local, args=('.', 1, w2t))
t_excel.start()
t_wait.start()
t_excel.join()
wb = t_excel.get_result()
stop = 1
sleep(1.1)
w2t('')
if hold != []:
avg = current_avg(hold, rcs, trqh, w2t)
for axis, cur_value in avg.items():
try:
shtname = f"J{axis}"
wb[shtname]["J4"].value = float(cur_value)
except:
pass
if dur == 0:
p_single(wb, single, vel, trq, rpm, w2t)
else:
p_scenario(wb, single, vel, trq, rpm, dur, w2t)
w2t(f"正在保存文件 {result}，需要 10s 左右", 1, 0, 'orange')
stop = 0
t_excel = Thread(target=wb.save, args=(result, ))
t_wait = Thread(target=w2t_local, args=('.', 1, w2t))
t_excel.start()
t_wait.start()
t_excel.join()
stop = 1
sleep(1.1)
w2t('\n')
w2t("----------------------------------------------------------")
w2t("全部处理完毕")
def find_point(data_file, pos, flag, df, _row_s, _row_e, w2t, exitcode, threshold, step, end_point):
if flag == 'lt':
while _row_e > end_point:
speed_avg = df.iloc[_row_s:_row_e, 0].abs().mean()
if speed_avg < threshold:
_row_e -= step
_row_s -= step
continue
else:
return _row_s, _row_e
else:
w2t(f"[{pos}] {data_file}数据有误，需要检查，无法找到第{exitcode}个有效点...", 0, exitcode, 'red')
elif flag == 'gt':
while _row_e > end_point:
speed_avg = df.iloc[_row_s:_row_e, 0].abs().mean()
if speed_avg > threshold:
_row_e -= step
_row_s -= step
continue
else:
return _row_s, _row_e
else:
w2t(f"[{pos}] {data_file}数据有误，需要检查，无法找到有效起始点或结束点...", 0, exitcode, 'red')
def p_single(wb, single, vel, trq, rpm, w2t):
# 1. 先找到第一个速度为零的点，数据从后往前找，一开始就是零的情况不予考虑
# 2. 记录第一个点的位置，继续向前查找第二个速度为零的点，同理，一开始为零的点不予考虑
# 3. 记录第二个点的位置，并将其中的数据拷贝至对应位置
for data_file in single:
rpm = 1 if rpm == 0 else rpm
scale = 1000 if data_file.endswith('.csv') else 1
axis = int(data_file.split('\\')[-1].split('_')[0].removeprefix('j'))
shtname = f"J{axis}"
ws = wb[shtname]
set_option("display.precision", 2)
if data_file.endswith('.data'):
df = read_csv(data_file, sep='\t')
elif data_file.endswith('.csv'):
df = read_csv(data_file, sep=',', encoding='gbk', header=8)
csv_reader = reader(open(data_file))
i = 0
cycle = 0.001
for row in csv_reader:
i += 1
if i == 3:
cycle = float(row[0].split(':')[1].split('ms')[0]) / 1000
break
ws["H11"] = cycle
col_names = list(df.columns)
df_1 = df[col_names[vel-1]].multiply(rpm)
df_2 = df[col_names[trq-1]].multiply(scale)
df = concat([df_1, df_2], axis=1)
_step = 5 if data_file.endswith('.csv') else 50
_end_point = 30 if data_file.endswith('.csv') else 200
_adjust = 0 if data_file.endswith('.csv') else 150
_row_e = df.index[-1]
_row_s = _row_e - _end_point
speed_avg = df.iloc[_row_s:_row_e, 0].abs().mean()
if speed_avg < 2:
# 过滤尾部为零无效数据
_row_s, _row_e = find_point(data_file, 'a1', 'lt', df, _row_s, _row_e, w2t, 1, threshold=5, step=_step, end_point=_end_point)
# 找到第一个起始点 row_end，继续找到有数据的部分，后面有一段有效数据区
row_end = _row_e - _adjust
_row_e -= _end_point
_row_s -= _end_point
_row_s, _row_e = find_point(data_file, 'a2', 'gt', df, _row_s, _row_e, w2t, 3, threshold=5, step=_step, end_point=_end_point)
# 速度已经快要降为零了，继续寻找下一个速度上升点
_row_e -= _end_point
_row_s -= _end_point
_row_s, _row_e = find_point(data_file, 'a3', 'lt', df, _row_s, _row_e, w2t, 3, threshold=5, step=_step, end_point=_end_point)
elif speed_avg > 2:
# 过滤尾部非零无效数据
_row_s, _row_e = find_point(data_file, 'b1', 'gt', df, _row_s, _row_e, w2t, 2, threshold=5, step=_step, end_point=_end_point)
# 找到第一个起始点 row_end，继续找到有数据的部分，后面有一段零数据区
row_end = _row_e - _adjust
_row_e -= _end_point
_row_s -= _end_point
_row_s, _row_e = find_point(data_file, 'b2', 'lt', df, _row_s, _row_e, w2t, 4, threshold=5, step=_step, end_point=_end_point)
# 目前已经有一点的速度值了，继续往前搜寻下一个速度为零的点
_row_e -= _end_point
_row_s -= _end_point
_row_s, _row_e = find_point(data_file, 'b3', 'gt', df, _row_s, _row_e, w2t, 4, threshold=5, step=_step, end_point=_end_point)
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])
i = 0
for row in ws.iter_rows(min_row=2, min_col=2, max_row=70000, max_col=3):
for cell in row:
try:
_ = f"{data[i]:.2f}"
cell.value = float(_)
i += 1
except:
cell.value = None
def p_scenario(wb, single, vel, trq, rpm, dur, w2t):
for data_file in single:
cycle = 0.001
rpm = 1 if rpm == 0 else rpm
scale = 1000 if data_file.endswith('.csv') else 1
axis = int(data_file.split('\\')[-1].split('_')[0].removeprefix('j'))
shtname = f"J{axis}"
ws = wb[shtname]
set_option("display.precision", 2)
if data_file.endswith('.data'):
df = read_csv(data_file, sep='\t')
elif data_file.endswith('.csv'):
df = read_csv(data_file, sep=',', encoding='gbk', header=8)
csv_reader = reader(open(data_file))
i = 0
for row in csv_reader:
i += 1
if i == 3:
cycle = float(row[0].split(':')[1].split('ms')[0]) / 1000
break
ws["H11"] = cycle
col_names = list(df.columns)
df_1 = df[col_names[vel-1]].multiply(rpm)
df_2 = df[col_names[trq-1]].multiply(scale)
df = concat([df_1, df_2], axis=1)
row_start = 300
row_end = row_start + int(dur/cycle)
if row_end > df.index[-1]:
w2t(f"位置超限：{data_file} 共有 {df.index[-1]} 条数据，无法取到第 {row_end} 条数据，需要确认场景周期时间...", 0, 9, 'red')
data = []
for row in range(row_start, row_end):
data.append(df.iloc[row, 0])
data.append(df.iloc[row, 1])
i = 0
for row in ws.iter_rows(min_row=2, min_col=2, max_row=70000, max_col=3):
for cell in row:
try:
_ = f"{data[i]:.2f}"
cell.value = float(_)
i += 1
except:
cell.value = None
# =======================================
def main(path, sub, rcs, vel, trq, trqh, dur, rpm, w2t):
data_files = initialization(path, sub, w2t)
if sub == 'max':
current_max(data_files, rcs, trqh, w2t)
elif sub == 'avg':
current_avg(data_files, rcs, trqh, w2t)
elif sub == 'cycle':
current_cycle(dur, data_files, rcs, vel, trq, trqh, rpm, w2t)
else:
pass
if __name__ == '__main__':
stop = 0
main(*argv[1:])

198
aio/code/iso.py Normal file
View File

@ -0,0 +1,198 @@
# _*_ encoding：utf-8 _*_
import pdfplumber
from openpyxl import load_workbook
from os import scandir, remove
from os.path import exists
from sys import argv
def traversal_files(path, w2t):
# 功能：以列表的形式分别返回指定路径下的文件和文件夹，不包含子目录
# 参数：路径
# 返回值：路径下的文件夹列表 路径下的文件列表
if not exists(path):
msg = f'数据文件夹{path}不存在，请确认后重试......'
w2t(msg, 0, 1, 'red')
else:
dirs = files = []
for item in scandir(path):
if item.is_dir():
dirs.append(item.path)
elif item.is_file():
files.append(item.path)
return dirs, files
def p_iso(file, p_files, ws, tmpfile):
p_files.append(file)
pdf = pdfplumber.open(file)
with open(tmpfile, mode='w', encoding='utf-8') as fb:
for page in pdf.pages:
fb.write(page.extract_text())
with open(tmpfile, mode='r', encoding='utf-8') as fb:
lines = fb.readlines()
lines = [line for line in lines if not line.startswith('Page ')]
for line in lines:
if line.strip() == "Pose Accuracy and Repeatability":
index = lines.index(line)
ws.cell(row=3, column=7).value = float(lines[index+4].split()[1])
ws.cell(row=4, column=7).value = float(lines[index+5].split()[1])
ws.cell(row=5, column=7).value = float(lines[index+6].split()[1])
ws.cell(row=6, column=7).value = float(lines[index+7].split()[1])
ws.cell(row=7, column=7).value = float(lines[index+8].split()[1])
ws.cell(row=8, column=7).value = float(lines[index+4].split()[2])
ws.cell(row=9, column=7).value = float(lines[index+5].split()[2])
ws.cell(row=10, column=7).value = float(lines[index+6].split()[2])
ws.cell(row=11, column=7).value = float(lines[index+7].split()[2])
ws.cell(row=12, column=7).value = float(lines[index+8].split()[2])
elif line.strip() == 'Pose Accuracy Variation':
index = lines.index(line)
ws.cell(row=13, column=7).value = float(lines[index+4].split()[1])
ws.cell(row=14, column=7).value = float(lines[index+5].split()[1])
ws.cell(row=15, column=7).value = float(lines[index+6].split()[1])
elif line.strip() == 'Distance Accuracy':
index = lines.index(line)
ws.cell(row=16, column=7).value = float(lines[index + 4].split()[1])
ws.cell(row=17, column=7).value = float(lines[index + 4].split()[2])
elif line.strip() == 'Stabilisation Time and Overshoot':
index = lines.index(line)
ws.cell(row=18, column=7).value = float(lines[index + 7].split()[3])
ws.cell(row=19, column=7).value = float(lines[index + 7].split()[2])
elif line.strip() == 'Velocity Accuracy and Repeatability':
index = lines.index(line)
ws.cell(row=20, column=7).value = float(lines[index + 4].split()[1])
ws.cell(row=21, column=7).value = float(lines[index + 4].split()[2])
ws.cell(row=22, column=7).value = float(lines[index + 4].split()[3])
elif line.strip()[:31] == 'Path Accuracy and Repeatability':
index = lines.index(line)
ws.cell(row=29, column=7).value = float(lines[index + 4].split()[1])
ws.cell(row=30, column=7).value = float(lines[index + 4].split()[2])
elif line.strip() == 'Corner Overshoot and Roundoff':
index = lines.index(line)
ws.cell(row=35, column=7).value = float(lines[index + 4].split()[1])
ws.cell(row=36, column=7).value = float(lines[index + 4].split()[2])
elif line.strip() == 'Robot Weaving':
index = lines.index(line)
ws.cell(row=41, column=7).value = float(lines[index + 4].split()[2])
ws.cell(row=42, column=7).value = float(lines[index + 4].split()[3])
ws.cell(row=43, column=7).value = float(lines[index + 4].split()[4])
else:
pass
def p_iso_100(file, p_files, ws, tmpfile):
p_files.append(file)
pdf = pdfplumber.open(file)
with open(tmpfile, mode='w', encoding='utf-8') as fb:
for page in pdf.pages:
fb.write(page.extract_text())
with open(tmpfile, mode='r', encoding='utf-8') as fb:
lines = fb.readlines()
lines = [line for line in lines if not line.startswith('Page ')]
for line in lines:
if line.strip() == 'Velocity Accuracy and Repeatability':
index = lines.index(line)
ws.cell(row=26, column=7).value = float(lines[index + 4].split()[1])
ws.cell(row=27, column=7).value = float(lines[index + 4].split()[2])
ws.cell(row=28, column=7).value = float(lines[index + 4].split()[3])
elif line.strip()[:31] == 'Path Accuracy and Repeatability':
index = lines.index(line)
ws.cell(row=33, column=7).value = float(lines[index + 4].split()[1])
ws.cell(row=34, column=7).value = float(lines[index + 4].split()[2])
elif line.strip() == 'Corner Overshoot and Roundoff':
index = lines.index(line)
ws.cell(row=39, column=7).value = float(lines[index + 4].split()[1])
ws.cell(row=40, column=7).value = float(lines[index + 4].split()[2])
elif line.strip() == 'Robot Weaving':
index = lines.index(line)
ws.cell(row=47, column=7).value = float(lines[index + 4].split()[2])
ws.cell(row=48, column=7).value = float(lines[index + 4].split()[3])
ws.cell(row=49, column=7).value = float(lines[index + 4].split()[4])
else:
pass
def p_iso_1000(file, p_files, ws, tmpfile):
p_files.append(file)
pdf = pdfplumber.open(file)
with open(tmpfile, mode='w', encoding='utf-8') as fb:
for page in pdf.pages:
fb.write(page.extract_text())
with open(tmpfile, mode='r', encoding='utf-8') as fb:
lines = fb.readlines()
lines = [line for line in lines if not line.startswith('Page ')]
for line in lines:
if line.strip() == 'Velocity Accuracy and Repeatability':
index = lines.index(line)
ws.cell(row=23, column=7).value = float(lines[index + 4].split()[1])
ws.cell(row=24, column=7).value = float(lines[index + 4].split()[2])
ws.cell(row=25, column=7).value = float(lines[index + 4].split()[3])
elif line.strip()[:31] == 'Path Accuracy and Repeatability':
index = lines.index(line)
ws.cell(row=31, column=7).value = float(lines[index + 4].split()[1])
ws.cell(row=32, column=7).value = float(lines[index + 4].split()[2])
elif line.strip() == 'Corner Overshoot and Roundoff':
index = lines.index(line)
ws.cell(row=37, column=7).value = float(lines[index + 4].split()[1])
ws.cell(row=38, column=7).value = float(lines[index + 4].split()[2])
elif line.strip() == 'Robot Weaving':
index = lines.index(line)
ws.cell(row=44, column=7).value = float(lines[index + 4].split()[2])
ws.cell(row=45, column=7).value = float(lines[index + 4].split()[3])
ws.cell(row=46, column=7).value = float(lines[index + 4].split()[4])
else:
pass
def main(path, w2t):
dirs, files = traversal_files(path, 1)
try:
wb = load_workbook(path + "/iso-results.xlsx")
ws = wb.active
for i in range(3, 50):
ws.cell(row=i, column=7).value = None
tmpfile = f"{path}\\data.txt"
except Exception as Err:
w2t(f"发生错误：{Err}", 0, 2, 'red')
p_files = []
for file in files:
if file.endswith('.pdf') and file.split('\\')[-1] == 'ISO.pdf':
w2t(f"正在处理{file}......")
p_iso(file, p_files, ws, tmpfile)
w2t(f"文件{file}已处理完毕。\n")
elif file.endswith('.pdf') and file.split('\\')[-1] == 'ISO-V100.pdf':
w2t(f"正在处理{file}......")
p_iso_100(file, p_files, ws, tmpfile)
w2t(f"文件{file}已处理完毕。\n")
elif file.endswith('.pdf') and file.split('\\')[-1] == 'ISO-V1000.pdf':
w2t(f"正在处理{file}......")
p_iso_1000(file, p_files, ws, tmpfile)
w2t(f"文件{file}已处理完毕。\n")
else:
pass
wb.save(path + '/iso-results.xlsx')
wb.close()
if len(p_files) == 0:
w2t(f"目录 {path} 下没有需要处理的文件，需要确认......", 0, 3, 'red')
else:
remove(tmpfile)
w2t("------------------------------------------")
w2t("所有文件均已处理完毕！")
if __name__ == '__main__':
main(path=argv[1], w2t=argv[2])

171
aio/code/wavelogger.py Normal file
View File

@ -0,0 +1,171 @@
import os
from pandas import read_csv
from csv import reader
from sys import argv
from os.path import exists
from os import scandir, remove
from openpyxl import Workbook
def traversal_files(path, w2t):
# 功能：以列表的形式分别返回指定路径下的文件和文件夹，不包含子目录
# 参数：路径
# 返回值：路径下的文件夹列表 路径下的文件列表
if not exists(path):
msg = f'数据文件夹{path}不存在，请确认后重试......'
w2t(msg, 0, 1, 'red')
else:
dirs = []
files = []
for item in scandir(path):
if item.is_dir():
dirs.append(item.path)
elif item.is_file():
files.append(item.path)
return dirs, files
def find_point(bof, step, pos, data_file, flag, df, row, w2t):
# bof: backward or forward
# pos: used for debug
# flag: greater than or lower than
if flag == 'gt':
while 0 < row < df.index[-1]-100:
if df.iloc[row, 2] > 2:
if bof == 'backward':
row -= step
elif bof == 'forward':
row += step
continue
else:
row_target = row
break
else:
if bof == 'backward':
w2t(f"[{pos}] 在 {data_file} 中，无法正确识别数据，需要确认...", 0, 2, 'red')
elif bof == 'forward':
row_target = row + 100
elif flag == 'lt':
while 0 < row < df.index[-1]-100:
if df.iloc[row, 2] < 2:
if bof == 'backward':
row -= step
elif bof == 'forward':
row += step
continue
else:
row_target = row
break
else:
if bof == 'backward':
w2t(f"[{pos}] 在 {data_file} 中，无法正确识别数据，需要确认...", 0, 3, 'red')
elif bof == 'forward':
row_target = row + 100
return row_target
def get_cycle_info(data_file, df, row, step, w2t):
# end -> middle: low
# middle -> start: high
# 1. 从最后读取数据，无论是大于1还是小于1，都舍弃，找到相反的值的起始点
# 2. 从起始点，继续往前寻找，找到与之数值相反的中间点
# 3. 从中间点，继续往前寻找，找到与之数值相反的结束点，至此，得到了高低数值的时间区间以及一轮的周期时间
if df.iloc[row, 2] < 2:
row = find_point('backward', step, 'a1', data_file, 'lt', df, row, w2t)
_row = find_point('backward', step, 'a2', data_file, 'gt', df, row, w2t)
_row = find_point('backward', step, 'a3', data_file, 'lt', df, _row, w2t)
row_end = find_point('backward', step, 'a4', data_file, 'gt', df, _row, w2t)
row_middle = find_point('backward', step, 'a5', data_file, 'lt', df, row_end, w2t)
row_start = find_point('backward', step, 'a6', data_file, 'gt', df, row_middle, w2t)
return row_end-row_middle, row_middle-row_start, row_end-row_start
def initialization(path, w2t):
_, data_files = traversal_files(path, w2t)
for data_file in data_files:
if not data_file.lower().endswith('.csv'):
w2t(f"{data_file} 文件后缀错误，只允许 .csv 文件，需要确认！", 0, 1, 'red')
return data_files
def preparation(data_file, wb, w2t):
shtname = data_file.split('\\')[-1].split('.')[0]
ws = wb.create_sheet(shtname)
csv_reader = reader(open(data_file))
i = 0
begin = 70
for row in csv_reader:
i += 1
if i == 1:
begin = int(row[1])
break
df = read_csv(data_file, sep=',', encoding='gbk', skip_blank_lines=False, header=begin - 1, on_bad_lines='warn')
low, high, cycle = get_cycle_info(data_file, df, df.index[-1]-110, 5, w2t)
return ws, df, low, high, cycle
def single_file_proc(ws, data_file, df, low, high, cycle, w2t):
_row = _row_lt = _row_gt = count = 1
_step = 5
_data = {}
row_max = df.index[-1]-100
while _row < row_max:
if count not in _data.keys():
_data[count] = []
if df.iloc[_row, 2] < 2:
_row_lt = find_point('forward', _step, 'c'+str(_row), data_file, 'lt', df, _row, w2t)
_start = int((_row_lt - _row_gt - 50) / 2)
_end = _start + 50
value = df.iloc[_start:_end, 2].mean() + df.iloc[_start:_end, 2].std()
_data[count].append(value)
else:
_row_gt = find_point('forward', _step, 'c'+str(_row), data_file, 'gt', df, _row, w2t)
if _row_gt - _row_lt > cycle * 2:
count += 1
_row = max(_row_gt, _row_lt)
for i in range(2, 10):
ws.cell(row=1, column=i).value = f"第{i-1}次测试"
ws.cell(row=i, column=1).value = f"第{i-1}次精度变化"
for i in sorted(_data.keys()):
_row = 2
_column = i + 1
for value in _data[i]:
ws.cell(row=_row, column=_column).value = float(value)
_row += 1
def execution(data_files, w2t):
wb = Workbook()
for data_file in data_files:
ws, df, low, high, cycle = preparation(data_file, wb, w2t)
single_file_proc(ws, data_file, df, low, high, cycle, w2t)
wd = data_files[0].split('\\')
del wd[-1]
wd = '\\'.join(wd)
filename = wd + '\\result.xlsx'
wb.save(filename)
wb.close()
w2t('----------------------------------------')
w2t('所有文件均已处理完毕')
def main(path, w2t):
data_files = initialization(path, w2t)
execution(data_files, w2t)
if __name__ == '__main__':
main(path=argv[1], w2t=argv[2])