data_analyze_platform/LIB/MIDDLE/PingFeng/main.py

__author__ = 'ERIC'
#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 soc
import time, datetime
import traceback
from LIB.MIDDLE.CellStateEstimation.Common import log

from urllib import parse
import pymysql
import pdb
from apscheduler.schedulers.blocking import BlockingScheduler
import datacompy
import logging
import multiprocessing
#...................................电池包SOC计算......................................................................................................................

def SOC_cal():

    start=time.time()
    now_time=datetime.datetime.now()
    start_time=now_time-datetime.timedelta(seconds=300)
    start_time=start_time.strftime('%Y-%m-%d %H:%M:%S')
    end_time=now_time.strftime('%Y-%m-%d %H:%M:%S')

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


    db_res_engine = create_engine(
        "mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8".format(
            user, parse.quote_plus(password), host, port, db
        ))


    for i in range(0, len(df_sn)):
        factory = "骑享"
        sn = df_sn.loc[i, '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三元电芯
            elif df_sn.loc[i, 'imei'][3:9] == 'CLL128':    
                celltype=100 # 重卡
                factory = "金茂换电"
            elif df_sn.loc[i, 'imei'][0:6] == 'SPFPFL':   
                celltype=200#储能电站
                factory = "平峰"
            else:
                logger.info("pid-{} celltype-{} SN: {} SKIP!".format(os.getpid(), "未知", 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']


            #电池诊断................................................................................................................................................................
            if not df_bms.empty:
                SOC_result=soc.soc_test(sn,df_bms)
                SOC_result.columns=['sn','soc','time']
                SOC_result.reset_index(inplace=True,drop=True) 
                print(SOC_result)
                    #新增故障筛选并存入数据库.....................................................................
                if not SOC_result.empty:  #新增写入数据库
                    SOC_result.to_sql('pingfeng_soc_cal',con=db_res_engine, if_exists='append',index=False)

            end=time.time()
            print(end-start)  
            
        except Exception as e:
            logger.warning(e)
            logger.error(traceback.format_exc)
            logger.error(u"{} ：{},{} 任务运行错误\n".format(sn,start_time,end_time), exc_info=True)


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 # 历史数据运行标志
    
    def get_sn():
        global df_sn
        # # 更新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 where status in (1,2,3)")
        res = cursor.fetchall()
        df_sn = pd.DataFrame(res, columns=['sn', 'imei', 'add_time'])
        df_sn = df_sn[df_sn['sn'].str.contains('SPFPFL')]
        #df_sn = df_sn[df_sn['sn'].str.contains('PK500A20100000941')]
        
        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='batDiag', 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
    #     ))

        
    #............................模块运行前，先读取数据库中所有结束时间为0的数据，需要从数据库中读取................
    # print("select start_time, end_time, product_id, code, level, info, advice, factory from {}".format(tablename))

    # 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'])
    get_sn()
    SOC_cal()
    #定时任务.......................................................................................................................................................................
    scheduler = BlockingScheduler()
    scheduler.add_job(get_sn, 'interval', days=1, id='get_sn')
    scheduler.add_job(SOC_cal, 'interval', seconds=300, id='SOC_cal')
    
    try:  
        scheduler.start()
    except Exception as e:
        scheduler.shutdown()
        logger.error(str(e))