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

import os
import pandas as pd
from sqlalchemy import create_engine
import logging.handlers
import traceback
import re
import CBMSBatInterShort
import CBMSBatUniform
import VoltStray
import CBMSSafetyWarning
import datetime
import dateutil.relativedelta
from urllib import parse
from apscheduler.schedulers.blocking import BlockingScheduler
import warnings
import pymysql
from multiprocessing import Process
import time


#电池热安全预警核心算法函数
def fun(host, port, db, user, password, runenv, logger, period_second):
    logger.info("pid is {}".format(os.getpid()))
    global df_warning_ram
    global df_warning_ram1
    global df_warning_ram2
    global df_warning_ram3
    global df_lfp_ram
    global df_lfp_ram1
    try:
        db_engine = create_engine(
            "mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8".format(
                user, parse.quote_plus(password), host, port, db
            ))

        conn = pymysql.connect(host=host, port=int(port), user=user, password=password, database=db)
        cursor = conn.cursor()
    except:
        logger.error(u"数据库连接错误", exc_info=True)
        logger.error(traceback.format_exc)
        return

    now_time=datetime.datetime.now()
    start_time=now_time-datetime.timedelta(seconds=6*3600) #6*3600
    start_time1=now_time-datetime.timedelta(seconds=7*24*3600) #7*24*3600
    start_time2=now_time-datetime.timedelta(seconds=3*24*3600) #3*24*3600
    start_time3=now_time-datetime.timedelta(seconds=1*24*3600) #1*24*3600
    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 = datetime.datetime.now()
    # pre_time = now_time + dateutil.relativedelta.relativedelta(seconds=-period_second)  #1min
    # end_time = datetime.datetime.strftime(now_time, "%Y-%m-%d %H:%M:%S")
    # start_time = datetime.datetime.strftime(pre_time, "%Y-%m-%d %H:%M:%S")
    # 获取配置信息
    df_confs = pd.read_sql('select * from conf where status=1 order by factory,device_name', db_engine)
    # db_engine.dispose()

    # 获取历史报警数据
    df_fault_ram = pd.read_sql("select time_st,time_sp,sn,faultcode,faultlv,faultinfo,faultadvice from fault_result where faultcode = 110 and time_sp='0000-00-00 00:00:00'",db_engine);




    logger.info("time range from {} to {} ".format(start_time, end_time))
    for i in range(len(df_confs)):
        try:
            factory = df_confs.loc[i, 'factory']
            sn = df_confs.loc[i, 'device_name']
            if '康普盾' in factory:
                celltype = 1
                cell_volt_count = 120
                cell_temp_count = 40
            elif '华霆' in factory:
                celltype = 2
                cell_volt_count = 68
                cell_temp_count = 40
            elif '力神' in factory:
                celltype = 99
                cell_volt_count = 72
                cell_temp_count = 32

            logger.info("pid-{} FACTORY:{} - SN: {} START!".format(os.getpid(), factory, sn))
            sql = "select * from bms_data where factory ='{}' and sn = '{}' and time > '{}' and time <= '{}'".format(factory, sn, start_time, end_time)
            logger.info(sql)
            df_bms = pd.read_sql(sql, db_engine)
            if df_bms.empty:
                continue
            df_bms.columns = ['sn', 'factory', 'add_time', 'update_time', 'time', 'BMSStat', 'PackVolt', 'PackCrnt',
                              'PackSOC', 'PackSOH', 'alarm1', 'alarm2', 'alarm3', 'CellTemp',
                              'CellVolt', 'id']
            # 单体数据拆分
            cell_temp = df_bms['CellTemp']
            cell_volt = df_bms['CellVolt']
            cell_temps = []
            [cell_temps.append(list(map(float,x.split(',')))) for x in cell_temp]
            cell_volts = []
            [cell_volts.append(list(map(float,x.split(',')))) for x in cell_volt]
            del_indexes = []
            for i, cell_volt in enumerate(cell_volts):
                if len(cell_volt) != cell_volt_count:
                    del_indexes.append(i)
            for i, cell_temp in enumerate(cell_temps):
                if len(cell_temp) != cell_temp_count:
                    del_indexes.append(i)

            del_indexes = list(set(del_indexes))
            for i, del_index in enumerate(del_indexes):
                del_index = del_index - i
                cell_volts.pop(del_index)
                cell_temps.pop(del_index)
            df_bms = df_bms.drop(index=del_indexes, axis=1)
            cellvolt_name = ['CellVolt' + str(x) for x in range(1, cell_volt_count + 1)]
            celltemp_name = ['CellTemp' + str(x) for x in range(1, cell_temp_count + 1)]
            df_bms[cellvolt_name] = cell_volts
            df_bms[celltemp_name] = cell_temps
            df_bms = df_bms.drop(columns='CellVolt', axis=0)
            df_bms = df_bms.drop(columns='CellTemp', axis=0)
            df_bms = df_bms.reset_index(drop=True)
            logger.info("pid-{} FACTORY:{} - SN: {} 去除单体异常行{}数据!".format(os.getpid(), factory, sn, str(del_indexes)))

            # 算法执行
            df_soh = pd.read_sql(
                "select time_st,sn,soh,cellsoh from soh_result where factory ='{}' and sn = '{}' order by time_st desc limit 1".format(factory, sn),
                db_engine);
            df_uniform = pd.read_sql(
                "select time,sn,cellsoc_diff,cellmin_num,cellmax_num,cellvolt_rank from uniform_result where factory ='{}' and sn = '{}' order by time desc limit 1".format(factory, sn),
                db_engine);

            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:
                    df_short_res['add_time'] = datetime.datetime.now()
                    df_short_res['factory'] = factory
                    df_short_res.to_sql("intershort_result",con=db_engine, if_exists="append",index=False)
                    logger.info(u"{} intershort_result写入成功！！！\n".format(sn), exc_info=True)

                #!!!!!!!!!!!!!!往下还未进行部署修改

                #静置电压排名..................................................................................................................................................
                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_rank_res['add_time'] = datetime.datetime.now()
                    df_rank_res['factory'] = factory
                    df_rank_res.to_sql("uniform_result",con=db_engine, if_exists="append",index=False)
                    logger.info(u"{} uniform_result写入成功！！！\n".format(sn), exc_info=True)

                #电压离群.....................................................................................................................................................
                df_voltsigma=VoltStray.main(sn,df_bms,celltype)
                if not df_voltsigma.empty:
                    df_voltsigma['add_time'] = datetime.datetime.now()
                    df_voltsigma['factory'] = factory
                    df_voltsigma.to_sql("outlier_voltchangeratio_result",con=db_engine, if_exists="append",index=False)
                    logger.info(u"{} outlier_voltchangeratio_result写入成功！！！\n".format(sn), exc_info=True)

                #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 intershort_result where sn = '{}' and time_sp between '{}' and '{}'".format(sn,start_time1,end_time), db_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 uniform_result where sn = '{}' and time between '{}' and '{}'".format(sn,start_time2,end_time), db_engine)
            df_voltsigma = pd.read_sql("select time,sn,VolOl_Uni,VolChng_Uni from outlier_voltchangeratio_result where sn = '{}' and time between '{}' and '{}'".format(sn,start_time3,end_time), db_engine)
            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:
                    df_warning_res['add_time'] = datetime.datetime.now()
                    df_warning_res['factory'] = factory
                    df_warning_res.to_sql("fault_result",con=db_engine, if_exists="append",index=False)
                    logger.info(u"{} fault_result写入成功！！！\n".format(sn), exc_info=True)

            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_sp']=end_time
                    try:
                        cursor.execute('''
                                update fault_result set update_time='{}',time_sp='{}' where sn='{}' and faultcode={} and time_sp='0000-00-00 00:00:00'
                                '''.format(datetime.datetime.now(), end_time, sn, 110))
                        conn.commit()
                    except:
                        logger.error(traceback.format_exc)
                        logger.error(u"{} ：{},{} 任务运行错误\n".format(sn,start_time,end_time), exc_info=True)

            logger.info("pid-{} FACTORY:{} - SN: {} DONE!".format(os.getpid(), factory, sn))
        except:
            logger.error(u"pid-{} FACTORY:{} - SN: {} ERROR!\n".format(os.getpid(), factory, sn), exc_info=True)
            logger.error(traceback.format_exc)
    db_engine.dispose()
    logger.info("time range from {} to {} done!!!!!!!!! ".format(start_time, end_time))


