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

import pandas as pd
import numpy as np
import pymysql
from LIB.BACKEND import DBManager, Log
from apscheduler.schedulers.blocking import BlockingScheduler
import time, 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 now_time
    global df_data
    
    start=time.time()

    mylog=log.Mylog('log_warning.txt','error')
    mylog.logcfg()

    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'

    #读取故障结果库中code==110且end_time='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='start_time,end_time,product_id,code,level,info,advice'
    # tablename='all_fault_info'
    # sql =  "select %s from %s where code=110 and end_time='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:
        try:
            if 'KPD' in sn:
                celltype=1 
            elif 'LS' in sn:
                celltype=99 
            else:
                celltype=2


            #读取原始数据库数据........................................................................................................................................................
            df_bms= df_data[(df_data['time']>=start_time) & (df_data['time']<end_time)]
            df_bms=df_bms.reset_index(drop=True)
            if not df_bms.empty:
                cellvolt_name=['CellVolt'+str(x) for x in range(1,73)]
                celltemp_name=['CellTemp'+str(x) for x in range(1,33)]
                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)
        
            #读取结果数据库数据........................................................................................................................................................
            # db='qx_cas'
            # mode=1
            # DBRead=DBDownload.DBDownload(host, port, db, user, password,mode)
            # with DBRead as DBRead:
            #     df_soh=DBRead.getdata('time_st,sn,soh,cellsoh', tablename='cellstateestimation_soh', sn=sn, timename='time_sp', st=start_time, sp=end_time)
            #     df_uniform=DBRead.getdata('time,sn,cellsoc_diff,cellvolt_diff,cellmin_num,cellmax_num', tablename='cellstateestimation_uniform_socvoltdiff', sn=sn, timename='time', st=start_time1, sp=end_time)
            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())
                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())

                #内短路计算..................................................................................................................................................
                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:\00WorkSpace\01Python\data_analyze_platform\USER\alibaba\07BatSafetyWarning\热失控预警.txt','a',encoding="utf-8") as file:
                        file.write(str(tuple(df_warning_res.iloc[-1]))+'\n')
            
            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
                    with open(r'D:\00WorkSpace\01Python\data_analyze_platform\USER\alibaba\07BatSafetyWarning\热失控预警.txt','a',encoding="utf-8") as file:
                        file.write(str(tuple(df_warning_res.iloc[-1]))+'\n')

            end=time.time()
            print(end-start)

        except Exception as e:
            print(repr(e))
            mylog.logopt(e) 

#...............................................主函数起定时作用.......................................................................................................................
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=['LS002']
    
    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()

    now_time=datetime.datetime.strptime('2022-4-26 18:00:00','%Y-%m-%d %H:%M:%S')
    df_data= pd.read_excel(r'D:\01WorkSpace\项目\Alibaba\data\原始数据_2022-01-10_2022-02-10\力神_LS002.xlsx')
    df_data['time']=pd.to_datetime(df_data['time'], format='%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)