data_analyze_platform/USER/SPF/alibaba/07BatSafetyWarning/main.py

import pandas as pd
import numpy as np
import pymysql
from LIB.BACKEND import DBManager, Log
from apscheduler.schedulers.blocking import BlockingScheduler
import datetime
from USER.SPF.alibaba.Common import DBDownload
from USER.SPF.alibaba.Common import log
import CBMSBatInterShort
import CBMSBatUniform
import VoltStray
import CBMSSafetyWarning

#电池热安全预警核心算法函数......................................................................................................................
def saftywarning_cal():
    global SNnums
    global df_warning_ram
    global df_warning_ram1
    global df_warning_ram2
    global df_warning_ram3
    global df_lfp_ram
    global df_lfp_ram1
    global df_data,now_time
    

    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')
    now_time=now_time+datetime.timedelta(hours=6)

    #数据库配置
    host='rm-bp10j10qy42bzy0q77o.mysql.rds.aliyuncs.com'
    port=3306
    user='qx_read'
    password='Qx@123456'

    #读取故障结果库中faultcode==110且time_sp='0000-00-00 00:00:00'...............................
    # db='safety_platform'
    # mysql = pymysql.connect (host=host, user=user, password=password, port=port, database=db)
    # cursor = mysql.cursor()
    # param='time_st,time_sp,sn,faultcode,faultlv,faultinfo,faultadvice'
    # tablename='all_fault_info'
    # sql =  "select %s from %s where faultcode=110 and time_sp='0000-00-00 00:00:00'" %(param,tablename)
    # cursor.execute(sql)
    # res = cursor.fetchall()
    # df_fault_ram= pd.DataFrame(res,columns=param.split(','))
    # cursor.close()
    # mysql.close()
    df_fault_ram=pd.DataFrame(columns=['time_st','time_sp','sn','faultcode','faultlv','faultinfo','faultadvice'])

    for sn in SNnums:
        if 'PK500' in sn:
            celltype=1 #6040三元电芯
        elif 'PK502' in sn:
            celltype=2 #4840三元电芯
        elif 'K504B' in sn:
            celltype=99    #60ah林磷酸铁锂电芯
        elif 'MGMLXN750' in sn:
            celltype=3 #力信50ah三元电芯
        elif 'MGMCLN750' or 'UD' in sn: 
            celltype=4 #CATL 50ah三元电芯
        else:
            print('SN：{}，未找到对应电池类型！！！'.format(sn))
            continue
            # sys.exit()
        celltype=1

        #读取原始数据库数据........................................................................................................................................................
        df_bms= df_data[(df_data['time']>=start_time) & (df_data['time']<end_time)]
        df_bms=df_bms.reset_index(drop=True)
        cellvolt_name=['CellVolt'+str(x) for x in range(1,121)]
        celltemp_name=['CellTemp'+str(x) for x in range(1,41)]
        df_volt=pd.DataFrame([x[0].split(",") for x in np.array(df_bms[['cell_volt']])])
        df_volt.columns = cellvolt_name
        df_volt=df_volt.astype('float')
        df_temp=pd.DataFrame([x[0].split(",") for x in np.array(df_bms[['cell_temp']])])
        df_temp.columns = celltemp_name
        df_temp=df_temp.astype('float')
        df_bms=pd.concat([df_bms,df_volt,df_temp],axis=1)
        df_bms.dropna(axis=0, inplace=True)
        df_bms=df_bms.reset_index(drop=True)

        #读取结果数据库数据........................................................................................................................................................
        df_soh=pd.DataFrame()
        df_uniform=pd.DataFrame()

        if not df_bms.empty:
            #ram处理...............................................................................................................
            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:
                with open(r'D:\01WorkSpace\python\data_analyze_platform\USER\SPF\alibaba\07BatSafetyWarning\内短路.txt','a') as file:
                    file.write(str(df_short_res)+'\n')
            
            #静置电压排名..................................................................................................................................................
            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
                with open(r'D:\01WorkSpace\python\data_analyze_platform\USER\SPF\alibaba\07BatSafetyWarning\电压排名.txt','a') as file:
                    file.write(str(df_rank_res)+'\n')
            
            #电压离群.....................................................................................................................................................
            df_voltsigma=VoltStray.main(sn,df_bms,celltype)
            if not df_voltsigma.empty:
                with open(r'D:\01WorkSpace\python\data_analyze_platform\USER\SPF\alibaba\07BatSafetyWarning\电压离群.txt','a') as file:
                    file.write(str(df_voltsigma)+'\n')

            #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)
            

        #电池热安全预警..............................................................................................................................................................
        #读取内短路、析锂和一致性结果数据库数据
        db='qx_cas'
        mode=2
        tablename1='cellstateestimation_intershort'
        tablename3='cellstateestimation_uniform_socvoltdiff'
        tablename4='outlier_voltchangeratio'   #电压离群表单
        DBRead=DBDownload.DBDownload(host, port, db, user, password,mode)
        with DBRead as DBRead:
            df_short=DBRead.getdata('time_sp,sn,short_current', tablename=tablename1, sn=sn, timename='time_sp', st=start_time1, sp=end_time)
            # df_liplated=DBRead.getdata('time,sn,liplated,liplated_amount', tablename=tablename2, sn=sn, timename='time', st=start_time2, sp=end_time)
            df_uniform=DBRead.getdata('time,sn,cellsoc_diff,cellvolt_diff,cellmin_num,cellmax_num,cellvolt_rank', tablename=tablename3, sn=sn, timename='time', st=start_time2, sp=end_time)
            df_voltsigma=DBRead.getdata('time,sn,VolOl_Uni,VolChng_Uni', tablename=tablename4, sn=sn, timename='time', st=start_time3, sp=end_time)
            df_uniform=df_uniform.dropna(axis=0,how='any')
        
        #获取sn的故障RAM
        df_fault_ram_sn=df_fault_ram[df_fault_ram['sn']==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:
                with open(r'D:\Platform\platform_python\data_analyze_platform\USER\spf\01qixiang\07BatSafetyWarning\热失控预警.txt','a',encoding="utf-8") as file:
                    file.write(str(tuple(df_warning_res.iloc[-1]))+'\n')
        
        else:
            fault_time=df_fault_ram_sn.iloc[-1]['time_st']
            if (now_time-fault_time).total_seconds()>7*24*3600:   #df_warning_end历史故障筛选并更改数据库故障结束时间
                df_fault_ram_sn['time_st']=end_time
                with open(r'D:\Platform\platform_python\data_analyze_platform\USER\spf\01qixiang\07BatSafetyWarning\热失控预警.txt','a',encoding="utf-8") as file:
                    file.write(str(tuple(df_warning_res.iloc[-1]))+'\n')

#...............................................主函数起定时作用.......................................................................................................................
if __name__ == "__main__":
    
    # excelpath=r'D:\Platform\platform_python\data_analyze_platform\USER\spf\01qixiang\sn-20210903.xlsx'
    # SNdata_6060 = pd.read_excel(excelpath, sheet_name='科易6060')
    # SNdata_6040 = pd.read_excel(excelpath, sheet_name='科易6040')
    # SNdata_4840 = pd.read_excel(excelpath, sheet_name='科易4840')
    # SNdata_L7255 = pd.read_excel(excelpath, sheet_name='格林美-力信7255')
    # SNdata_C7255 = pd.read_excel(excelpath, sheet_name='格林美-CATL7255')
    # SNdata_U7255 = pd.read_excel(excelpath, sheet_name='优旦7255')
    # SNnums_6060=SNdata_6060['SN号'].tolist()
    # SNnums_6040=SNdata_6040['SN号'].tolist()
    # SNnums_4840=SNdata_4840['SN号'].tolist()
    # SNnums_L7255=SNdata_L7255['SN号'].tolist()
    # SNnums_C7255=SNdata_C7255['SN号'].tolist()
    # SNnums_U7255=SNdata_U7255['SN号'].tolist()
    # SNnums=SNnums_L7255 + SNnums_C7255 + SNnums_U7255 + SNnums_6040 + SNnums_4840 + SNnums_6060
    SNnums=['KPD001']
    
    mylog=log.Mylog('log_warning.txt','error')
    mylog.logcfg()

    #............................模块运行前，先读取数据库中所有结束时间为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()

    df_data= pd.read_excel(r'D:\01WorkSpace\项目\Alibaba\data\原始数据_2022-01-10_2022-02-10\康普盾_KPD001.xlsx')

    now_time='2022-1-10 00:00:00'
    now_time=datetime.datetime.strptime(now_time,'%Y-%m-%d %H:%M:%S')

    #定时任务.......................................................................................................................................................................
    scheduler = BlockingScheduler()
    scheduler.add_job(saftywarning_cal, 'interval', seconds=5)

    try:  
        scheduler.start()
    except Exception as e:
        scheduler.shutdown()
        print(repr(e))
        mylog.logopt(e)