__author__ = 'lmstack' #coding=utf-8 import os import datetime import pandas as pd from LIB.BACKEND import DBManager, Log from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker import time, datetime import traceback from LIB.MIDDLE.CellStateEstimation.Common import log from LIB.MIDDLE.CellStateEstimation.BatSafetyAlarm.V1_0_1 import CBMSSafetyAlarm from LIB.MIDDLE.CellStateEstimation.Common import DBDownload from urllib import parse import pymysql import pdb from apscheduler.schedulers.blocking import BlockingScheduler import datacompy import logging import multiprocessing #...................................电池包电芯安全诊断函数...................................................................................................................... def diag_cal(df_sn, df_diag_ram, df_bms_ram, log_name): # 日志 logger = logging.getLogger() fh = logging.FileHandler(log_name + ".log", encoding="utf-8",mode="a") formatter = logging.Formatter("%(asctime)s - %(name)s-%(levelname)s %(message)s") fh.setFormatter(formatter) logger.addHandler(fh) logger.setLevel(logging.INFO) logger.info("pid is {}".format(os.getpid())) # 读取结果数据库 host2='rm-bp10j10qy42bzy0q77o.mysql.rds.aliyuncs.com' port2=3306 db2='safety_platform' user2='qx_read' password2='Qx@123456' db_res_engine = create_engine( "mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8".format( user2, parse.quote_plus(password2), host2, port2, db2 )) start=time.time() now_time=datetime.datetime.now() start_time=now_time-datetime.timedelta(seconds=70) start_time=start_time.strftime('%Y-%m-%d %H:%M:%S') end_time=now_time.strftime('%Y-%m-%d %H:%M:%S') for i in range(0, len(df_sn)): try: if df_sn.loc[i, 'imei'][5:9] == 'N640': celltype=1 #6040三元电芯 elif df_sn.loc[i, 'imei'][5:9] == 'N440': celltype=2 #4840三元电芯 elif df_sn.loc[i, 'imei'][5:9] == 'L660': celltype=99 # 6060锂电芯 elif df_sn.loc[i, 'imei'][3:5] == 'LX' and df_sn.loc[i, 'imei'][5:9] == 'N750': celltype=3 #力信 50ah三元电芯 elif df_sn.loc[i, 'imei'][3:5] == 'CL' and df_sn.loc[i, 'imei'][5:9] == 'N750': celltype=4 #CATL 50ah三元电芯 else: logger.info("pid-{} celltype-{} SN: {} SKIP!".format(os.getpid(), "未知", sn)) continue sn = df_sn.loc[i, 'sn'] logger.info("pid-{} celltype-{} SN: {} START!".format(os.getpid(), celltype, sn)) #读取原始数据库数据........................................................................................................................................................ dbManager = DBManager.DBManager() df_data = dbManager.get_data(sn=sn, start_time=start_time, end_time=end_time, data_groups=['bms']) df_bms = df_data['bms'] # print(df_bms) #电池诊断................................................................................................................................................................ if not df_bms.empty: df_diag_ram_sn=df_diag_ram[df_diag_ram['product_id']==sn] df_bms_ram_sn=df_bms_ram[df_bms_ram['sn']==sn] if df_diag_ram_sn.empty: SafetyAlarm=CBMSSafetyAlarm.SafetyAlarm(sn,celltype,df_bms, df_bms_ram_sn) df_diag_res, df_bms_res=SafetyAlarm.diag() #更新bms的ram数据 和 diag的Ram数据 sn_index=df_bms_ram.loc[df_bms_ram['sn']==sn].index df_bms_ram=df_bms_ram.drop(index=sn_index) df_bms_ram=df_bms_ram.append(df_bms_res) sn_index=df_diag_ram.loc[df_diag_ram['product_id']==sn].index df_diag_ram=df_diag_ram.drop(index=sn_index) df_diag_ram=df_diag_ram.append(df_diag_res) df_diag_ram.reset_index(inplace=True,drop=True) #重置索引 #当前热失控故障写入数据库 if not df_diag_res.empty: df_diag_res.columns = ['start_time', 'end_time', 'product_id', 'code', 'level', 'info', 'advice'] df_diag_res['factory'] = '骑享' df_diag_res.to_sql("all_fault_info",con=db_res_engine, if_exists="append",index=False) #当前热失控已超过三天变为历史故障并写入数据库，并删除原有数据库中的当前故障和ram中的当前故障 elif (now_time-df_bms_ram_sn.iloc[-1]['time']).total_seconds()>3*24*3600: df_diag_ram=df_diag_ram.drop(df_diag_ram['sn']==sn) #删除ram中的当前故障 df_bms_ram_sn.iloc[-1]['end_time']=now_time try: conn = pymysql.connect(host=host2, port=port2, user=user2, password=password2, database=db2) cursor = conn.cursor() cursor.execute(''' update all_fault_info set end_time={} where product_id='{}' and code={} and factory='骑享' '''.format(now_time, sn, 119)) conn.commit() conn.close(); except: logger.error(traceback.format_exc) logger.error(u"{} ：{},{} 任务运行错误\n".format(sn,start_time,end_time), exc_info=True) conn.close(); except: logger.error(traceback.format_exc) logger.error(u"{} ：{},{} 任务运行错误\n".format(sn,start_time,end_time), exc_info=True) logger.info("pid-{} Done!".format(os.getpid())) #...................................................主进程........................................................................................................... def mainprocess(): global SNnums global df_diag_ram global df_bms_ram global log_path # 更新sn列表 host='rm-bp10j10qy42bzy0q7.mysql.rds.aliyuncs.com' port=3306 db='qixiang_oss' user='qixiang_oss' password='Qixiang2021' conn = pymysql.connect(host=host, port=port, user=user, password=password, database=db) cursor = conn.cursor() cursor.execute("select sn, imei, add_time from app_device") res = cursor.fetchall() df_sn = pd.DataFrame(res, columns=['sn', 'imei', 'add_time']) df_sn = df_sn.reset_index(drop=True) conn.close(); process = 2 pool = multiprocessing.Pool(processes = process) for i in range(process): sn_list = df_sn[int(len(df_sn)*i/process):int(len(df_sn)*(i+1)/process)] sn_list = sn_list.reset_index(drop=True) log_name = log_path + '/log_' + str(i) pool.apply_async(diag_cal, (sn_list,df_diag_ram,df_bms_ram,log_name)) pool.close() pool.join() if __name__ == "__main__": # 时间设置 # now_time = datetime.datetime.now() # pre_time = now_time + dateutil.relativedelta.relativedelta(days=-1)# 前一日 # end_time=datetime.datetime.strftime(now_time,"%Y-%m-%d 00:00:00") # start_time=datetime.datetime.strftime(pre_time,"%Y-%m-%d 00:00:00") history_run_flag = False # 历史数据运行标志 # # 更新sn列表 # host='rm-bp10j10qy42bzy0q7.mysql.rds.aliyuncs.com' # port=3306 # db='qixiang_oss' # user='qixiang_oss' # password='Qixiang2021' # conn = pymysql.connect(host=host, port=port, user=user, password=password, database=db) # cursor = conn.cursor() # cursor.execute("select sn, imei, add_time from app_device") # res = cursor.fetchall() # df_sn = pd.DataFrame(res, columns=['sn', 'imei', 'add_time']) # df_sn = df_sn.reset_index(drop=True) # conn.close(); # 数据库配置 host = 'rm-bp10j10qy42bzy0q77o.mysql.rds.aliyuncs.com' port = 3306 user = 'qx_cas' password = parse.quote_plus('Qx@123456') database = 'qx_cas' db_engine = create_engine( "mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8".format( user, password, host, port, database )) DbSession = sessionmaker(bind=db_engine) # 运行历史数据配置 df_first_data_time = pd.read_sql("select * from bat_first_data_time", db_engine) # 日志配置 now_str = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()).replace(":","_") log_path = 'log/' + now_str if not os.path.exists(log_path): os.makedirs(log_path) log = Log.Mylog(log_name='batsafetyAlarm', log_level = 'info') log.set_file_hl(file_name='{}/info.log'.format(log_path), log_level='info', size=1024* 1024 * 100) log.set_file_hl(file_name='{}/error.log'.format(log_path), log_level='error', size=1024* 1024 * 100) logger = log.get_logger() logger.info("pid is {}".format(os.getpid())) # 算法参数 host='rm-bp10j10qy42bzy0q77o.mysql.rds.aliyuncs.com' port=3306 db='safety_platform' user='qx_read' password=parse.quote_plus('Qx@123456') tablename='all_fault_info' db_res_engine = create_engine( "mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8".format( user, password, host, port, db )) #............................模块运行前，先读取数据库中所有结束时间为0的数据，需要从数据库中读取................ # print("select start_time, end_time, product_id, code, level, info, advice, factory from {}".format(tablename)) 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) result = result[['start_time', 'end_time', 'product_id', 'code', 'level', 'info', 'advice']] df_diag_ram=result[(result['end_time']=='0000-00-00 00:00:00') & (result['code']==119)] df_bms_ram=pd.DataFrame(columns=['time', 'sn', 'packvolt', 'cellvolt', 'celltemp']) #定时任务....................................................................................................................................................................... scheduler = BlockingScheduler() scheduler.add_job(mainprocess, 'interval', seconds=60, id='diag_job') try: scheduler.start() except Exception as e: scheduler.shutdown() logger.error(str(e))