andy/python_tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

实现根据方案自动从数据库下载检测数据并分析

Browse Source
This commit is contained in:
ranchuan
2023-09-18 18:16:26 +08:00
parent 769dfc3734
commit 125c199756
4 changed files with 228 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
ReadMe.txt
View File

@@ -44,4 +44,6 @@
2023.9.15
新增analysis工具
2023.9.17
实现对检测数据的分析,自动填充上下限,一键导出方案
实现对检测数据的分析,自动填充上下限,一键导出方案
2023.9.18
实现根据方案自动从数据库下载检测数据并分析

11
analysis/analysis.py
View File

@@ -151,13 +151,14 @@ class Analysis(QWidget):
# 连接的绘制的方法
def import_but_clicked(self):
# t = threading.Thread(target=self.__anslysis, args=())
# t.start()
# def __anslysis(self):
self.sch_data=scheme_data.sch_data()
self.__import_but.setEnabled(False)
widget=QWidget()
widget.setGeometry(QRect(0,0,1200,20000))
self.titles,datas=scheme_data.datas()
self.titles,datas=self.sch_data.datas_sql()
if(self.titles==None):
self.__import_but.setEnabled(True)
return
for i in range(len(self.titles)):
self.addItem(datas[i],self.titles[i],[self.titles[i]])
widget.setLayout(self.__layout)
@@ -168,7 +169,7 @@ class Analysis(QWidget):
max=i[1].item_value(t+":LimitMax")
min=i[1].item_value(t+":LimitMin")
ret_limit.append((max,min))
scheme_data.export_scheme(ret_limit)
self.sch_data.export_scheme(ret_limit)
# 运行程序
if __name__ == '__main__':
app = QApplication(sys.argv)

270
analysis/scheme_data.py
View File

