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

[modify]: 完成了电流处理的基本工作，包括框架搭建和思路整理，重写了max/avg以支持工业数据，新增加了duration的逻辑，中间状态，先保存一下

Browse Source
This commit is contained in:
gitea
2024-06-04 08:29:23 +08:00
parent afb862201a
commit 55514abc93
4 changed files with 223 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
aio/current.py
View File

@ -1,8 +1,8 @@
import openpyxl
from openpyxl import load_workbook
from os import scandir
from os.path import exists
from sys import argv
import pandas
from pandas import read_csv
def traversal_files(path, w2t):
@ -24,47 +24,114 @@ def traversal_files(path, w2t):
return dirs, files
def initialization(path, w2t):
def initialization(path, sub, w2t):
_, data_files = traversal_files(path, w2t)
count = 0
for data_file in data_files:
if not data_file.endswith('.data'):
msg = f"所有文件必须以 .data 结尾，请检查后重新运行。"
w2t(msg, 0, 2)
if sub != 'cycle':
if not (data_file.endswith('.data') or data_file.endswith('.csv')):
msg = f"所有文件必须以 .data 结尾，请检查后重新运行。"
w2t(msg, 0, 2)
else:
if data_file.endswith('.xlsx'):
count += 1
elif not (data_file.endswith('.data') or data_file.endswith('.csv')):
msg = f"所有文件必须以 .data 结尾，请检查后重新运行。"
w2t(msg, 0, 3)
if sub == 'cycle' and count != 1:
w2t("未找到电机电流数据处理excel表格，确认后重新运行！", 0, 4)
return data_files
def current_max(data_files, rc, trq, w2t):
current = {1: 0, 2: 0, 3: 0, 4: 0, 5: 0, 6: 0, 7: 0}
for data_file in data_files:
df = pandas.read_csv(data_file, sep='\t')
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 = rc[axis-1]
col = df.columns.values[trq-1]
c_max = df[col].max()
w2t(f"{data_file}: {c_max/1000*rc:.4f}")
scale = 1 if data_file.endswith('.csv') else 1000
_ = abs(c_max/scale*rca)
current[axis] = _
w2t(f"{data_file}: {_:.4f}")
w2t("【MAX】数据处理完毕......")
return current
def current_avg(data_files, rc, trq, w2t):
current = {1: 0, 2: 0, 3: 0, 4: 0, 5: 0, 6: 0, 7: 0}
for data_file in data_files:
df = pandas.read_csv(data_file, sep='\t')
col = df.columns.values[trq-1]
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 = rc[axis-1]
col = df.columns.values[trq - 1]
c_std = df[col].std()
c_avg = df[col].mean()
w2t(f"{data_file}: {(abs(c_avg)+c_std)/1000*rc:.4f}")
scale = 1 if data_file.endswith('.csv') else 1000
_ = (abs(c_avg)+c_std)/scale*rca
current[axis] = _
w2t(f"{data_file}: {_:.4f}")
w2t("【AVG】数据处理完毕......")
return current
def main(path, sub, rc, vel, trq, w2t):
data_files = initialization(path, w2t)
def current_cycle(dur, data_files, rc, vel, trq, w2t):
result = None
hold = []
single = []
for data_file in data_files:
if data_file.endswith('.xlsx'):
result = data_file
elif (data_file.endswith('.csv') or data_file.endswith('.data')) and data_file.startswith('hold', 3, 10):
hold.append(data_file)
else:
single.append(data_file)
if hold != []:
avg = current_avg(hold, rc, trq, w2t)
wb = load_workbook(result)
for k, v in avg.items():
try:
shtname = f"J{k}"
wb[shtname]["J4"].value = v
except:
pass
for data_file in single:
axis = int(data_file.split('\\')[-1].split('_')[0].removeprefix('j'))
rcs = rc[axis-1]
pass
# =======================================
def main(path, sub, rc, vel, trq, dur, w2t):
data_files = initialization(path, sub, w2t)
if sub == 'max':
current_max(data_files, rc, trq, w2t)
elif sub == 'avg':
current_avg(data_files, rc, trq, w2t)
elif sub == 'cycle':
current_cycle(dur, data_files, rc, vel, trq, w2t)
else:
pass
if __name__ == '__main__':
main(path=argv[1], sub=argv[2], rc=argv[3], vel=argv[4], trq=argv[5], w2t=argv[6])
main(*argv[1:])