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

[modify] minor modification, change write2textbox and using shorter func name

Browse Source
This commit is contained in:
gitea 2024-05-30 08:28:59 +08:00
parent 05f461f8c1
commit 9b849897e2
3 changed files with 56 additions and 65 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
rokae/aio/aio.py
View File

@ -1,10 +1,10 @@
import os.path from os.path import exists
from threading import Thread from threading import Thread
import tkinter.messagebox import tkinter.messagebox
import customtkinter import customtkinter
import brake import brake
import current import current
import time from time import time, strftime, localtime
customtkinter.set_appearance_mode("System") # Modes: "System" (standard), "Dark", "Light" customtkinter.set_appearance_mode("System") # Modes: "System" (standard), "Dark", "Light"
customtkinter.set_default_color_theme("blue") # Themes: "blue" (standard), "green", "dark-blue" customtkinter.set_default_color_theme("blue") # Themes: "blue" (standard), "green", "dark-blue"
@ -125,7 +125,6 @@ class App(customtkinter.CTk):
self.myThread.daemon = True # 主线程退出就直接让子线程跟随退出,不论是否运行完成。 self.myThread.daemon = True # 主线程退出就直接让子线程跟随退出,不论是否运行完成。
self.myThread.start() self.myThread.start()
def initialization(self): def initialization(self):
self.label_path.configure(text="Path", text_color="black") self.label_path.configure(text="Path", text_color="black")
self.label_av.configure(text="AV", text_color="black") self.label_av.configure(text="AV", text_color="black")
@ -198,16 +197,8 @@ class App(customtkinter.CTk):
elif func_name == 'avg': elif func_name == 'avg':
pass pass
def write2textbox(self, text, exitcode=-1, wait=0): def write2textbox(self, text, wait=0):
if exitcode != -1: if wait != 0:
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.configure(state="normal")
self.textbox.insert(index='end', text=text) self.textbox.insert(index='end', text=text)
self.textbox.update() self.textbox.update()
@ -231,8 +222,7 @@ class App(customtkinter.CTk):
trq = self.option_trq.get() trq = self.option_trq.get()
sub_func = self.menu_sub.get() sub_func = self.menu_sub.get()
c1 = exists(path)
c1 = os.path.exists(path)
c2 = av.isdigit() c2 = av.isdigit()
c3 = rr.isdigit() c3 = rr.isdigit()
c4 = rpm = 1 c4 = rpm = 1
@ -255,7 +245,7 @@ class App(customtkinter.CTk):
vel = self.option_vel.get() vel = self.option_vel.get()
trq = self.option_trq.get() trq = self.option_trq.get()
sub_func = self.menu_sub.get() sub_func = self.menu_sub.get()
c1 = os.path.exists(path) c1 = exists(path)
c2 = sub_func in ['max', 'avg'] c2 = sub_func in ['max', 'avg']
try: try:
@ -296,7 +286,7 @@ class App(customtkinter.CTk):
content = self.textbox.get(index1='1.0', index2='end') content = self.textbox.get(index1='1.0', index2='end')
if len(content) > 1: if len(content) > 1:
try: try:
now = time.strftime('%Y%m%d%H%M%S', time.localtime(time.time())) now = strftime('%Y%m%d%H%M%S', localtime(time()))
log_name = f"{now}_aio.log" log_name = f"{now}_aio.log"
with open(f'{log_name}', 'w', encoding='utf-8') as objlog: with open(f'{log_name}', 'w', encoding='utf-8') as objlog:
objlog.write(content) objlog.write(content)

73
rokae/aio/brake.py
View File

