321 lines
12 KiB
Python
321 lines
12 KiB
Python
|
from PyQt6.QtWidgets import QApplication, QMainWindow
|
||
|
|
from PyQt6.QtGui import QKeySequence, QShortcut
|
||
|
|
import datetime, time, sys, os
|
||
|
|
import numpy as np
|
||
|
|
import pyqtgraph as pg
|
||
|
|
|
||
|
|
def csv_to_nparray(fileName, ContainDateTime, timeRange):
|
||
|
|
itemMaxNum = 60
|
||
|
|
if ContainDateTime:
|
||
|
|
timeRange = timeRange.split(',')
|
||
|
|
if len(timeRange) == 2:
|
||
|
|
Filter = 1
|
||
|
|
startTime = datetime.datetime.strptime(timeRange[0].strip(),'%Y/%m/%d %H:%M:%S')
|
||
|
|
start_timestamp=time.mktime(startTime.timetuple())
|
||
|
|
stopTime = datetime.datetime.strptime(timeRange[1].strip(),'%Y/%m/%d %H:%M:%S')
|
||
|
|
stop_timestamp=time.mktime(stopTime.timetuple())
|
||
|
|
else:
|
||
|
|
Filter = 0
|
||
|
|
else:
|
||
|
|
Filter = 0
|
||
|
|
|
||
|
|
#读取文件
|
||
|
|
with open(fileName, "r") as f:
|
||
|
|
recordTotalNum = len(f.readlines())
|
||
|
|
f.seek(0)
|
||
|
|
|
||
|
|
#读取数据文件中的标题行
|
||
|
|
itemNames = []
|
||
|
|
itemNames.append('SampleNO')
|
||
|
|
if ContainDateTime == False:
|
||
|
|
itemNames.append('DateTime')
|
||
|
|
|
||
|
|
tempStr = f.readline()
|
||
|
|
tempStr = tempStr.replace(' ','')
|
||
|
|
tempStr = tempStr.replace('.','_')
|
||
|
|
tempStr = tempStr.replace('"','')
|
||
|
|
tempStr = tempStr.replace('\t','')
|
||
|
|
tempStr = tempStr.replace('\n','')
|
||
|
|
temp = tempStr.split(',')
|
||
|
|
for i in range(len(temp)):
|
||
|
|
temp[i] = temp[i].strip()
|
||
|
|
if ContainDateTime == True:
|
||
|
|
temp[0] = 'DateTime'
|
||
|
|
itemNames.extend(temp)
|
||
|
|
# print(itemNames)
|
||
|
|
temp = [item for item in itemNames if len(item)>0]
|
||
|
|
itemNames = temp[0:itemMaxNum]
|
||
|
|
print(f"itemNames:{itemNames}")
|
||
|
|
|
||
|
|
#读取数据文件中的数据，以浮点数的形式读取
|
||
|
|
rawData = np.array([])
|
||
|
|
rawData_list = list(rawData)
|
||
|
|
recordNum = 1
|
||
|
|
for i in range(1, recordTotalNum-1):
|
||
|
|
record = []
|
||
|
|
line = f.readline()
|
||
|
|
line = line.replace('"','')
|
||
|
|
line = line.replace('/', ' ')
|
||
|
|
line = line.replace('T', ' ')
|
||
|
|
line = line.replace('Z', '')
|
||
|
|
line = line.replace('\t', '')
|
||
|
|
line = line.replace('\n', '')
|
||
|
|
substrings = line.split(',')
|
||
|
|
|
||
|
|
if ContainDateTime:
|
||
|
|
substrings[0] = substrings[0].split('.')[0]
|
||
|
|
substrings[0] = substrings[0].replace('-',' ')
|
||
|
|
substrings[0] = substrings[0].replace(':',' ')
|
||
|
|
temp = substrings[0].split()
|
||
|
|
date_str = ""
|
||
|
|
for value in temp:
|
||
|
|
if value == '':
|
||
|
|
temp.remove(value)
|
||
|
|
datetime_str = temp[0]+'-'+temp[1]+'-'+temp[2]+' '+temp[3]+':'+temp[4]+':'+temp[5]
|
||
|
|
if datetime_str == '1970-1-1 1:1:1' or datetime_str == '1970-1-1 1:01:01':
|
||
|
|
datetime_str = temp[0]+'-'+temp[1]+'-'+'2'+' '+temp[3]+':'+temp[4]+':'+temp[5]
|
||
|
|
dt = datetime.datetime.strptime(datetime_str,'%Y-%m-%d %H:%M:%S')
|
||
|
|
timestamp=time.mktime(dt.timetuple())
|
||
|
|
else:
|
||
|
|
datetime_str = '1971-1-2 1:1:1'
|
||
|
|
dt = datetime.datetime.strptime(datetime_str,'%Y-%m-%d %H:%M:%S')
|
||
|
|
timestamp=time.mktime(dt.timetuple())
|
||
|
|
|
||
|
|
record.append(recordNum)
|
||
|
|
record.append(timestamp)
|
||
|
|
|
||
|
|
if ContainDateTime:
|
||
|
|
for j in range(1, min(substrings.__len__(), itemMaxNum)):
|
||
|
|
try:
|
||
|
|
record.append(float(substrings[j].replace(' ','')))
|
||
|
|
except:
|
||
|
|
record.append(-1)
|
||
|
|
pass
|
||
|
|
else:
|
||
|
|
for j in range(0, min(substrings.__len__(), itemMaxNum)):
|
||
|
|
try:
|
||
|
|
record.append(float(substrings[j].replace(' ','')))
|
||
|
|
except:
|
||
|
|
record.append(-1)
|
||
|
|
pass
|
||
|
|
if Filter:
|
||
|
|
if record[1] >= start_timestamp and record[1] <= stop_timestamp:
|
||
|
|
recordNum += 1
|
||
|
|
rawData_list.append(record)
|
||
|
|
else:
|
||
|
|
recordNum += 1
|
||
|
|
rawData_list.append(record)
|
||
|
|
print(f"record:{record}")
|
||
|
|
rawData = np.array(rawData_list)
|
||
|
|
return itemNames, rawData
|
||
|
|
|
||
|
|
def updateStatistic():
|
||
|
|
global showStatistic, statisticData
|
||
|
|
if showStatistic:
|
||
|
|
indexRange = vbs[0].viewRange()
|
||
|
|
for i in range(0,len(itemNames),1):
|
||
|
|
data = np.array(records[int(indexRange[0][0]):int(indexRange[0][1]),i])
|
||
|
|
statisticData[itemNamesRaw[i]] = [round(np.min(data),2), round(np.max(data),2), round(np.mean(data),2), round(np.std(data),2)]
|
||
|
|
label1.setText(tempStr1.format(**statisticData))
|
||
|
|
|
||
|
|
def updateViews():
|
||
|
|
# setGeometry设置几何图形
|
||
|
|
# sceneBoundingRect场景边界矩形
|
||
|
|
for i in range(0, itemsNum-1):
|
||
|
|
vbs[i].setGeometry(vbs[i].sceneBoundingRect())
|
||
|
|
|
||
|
|
def mouseMoved(evt):
|
||
|
|
global tempStr, itemNamesRaw, recordsRaw, BiasID
|
||
|
|
index = 0
|
||
|
|
dic = {}
|
||
|
|
pos = evt[0]
|
||
|
|
# if l.sceneBoundingRect().contains(pos):
|
||
|
|
for i in range(0, rowNum):
|
||
|
|
if vbs[i*curveNum-1].sceneBoundingRect().contains(pos):
|
||
|
|
mousePoint = vbs[i*curveNum-1].mapSceneToView(pos)
|
||
|
|
index = int(mousePoint.x())
|
||
|
|
|
||
|
|
if index > 0 and index < recordsNum:
|
||
|
|
dic[itemNamesRaw[0]] = index
|
||
|
|
dic[itemNamesRaw[1]] = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(recordsRaw[index-1,1]))
|
||
|
|
for k in range(2, itemsNum+BiasID):
|
||
|
|
dic[itemNamesRaw[k]] =recordsRaw[index-1,k]
|
||
|
|
# print(dic)
|
||
|
|
label.setText(tempStr.format(**dic))
|
||
|
|
for i in range(0, rowNum):
|
||
|
|
vLine[i].setPos(index)
|
||
|
|
|
||
|
|
def draw():
|
||
|
|
global l, itemNames, itemsNum, axsX, vbs, vLine, axsY, tempStr, label, curveList, rowNum, label1, BiasID, tempStr1
|
||
|
|
label = pg.LabelItem(justify='left')
|
||
|
|
l.addItem(label, row=0, col=0, rowspan=1, colspan=11)
|
||
|
|
label1 = pg.LabelItem(justify='left')
|
||
|
|
l.addItem(label1, row=1, col=0, rowspan=1, colspan=11)
|
||
|
|
labelStyle = {'font-size': '9pt'}
|
||
|
|
for i in range(0, itemsNum-1):
|
||
|
|
axsY.append(pg.AxisItem("left", pen=colors[i%10]))
|
||
|
|
axsY[i].setLabel(itemNames[i+1], color=colors[i%10],**labelStyle)
|
||
|
|
vbs.append(pg.ViewBox())
|
||
|
|
l.addItem(axsY[i],row = int(i/curveNum)+2, col = i%curveNum, rowspan=1, colspan=1)
|
||
|
|
l.addItem(vbs[i],row = int(i/curveNum)+2, col = 10, rowspan=1, colspan=1)
|
||
|
|
axsY[i].linkToView(vbs[i])
|
||
|
|
|
||
|
|
if(i>0):
|
||
|
|
vbs[i].setXLink(vbs[0])
|
||
|
|
if(i%curveNum == 0):
|
||
|
|
axsX.append(pg.AxisItem("bottom", pen=colors[10]))
|
||
|
|
axsX[int(i/curveNum)].setLabel(itemNames[0], color=colors[10],**labelStyle)
|
||
|
|
l.addItem(axsX[int(i/curveNum)],row = int(i/curveNum)+3, col = 10, rowspan=1, colspan=1)
|
||
|
|
axsX[int(i/curveNum)].linkToView(vbs[0])
|
||
|
|
|
||
|
|
tempStr = ""
|
||
|
|
if DisplayTime == True:
|
||
|
|
tempStr += "<span style='font-size: 9pt; word-wrap: break-word; white-space: normal;'>NO.={SampleNO}, "
|
||
|
|
BiasID = 0
|
||
|
|
else:
|
||
|
|
tempStr += "<span style='font-size: 9pt; word-wrap: break-word; white-space: normal;'>NO.={SampleNO}, <span style='color:white;'>DateTime={DateTime}</span>, "
|
||
|
|
BiasID = 1
|
||
|
|
for i in range(1, itemsNum):
|
||
|
|
tempStr += "<span style='color:"
|
||
|
|
tempStr += str(colors[(i-1)%10])
|
||
|
|
tempStr += "'>"
|
||
|
|
tempStr += itemNames[i]
|
||
|
|
tempStr += "={"
|
||
|
|
tempStr += itemNames[i]
|
||
|
|
tempStr += "}</span>, "
|
||
|
|
if(i%10 == 0):
|
||
|
|
tempStr +="<br>"
|
||
|
|
tempStr += "</span>"
|
||
|
|
# print(tempStr)
|
||
|
|
for i in range(0, rowNum):
|
||
|
|
vLine.append(pg.InfiniteLine(angle=90, movable=True))
|
||
|
|
vbs[i*curveNum-1].addItem(vLine[i])
|
||
|
|
# pI.addItem(vLine)
|
||
|
|
|
||
|
|
tempStr1 = "<span style='font-size: 9pt; word-wrap: break-word; white-space: normal;'>"
|
||
|
|
for i in range(1, itemsNum):
|
||
|
|
tempStr1 += "<span style='color:"
|
||
|
|
tempStr1 += str(colors[(i-1)%10])
|
||
|
|
tempStr1 += "'>"
|
||
|
|
tempStr1 += itemNames[i]
|
||
|
|
tempStr1 += "={"
|
||
|
|
tempStr1 += itemNames[i]
|
||
|
|
tempStr1 += "}</span>, "
|
||
|
|
if(i%5 == 0):
|
||
|
|
tempStr1 +="<br>"
|
||
|
|
tempStr1 += "</span>"
|
||
|
|
|
||
|
|
curveList = []
|
||
|
|
for i in range(0, itemsNum-1):
|
||
|
|
curve = pg.PlotCurveItem(records[:,0], records[:,i+1], pen=colors[i%10])
|
||
|
|
curveList.append(curve)
|
||
|
|
vbs[i].addItem(curve)
|
||
|
|
#vbs[i].addItem(pg.ScatterPlotItem(records[:,0], records[:,i+1], pen=colors[i%10], symbolBrush=(255,0,0), symbolPen='w', size=2))
|
||
|
|
|
||
|
|
# vbs[0].sigResized.connect(updateStatistic)
|
||
|
|
|
||
|
|
for i in range(0, itemsNum-1):
|
||
|
|
vbs[i].enableAutoRange(axis= pg.ViewBox.XYAxes, enable=True)
|
||
|
|
|
||
|
|
def saveStatisticsData():
|
||
|
|
global statisticData, fileName, itemNames
|
||
|
|
if os.path.exists(fileName.split('.')[0]+'.sta'):
|
||
|
|
fo = open(fileName.split('.')[0]+'.sta', 'a+')
|
||
|
|
else:
|
||
|
|
fo = open(fileName.split('.')[0]+'.sta', 'w')
|
||
|
|
fo.write('; '.join(itemNames[0:])+'\n')
|
||
|
|
tempStr = ''
|
||
|
|
for item in statisticData.values():
|
||
|
|
tempStr += f"{str(item).replace('[','').replace(']','')}; "
|
||
|
|
fo.write(tempStr[:-1]+'\n')
|
||
|
|
fo.close()
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
#解析参数
|
||
|
|
# fileName = sys.argv[1]
|
||
|
|
# timeRange = sys.argv[2]
|
||
|
|
# ContainDateTime = sys.argv[3]
|
||
|
|
# curveNum = int(sys.argv[4])
|
||
|
|
# showStatistic =sys.argv[5]
|
||
|
|
|
||
|
|
# if showStatistic == 'True':
|
||
|
|
# showStatistic = True
|
||
|
|
# else:
|
||
|
|
# showStatistic = False
|
||
|
|
|
||
|
|
# if ContainDateTime == 'True':
|
||
|
|
# ContainDateTime = True
|
||
|
|
# else:
|
||
|
|
# ContainDateTime = False
|
||
|
|
|
||
|
|
# if ContainDateTime == True:
|
||
|
|
# DisplayTime = True
|
||
|
|
# else:
|
||
|
|
# DisplayTime = False
|
||
|
|
|
||
|
|
fileName = r'D:\DESIGN\JupterLab\TestDataAnalyzor\output\加表_温漂负一倍_output.csv'
|
||
|
|
# fileName = r'D:\DESIGN\JupterLab\TestDataAnalyzor\output\加表温度_温漂负一倍_output.csv'
|
||
|
|
|
||
|
|
timeRange = ''
|
||
|
|
ContainDateTime = False
|
||
|
|
curveNum = 10
|
||
|
|
showStatistic = False
|
||
|
|
DisplayTime = False
|
||
|
|
|
||
|
|
itemNamesRaw, recordsRaw = csv_to_nparray(fileName, ContainDateTime, timeRange)
|
||
|
|
itemNames = itemNamesRaw
|
||
|
|
records = recordsRaw
|
||
|
|
|
||
|
|
if DisplayTime == False:
|
||
|
|
itemNames = np.delete(itemNames, 1, 0)
|
||
|
|
# print(itemNames)
|
||
|
|
itemsNum = itemNames.__len__()
|
||
|
|
# print(itemsNum)
|
||
|
|
|
||
|
|
if DisplayTime == False:
|
||
|
|
records = np.delete(records, 1, 1)
|
||
|
|
# print(records[:,:])
|
||
|
|
|
||
|
|
recordsNum = records.__len__()
|
||
|
|
|
||
|
|
if(itemsNum%curveNum == 0):
|
||
|
|
rowNum = int(itemsNum/curveNum)
|
||
|
|
else:
|
||
|
|
rowNum = int(itemsNum/curveNum)+1
|
||
|
|
|
||
|
|
colors = ['#FFFFFF', '#FF0000', '#FFFF00', '#FF00FF', '#00FF00', '#00FFFF', '#C0FF00', '#FF0080', '#FF7F7F', '#7FFF7F','#FFFFFF']
|
||
|
|
app = QApplication([])
|
||
|
|
axsY=[]
|
||
|
|
axsX=[]
|
||
|
|
vbs=[]
|
||
|
|
vLine=[]
|
||
|
|
# 创建视图
|
||
|
|
# pw = pg.GraphicsView()
|
||
|
|
|
||
|
|
pw = pg.PlotWidget()
|
||
|
|
pw.setWindowTitle(fileName)
|
||
|
|
pw.show()
|
||
|
|
|
||
|
|
# 创建图形布局
|
||
|
|
l = pg.GraphicsLayout()
|
||
|
|
# 设置视图中心小部件 为该布局
|
||
|
|
pw.setCentralWidget(l)
|
||
|
|
|
||
|
|
draw()
|
||
|
|
|
||
|
|
statisticData = {}
|
||
|
|
proxy = pg.SignalProxy(l.scene().sigMouseMoved, rateLimit=60, slot=mouseMoved)
|
||
|
|
proxy1 = pg.SignalProxy(vbs[0].sigRangeChanged, rateLimit=60, slot=updateStatistic)
|
||
|
|
|
||
|
|
shortcut1 = QShortcut(QKeySequence(f"Ctrl+S"), pw)
|
||
|
|
shortcut1.activated.connect(lambda: saveStatisticsData())
|
||
|
|
# shortcut1.activated.connect(lambda: print("保存操作触发"))
|
||
|
|
|
||
|
|
# qs1 = QKeySequence(f"Ctrl+Q")
|
||
|
|
# qsc = QShortcut(qs1, pw)
|
||
|
|
# QShortcut(QKeySequence(qs1), pw).activated.connect(app.quit)
|
||
|
|
# QShortcut(QKeySequence(f"Ctrl+Q"), pw).activated.connect(app.quit)
|
||
|
|
|
||
|
|
updateViews()
|
||
|
|
app.exec()
|