@@ -5,9 +5,105 @@ import numpy as np
import datetime
from datetime import datetime, timedelta
import os
import sys
import pymysql as mysql
from PyQt5.QtCore import *
from PyQt5.QtGui import *
from PyQt5.QtWidgets import *
__INFO_NAME="scheme_export_info.txt"
class quest_text(QObject):
def __init__(self,title:str,info:str):
QObject.__init__(self)
# self.app = QApplication(sys.argv)
self.w=QDialog()
self.w.resize(320,80)
self.w.setWindowTitle(title)
self.w.setAttribute(Qt.WidgetAttribute.WA_DeleteOnClose)
self.w.setWindowModality(Qt.WindowModality.ApplicationModal)
self.text=info
self.text_edit = QTextEdit(self.w)
self.text_edit.setObjectName(u"text")
self.text_edit.setGeometry(QRect(10, 10, 300, 30))
self.text_edit.setFrameShape(QFrame.Shape.Box)
self.text_edit.setMidLineWidth(1)
self.text_edit.setVerticalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAsNeeded)
self.text_edit.setPlainText(self.text)
self.text_edit.moveCursor(QTextCursor.MoveOperation.End, QTextCursor.MoveMode.MoveAnchor)
self.ok_but_init()
self.cancel_but_init()
def ok_but_init(self):
self.ok_but=QPushButton(self.w)
self.ok_but.setObjectName(u"ok_but")
self.ok_but.setGeometry(QRect(210, 50, 90, 28))
self.ok_but.clicked.connect(self.ok_but_clicked)
self.ok_but.setText("确认")
self.ok_but.setCheckable(True)
self.ok_but.setChecked(True)
self.ok_state=False
def cancel_but_init(self):
self.cancel_but=QPushButton(self.w)
self.cancel_but.setObjectName(u"cancel_but")
self.cancel_but.setGeometry(QRect(100, 50, 90, 28))
self.cancel_but.clicked.connect(self.cancel_but_clicked)
self.cancel_but.setText("取消")
self.ok_but.setCheckable(True)
self.ok_state=False
def ok_but_clicked(self):
self.text=self.text_edit.toPlainText()
self.ok_state=True
self.w.done(QDialog.DialogCode.Accepted)
self.w.close()
def cancel_but_clicked(self):
self.text=self.text_edit.toPlainText()
self.ok_state=False
self.w.done(QDialog.DialogCode.Accepted)
self.w.close()
def show(self):
self.w.exec()
return self.ok_state,self.text
def _connect_sql():
try:
db=mysql.connect(host='47.108.119.117',user='ymp',passwd='Abc@12345',port=15319)
cur=db.cursor()
cur.execute("use datacenter")
print("connect mysql success.")
return db,cur
except Exception as e:
print("can not connect service.")
return None,None
# 生成一个任务的参数
def __scheme_task_to_retinfo(j:json):
def _scheme_task_to_retinfo(j:json):
texts=[]
length=len(j["TestStandard"])
for i in j["ReturnInfo"]:
@@ -15,7 +111,7 @@ def __scheme_task_to_retinfo(j:json):
return texts
# 生成任务参数序列
def __scheme_tasks_to_retinfo(j:json):
def _scheme_tasks_to_retinfo(j:json):
# texts=[]
# for i in j["TaskArray"]:
# texts+=__scheme_task_to_retinfo(i)
@@ -24,36 +120,66 @@ def __scheme_tasks_to_retinfo(j:json):
return texts
def __json_extract_retinfo(name:str):
def _json_extract_retinfo(name:str):
with open(name,"rb") as f:
json_obj=json.loads(f.read())
return __scheme_tasks_to_retinfo(json_obj)
return None
return _scheme_tasks_to_retinfo(json_obj),json_obj["PlanID"]
return None,None
def __get_date():
def _get_date():
now_time = datetime.utcnow()
utc_time = now_time + timedelta(hours=8)
utc_time = utc_time.strftime("%Y%m%d")
return utc_time
# 获取本次生成方案的id号
def __get_scheme_id():
info_name="scheme_export_info.txt"
id=1
# 保存设置
def _save_info(info):
info_name=__INFO_NAME
with open(info_name,"w+") as f:
# print("saving info:",date,id,save_name)
f.write(info[0]+'\n')
f.write(str(info[1])+'\n')
f.write(info[2]+'\n')
f.write(info[3]+'\n')
f.write(info[4]+'\n')
def _get_info():
info_name=__INFO_NAME
if os.path.exists(info_name):
info_f=open(info_name,'r')
lines=info_f.readlines()
if(__get_date()==lines[0].strip()):
id=int(lines[1].strip())+1
date=lines[0].strip()
id=int(lines[1].strip())
name=lines[2].strip()
date_start=lines[3].strip()
date_end=lines[4].strip()
info_f.close()
with open(info_name,"w+") as f:
f.write(__get_date()+'\n')
f.write(str(id)+'\n')
else:
date="20230101"
id=1
name="方案参数校准"
date_start="2023-01-01"
date_end="2023-01-01"
# print("get info:",date,id,name)
return (date,id,name,date_start,date_end)
# 获取本次生成方案的id号
def _get_scheme_id():
info=_get_info()
date=info[0]
id=info[1]
if(date!=_get_date()):
date=_get_date()
id=1
else:
id+=1
_save_info((date,id,info[2],info[3],info[4]))
return id
def __reflush_scheme_id(id):
def _reflush_scheme_id(id):
# 日bit0~bit4 月bit5~bit8 年bit9~bit15
sid=(id>>0)&0x7f
modele=(id>>7)&0x1f
@@ -63,7 +189,7 @@ def __reflush_scheme_id(id):
year=((id>>25)&0x7f)+2022
now_time = datetime.utcnow()
utc_time = now_time + timedelta(hours=8)
sid=__get_scheme_id()
sid=_get_scheme_id()
day=utc_time.day
month=utc_time.month
year=utc_time.year-2022
@@ -71,9 +197,9 @@ def __reflush_scheme_id(id):
return id,sid
def main():
def _main():
save_name="check_data.csv"
title=__json_extract_retinfo("代工厂数据采集EJ旧68uF22um08211.json")
title,=_json_extract_retinfo("代工厂数据采集EJ旧68uF22um08211.json")
save_f=open(save_name,'+w')
save_f.write(','.join(title)+'\n')
err_num=0
@@ -88,43 +214,73 @@ def main():
print("检测异常数量为：",err_num)
def datas():
title=__json_extract_retinfo("代工厂数据采集EJ旧68uF22um08211.json")
data_list=[]
for i in title:
t=[]
data_list.append(t)
with open("EJ09C测试数据.csv",newline="") as f:
data=csv.reader(f)
for row in data:
if(row[3]=='0'):
s=row[4].split(',')
for i in range(len(s)):
data_list[i].append(int(s[i]))
return title,data_list
class sch_data(object):
def __init__(self):
self.scheme_name=""
def datas_sql(self):
fileName,fileType = QFileDialog.getOpenFileNames(None, "选取文件", os.getcwd(),
"检测方案(*.json)")
if(len(fileName)==0):
print("user cancelled.")
return None,None
print(fileName,fileType)
scheme_name=fileName[0]
info=_get_info()
quest=quest_text("请输入要获取数据的日期",info[3]+','+info[4])
ack,date_interval=quest.show()
if(ack!=True):
print("user cancelled.")
return None,None
date_interval=date_interval.split(',')
date_start=date_interval[0]+" 00:00:00"
date_end=date_interval[1]+" 23:59:59"
_save_info((info[0],info[1],info[2],date_interval[0],date_interval[1]))
self.scheme_name=scheme_name
title,scheme_id=_json_extract_retinfo(scheme_name)
data_list=[]
for i in title:
t=[]
data_list.append(t)
db,cur=_connect_sql()
if(db!=None):
cmd="""SELECT id,create_time,addr,err_code,valuez FROM check_result_detail
where create_time > %s and create_time < %s
and result_id in (select pk from check_result where plan_id =%s) ;"""
cur.execute(cmd,(date_start,date_end,scheme_id))
check_data=cur.fetchall()
print("check_data_len=",len(check_data))
for row in check_data:
if(row[3]==0):
s=row[4].split(',')
for i in range(len(s)):
data_list[i].append(int(s[i]))
return title,data_list
def export_scheme(ret_limit):
name="代工厂数据采集EJ旧68uF22um08211.json"
save_name="代工厂参数标定测试EJ"
index=0
with open(name,"r",encoding="utf-8") as f:
json_obj=json.loads(f.read())
for i in json_obj["TaskArray"]:
for j in i["TestStandard"]:
j["Max"]=ret_limit[index][0]
j["Min"]=ret_limit[index][1]
json_obj["PlanID"],sid=__reflush_scheme_id(json_obj["PlanID"])
save_name=save_name+"_"+__get_date()+"{d:02d}".format(d=sid)
json_obj["PlanBrief"]=save_name
json_str=json.dumps(json_obj, sort_keys=True, indent=4, separators=(',', ': '),ensure_ascii=False)
save_f=open(save_name+".json",'+wb')
save_f.write(json_str.encode("utf-8"))
save_f.close()
def export_scheme(self,ret_limit):
name=self.scheme_name
info=_get_info()
date=info[0]
id=info[1]
save_name=info[2]
quest=quest_text("请输入要导出的文件名",save_name)
ack,save_name=quest.show()
if(ack!=True):
print("user cancelled.")
return
_save_info((date,id,save_name,info[3],info[4]))
index=0
with open(name,"r",encoding="utf-8") as f:
json_obj=json.loads(f.read())
for i in json_obj["TaskArray"]:
for j in i["TestStandard"]:
j["Max"]=ret_limit[index][0]
j["Min"]=ret_limit[index][1]
json_obj["PlanID"],sid=_reflush_scheme_id(json_obj["PlanID"])
save_name=save_name+"_"+date+"{d:02d}".format(d=sid)
json_obj["PlanBrief"]=save_name
json_str=json.dumps(json_obj, sort_keys=True, indent=4, separators=(',', ': '),ensure_ascii=False)
save_f=open(save_name+".json",'+wb')
save_f.write(json_str.encode("utf-8"))
save_f.close()
if __name__ == "__main__":
# main()
x=[1,2,3,4,5,6,7]
avg=np.average(x)
print(type(x),type(avg))
sub=np.abs(x-avg)
print(sub)
_main()

