__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.BatSafetyWarning.V1_0_1 import CBMSSafetyWarning from LIB.MIDDLE.CellStateEstimation.BatSafetyWarning.V1_0_1 import CBMSBatInterShort from LIB.MIDDLE.CellStateEstimation.BatSafetyWarning.V1_0_1 import CBMSBatUniform from LIB.MIDDLE.CellStateEstimation.BatSafetyWarning.V1_0_1 import VoltStray from urllib import parse import pymysql from apscheduler.schedulers.blocking import BlockingScheduler #...................................电池包电芯安全诊断函数...................................................................................................................... def saftywarning_cal(): global df_sn global df_warning_ram global df_warning_ram1 global df_warning_ram2 global df_warning_ram3 global df_lfp_ram global df_lfp_ram1 # 读取结果数据库 host2='rm-bp10j10qy42bzy0q77o.mysql.rds.aliyuncs.com' port2=3306 db2='safety_platform' user2='qx_algo_rw' password2='qx@123456' db_safety_platform_engine = create_engine( "mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8".format( user2, parse.quote_plus(password2), host2, port2, db2 )) conn_platform = pymysql.connect(host=host2, port=port2, user=user2, password=password2, database=db2) cursor_platform = conn_platform.cursor() host='rm-bp10j10qy42bzy0q77o.mysql.rds.aliyuncs.com' port=3306 db='qx_cas' user='qx_algo_rw' password='qx@123456' db_qxcas_engine = create_engine( "mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8".format( user, parse.quote_plus(password), host, port, db )) conn_cas = pymysql.connect(host=host, port=port, user=user, password=password, database=db) cursor_cas = conn_cas.cursor() result=pd.read_sql("select start_time, end_time, product_id, code, level, info, advice from all_fault_info_copy where factory = '{}'".format('骑享'), db_safety_platform_engine) result = result[['start_time', 'end_time', 'product_id', 'code', 'level', 'info', 'advice']] df_fault_ram=result[(result['end_time']=='0000-00-00 00:00:00') & (result['code']==110)] now_time=datetime.datetime.now() start_time=now_time-datetime.timedelta(hours=6) start_time1=now_time-datetime.timedelta(days=7) start_time2=now_time-datetime.timedelta(days=3) start_time3=now_time-datetime.timedelta(days=1) start_time=start_time.strftime('%Y-%m-%d %H:%M:%S') start_time1=start_time1.strftime('%Y-%m-%d %H:%M:%S') start_time2=start_time2.strftime('%Y-%m-%d %H:%M:%S') start_time3=start_time3.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)) # 读取soh数据 df_soh = pd.read_sql("select time_st,sn,soh,cellsoh from cellstateestimation_soh where sn = '{}' order by id desc limit 1".format(sn), db_qxcas_engine) df_uniform = pd.read_sql("select time,sn,cellsoc_diff,cellvolt_diff,cellmin_num,cellmax_num from cellstateestimation_uniform_socvoltdiff where sn = '{}' order by id desc limit 1".format(sn), db_qxcas_engine) df_uniform.loc[0,'cellvolt_rank']=str([1]*20) df_voltsigma=pd.DataFrame() #读取原始数据库数据................................................................................id........................................................................ 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_warning_ram_sn=df_warning_ram[df_warning_ram['sn']==sn] df_warning_ram_sn1=df_warning_ram1[df_warning_ram1['sn']==sn] df_warning_ram_sn2=df_warning_ram2[df_warning_ram2['sn']==sn] df_warning_ram_sn3=df_warning_ram3[df_warning_ram3['sn']==sn] df_warning_ram_sn.reset_index(inplace=True,drop=True) #重置索引 df_warning_ram_sn1.reset_index(inplace=True,drop=True) #重置索引 df_warning_ram_sn2.reset_index(inplace=True,drop=True) #重置索引 df_warning_ram_sn3.reset_index(inplace=True,drop=True) #重置索引 if celltype>50 and (not df_lfp_ram.empty): df_lfp_ram_sn=df_lfp_ram[df_lfp_ram['sn']==sn] df_lfp_ram_sn.reset_index(inplace=True,drop=True) #重置索引 else: df_lfp_ram_sn=pd.DataFrame() df_lfp_ram=pd.DataFrame(columns=df_bms.columns.tolist()+['sn']) if celltype>50 and (not df_lfp_ram1.empty): df_lfp_ram_sn1=df_lfp_ram1[df_lfp_ram1['sn']==sn] df_lfp_ram_sn1.reset_index(inplace=True,drop=True) #重置索引 else: df_lfp_ram_sn1=pd.DataFrame() df_lfp_ram1=pd.DataFrame(columns=df_bms.columns.tolist()+['sn']) #内短路计算.................................................................................................................................................. BatShort=CBMSBatInterShort.BatInterShort(sn,celltype,df_bms,df_soh,df_warning_ram_sn,df_warning_ram_sn1,df_warning_ram_sn2,df_warning_ram_sn3,df_lfp_ram_sn) df_short_res, df_ram_res, df_ram_res1, df_ram_res2, df_ram_res3, df_ram_res4=BatShort.intershort() if not df_short_res.empty: df_short_res.columns = ['time_st', 'time_sp', 'sn', 'method', 'short_current', 'baltime'] df_short_res.to_sql("cellStateEstimation_interShort",con=db_qxcas_engine, if_exists="append",index=False) #静置电压排名.................................................................................................................................................. BatUniform=CBMSBatUniform.BatUniform(sn,celltype,df_bms,df_uniform,df_ram_res3,df_lfp_ram_sn1) df_rank_res, df_ram_res3, df_ram_res5=BatUniform.batuniform() if not df_rank_res.empty: df_uniform=df_rank_res df_rank_res.to_sql("cellstateestimation_uniform_socvoltdiff",con=db_qxcas_engine, if_exists="append",index=False) #电压离群..................................................................................................................................................... df_voltsigma=VoltStray.main(sn,df_bms,celltype) if not df_voltsigma.empty: df_voltsigma.to_sql("outlier_voltchangeratio",con=db_qxcas_engine, if_exists="append",index=False) #内短路ram处理......................................................... df_warning_ram=df_warning_ram.drop(df_warning_ram[df_warning_ram.sn==sn].index) df_warning_ram1=df_warning_ram1.drop(df_warning_ram1[df_warning_ram1.sn==sn].index) df_warning_ram2=df_warning_ram2.drop(df_warning_ram2[df_warning_ram2.sn==sn].index) df_warning_ram3=df_warning_ram3.drop(df_warning_ram3[df_warning_ram3.sn==sn].index) df_warning_ram=pd.concat([df_warning_ram,df_ram_res],ignore_index=True) df_warning_ram1=pd.concat([df_warning_ram1,df_ram_res1],ignore_index=True) df_warning_ram2=pd.concat([df_warning_ram2,df_ram_res2],ignore_index=True) df_warning_ram3=pd.concat([df_warning_ram3,df_ram_res3],ignore_index=True) if celltype>50: df_lfp_ram=df_lfp_ram.drop(df_lfp_ram[df_lfp_ram.sn==sn].index) df_lfp_ram=pd.concat([df_lfp_ram,df_ram_res4],ignore_index=True) df_lfp_ram1=df_lfp_ram1.drop(df_lfp_ram1[df_lfp_ram1.sn==sn].index) df_lfp_ram1=pd.concat([df_lfp_ram1,df_ram_res5],ignore_index=True) #电池热安全预警.............................................................................................................................................................. #读取内短路、析锂和一致性结果数据库数据 df_short = pd.read_sql("select time_sp,sn,short_current from cellstateestimation_intershort where sn = '{}' and time_sp between '{}' and '{}'".format(sn,start_time1,end_time), db_qxcas_engine) # df_liplated = pd.read_sql("select time,sn,liplated,liplated_amount from mechanism_liplated where sn = '{}' and time between '{}' and '{}'".format(sn,start_time2,end_time), db_qxcas_engine) df_uniform = pd.read_sql("select time,sn,cellsoc_diff,cellvolt_diff,cellmin_num,cellmax_num,cellvolt_rank from cellstateestimation_uniform_socvoltdiff where sn = '{}' and time between '{}' and '{}'".format(sn,start_time3,end_time), db_qxcas_engine) df_voltsigma = pd.read_sql("select time,sn,VolOl_Uni,VolChng_Uni from outlier_voltchangeratio where sn = '{}' and time between '{}' and '{}'".format(sn,start_time3,end_time), db_qxcas_engine) #获取sn的故障RAM df_fault_ram_sn=df_fault_ram[df_fault_ram['product_id']==sn] #热安全预警 if df_fault_ram_sn.empty: BatWarning=CBMSSafetyWarning.SafetyWarning(sn,celltype,df_short,df_uniform,df_voltsigma,df_soh) df_warning_res=BatWarning.diag() #当前热失控故障写入数据库 if not df_warning_res.empty: df_warning_res['add_time'] = datetime.datetime.now() df_warning_res['factory'] = '骑享' df_warning_res.to_sql("all_fault_info_copy",con=db_safety_platform_engine, if_exists="append",index=False) conn_platform.commit() else: fault_time=datetime.datetime.strptime(df_fault_ram_sn.iloc[-1]['start_time'], '%Y-%m-%d %H:%M:%S') if (now_time-fault_time).total_seconds()>3*24*3600: #df_warning_end历史故障筛选并更改数据库故障结束时间 df_fault_ram_sn['end_time']=end_time df_fault_ram_sn['Batpos']=1 try: cursor.execute(''' update all_fault_info_copy set update_time='{}',end_time='{}', Batpos={} where product_id='{}' and code={} and end_time='0000-00-00 00:00:00' and factory='骑享' '''.format(datetime.datetime.now(), df_fault_ram_sn.loc[0,'end_time'], df_fault_ram_sn.loc[0,'Batpos'],sn, 110)) conn_platform.commit() except: logger.error(traceback.format_exc) logger.error(u"{} ：{},{} 任务运行错误\n".format(sn,start_time,end_time), exc_info=True) except: logger.error(traceback.format_exc) logger.error(u"{} ：{},{} 任务运行错误\n".format(sn,start_time,end_time), exc_info=True) cursor_platform.close() conn_platform.close() cursor_cas.close() conn_cas.close() db_qxcas_engine.dispose() db_safety_platform_engine.dispose() logger.info("pid-{} Done!".format(os.getpid())) #...................................................主进程........................................................................................................... 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_test/' + now_str if not os.path.exists(log_path): os.makedirs(log_path) log = Log.Mylog(log_name='batsafetyWarning', 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_copy' db_res_engine = create_engine( "mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8".format( user, password, host, port, db )) #............................模块运行前，先读取数据库中所有结束时间为0的数据，需要从数据库中读取................ df_warning_ram=pd.DataFrame(columns=['sn','time','deltsoc','cellsoc']) df_warning_ram1=pd.DataFrame(columns=['sn','time1','deltsoc1']) df_warning_ram2=pd.DataFrame(columns=['sn','time2','deltAs2']) df_warning_ram3=pd.DataFrame(columns=['sn','time3','standingtime','standingtime1','standingtime2']) df_lfp_ram=pd.DataFrame() df_lfp_ram1=pd.DataFrame() # now_time='2021-10-15 00:00:00' # now_time=datetime.datetime.strptime(now_time,'%Y-%m-%d %H:%M:%S') saftywarning_cal() #定时任务....................................................................................................................................................................... scheduler = BlockingScheduler() scheduler.add_job(saftywarning_cal, 'interval', hours=6, id='saftywarning_cal') try: scheduler.start() except Exception as e: scheduler.shutdown() logger.error(str(e))