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