6
mysql/connections.ncx Normal file
View File

@@ -0,0 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<Connections Ver="1.5">
<Connection ConnectionName="data_service" ProjectUUID="" ConnType="MYSQL" OraConnType="" ServiceProvider="Default" Host="124.70.178.159" Port="3306" Database="" OraServiceNameType="" TNS="" MSSQLAuthenMode="" MSSQLAuthenWindowsDomain="" DatabaseFileName="" UserName="admin" Password="3BFE1EF86FA1343C221914D12EDD2026" SavePassword="true" SettingsSavePath="C:\Users\Andy\Documents\Navicat\MySQL\Servers\data_service" SessionLimit="0" ClientCharacterSet="" ClientEncoding="65001" Keepalive="false" KeepaliveInterval="240" Encoding="65001" MySQLCharacterSet="true" Compression="false" AutoConnect="false" NamedPipe="false" NamedPipeSocket="" OraRole="" OraOSAuthen="false" SQLiteEncrypt="false" SQLiteEncryptPassword="" SQLiteSaveEncryptPassword="false" UseAdvanced="false" SSL="false" SSL_Authen="false" SSL_PGSSLMode="REQUIRE" SSL_ClientKey="" SSL_ClientCert="" SSL_CACert="" SSL_Clpher="" SSL_PGSSLCRL="" SSL_WeakCertValidation="false" SSL_AllowInvalidHostName="false" SSL_PEMClientKeyPassword="" SSH="false" SSH_Host="" SSH_Port="22" SSH_UserName="" SSH_AuthenMethod="PASSWORD" SSH_Password="" SSH_SavePassword="false" SSH_PrivateKey="" SSH_Passphrase="" SSH_SavePassphrase="false" SSH_Compress="false" HTTP="false" HTTP_URL="" HTTP_PA="" HTTP_PA_UserName="" HTTP_PA_Password="" HTTP_PA_SavePassword="" HTTP_EQ="" HTTP_CA="" HTTP_CA_ClientKey="" HTTP_CA_ClientCert="" HTTP_CA_CACert="" HTTP_CA_Passphrase="" HTTP_Proxy="" HTTP_Proxy_Host="" HTTP_Proxy_Port="" HTTP_Proxy_UserName="" HTTP_Proxy_Password="" HTTP_Proxy_SavePassword="" Compatibility="false" Compatibility_OverrideLowerCaseTableNames="false" Compatibility_LowerCaseTableNames="" Compatibility_OverrideSQLMode="false" Compatibility_SQLMode="" Compatibility_OverrideIsSupportNDB="false" Compatibility_IsSupportNDB="false" Compatibility_OverrideDatabaseListingMethod="false" Compatibility_DatabaseListingMethod="" Compatibility_OverrideViewListingMethod="false" Compatibility_ViewListingMethod=""/>
<Connection ConnectionName="生成数据" ProjectUUID="" ConnType="MYSQL" OraConnType="" ServiceProvider="Default" Host="47.108.119.117" Port="15319" Database="" OraServiceNameType="" TNS="" MSSQLAuthenMode="" MSSQLAuthenWindowsDomain="" DatabaseFileName="" UserName="ymp" Password="49EE49F2A92E761E13A35B7946DFABED" SavePassword="true" SettingsSavePath="C:\Users\Andy\Documents\Navicat\MySQL\Servers\生成数据" SessionLimit="0" ClientCharacterSet="" ClientEncoding="65001" Keepalive="false" KeepaliveInterval="240" Encoding="65001" MySQLCharacterSet="true" Compression="false" AutoConnect="false" NamedPipe="false" NamedPipeSocket="" OraRole="" OraOSAuthen="false" SQLiteEncrypt="false" SQLiteEncryptPassword="" SQLiteSaveEncryptPassword="false" UseAdvanced="false" SSL="false" SSL_Authen="false" SSL_PGSSLMode="REQUIRE" SSL_ClientKey="" SSL_ClientCert="" SSL_CACert="" SSL_Clpher="" SSL_PGSSLCRL="" SSL_WeakCertValidation="false" SSL_AllowInvalidHostName="false" SSL_PEMClientKeyPassword="" SSH="false" SSH_Host="" SSH_Port="22" SSH_UserName="" SSH_AuthenMethod="PASSWORD" SSH_Password="" SSH_SavePassword="false" SSH_PrivateKey="" SSH_Passphrase="" SSH_SavePassphrase="false" SSH_Compress="false" HTTP="false" HTTP_URL="" HTTP_PA="" HTTP_PA_UserName="" HTTP_PA_Password="" HTTP_PA_SavePassword="" HTTP_EQ="" HTTP_CA="" HTTP_CA_ClientKey="" HTTP_CA_ClientCert="" HTTP_CA_CACert="" HTTP_CA_Passphrase="" HTTP_Proxy="" HTTP_Proxy_Host="" HTTP_Proxy_Port="" HTTP_Proxy_UserName="" HTTP_Proxy_Password="" HTTP_Proxy_SavePassword="" Compatibility="false" Compatibility_OverrideLowerCaseTableNames="false" Compatibility_LowerCaseTableNames="" Compatibility_OverrideSQLMode="false" Compatibility_SQLMode="" Compatibility_OverrideIsSupportNDB="false" Compatibility_IsSupportNDB="false" Compatibility_OverrideDatabaseListingMethod="false" Compatibility_DatabaseListingMethod="" Compatibility_OverrideViewListingMethod="false" Compatibility_ViewListingMethod=""/>
</Connections>