data_analyze_platform/USER/SPF/01qixiang/06BatSafetyAlarm/deploy - 副本.py

__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
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_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_algo_rw'
    password2='qx@123456'
    
    db_res_engine = create_engine(
        "mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8".format(
            user2, parse.quote_plus(password2), host2, port2, db2
        ))
    
    conn = pymysql.connect(host=host2, port=port2, user=user2, password=password2, database=db2)
    cursor = conn.cursor()
    
    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'])
    
    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:
            sn = df_sn.loc[i, 'sn']
            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
            
            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)

            #电池诊断................................................................................................................................................................
            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:   
                if not df_bms.empty:
                    SafetyAlarm=CBMSSafetyAlarm.SafetyAlarm(sn,celltype,df_bms, df_bms_ram_sn)
                    df_diag_res, df_bms_res=SafetyAlarm.diag() 

                    #更新bms的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['add_time'] = datetime.datetime.now()
                        df_diag_res['factory'] = '骑享'
                        df_diag_res.to_sql("all_fault_info",con=db_res_engine, if_exists="append",index=False)

                    
                #当前热失控已超过1天变为历史故障并写入数据库，并删除原有数据库中的当前故障和ram中的当前故障
            else:
                fault_time=datetime.datetime.strptime(df_diag_ram_sn.iloc[-1]['start_time'], '%Y-%m-%d %H:%M:%S')
                if (now_time-fault_time).total_seconds()>24*3600:
                    df_diag_ram_sn['end_time']=end_time
                    df_diag_ram_sn['Batpos']=1
                
                    try:

                        cursor.execute('''
                                    update all_fault_info set update_time='{}',end_time='{}', Batpos={} where product_id='{}' and end_time='0000-00-00 00:00:00' and code={} and factory='骑享'
                                    '''.format(datetime.datetime.now(), end_time, 1,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)
    cursor.close()
    conn.close()
    db_res_engine.dispose()
    logger.info("pid-{} Done!".format(os.getpid()))

#...................................................主进程...........................................................................................................
def mainprocess():
    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)
    # cursor.close()
    # conn.close();
    df_sn=pd.DataFrame(columns=['sn'])
    df_sn['sn']=['PK504B10100004434', 'MGMCLN750N215N205', 'PK504B10100004447', 'PK504B10100004379', 'PK504B10100004483']
    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_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
        ))

        
    #ram参数初始化........................................................................................................
    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))