@ -1,24 +1,25 @@
# coding: utf-8 # coding: utf-8
import os from os import scandir
import sys from os.path import isfile, exists
import openpyxl from sys import argv
import time from openpyxl import load_workbook
from time import time, sleep, strftime, localtime
from threading import Thread from threading import Thread
import pandas from pandas import read_csv
def traversal_files(path, w2t): def traversal_files(path, w2t):
# 功能：以列表的形式分别返回指定路径下的文件和文件夹，不包含子目录 # 功能：以列表的形式分别返回指定路径下的文件和文件夹，不包含子目录
# 参数：路径 # 参数：路径
# 返回值：路径下的文件夹列表 路径下的文件列表 # 返回值：路径下的文件夹列表 路径下的文件列表
if not os.path.exists(path): if not exists(path):
msg = f'数据文件夹{path}不存在，请确认后重试......' msg = f'数据文件夹{path}不存在，请确认后重试......'
w2t(msg, 1) w2t(msg)
else: else:
dirs = [] dirs = []
files = [] files = []
for item in os.scandir(path): for item in scandir(path):
if item.is_dir(): if item.is_dir():
dirs.append(item.path) dirs.append(item.path)
elif item.is_file(): elif item.is_file():
@ -71,7 +72,7 @@ def find_row_start(data_file, df, conditions, av, rr, axis, vel, trq, w2t, rpm):
row_start -= step row_start -= step
else: else:
msg = f"可能是{data_file}这个文件数据采集有问题，比如未采集理论速度值，也有可能是程序步长设定问题，请检查......" msg = f"可能是{data_file}这个文件数据采集有问题，比如未采集理论速度值，也有可能是程序步长设定问题，请检查......"
w2t(msg, 2) w2t(msg)
return row_max, row_start return row_max, row_start
@ -119,10 +120,10 @@ def single_file_process(data_file, wb_result, count, av, rr, axis, vel, trq, w2t
# 返回值：- # 返回值：-
if data_file.endswith('.data'): if data_file.endswith('.data'):
sep = '\t' sep = '\t'
df = pandas.read_csv(data_file, sep=sep) df = read_csv(data_file, sep=sep)
elif data_file.endswith('.csv'): elif data_file.endswith('.csv'):
sep = ',' sep = ','
df = pandas.read_csv(data_file, sep=sep, encoding='gbk', header=8) df = 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)
@ -139,8 +140,8 @@ 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 = strftime('%Y-%m-%d %H:%M:%S', localtime(time()))
file_type = 'file' if os.path.isfile(docs) else 'dir' file_type = 'file' if isfile(docs) else 'dir'
if flag == 'start' and file_type == 'dir': if flag == 'start' and file_type == 'dir':
w2t(f"[{now}] 正在处理目录 {docs} 中的数据......") w2t(f"[{now}] 正在处理目录 {docs} 中的数据......")
elif flag == 'start' and file_type == 'file': elif flag == 'start' and file_type == 'file':
@ -159,7 +160,7 @@ class GetThreadResult(Thread):
self.result = 0 self.result = 0
def run(self): def run(self):
time.sleep(1) sleep(1)
self.result = self.func(*self.args) self.result = self.func(*self.args)
def get_result(self): def get_result(self):
@ -170,12 +171,12 @@ class GetThreadResult(Thread):
return None return None
def w2t_local(msg, exitcode, wait, w2t): def w2t_local(msg, wait, w2t):
while True: while True:
global stop global stop
if stop == 0 and wait != 0: if stop == 0 and wait != 0:
time.sleep(1) sleep(1)
w2t(msg, exitcode, wait) w2t(msg, wait)
else: else:
break break
@ -185,19 +186,19 @@ def data_process(result_file, raw_data_dirs, av, rr, axis, vel, trq, w2t, rpm):
# 参数：结果文件，数据目录，以及预读取的参数 # 参数：结果文件，数据目录，以及预读取的参数
# 返回值：- # 返回值：-
file_name = result_file.split('\\')[-1] file_name = result_file.split('\\')[-1]
w2t(f"打开文件 {file_name} 需要 1min 左右，请耐心等待", -1, 1) w2t(f"打开文件 {file_name} 需要 1min 左右", 1)
global stop global stop
stop = 0 stop = 0
t_excel = GetThreadResult(openpyxl.load_workbook, args=(result_file, )) t_excel = GetThreadResult(load_workbook, args=(result_file, ))
t_wait = Thread(target=w2t_local, args=('.', -1, 1, w2t)) t_wait = Thread(target=w2t_local, args=('.', 1, w2t))
t_excel.start() t_excel.start()
t_wait.start() t_wait.start()
t_excel.join() t_excel.join()
wb_result = t_excel.get_result() wb_result = t_excel.get_result()
stop = 1 stop = 1
time.sleep(1.1) sleep(1.1)
w2t('', -1, 0) w2t('')
prefix = result_file.split('\\')[-1].split('_')[0] prefix = result_file.split('\\')[-1].split('_')[0]
for raw_data_dir in raw_data_dirs: for raw_data_dir in raw_data_dirs:
@ -222,16 +223,16 @@ def data_process(result_file, raw_data_dirs, av, rr, axis, vel, trq, w2t, rpm):
# --------------------------------------------------- # ---------------------------------------------------
now_doing_msg(result_file, 'done', w2t) now_doing_msg(result_file, 'done', w2t)
w2t(f"保存文件 {file_name} 需要 1min 左右，请耐心等待", -1, 1) w2t(f"保存文件 {file_name} 需要 1min 左右", 1)
stop = 0 stop = 0
t_excel = Thread(target=wb_result.save, args=(result_file, )) t_excel = Thread(target=wb_result.save, args=(result_file, ))
t_wait = Thread(target=w2t_local, args=('.', -1, 1, w2t)) t_wait = Thread(target=w2t_local, args=('.', 1, w2t))
t_excel.start() t_excel.start()
t_wait.start() t_wait.start()
t_excel.join() t_excel.join()
stop = 1 stop = 1
time.sleep(1.1) sleep(1.1)
w2t('\n', -1, 0) w2t('\n')
def check_files(raw_data_dirs, result_files, w2t): def check_files(raw_data_dirs, result_files, w2t):
@ -242,7 +243,7 @@ def check_files(raw_data_dirs, result_files, w2t):
msg = "结果文件数目错误，结果文件有且只有三个，请确认！" msg = "结果文件数目错误，结果文件有且只有三个，请确认！"
for result_file in result_files: for result_file in result_files:
w2t(result_file) w2t(result_file)
w2t(msg, 3) w2t(msg)
prefix = [] prefix = []
for result_file in result_files: for result_file in result_files:
@ -255,7 +256,7 @@ def check_files(raw_data_dirs, result_files, w2t):
1. load33_自研_制动性能测试.xlsx 1. load33_自研_制动性能测试.xlsx
2. load66_自研_制动性能测试.xlsx 2. load66_自研_制动性能测试.xlsx
3. load100_自研_制动性能测试.xlsx""" 3. load100_自研_制动性能测试.xlsx"""
w2t(msg, 4) w2t(msg)
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('_')
@ -267,16 +268,16 @@ def check_files(raw_data_dirs, result_files, w2t):
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% 臂展情况下的测试结果文件夹"
w2t(msg, 5) w2t(msg)
_, raw_data_files = traversal_files(raw_data_dir, w2t) _, raw_data_files = traversal_files(raw_data_dir, w2t)
if len(raw_data_files) != 3: if len(raw_data_files) != 3:
msg = f"数据目录 {raw_data_dir} 下数据文件个数错误，每个数据目录下有且只能有三个以 .data 为后缀的数据文件" msg = f"数据目录 {raw_data_dir} 下数据文件个数错误，每个数据目录下有且只能有三个以 .data 为后缀的数据文件"
w2t(msg, 6) w2t(msg)
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') or raw_data_file.split('\\')[-1].endswith('.csv')): if not (raw_data_file.split('\\')[-1].endswith('.data') or raw_data_file.split('\\')[-1].endswith('.csv')):
msg = f"数据文件 {raw_data_file} 后缀错误，每个数据目录下有且只能有三个以 .data/csv 为后缀的数据文件" msg = f"数据文件 {raw_data_file} 后缀错误，每个数据目录下有且只能有三个以 .data/csv 为后缀的数据文件"
w2t(msg, 7) w2t(msg)
w2t("数据目录合规性检查结束，未发现问题......") w2t("数据目录合规性检查结束，未发现问题......")
@ -285,7 +286,7 @@ def execution(path, av, rr, rpm, axis,vel, trq, w2t):
# 功能：执行处理所有数据文件 # 功能：执行处理所有数据文件
# 参数：initialization函数的返回值 # 参数：initialization函数的返回值
# 返回值：- # 返回值：-
time_start = time.time() time_start = time()
raw_data_dirs, result_files = traversal_files(path, w2t) raw_data_dirs, result_files = traversal_files(path, w2t)
check_files(raw_data_dirs, result_files, w2t) check_files(raw_data_dirs, result_files, w2t)
@ -305,13 +306,13 @@ def execution(path, av, rr, rpm, axis,vel, trq, w2t):
# 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]
w2t("----------------------------------------------------------")
w2t("全部处理完毕")
except Exception as Err: except Exception as Err:
msg = "程序运行错误，请检查配置文件是否准确设定，以及数据文件组织是否正确，也有可能是结果文件损坏，尝试重新复制一份，再运行！" msg = "程序运行错误，请检查配置文件是否准确设定，以及数据文件组织是否正确，也有可能是结果文件损坏，尝试重新复制一份，再运行！"
w2t(msg) w2t(msg)
time_end = time.time() w2t("----------------------------------------------------------")
w2t("全部处理完毕")
time_end = 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"
w2t(msg) w2t(msg)
@ -323,4 +324,4 @@ def main(path, av, rr, rpm, axis,vel, trq, w2t):
if __name__ == "__main__": if __name__ == "__main__":
stop = 0 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]) main(path=argv[1], av=argv[2], rr=argv[3], rpm=argv[4], axis=argv[5], vel=argv[6], trq=argv[7], w2t=argv[8])

2
rokae/aio/readme.txt
View File

@ -113,5 +113,5 @@ v0.0.6(2024/05/23)
v0.1.0(2024/05/29) v0.1.0(2024/05/29)
1. 修改为customtkinter图形化界面 1. 修改为customtkinter图形化界面
2. 支持工业机器人制动数据处理（理论上支持，测试数据有问题，待验证） 2. 支持工业机器人制动数据处理（理论上支持，测试数据有问题，待验证）
3. 删除configs.xlsx配置表格，直接在界面配置 3. 删除configs.xlsx配置表格，直接在界面配置，新增layout.xlsx文件，存储customtkinter布局
4. 电流尚未支持 4. 电流尚未支持