deploy.py 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282
  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.BatSafetyAlarm.V1_0_1 import CBMSSafetyAlarm
  13. from LIB.MIDDLE.CellStateEstimation.Common 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(df_sn, df_bms_ram, df_alarm_ram, log_name):
  23. # 日志
  24. logger = logging.getLogger()
  25. fh = logging.handlers.RotatingFileHandler(log_name + ".log", encoding="utf-8",maxBytes=1024*1024*1024, backupCount=5, mode="a")
  26. # fh = logging.handlers.db_res_engine(log_name + ".log", encoding="utf-8",maxBytes=1024*1024*1024, backupCount=5, mode="a")
  27. formatter = logging.Formatter("%(asctime)s - %(name)s-%(levelname)s %(message)s")
  28. fh.setFormatter(formatter)
  29. logger.addHandler(fh)
  30. logger.setLevel(logging.INFO)
  31. logger.info("pid is {}".format(os.getpid()))
  32. # 读取结果数据库
  33. host2='rm-bp10j10qy42bzy0q77o.mysql.rds.aliyuncs.com'
  34. port2=3306
  35. db2='safety_platform'
  36. user2='qx_algo_rw'
  37. password2='qx@123456'
  38. db_res_engine = create_engine(
  39. "mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8".format(
  40. user2, parse.quote_plus(password2), host2, port2, db2
  41. ))
  42. conn = pymysql.connect(host=host2, port=port2, user=user2, password=password2, database=db2)
  43. cursor = conn.cursor()
  44. result=pd.read_sql("select start_time, end_time, product_id, code, level, info, advice from all_fault_info", db_res_engine)
  45. result = result[['start_time', 'end_time', 'product_id', 'code', 'level', 'info', 'advice']]
  46. df_diag_ram=result[(result['end_time']=='0000-00-00 00:00:00') & (result['code']=='C599')]
  47. start=time.time()
  48. now_time=datetime.datetime.now()
  49. start_time=now_time-datetime.timedelta(seconds=70)
  50. start_time=start_time.strftime('%Y-%m-%d %H:%M:%S')
  51. end_time=now_time.strftime('%Y-%m-%d %H:%M:%S')
  52. for i in range(0, len(df_sn)):
  53. factory = "骑享"
  54. try:
  55. if df_sn.loc[i, 'imei'][5:9] == 'N640':
  56. celltype=1 #6040三元电芯
  57. elif df_sn.loc[i, 'imei'][5:9] == 'N440':
  58. celltype=2 #4840三元电芯
  59. elif df_sn.loc[i, 'imei'][5:9] == 'L660':
  60. celltype=99 # 6060锂电芯
  61. elif df_sn.loc[i, 'imei'][3:5] == 'LX' and df_sn.loc[i, 'imei'][5:9] == 'N750':
  62. celltype=3 #力信 50ah三元电芯
  63. elif df_sn.loc[i, 'imei'][3:5] == 'CL' and df_sn.loc[i, 'imei'][5:9] == 'N750':
  64. celltype=4 #CATL 50ah三元电芯
  65. elif df_sn.loc[i, 'imei'][3:9] == 'CLL128':
  66. celltype=100 # 重卡
  67. factory = "金茂换电"
  68. else:
  69. logger.info("pid-{} celltype-{} SN: {} SKIP!".format(os.getpid(), "未知", sn))
  70. continue
  71. sn = df_sn.loc[i, 'sn']
  72. logger.info("pid-{} celltype-{} SN: {} START!".format(os.getpid(), celltype, sn))
  73. #读取原始数据库数据........................................................................................................................................................
  74. dbManager = DBManager.DBManager()
  75. df_data = dbManager.get_data(sn=sn, start_time=start_time, end_time=end_time, data_groups=['bms'])
  76. df_bms = df_data['bms']
  77. # print(df_bms)
  78. #电池诊断................................................................................................................................................................
  79. df_diag_ram_sn=df_diag_ram[df_diag_ram['product_id']==sn]
  80. df_bms_ram_sn=df_bms_ram[df_bms_ram['sn']==sn]
  81. df_alarm_ram_sn=df_alarm_ram[df_alarm_ram['sn']==sn]
  82. if df_diag_ram_sn.empty:
  83. SafetyAlarm=CBMSSafetyAlarm.SafetyAlarm(sn,celltype,df_bms, df_bms_ram_sn,df_alarm_ram_sn)
  84. df_diag_res, df_bms_res,df_ram_res=SafetyAlarm.safety_alarm_diag()
  85. #更新bms的ram数据 和 diag的Ram数据
  86. sn_index=df_bms_ram.loc[df_bms_ram['sn']==sn].index
  87. df_bms_ram=df_bms_ram.drop(index=sn_index)
  88. df_bms_ram=df_bms_ram.append(df_bms_res)
  89. sn_index=df_alarm_ram.loc[df_alarm_ram['sn']==sn].index
  90. df_alarm_ram=df_alarm_ram.drop(index=sn_index)
  91. df_alarm_ram=df_alarm_ram.append(df_ram_res)
  92. # sn_index=df_diag_ram.loc[df_diag_ram['product_id']==sn].index
  93. # df_diag_ram=df_diag_ram.drop(index=sn_index)
  94. # df_diag_ram=df_diag_ram.append(df_diag_res)
  95. # df_diag_ram.reset_index(inplace=True,drop=True) #重置索引
  96. #当前热失控故障写入数据库
  97. if not df_diag_res.empty:
  98. df_diag_res.columns = ['start_time', 'end_time', 'product_id', 'code', 'level', 'info', 'advice']
  99. df_diag_res['add_time'] = datetime.datetime.now()
  100. df_diag_res['factory'] = factory
  101. df_diag_res.to_sql("all_fault_info",con=db_res_engine, if_exists="append",index=False)
  102. logger.info(u"{} 写入成功!!!\n".format(sn), exc_info=True)
  103. #当前热失控已超过三天变为历史故障并写入数据库,并删除原有数据库中的当前故障和ram中的当前故障
  104. else:
  105. fault_time=datetime.datetime.strptime(df_diag_ram_sn.iloc[-1]['start_time'], '%Y-%m-%d %H:%M:%S')
  106. if (now_time-fault_time).total_seconds()>24*3600:
  107. df_diag_ram_sn['end_time']=end_time
  108. df_diag_ram_sn['Batpos']=1
  109. try:
  110. cursor.execute('''
  111. update all_fault_info set update_time='{}',end_time='{}', Batpos={} where product_id='{}' and end_time='0000-00-00 00:00:00' and code='{}'
  112. '''.format(datetime.datetime.now(), end_time, 1,sn, "C599"))
  113. conn.commit()
  114. logger.info(u"{} 更新成功\n".format(sn), exc_info=True)
  115. except:
  116. logger.error(traceback.format_exc)
  117. logger.error(u"{} :{},{} 任务运行错误\n".format(sn,start_time,end_time), exc_info=True)
  118. except:
  119. logger.error(traceback.format_exc)
  120. logger.error(u"{} :{},{} 任务运行错误\n".format(sn,start_time,end_time), exc_info=True)
  121. break
  122. cursor.close()
  123. conn.close()
  124. db_res_engine.dispose()
  125. logger.info("pid-{} Done!".format(os.getpid()))
  126. return df_bms_ram,df_alarm_ram
  127. #...................................................主进程...........................................................................................................
  128. def mainprocess():
  129. global SNnums
  130. global df_bms_ram1, df_bms_ram2, df_alarm_ram1, df_alarm_ram2
  131. global log_path
  132. # 更新sn列表
  133. host='rm-bp10j10qy42bzy0q7.mysql.rds.aliyuncs.com'
  134. port=3306
  135. db='qixiang_oss'
  136. user='qixiang_oss'
  137. password='Qixiang2021'
  138. conn = pymysql.connect(host=host, port=port, user=user, password=password, database=db)
  139. cursor = conn.cursor()
  140. cursor.execute("select sn, imei, add_time from app_device where status in (1,2,3)")
  141. res = cursor.fetchall()
  142. df_sn = pd.DataFrame(res, columns=['sn', 'imei', 'add_time'])
  143. df_sn = df_sn.reset_index(drop=True)
  144. cursor.close()
  145. conn.close();
  146. process = 2
  147. pool = multiprocessing.Pool(processes = process)
  148. res_list=[]
  149. ram_list1=[df_bms_ram1, df_bms_ram2]
  150. ram_list2=[df_alarm_ram1, df_alarm_ram1]
  151. for i in range(process):
  152. sn_list = df_sn[int(len(df_sn)*i/process):int(len(df_sn)*(i+1)/process)]
  153. sn_list = sn_list.reset_index(drop=True)
  154. log_name = log_path + '/log_' + str(i)
  155. df_bms_ram=ram_list1[i]
  156. df_alarm_ram=ram_list2[i]
  157. df_res = pool.apply_async(diag_cal, (sn_list,df_bms_ram,df_alarm_ram,log_name)).get()
  158. res_list.append(df_res)
  159. pool.close()
  160. pool.join()
  161. df_bms_ram1=res_list[0][0]
  162. df_bms_ram2=res_list[1][0]
  163. df_alarm_ram1=res_list[0][1]
  164. df_alarm_ram2=res_list[1][1]
  165. if __name__ == "__main__":
  166. # 时间设置
  167. # now_time = datetime.datetime.now()
  168. # pre_time = now_time + dateutil.relativedelta.relativedelta(days=-1)# 前一日
  169. # end_time=datetime.datetime.strftime(now_time,"%Y-%m-%d 00:00:00")
  170. # start_time=datetime.datetime.strftime(pre_time,"%Y-%m-%d 00:00:00")
  171. history_run_flag = False # 历史数据运行标志
  172. # # 更新sn列表
  173. # host='rm-bp10j10qy42bzy0q7.mysql.rds.aliyuncs.com'
  174. # port=3306
  175. # db='qixiang_oss'
  176. # user='qixiang_oss'
  177. # password='Qixiang2021'
  178. # conn = pymysql.connect(host=host, port=port, user=user, password=password, database=db)
  179. # cursor = conn.cursor()
  180. # cursor.execute("select sn, imei, add_time from app_device")
  181. # res = cursor.fetchall()
  182. # df_sn = pd.DataFrame(res, columns=['sn', 'imei', 'add_time'])
  183. # df_sn = df_sn.reset_index(drop=True)
  184. # conn.close();
  185. # 数据库配置
  186. host = 'rm-bp10j10qy42bzy0q77o.mysql.rds.aliyuncs.com'
  187. port = 3306
  188. user = 'qx_cas'
  189. password = parse.quote_plus('Qx@123456')
  190. database = 'qx_cas'
  191. db_engine = create_engine(
  192. "mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8".format(
  193. user, password, host, port, database
  194. ))
  195. DbSession = sessionmaker(bind=db_engine)
  196. # 运行历史数据配置
  197. df_first_data_time = pd.read_sql("select * from bat_first_data_time", db_engine)
  198. # 日志配置
  199. now_str = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()).replace(":","_")
  200. log_path = 'log/' + now_str
  201. if not os.path.exists(log_path):
  202. os.makedirs(log_path)
  203. log = Log.Mylog(log_name='batsafetyAlarm', log_level = 'info')
  204. log.set_file_hl(file_name='{}/info.log'.format(log_path), log_level='info', size=1024* 1024 * 100)
  205. log.set_file_hl(file_name='{}/error.log'.format(log_path), log_level='error', size=1024* 1024 * 100)
  206. logger = log.get_logger()
  207. logger.info("pid is {}".format(os.getpid()))
  208. # 算法参数
  209. host='rm-bp10j10qy42bzy0q77o.mysql.rds.aliyuncs.com'
  210. port=3306
  211. db='safety_platform'
  212. user='qx_read'
  213. password=parse.quote_plus('Qx@123456')
  214. tablename='all_fault_info'
  215. db_res_engine = create_engine(
  216. "mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8".format(
  217. user, password, host, port, db
  218. ))
  219. #............................模块运行前,先读取数据库中所有结束时间为0的数据,需要从数据库中读取................
  220. # print("select start_time, end_time, product_id, code, level, info, advice, factory from {}".format(tablename))
  221. # 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)
  222. # result = result[['start_time', 'end_time', 'product_id', 'code', 'level', 'info', 'advice']]
  223. # df_diag_ram=result[(result['end_time']=='0000-00-00 00:00:00') & (result['code']==119)]
  224. df_bms_ram1=pd.DataFrame(columns=['time', 'sn', 'packvolt', 'cellvolt', 'celltemp'])
  225. df_bms_ram2=pd.DataFrame(columns=['time', 'sn', 'packvolt', 'cellvolt', 'celltemp'])
  226. df_alarm_ram1=pd.DataFrame(columns=['sn','time','safetywarning1','safetywarning2'])
  227. df_alarm_ram2=pd.DataFrame(columns=['sn','time','safetywarning1','safetywarning2'])
  228. mainprocess()
  229. #定时任务.......................................................................................................................................................................
  230. scheduler = BlockingScheduler()
  231. scheduler.add_job(mainprocess, 'interval', seconds=60, id='diag_job')
  232. try:
  233. scheduler.start()
  234. except Exception as e:
  235. scheduler.shutdown()
  236. logger.error(str(e))