def heart_beat(host, port, db, user, password, log):
    while True:
        try:
            db_engine = create_engine(
                "mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8".format(
                    user, parse.quote_plus(password), host, port, db
                ))
            while True:
                now = datetime.datetime.now()
                db_engine.execute("update status set algo_safetywarning='{}'".format(now))
                log.info("safetywarning心跳信号" + now.strftime("%Y-%m-%d %H:%M:%S"))
                time.sleep(5)
        except:
            log.error(u"safetywarning心跳错误", exc_info=True)
            log.error(traceback.format_exc)
            time.sleep(5)
        finally:
            db_engine.dispose()

if __name__ == '__main__':
    warnings.filterwarnings("ignore")
    env_dist = os.environ
    # 配置信息信息
    host = env_dist.get("ALI_HOST", '120.25.223.1')
    port = env_dist.get("ALI_PORT", '4901')
    db = env_dist.get("ALI_DB", 'ali')
    user = env_dist.get("ALI_ROOT", 'root')
    password = env_dist.get("ALI_PASSWORD", '123456')
    # host = env_dist.get("ALI_HOST", '192.168.31.141')
    # port = env_dist.get("ALI_PORT", '3306')
    # db = env_dist.get("ALI_DB", 'ali')
    # user = env_dist.get("ALI_ROOT", 'root')
    # password = env_dist.get("ALI_PASSWORD", 'Ali@123456')
    runenv = env_dist.get("ALI_RUNENV", 'dev')
    period_second = env_dist.get("ALI_PERIOD_SECOND", '5,5,5,5').split(',')
    period_second = env_dist.get("ALI_PERIOD_SECOND", '10')



    algo_name = 'safetywarning'
    # 日志配置
    log_path = './log' + '/' + algo_name
    if not os.path.exists(log_path):
        os.makedirs(log_path)
    logger = logging.getLogger()
    logger.setLevel(logging.DEBUG)

    fh = logging.handlers.RotatingFileHandler(filename='{}/info.log'.format(log_path), maxBytes=1024 * 1024 * 1024, backupCount=5, encoding="utf-8", mode="a")
    formatter = logging.Formatter("%(asctime)s - %(name)s-%(levelname)s %(message)s")
    fh.suffix = "%Y-%m-%d_%H-%M.log"
    fh.extMatch = re.compile(r"^\d{4}-\d{2}-\d{2}_\d{2}-\d{2}.log$")
    fh.setFormatter(formatter)
    fh.setLevel(logging.DEBUG)
    logger.addHandler(fh)

    fh = logging.handlers.RotatingFileHandler(filename='{}/error.log'.format(log_path), maxBytes=1024 * 1024 * 1024, backupCount=5, encoding="utf-8", mode="a")
    formatter = logging.Formatter("%(asctime)s - %(name)s-%(levelname)s %(message)s")
    fh.setFormatter(formatter)
    fh.setLevel(logging.ERROR)
    logger.addHandler(fh)

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

    # 开启定时任务
    logger.info("算法初始化完毕，开始周期运行！！！！！！！！！！！！！！！！！！！！！")
    p = Process(target=heart_beat, args=(host, port, db, user, password, logger,))
    p.start()
    scheduler = BlockingScheduler()
    # heart_beat(host, port, db, user, password, logger)
    # scheduler.add_job(func=heart_beat, args=(host, port, db, user, password, logger), trigger='interval', seconds=5, max_instances=1, coalesce=True)
    fun(host, port, db, user, password, runenv, logger, period_second)
    scheduler.add_job(func=fun, args=(host, port, db, user, password, runenv, logger, period_second), trigger='interval', seconds=int(period_second),
                      max_instances=1, coalesce=True)


    try:
        scheduler.start()
    except Exception as e:
        scheduler.shutdown()
        logger.error(str(e))