import execjs import csv import numpy as np import pandas as pd import time import json import jenkspy import requests import pymssql import config class HealthyScoringSystem: def __init__(self): # 读取配置 ms = MSSQL(host="120.26.116.243", user="root", pwd="123456", database="alert") conditionlist = ms.ExecQuery(f"SELECT * FROM [ASSESS].[dbo].[conditionlist]") pointconfigs = ms.ExecQuery(f"SELECT * FROM [ASSESS].[dbo].[pointconfigs]") scorelist = ms.ExecQuery(f"SELECT * FROM [ASSESS].[dbo].[scorelist]") subsystemweightlist = ms.ExecQuery(f"SELECT * FROM [ASSESS].[dbo].[subsystemweightlist]") systemweightlist = ms.ExecQuery(f"SELECT * FROM [ASSESS].[dbo].[systemweightlist]") unitweightlist = ms.ExecQuery(f"SELECT * FROM [ASSESS].[dbo].[unitweightlist]") condi = list(range(len(conditionlist))) for i in range(len(condi)): condi[i] = str(condi[i]) point = list(range(len(pointconfigs))) for i in range(len(point)): point[i] = str(point[i]) score = list(range(len(scorelist))) for i in range(len(score)): score[i] = str(score[i]) subsy = list(range(len(subsystemweightlist))) for i in range(len(subsy)): subsy[i] = str(subsy[i]) syste = list(range(len(systemweightlist))) for i in range(len(syste)): syste[i] = str(syste[i]) unitw = list(range(len(unitweightlist))) for i in range(len(unitw)): unitw[i] = str(unitw[i]) self.conditionlist = dict(zip(list(dict(zip(condi,condi)).values()),conditionlist)) self.pointconfigs = dict(zip(list(dict(zip(point,point)).values()),pointconfigs)) self.scorelist = dict(zip(list(dict(zip(score,score)).values()),scorelist)) self.subsystemweightlist = dict(zip(list(dict(zip(subsy,subsy)).values()),subsystemweightlist)) self.systemweightlist = dict(zip(list(dict(zip(syste,syste)).values()),systemweightlist)) self.unitweightlist = dict(zip(list(dict(zip(unitw,unitw)).values()),unitweightlist)) # self.treeconfigs = self.indexMap("treeconfigs.csv") self.inputData = {} self.conditionMap = {} self.outputData = {} self.time = time.time() def indexMap(self, csvFileName): map = {} csvFile = open(csvFileName, "r") reader = csv.reader(csvFile) colnames = [] for item in reader: if reader.line_num == 1: for colname in item: colnames.append(colname) else: details = {} for i in range(len(colnames) - 1): details[colnames[i + 1]] = item[i + 1] map[item[0]] = details csvFile.close() return map def getInputDate(self): # 取实时值 for value in self.pointconfigs.values(): self.inputData[value['pointname']] = get_now_data(value['pointname']) # 上线之后把这个换成取实时值 self.time = time.time() def Scoring(self): # 测点级打分 for key, value in self.scorelist.items(): x1 = self.inputData[self.pointconfigs[value['firstpoint']]['pointname']] if value['formula'] == '0': # 分数等于 np.clip(x1-x2,0,1) x2 = self.inputData[self.pointconfigs[value['secondpoint']]['pointname']] y = np.clip(x1 - x2, 0, 1) elif value['formula'] == '1': # 分数等于np.clip(abs(x1-x2),0,1) x2 = self.inputData[self.pointconfigs[value['secondpoint']]['pointname']] y = np.clip(abs(x1 - x2), 0, 1) elif value['formula'] == '2': # 分数等于np.tanh(x1) y = np.tanh(x1) else: # 分数根据js字符串解析 x2 = self.inputData[self.pointconfigs[value['secondpoint']]['pointname']] y = execjs.eval(value['formula'].replace("x1", str(x1)).replace("x2", str(x2))) self.outputData[value['scoreoutput']] = y def Conditioning(self): # 求解计算条件 for key, value in self.conditionlist.items(): inputlist = value['inputlist'].split(',') formula = value['code'] for i in range(len(inputlist)): formula = formula.replace("x" + str(len(inputlist) - i), str(self.inputData[ self.pointconfigs[inputlist[len(inputlist) - i - 1]]['pointname']])) y = execjs.eval(formula) self.conditionMap[key] = y def weightedSumScoring(self, level): # 系统级打分 switch = { "subsystem": { 'currentlevellist': self.subsystemweightlist, 'nextlevellist': self.scorelist, 'nextlevelliststr': 'scorelist', }, "system": { 'currentlevellist': self.systemweightlist, 'nextlevellist': self.subsystemweightlist, 'nextlevelliststr': 'subsystemlist', }, "unit": { 'currentlevellist': self.unitweightlist, 'nextlevellist': self.systemweightlist, 'nextlevelliststr': 'systemlist', }, } currentlevellist = switch[level]['currentlevellist'] nextlevellist = switch[level]['nextlevellist'] nextlevelliststr = switch[level]['nextlevelliststr'] for key, value in currentlevellist.items(): readyFlag = True conditionlist = value['conditionlist'].split(',') for condition in conditionlist: readyFlag = readyFlag and self.conditionMap[condition] readyFlag = True # 后面要删掉 if readyFlag: scoreindexlist = value[nextlevelliststr].split(',') scorelist = [] for scoreindex in scoreindexlist: scorelist.append(self.outputData[nextlevellist[scoreindex]['scoreoutput']]) weightlist = list(map(float, (value['weightlist'].split(',')))) score = \ np.dot(np.array(scorelist).reshape(1, len(scorelist)), np.array(weightlist).reshape(len(scorelist), 1))[ 0, 0] / np.sum(weightlist) self.outputData[value['scoreoutput']] = score def getScore(self, unit=None, system=None, subsystem=None, type=None): # 获取指定级别分数的接口 if subsystem != None: for key, value in self.subsystemweightlist.items(): if value['unit'] == unit and value['system'] == system and value['subsystem'] == subsystem and value[ 'type'] == type: return self.outputData[value['scoreoutput']] elif system != None and subsystem == None: for key, value in self.systemweightlist.items(): if value['unit'] == unit and value['system'] == system and value['type'] == type: return self.outputData[value['scoreoutput']] elif unit != None and system == None: for key, value in self.unitweightlist.items(): if value['unit'] == unit and value['type'] == type: return self.outputData[value['scoreoutput']] def outputScore(self): nowtime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(self.time)) ms = MSSQL(host="172.28.137.230", user="sa", pwd="powerSIS#123", database="ASSESS") score = len(self.scorelist) subsy = len(self.subsystemweightlist) + score syste = len(self.systemweightlist) + subsy for table in ['conditionlist', 'scorelist', 'subsystemweightlist', 'systemweightlist', 'unitweightlist']: if table == 'conditionlist': for i in range(len(self.conditionlist)): setitem = f"result = '{self.conditionMap[f'{i}']}' , time = '{nowtime}'" whereitem = f"indice = {i}" Queryitem = f"UPDATE [ASSESS].[dbo].[{table}] set {setitem} where {whereitem}" ms.ExecNonQuery(Queryitem) elif table == 'scorelist': j = 0 for i in list(self.outputData.values())[:score]: setitem = f"result = '{i}' , time = '{nowtime}'" whereitem = f"indice = {j}" Queryitem = f"UPDATE [ASSESS].[dbo].[{table}] set {setitem} where {whereitem}" ms.ExecNonQuery(Queryitem) j = j + 1 elif table == 'subsystemweightlist': j = 0 for i in list(self.outputData.values())[score:subsy]: setitem = f"result = '{i}' , time = '{nowtime}'" whereitem = f"indice = {j}" Queryitem = f"UPDATE [ASSESS].[dbo].[{table}] set {setitem} where {whereitem}" ms.ExecNonQuery(Queryitem) j = j + 1 elif table == 'systemweightlist': j = 0 for i in list(self.outputData.values())[subsy:syste]: setitem = f"result = '{i}' , time = '{nowtime}'" whereitem = f"indice = {j}" Queryitem = f"UPDATE [ASSESS].[dbo].[{table}] set {setitem} where {whereitem}" ms.ExecNonQuery(Queryitem) j = j + 1 elif table == 'unitweightlist': j = 0 for i in list(self.outputData.values())[syste:]: setitem = f"result = '{i}' , time = '{nowtime}'" whereitem = f"indice = {j}" Queryitem = f"UPDATE [ASSESS].[dbo].[{table}] set {setitem} where {whereitem}" ms.ExecNonQuery(Queryitem) j = j + 1 class MSSQL: def __init__(self,host,user,pwd,database): self.host = host self.user = user self.pwd = pwd self.db = database def __GetConnect(self): """ 得到连接信息 返回: conn.cursor() """ if not self.db: raise(NameError,"没有设置数据库信息") self.conn = pymssql.connect(host=self.host,user=self.user,password=self.pwd,database=self.db,port=config._PORT, charset="utf8") cur = self.conn.cursor() if not cur: raise(NameError,"连接数据库失败") else: return cur def ExecQuery(self,sql): """ 执行查询语句 返回的是一个包含tuple的list，list的元素是记录行，tuple的元素是每行记录的字段 """ cur = self.__GetConnect() cur.execute(sql) resList = cur.fetchall() for i in range(len(resList)): resList[i] = resList[i][1:] #读取 sql 得到dict by lifan # 只用于select语句，返回一行的列名 desc = cur.description desc = desc[:0] + desc[1:] object_dict = [ dict(zip([col[0] for col in desc], row)) for row in resList ] #查询完毕后必须关闭连接 self.conn.close() return object_dict def ExecNonQuery(self,sql): """ 执行非查询语句 调用示例： cur = self.__GetConnect() cur.execute(sql) self.conn.commit() self.conn.close() """ cur = self.__GetConnect() cur.execute(sql) self.conn.commit() self.conn.close() def get_now_data(items): url = f"http://{config._EXA_IP}:9000/exawebapi/exanow/getfloatvaluebatch" headers = {"Content-Type": "application/json;charset=utf-8"} # ,"token":get_token() #point_array = items.split(',') data = {} data['ItemNames'] = items res = requests.get(url, headers=headers, params=data) response = res.text.replace("[","").replace("]","") return float(response) if __name__ == '__main__': a = "JL_D2_20DAS05A:LAV10CE101.PNT" #get_now_data(a) HSS = HealthyScoringSystem() HSS.getInputDate() HSS.Scoring() HSS.Conditioning() HSS.weightedSumScoring('subsystem') HSS.weightedSumScoring('system') HSS.weightedSumScoring('unit') HSS.outputScore() print(HSS.getScore(unit='1', type='1')) print(HSS.getScore(unit='1', system='1', type='1')) print(HSS.getScore(unit='1', system='1', subsystem='1', type='1')) # url = f"http://{config._CLEAN_IP}/exawebapi/exatime/GetCleaningData?ItemsInfo=%s&SamplingTimePeriods=%s&Constraint=%s&SamplingPeriod=%s&DCount=%d" % (ItemsInfo, SamplingTimePeriods, Constraint, interval, DCount) # response = requests.get(url) # content = json.loads(response.text) b = 1