deploy_test.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226
  1. __author__ = 'lmstack'
  2. #coding=utf-8
  3. import os
  4. import datetime
  5. import pandas as pd
  6. from LIB.BACKEND import DBManager, Log
  7. from sqlalchemy import create_engine
  8. from sqlalchemy.orm import sessionmaker
  9. import time, datetime
  10. import traceback
  11. from LIB.MIDDLE.CellStateEstimation.Common import log
  12. from LIB.MIDDLE.CellStateEstimation.BatDiag.V1_0_0 import CBMSBatDiag
  13. from LIB.MIDDLE.CellStateEstimation.Common.V1_0_1 import DBDownload
  14. from urllib import parse
  15. import pymysql
  16. import pdb
  17. from apscheduler.schedulers.blocking import BlockingScheduler
  18. import datacompy
  19. import logging
  20. import multiprocessing
  21. #...................................电池包电芯安全诊断函数......................................................................................................................
  22. def diag_cal():
  23. global df_sn, df_bms_ram
  24. start=time.time()
  25. now_time=datetime.datetime.now()
  26. start_time=now_time-datetime.timedelta(seconds=310)
  27. start_time=start_time.strftime('%Y-%m-%d %H:%M:%S')
  28. end_time=now_time.strftime('%Y-%m-%d %H:%M:%S')
  29. #数据库配置
  30. host='rm-bp10j10qy42bzy0q77o.mysql.rds.aliyuncs.com'
  31. port=3306
  32. db='safety_platform'
  33. user='qx_read'
  34. password='Qx@123456'
  35. #读取故障结果库中当前故障......................................................
  36. param='start_time,end_time,product_id,code,level,info,advice'
  37. tablename='all_fault_info_copy'
  38. mysql = pymysql.connect (host=host, user=user, password=password, port=port, database=db)
  39. cursor = mysql.cursor()
  40. sql = "select %s from %s where end_time='0000-00-00 00:00:00'" %(param,tablename)
  41. cursor.execute(sql)
  42. res = cursor.fetchall()
  43. df_diag_ram= pd.DataFrame(res,columns=param.split(','))
  44. db_res_engine = create_engine(
  45. "mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8".format(
  46. user, parse.quote_plus(password), host, port, db
  47. ))
  48. for i in range(0, len(df_sn)):
  49. factory = "骑享"
  50. try:
  51. if df_sn.loc[i, 'imei'][5:9] == 'N640':
  52. celltype=1 #6040三元电芯
  53. elif df_sn.loc[i, 'imei'][5:9] == 'N440':
  54. celltype=2 #4840三元电芯
  55. elif df_sn.loc[i, 'imei'][5:9] == 'L660':
  56. celltype=99 # 6060锂电芯
  57. elif df_sn.loc[i, 'imei'][3:5] == 'LX' and df_sn.loc[i, 'imei'][5:9] == 'N750':
  58. celltype=3 #力信 50ah三元电芯
  59. elif df_sn.loc[i, 'imei'][3:5] == 'CL' and df_sn.loc[i, 'imei'][5:9] == 'N750':
  60. celltype=4 #CATL 50ah三元电芯
  61. elif df_sn.loc[i, 'imei'][3:9] == 'CLL128':
  62. celltype=100 # 重卡
  63. factory = "金茂换电"
  64. else:
  65. logger.info("pid-{} celltype-{} SN: {} SKIP!".format(os.getpid(), "未知", sn))
  66. continue
  67. sn = df_sn.loc[i, 'sn']
  68. logger.info("pid-{} celltype-{} SN: {} START!".format(os.getpid(), celltype, sn))
  69. #读取原始数据库数据........................................................................................................................................................
  70. dbManager = DBManager.DBManager()
  71. df_data = dbManager.get_data(sn=sn, start_time=start_time, end_time=end_time, data_groups=['bms'])
  72. df_bms = df_data['bms']
  73. #读取结果数据库数据........................................................................................................................................................
  74. db='qx_cas'
  75. mode=1
  76. tablename1='cellstateestimation_soh'
  77. tablename2='cellstateestimation_uniform_socvoltdiff'
  78. # tablename3='cellstateestimation_soc'
  79. DBRead=DBDownload.DBDownload(host, port, db, user, password,mode)
  80. with DBRead as DBRead:
  81. print(sn,start_time,end_time)
  82. df_soh=DBRead.getdata('time_st,sn,soh,cellsoh', tablename=tablename1, sn=sn, timename='time_sp', st=start_time, sp=end_time)
  83. df_uniform=DBRead.getdata('time,sn,cellsoc_diff,cellmin_num,cellmax_num', tablename=tablename2, sn=sn, timename='time', st=start_time, sp=end_time)
  84. # df_soc=DBRead.getdata('time','sn','packsoc', tablename=tablename3, sn=sn)
  85. #电池诊断................................................................................................................................................................
  86. if not df_bms.empty:
  87. df_diag_ram_sn=df_diag_ram[df_diag_ram['product_id']==sn]
  88. df_diag_ram_sn.reset_index(inplace=True,drop=True)
  89. df_bms_ram_sn=df_bms_ram[df_bms_ram['sn']==sn]
  90. df_bms_ram_sn.reset_index(inplace=True,drop=True)
  91. batquality=CBMSBatDiag.BatDiag(sn,celltype,df_bms,df_diag_ram_sn,df_bms_ram_sn,df_soh,df_uniform)
  92. df_diag_res, df_bms_res=batquality.diag() #获取电池故障结果和电池评分结果
  93. #更新bms的ram数据
  94. sn_index=df_bms_ram.loc[df_bms_ram['sn']==sn].index
  95. df_bms_ram=df_bms_ram.drop(index=sn_index)
  96. df_bms_ram=df_bms_ram.append(df_bms_res)
  97. #历史故障筛选并更改数据库故障结束时间.........................................................
  98. if not df_diag_res.empty:
  99. df_diag_new = pd.concat([df_diag_res,df_diag_ram_sn,df_diag_ram_sn]).drop_duplicates(subset=['start_time','code'],keep=False)#此次判断中新增故障
  100. df_diag_new.reset_index(inplace=True,drop=True) #重置索引
  101. df_diag_end=pd.concat([df_diag_res,df_diag_new,df_diag_new]).drop_duplicates(subset=['start_time','code'],keep=False)#此次判断中新增故障
  102. df_diag_end=df_diag_end[df_diag_end['end_time'] != '0000-00-00 00:00:00']
  103. df_diag_end.reset_index(inplace=True,drop=True) #重置索引
  104. if not df_diag_end.empty: #变为历史故障更改数据库
  105. for j in range(len(df_diag_end)):
  106. cursor.execute('''
  107. update all_fault_info_copy set update_time='{}',end_time='{}', Batpos={} where product_id='{}' and end_time='0000-00-00 00:00:00' and code='{}'
  108. '''.format(datetime.datetime.now(), df_diag_end.loc[j,'end_time'], 1 ,sn, df_diag_end.loc[j,'code']))
  109. mysql.commit()
  110. #新增故障筛选并存入数据库.....................................................................
  111. if not df_diag_new.empty: #新增写入数据库
  112. df_diag_new.to_sql("all_fault_info_copy",con=db_res_engine, if_exists="append",index=False)
  113. end=time.time()
  114. print(end-start)
  115. except Exception as e:
  116. logger.error(traceback.format_exc)
  117. logger.error(u"{} :{},{} 任务运行错误\n".format(sn,start_time,end_time), exc_info=True)
  118. cursor.close()
  119. mysql.close()
  120. if __name__ == "__main__":
  121. # 时间设置
  122. # now_time = datetime.datetime.now()
  123. # pre_time = now_time + dateutil.relativedelta.relativedelta(days=-1)# 前一日
  124. # end_time=datetime.datetime.strftime(now_time,"%Y-%m-%d 00:00:00")
  125. # start_time=datetime.datetime.strftime(pre_time,"%Y-%m-%d 00:00:00")
  126. history_run_flag = False # 历史数据运行标志
  127. # # 更新sn列表
  128. host='rm-bp10j10qy42bzy0q7.mysql.rds.aliyuncs.com'
  129. port=3306
  130. db='qixiang_oss'
  131. user='qixiang_oss'
  132. password='Qixiang2021'
  133. conn = pymysql.connect(host=host, port=port, user=user, password=password, database=db)
  134. cursor = conn.cursor()
  135. cursor.execute("select sn, imei, add_time from app_device where status in (1,2,3)")
  136. res = cursor.fetchall()
  137. df_sn = pd.DataFrame(res, columns=['sn', 'imei', 'add_time'])
  138. df_sn = df_sn.reset_index(drop=True)
  139. conn.close();
  140. # # 数据库配置
  141. # host = 'rm-bp10j10qy42bzy0q77o.mysql.rds.aliyuncs.com'
  142. # port = 3306
  143. # user = 'qx_cas'
  144. # password = parse.quote_plus('Qx@123456')
  145. # database = 'qx_cas'
  146. # db_engine = create_engine(
  147. # "mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8".format(
  148. # user, password, host, port, database
  149. # ))
  150. # DbSession = sessionmaker(bind=db_engine)
  151. # # 运行历史数据配置
  152. # df_first_data_time = pd.read_sql("select * from bat_first_data_time", db_engine)
  153. # 日志配置
  154. now_str = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()).replace(":","_")
  155. log_path = 'log_test/' + now_str
  156. if not os.path.exists(log_path):
  157. os.makedirs(log_path)
  158. log = Log.Mylog(log_name='batDiag', log_level = 'info')
  159. log.set_file_hl(file_name='{}/info.log'.format(log_path), log_level='info', size=1024* 1024 * 100)
  160. log.set_file_hl(file_name='{}/error.log'.format(log_path), log_level='error', size=1024* 1024 * 100)
  161. logger = log.get_logger()
  162. logger.info("pid is {}".format(os.getpid()))
  163. # # 算法参数
  164. # host='rm-bp10j10qy42bzy0q77o.mysql.rds.aliyuncs.com'
  165. # port=3306
  166. # db='safety_platform'
  167. # user='qx_read'
  168. # password=parse.quote_plus('Qx@123456')
  169. # tablename='all_fault_info'
  170. # db_res_engine = create_engine(
  171. # "mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8".format(
  172. # user, password, host, port, db
  173. # ))
  174. #............................模块运行前,先读取数据库中所有结束时间为0的数据,需要从数据库中读取................
  175. # print("select start_time, end_time, product_id, code, level, info, advice, factory from {}".format(tablename))
  176. # result=pd.read_sql("select start_time, end_time, product_id, code, level, info, advice from all_fault_info where factory = '{}'".format('骑享'), db_res_engine)
  177. # result = result[['start_time', 'end_time', 'product_id', 'code', 'level', 'info', 'advice']]
  178. # df_diag_ram=result[(result['end_time']=='0000-00-00 00:00:00') & (result['code']==119)]
  179. df_bms_ram=pd.DataFrame(columns=['time', 'sn', 'packvolt', 'cellvolt', 'celltemp'])
  180. diag_cal()
  181. #定时任务.......................................................................................................................................................................
  182. scheduler = BlockingScheduler()
  183. scheduler.add_job(diag_cal, 'interval', seconds=300, id='diag_job')
  184. try:
  185. scheduler.start()
  186. except Exception as e:
  187. scheduler.shutdown()
  188. logger.error(str(e))