update

LIB/FRONTEND/ALLRECORD/record.md

@@ -0,0 +1,7 @@
+实时运行程序：
+1、故障码转换
+2、里程实时计算
+3、低电量报警
+4、cal_DailyMileage
+5、cal_batsafetyalarm
+6、cal_safecenter_DataSta

LIB/FRONTEND/SaftyCenter/DataSta/deploy.py

@@ -44,34 +44,33 @@ def Week_Task():
     "mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8".format(
         user, parse.quote_plus(password), host, port, db
     ))
-    
-    DBRead=DBDownload.DBDownload(host, port, db, user, password,mode)
-    with DBRead as DBRead:
-        df_fltinfo=DBRead.getdata('product_id','level','code','start_time',tablename=tablename1,factory='骑享',sn='',timename='',st='',sp='')#dbdownload经过了改编
-    #............................获取时间................................      
-    end_time=datetime.datetime.now()
-    # end_time=datetime.datetime.strptime(end_time,'%Y-%m-%d')
-    start_time=end_time-datetime.timedelta(days=7)
-    start_time=start_time.strftime('%Y-%m-%d')
-    end_time=end_time.strftime('%Y-%m-%d')
-    FltAlarmInfo,Celltype=DataSta.SaftyWarningSta(CS_Data,df_fltinfo,start_time,end_time)
-    FaultLvlCount=DataSta.WeekInfoSta(df_fltinfo,start_time,end_time)
-    lvl1=FaultLvlCount[FaultLvlCount['level']==1]['product_id'].values
-    lvl2=FaultLvlCount[FaultLvlCount['level']==2]['product_id'].values
-    lvl3=FaultLvlCount[FaultLvlCount['level']==3]['product_id'].values
-    lvl4=FaultLvlCount[FaultLvlCount['level']==4]['product_id'].values
-    lvl5=FaultLvlCount[FaultLvlCount['level']==5]['product_id'].values
-    all_period_fault_info.loc[0,'factory']='骑享'
-    all_period_fault_info.loc[0,'week']=toweek
-    all_period_fault_info.loc[0,'level1_count']=lvl1
-    all_period_fault_info.loc[0,'level2_count']=lvl2
-    all_period_fault_info.loc[0,'level3_count']=lvl3
-    all_period_fault_info.loc[0,'level4_count']=lvl4
-    all_period_fault_info.loc[0,'level5_count']=lvl5
-    all_period_fault_info.loc[0,'solve_rate']=FltAlarmInfo.loc[0,'OprationManageRate']
-    all_period_fault_info.to_sql('all_period_fault_info', db_engine, if_exists='append', index=False)
-    print(all_period_fault_info)
-    
+    try:
+        DBRead=DBDownload.DBDownload(host, port, db, user, password,mode)
+        with DBRead as DBRead:
+            df_fltinfo=DBRead.getdata('product_id','level','code','start_time',tablename=tablename1,factory='骑享',sn='',timename='',st='',sp='')#dbdownload经过了改编
+        #............................获取时间................................      
+        end_time=datetime.datetime.now()
+        # end_time=datetime.datetime.strptime(end_time,'%Y-%m-%d')
+        start_time=end_time-datetime.timedelta(days=7)
+        start_time=start_time.strftime('%Y-%m-%d')
+        end_time=end_time.strftime('%Y-%m-%d')
+        FltAlarmInfo,Celltype=DataSta.SaftyWarningSta(CS_Data,df_fltinfo,start_time,end_time)
+        FaultLvlCount=DataSta.WeekInfoSta(df_fltinfo,start_time,end_time)
+        for i in range(1,6):
+            if not FaultLvlCount[FaultLvlCount['level']==i]['product_id'].empty:
+                all_period_fault_info.loc[0,'level'+str(i)+'_count']=int(FaultLvlCount[FaultLvlCount['level']==i]['product_id'].values)
+            else:
+                all_period_fault_info.loc[0,'level'+str(i)+'_count']=int(0)
+        all_period_fault_info.loc[0,'factory']='骑享'
+        all_period_fault_info.loc[0,'week']=toweek
+
+        all_period_fault_info.loc[0,'solve_rate']=FltAlarmInfo.loc[0,'OprationManageRate']
+        all_period_fault_info.fillna(0,inplace=False)
+        all_period_fault_info.to_sql('all_period_fault_info', db_engine, if_exists='append', index=False)
+    except Exception as e:
+        logger.error(traceback.format_exc)
+        logger.error(u" 任务运行错误\n".format(), exc_info=True)
+        
 def Minutes_Task():
     
     #............................获取数据................................
@@ -107,10 +106,10 @@ def Minutes_Task():
     mode=2
     tablename1='all_fault_info'
     
-    db_engine = create_engine(
-    "mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8".format(
-        user, parse.quote_plus(password), host, port, db
-    ))
+    # db_engine = create_engine(
+    # "mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8".format(
+    #     user, parse.quote_plus(password), host, port, db
+    # ))
     DBRead=DBDownload.DBDownload(host, port, db, user, password,mode)
     with DBRead as DBRead:
         df_fltinfo=DBRead.getdata('product_id','level','code','start_time',tablename=tablename1,factory='骑享',sn='',timename='',st='',sp='')#dbdownload经过了改编
@@ -125,7 +124,7 @@ def Minutes_Task():
     SatftyCount=DataSta.SftyWrngClsfy(df_fltinfo)
     MaxAccumAh,TotalAccumAh,MaxCycle,MaxRunningHour,TotalRunHour=DataSta.AccumInfo(df_last_accum,df_FirstDataTime,end_time)
 
-    all_statistic_info=DataFrame(columns=['factory','total_alarm','alarm_total_today','alarm_not_close_today','alarm_close_today','alarm_uregent_total_today','alarm_uregent_close_today','alarm_uregent_not_close_today','alarm_close_total','run_time_total','dischrg_total','odo_total','max_dischrg_one','max_runtime_one','max_cycle_one','max_odo_one','alarm_close_total','alarm_total','cell_type','cell_type_count','cell_safety_risk_count','data_safety_risk_count','status_safety_risk_count','hv_safety_risk_count','system_safety_risk_count','sample_safety_risk_count','controller_safety_risk_count','design_safety_risk_count'])
+    all_statistic_info=DataFrame(columns=['factory','total_alarm','alarm_total_today','alarm_not_close_today','alarm_close_today','alarm_uregent_total_today','alarm_uregent_close_today','alarm_uregent_not_close_today','run_time_total','dischrg_total','odo_total','max_dischrg_one','max_runtime_one','max_cycle_one','max_odo_one','alarm_close_total','alarm_total','cell_type','cell_type_count','cell_safety_risk_count','data_safety_risk_count','status_safety_risk_count','hv_safety_risk_count','system_safety_risk_count','sample_safety_risk_count','controller_safety_risk_count','design_safety_risk_count'])
     all_statistic_info.loc[0,'factory']='骑享'
     all_statistic_info.loc[0,'total_alarm']=FltAlarmInfo.loc[0,'SftyPlt_Data_Total']
     all_statistic_info.loc[0,'alarm_total_today']=FltAlarmInfo.loc[0,'SftyPlt_Data_day']
@@ -146,8 +145,9 @@ def Minutes_Task():
     CellType=Celltype.columns.tolist()
     CellType=','.join(CellType) 
     all_statistic_info.loc[0,'cell_type']=str(CellType)
-    CellTypeCount=str(Celltype.loc[0].values)
-    CellTypeCount=''.join(CellTypeCount) 
+    CellTypeCount=Celltype.loc[0].tolist()
+    CellTypeCount=[str(x) for x in CellTypeCount]
+    CellTypeCount=','.join(CellTypeCount)  
     all_statistic_info.loc[0,'cell_type_count']=str(CellTypeCount)
     all_statistic_info.loc[0,'cell_safety_risk_count']=SatftyCount.loc[0,'CellSaftyCount']
     all_statistic_info.loc[0,'data_safety_risk_count']=SatftyCount.loc[0,'DataSaftyCodeCount']
@@ -157,9 +157,32 @@ def Minutes_Task():
     all_statistic_info.loc[0,'sample_safety_risk_count']=SatftyCount.loc[0,'SamplingSatyCount']
     all_statistic_info.loc[0,'controller_safety_risk_count']=SatftyCount.loc[0,'CtrlSaftyCodeCount']
     all_statistic_info.loc[0,'design_safety_risk_count']=SatftyCount.loc[0,'DsnSaftyCodeCount']
-    all_statistic_info.to_sql('all_statistic_info', db_engine, if_exists='append', index=False)
+    try:
+        conn = pymysql.connect(host=host, port=port, user=user, password=password, database=db)
+        cursor = conn.cursor()
+        sql = ''' update all_statistic_info set update_time='{}', total_alarm={}, alarm_total_today={}, alarm_close_today={}, alarm_not_close_today={},
+                    alarm_uregent_total_today={}, alarm_uregent_close_today={},alarm_uregent_not_close_today={},run_time_total={},dischrg_total={},
+                    odo_total={},max_dischrg_one={},max_runtime_one={},max_cycle_one={},max_odo_one={},alarm_close_total={},alarm_total={},cell_type='{}',
+                    cell_type_count='{}',cell_safety_risk_count={}, data_safety_risk_count={},status_safety_risk_count={},hv_safety_risk_count={},
+                    system_safety_risk_count={},sample_safety_risk_count={},controller_safety_risk_count={},design_safety_risk_count={}
+                    where factory='{}' 
+                    '''.format(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), all_statistic_info.loc[0,'total_alarm'], all_statistic_info.loc[0,'alarm_total_today'],all_statistic_info.loc[0,'alarm_close_today'],
+                               all_statistic_info.loc[0,'alarm_not_close_today'],all_statistic_info.loc[0,'alarm_uregent_total_today'],all_statistic_info.loc[0,'alarm_uregent_close_today'],
+                               all_statistic_info.loc[0,'alarm_uregent_not_close_today'],all_statistic_info.loc[0,'run_time_total'],all_statistic_info.loc[0,'dischrg_total'],
+                               all_statistic_info.loc[0,'odo_total'],all_statistic_info.loc[0,'max_dischrg_one'],all_statistic_info.loc[0,'max_runtime_one'],
+                               all_statistic_info.loc[0,'max_cycle_one'],all_statistic_info.loc[0,'max_odo_one'],all_statistic_info.loc[0,'alarm_close_total'],
+                               all_statistic_info.loc[0,'alarm_total'],all_statistic_info.loc[0,'cell_type'],all_statistic_info.loc[0,'cell_type_count'],
+                               all_statistic_info.loc[0,'cell_safety_risk_count'],all_statistic_info.loc[0,'data_safety_risk_count'],all_statistic_info.loc[0,'status_safety_risk_count'],
+                               all_statistic_info.loc[0,'hv_safety_risk_count'],all_statistic_info.loc[0,'system_safety_risk_count'],all_statistic_info.loc[0,'sample_safety_risk_count'],
+                                all_statistic_info.loc[0,'controller_safety_risk_count'],all_statistic_info.loc[0,'design_safety_risk_count'],'骑享')
+        cursor.execute(sql)
 
-    print(all_statistic_info)
+        conn.commit()
+        conn.close();
+    except:
+        logger.error(traceback.format_exc)
+        logger.error(u"{},{} 任务运行错误\n".format(start_time,end_time), exc_info=True)
+        conn.close();
 
 if __name__ == "__main__":
     
@@ -234,10 +257,11 @@ if __name__ == "__main__":
     # drive_count = 0
     # stand_count = 0
     # for sn in df_sn['sn'].tolist():
+    Week_Task()
     Minutes_Task()
     scheduler = BlockingScheduler()
-    scheduler.add_job(Week_Task, 'interval', seconds=10, id='Week_Task')
-    scheduler.add_job(Minutes_Task, 'interval', seconds=10, id='Hour_Task')
+    scheduler.add_job(Week_Task, 'interval', days=7, id='Week_Task')
+    scheduler.add_job(Minutes_Task, 'interval', seconds=300, id='Hour_Task')
     
     try:
         logger.info("周期任务 START!")

LIB/FRONTEND/SaftyCenter/diagfault/deploy.py

@@ -0,0 +1,259 @@
+
+__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 dateutil.relativedelta
+import traceback
+from LIB.MIDDLE.CellStateEstimation.Common import log
+from pandas.core.frame import DataFrame
+from LIB.MIDDLE.SaftyCenter.Common import QX_BatteryParam
+from LIB.MIDDLE.SaftyCenter.diagfault.SC_SamplingSafty import SamplingSafty
+
+from LIB.MIDDLE.CellStateEstimation.Common.V1_0_1 import DBDownload
+from urllib import parse
+import pymysql
+import pdb
+from apscheduler.schedulers.blocking import BlockingScheduler
+import datacompy
+from LIB.MIDDLE.SaftyCenter.Common import FeiShuData
+from LIB.MIDDLE.SaftyCenter.Common import QX_BatteryParam
+from LIB.MIDDLE.SaftyCenter.diagfault import  CBMSBatDiag
+from LIB.MIDDLE.SaftyCenter.diagfault.SC_SamplingSafty import SamplingSafty
+
+def fun():
+    global df_sn
+    global db_res_engine
+    global logger
+    global df_Diag_Ram
+    # 读取结果数据库
+    host2='rm-bp10j10qy42bzy0q77o.mysql.rds.aliyuncs.com'
+    port2=3306
+    db2='safety_platform'
+    user2='qx_read'
+    password2='Qx@123456'
+    start=time.time()
+    end_time=datetime.datetime.now()
+    start_time=end_time-datetime.timedelta(seconds=120)
+    start_time=start_time.strftime('%Y-%m-%d %H:%M:%S')
+    end_time=end_time.strftime('%Y-%m-%d %H:%M:%S')
+    logger.info("cycle start !!!!!!!!!!!!!!!!!!!!")
+    
+    start=time.time()
+    end_time=datetime.datetime.now()
+    start_time=end_time-datetime.timedelta(seconds=130)
+    start_time=start_time.strftime('%Y-%m-%d %H:%M:%S')
+    end_time=end_time.strftime('%Y-%m-%d %H:%M:%S')
+    df_read_Yunw = FeiShuData.getFeiShuDATA()#运维表格数据
+    df_read_Yunw.rename(columns={'电池编码':'product_id'},inplace=True)
+    df_read_Yunw.rename(columns={'内容描述':'info'},inplace=True)
+    df_read_Yunw.rename(columns={'发生时间':'start_time'},inplace=True)
+    df_read_Yunw.rename(columns={'维修信息':'advice'},inplace=True)
+
+    for i in range(0, len(df_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三元电芯
+            else:
+                logger.info("pid-{} celltype-{} SN: {} SKIP!".format(os.getpid(), "未知", sn))
+                continue
+            sn = df_sn.loc[i, 'sn']
+            
+            logger.info("pid-{} celltype-{} SN: {} START!".format(os.getpid(), celltype, sn))
+                
+                
+            param=QX_BatteryParam.BatteryInfo(celltype)   
+            #读取原始数据库数据........................................................................................................................................................
+            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']
+
+            #读取结果数据库数据........................................................................................................................................................
+            host='rm-bp10j10qy42bzy0q77o.mysql.rds.aliyuncs.com'
+            port=3306
+            db='qx_cas'
+            user='qx_read'
+            password='Qx@123456'
+            mode=1
+            tablename1='cellstateestimation_soh'
+            tablename2='cellstateestimation_uniform_socvoltdiff'       
+            #电池诊断................................................................................................................................................................
+            DBRead=DBDownload.DBDownload(host, port, db, user, password,mode)
+            with DBRead as DBRead:
+                df_soh=DBRead.getdata('time_st,sn,soh,cellsoh', tablename=tablename1, sn=sn, timename='time_sp', st=start_time, sp=end_time)
+                df_uniform=DBRead.getdata('time,sn,cellsoc_diff,cellmin_num,cellmax_num', tablename=tablename2, sn=sn, timename='time', st=start_time, sp=end_time)
+
+            #电池诊断................................................................................................................................................................
+            CellFltInfo=DataFrame(columns=['start_time', 'end_time', 'product_id', 'code', 'level', 'info','advice'])
+            df_Diag_Ram_sn = df_Diag_Ram.loc[df_Diag_Ram['product_id']==sn]#历史故障
+            df_Diag_Ram_sn_else = pd.concat([df_Diag_Ram,df_Diag_Ram_sn,df_Diag_Ram_sn]).drop_duplicates(subset=['product_id','code','start_time','Batpos','info'],keep=False)#sn之外的故障
+            CellFltInfo = df_Diag_Ram_sn.drop('Batpos',axis=1)
+            df_Diag_Ram_add = pd.DataFrame()
+            df_Diag_Ram_Update_change = pd.DataFrame()
+            if not df_bms.empty:
+                df_Diag_Batdiag_update_xq=SamplingSafty.main(sn,param,df_bms,CellFltInfo)#学琦计算故障   
+                BatDiag=CBMSBatDiag.BatDiag(sn,celltype,df_bms, df_soh, df_uniform, CellFltInfo)#鹏飞计算
+                df_Diag_Batdiag_update=BatDiag.diag() 
+                df_Diag_Cal_Update_add = pd.concat([CellFltInfo,df_Diag_Batdiag_update_xq,df_Diag_Batdiag_update])#重新计算的该SN下的故障
+                df_Diag_Cal_Update_temp = df_Diag_Cal_Update_add.drop_duplicates(subset=['product_id','start_time','end_time','code','info'], keep='first', inplace=False, ignore_index=False)#去除相同故障
+                df_Diag_cal_early_unfix = pd.DataFrame()
+                df_Diag_Cal_finish = pd.DataFrame()
+                df_Diag_cal_early_fix = pd.DataFrame()
+                if not df_Diag_Cal_Update_temp.empty:
+                    #------------------------------合并两者故障，并将同一sn号下的车辆故障放一起----------------------------------------------
+                    df_Diag_Cal_Update = df_Diag_Cal_Update_temp#替换上一行
+                    df_Diag_Cal_finish = df_Diag_Cal_Update.loc[df_Diag_Cal_Update['end_time'] != '0000-00-00 00:00:00']
+                    df_Diag_Cal_new = df_Diag_Cal_Update.loc[df_Diag_Cal_Update['end_time'] == '0000-00-00 00:00:00']
+                    df_Diag_Cal_finish['Batpos'] = 1
+                    df_Diag_Cal_new['Batpos'] = 0
+                    df_feishu_sta = df_read_Yunw.loc[(df_read_Yunw['product_id'] == sn)]#飞书中该sn车辆状态
+                    if df_feishu_sta.empty:#飞书中没有该sn记录故障的新增
+                        df_Diag_cal_early_unfix = df_Diag_Cal_new#如果为新出故障，则直接记录在df_diag_frame中
+                    else:
+                        df_Diag_cal_later = df_Diag_Cal_new.loc[pd.to_datetime(df_Diag_Cal_new['start_time']) > max(pd.to_datetime(df_feishu_sta['start_time']))]#故障表中故障时间晚于飞书记录时间的新增
+                        df_Diag_cal_early = pd.concat([df_Diag_Cal_new,df_Diag_cal_later,df_Diag_cal_later]).drop_duplicates(subset=['product_id','code','start_time'],keep=False)#故障表中故障时间早于飞书记录时间
+                        df_feishu_sta_latest = df_feishu_sta.loc[pd.to_datetime(df_feishu_sta['start_time']) == max(pd.to_datetime(df_feishu_sta['start_time']))]#飞书中该SN下的最新故障
+                        df_feishu_diag_unfix = (df_feishu_sta_latest['advice'] == '需正常返仓') | (df_feishu_sta_latest['advice'] == '需紧急返仓')
+                        if any(df_feishu_diag_unfix):
+                            df_Diag_cal_early_unfix = df_Diag_Cal_new
+                        else:
+                            df_Diag_cal_early_fix = df_Diag_cal_early
+                            df_Diag_cal_early_unfix = df_Diag_cal_later
+                    if not df_Diag_cal_early_fix.empty:
+                        df_Diag_cal_early_fix['Batpos'] = 1
+                df_Diag_Ram_Update = pd.concat([df_Diag_cal_early_unfix,df_Diag_cal_early_fix,df_Diag_Cal_finish])
+                df_Diag_Ram_Update.sort_values(by = ['start_time'], axis = 0, ascending=True,inplace=True)#该sn下当次诊断的故障状态
+                df_Diag_Ram_add = pd.concat([df_Diag_Ram_Update,df_Diag_Ram_sn,df_Diag_Ram_sn]).drop_duplicates(subset=['start_time','code'],keep=False)#此次判断中新增故障
+                df_Diag_Ram_Update_old = pd.concat([df_Diag_Ram_Update,df_Diag_Ram_add,df_Diag_Ram_add]).drop_duplicates(subset=['start_time','code'],keep=False)#此次判断中新增故障
+                df_Diag_Ram_Update_change = pd.concat([df_Diag_Ram_Update_old,df_Diag_Ram_sn,df_Diag_Ram_sn]).drop_duplicates(subset=['start_time','code','Batpos'],keep=False)#此次判断中新增故障
+                df_Diag_Ram = pd.concat([df_Diag_Ram_sn_else,df_Diag_Cal_new])
+
+            # if (len(df_Diag_Ram_add) > 0) | (len(df_Diag_Ram_Update_change) > 0):#历史及现有故障
+            #     df_Diag_Ram.to_csv(r'D:\Work\Code_write\data_analyze_platform\USER\01Screen_Problem\result.csv',index=False,encoding='GB18030')
+            if len(df_Diag_Ram_add) > 0:#新增故障
+                df_Diag_Ram_add.columns = ['product_id', 'start_time', 'end_time', 'code', 'level', 'info', 'advice', 'Batpos']
+                df_Diag_Ram_add['factory'] = '骑享'
+                df_Diag_Ram_add.to_sql("all_fault_info",con=db_res_engine, if_exists="append",index=False)
+                # df_Diag_Ram_add.to_csv(r'D:\Work\Code_write\data_analyze_platform\USER\01Screen_Problem\result_add.csv',index=False,encoding='GB18030')
+            if len(df_Diag_Ram_Update_change) > 0:#更改故障
+                df_Diag_Ram_Update_change = df_Diag_Ram_Update_change.reset_index(drop=True)
+                try:
+                    conn = pymysql.connect(host=host2, port=port2, user=user2, password=password2, database=db2)
+                    cursor = conn.cursor()
+                    for i in range(0,len(df_Diag_Ram_Update_change)):
+                        cursor.execute('''
+                                    update all_fault_info set end_time='{}', Batpos={}, where product_id='{}' and code={} and start_time='{}'
+                                    '''.format(df_Diag_Ram_Update_change.loc[i,'end_time'], df_Diag_Ram_Update_change.loc[i, 'Batpos'],
+                                               df_Diag_Ram_Update_change.loc[i,'product_id'], df_Diag_Ram_Update_change.loc[i,'code'], 
+                                               df_Diag_Ram_Update_change.loc[i,'start_time']))
+                        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();
+                # df_Diag_Ram_Update_change.to_csv(r'D:\Work\Code_write\data_analyze_platform\USER\01Screen_Problemm\result_change.csv',index=False,encoding='GB18030')
+            end=time.time()
+            logger.info("pid-{} celltype-{} SN: {} DONE!".format(os.getpid(), celltype, sn))
+        except:
+            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 # 历史数据运行标志
+    
+
+    # 更新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='saftyCenter_diagfault', 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))
+    df_Diag_Ram=pd.read_sql("select start_time, end_time, product_id, code, level, info, advice, Batpos 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']
+
+
+    #定时任务.......................................................................................................................................................................
+    scheduler = BlockingScheduler()
+
+    scheduler.add_job(fun, 'interval', seconds=300, id='diag_job')
+    try:  
+        scheduler.start()
+    except Exception as e:
+        scheduler.shutdown()
+        logger.error(str(e))
+    

LIB/FRONTEND/SaftyCenter/diagfault/main.py

@@ -0,0 +1,177 @@
+import CBMSBatDiag
+from SC_SamplingSafty import SamplingSafty
+import datetime
+import pandas as pd
+from LIB.BACKEND import DBManager, Log
+from sqlalchemy import create_engine
+import time, datetime
+from apscheduler.schedulers.blocking import BlockingScheduler
+from LIB.MIDDLE.CellStateEstimation.Common.V1_0_1 import DBDownload
+from LIB.MIDDLE.CellStateEstimation.Common.V1_0_1 import log
+from pandas.core.frame import DataFrame
+import datacompy
+from SaftyCenter.Common import FeiShuData
+from SaftyCenter.Common import QX_BatteryParam
+
+#...................................电池包电芯安全诊断函数......................................................................................................................
+def diag_cal():
+    global SNnums
+    global df_Diag_Ram
+  
+    start=time.time()
+    end_time=datetime.datetime.now()
+    start_time=end_time-datetime.timedelta(seconds=130)
+    start_time=start_time.strftime('%Y-%m-%d %H:%M:%S')
+    end_time=end_time.strftime('%Y-%m-%d %H:%M:%S')
+    df_read_Yunw = FeiShuData.getFeiShuDATA()#运维表格数据
+    df_read_Yunw.rename(columns={'电池编码':'product_id'},inplace=True)
+    df_read_Yunw.rename(columns={'内容描述':'info'},inplace=True)
+    df_read_Yunw.rename(columns={'发生时间':'start_time'},inplace=True)
+    df_read_Yunw.rename(columns={'维修信息':'advice'},inplace=True)
+    for sn in SNnums:
+        print(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
+            # sys.exit()
+        param=QX_BatteryParam.BatteryInfo(celltype) 
+        print(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']
+
+        #读取结果数据库数据........................................................................................................................................................
+        host='rm-bp10j10qy42bzy0q77o.mysql.rds.aliyuncs.com'
+        port=3306
+        db='qx_cas'
+        user='qx_read'
+        password='Qx@123456'
+        mode=1
+        tablename1='cellstateestimation_soh'
+        tablename2='cellstateestimation_uniform_socvoltdiff'       
+        #电池诊断................................................................................................................................................................
+        DBRead=DBDownload.DBDownload(host, port, db, user, password,mode)
+        with DBRead as DBRead:
+            df_soh=DBRead.getdata('time_st,sn,soh,cellsoh', tablename=tablename1, sn=sn, timename='time_sp', st=start_time, sp=end_time)
+            df_uniform=DBRead.getdata('time,sn,cellsoc_diff,cellmin_num,cellmax_num', tablename=tablename2, sn=sn, timename='time', st=start_time, sp=end_time)
+
+        #电池诊断................................................................................................................................................................
+        CellFltInfo=DataFrame(columns=['start_time', 'end_time', 'product_id', 'code', 'level', 'info','advice'])
+        df_Diag_Ram_sn = df_Diag_Ram.loc[df_Diag_Ram['product_id']==sn]#历史故障
+        df_Diag_Ram_sn_else = pd.concat([df_Diag_Ram,df_Diag_Ram_sn,df_Diag_Ram_sn]).drop_duplicates(subset=['product_id','code','start_time','Batpos','info'],keep=False)#sn之外的故障
+        CellFltInfo = df_Diag_Ram_sn.drop('Batpos',axis=1)
+        df_Diag_Ram_fix = df_Diag_Ram.loc[df_Diag_Ram['Batpos'] == 1]
+        df_Diag_Ram_unfix = df_Diag_Ram.loc[df_Diag_Ram['Batpos'] == 0]
+        if not df_bms.empty:
+            df_Diag_Batdiag_update_xq=SamplingSafty.main(sn,param,df_bms,CellFltInfo)#学琦计算故障   
+            BatDiag=CBMSBatDiag.BatDiag(sn,celltype,df_bms, df_soh, df_uniform, CellFltInfo)#鹏飞计算
+            df_Diag_Batdiag_update=BatDiag.diag() 
+            df_Diag_Cal_Update_add = pd.concat([CellFltInfo,df_Diag_Batdiag_update_xq,df_Diag_Batdiag_update])#重新计算的该SN下的故障
+            df_Diag_Cal_Update_temp = df_Diag_Cal_Update_add.drop_duplicates(subset=['product_id','start_time','end_time','code','info'], keep='first', inplace=False, ignore_index=False)#去除相同故障
+            df_Diag_cal_early_unfix = pd.DataFrame()
+            df_sn_car_fix = pd.DataFrame()
+            df_Diag_Cal_finish = pd.DataFrame()
+            df_Diag_cal_early_fix = pd.DataFrame()
+            if not df_Diag_Cal_Update_temp.empty:
+                #------------------------------合并两者故障，并将同一sn号下的车辆故障放一起----------------------------------------------
+                df_Diag_Cal_Update = df_Diag_Cal_Update_temp#替换上一行
+                df_Diag_Cal_finish = df_Diag_Cal_Update.loc[df_Diag_Cal_Update['end_time'] != '0000-00-00 00:00:00']
+                df_Diag_Cal_new = df_Diag_Cal_Update.loc[df_Diag_Cal_Update['end_time'] == '0000-00-00 00:00:00']
+                df_Diag_Cal_finish['Batpos'] = 1
+                df_Diag_Cal_new['Batpos'] = 0
+                df_feishu_sta = df_read_Yunw.loc[(df_read_Yunw['product_id'] == sn)]#飞书中该sn车辆状态
+                if df_feishu_sta.empty:
+                    df_Diag_cal_early_unfix = df_Diag_Cal_new#如果为新出故障，则直接记录在df_diag_frame中
+                else:
+                    df_Diag_cal_later = df_Diag_Cal_new.loc[df_Diag_Cal_new['start_time'] > max(df_feishu_sta['start_time'])]#故障表中故障时间晚于飞书记录时间
+                    df_Diag_cal_early = pd.concat([df_Diag_Cal_new,df_Diag_cal_later,df_Diag_cal_later]).drop_duplicates(subset=['product_id','code','start_time'],keep=False)#故障表中故障时间早于飞书记录时间
+                    df_feishu_sta_latest = df_feishu_sta.loc[df_feishu_sta['start_time'] == max(df_feishu_sta['start_time'])]#飞书中该SN下的最新故障
+                    df_feishu_diag_unfix = (df_feishu_sta_latest['advice'] == '需正常返仓') | (df_feishu_sta_latest['advice'] == '需紧急返仓')
+                    df_sn_car_unfix = pd.DataFrame()
+                    if any(df_feishu_diag_unfix):
+                        df_Diag_cal_early_unfix = df_Diag_Cal_new
+                    else:
+                        df_Diag_cal_early_fix = df_Diag_cal_early
+                        df_Diag_cal_early_unfix = df_Diag_cal_later
+                if not df_Diag_cal_early_fix.empty:
+                    df_Diag_cal_early_fix['Batpos'] = 1
+            df_Diag_Ram_Update = pd.concat([df_Diag_cal_early_unfix,df_Diag_cal_early_fix,df_Diag_Cal_finish])
+            df_Diag_Ram_Update.sort_values(by = ['start_time'], axis = 0, ascending=True,inplace=True)#对故障信息按照时间进行排序
+            df_temp5 = pd.concat([df_Diag_Ram_Update,df_Diag_Ram_sn_else])
+            df_Diag_Ram_sum = df_temp5.drop_duplicates(subset=['product_id','start_time','end_time','code','info'], keep='first', inplace=False, ignore_index=False)#去除相同故障
+            df_tempnum = df_Diag_Ram_sum.groupby(['product_id']).size()#获取每个sn的故障总数
+            col1 = df_tempnum[df_tempnum>1].reset_index()[['product_id']]#多故障sn号
+            col2 = df_tempnum[df_tempnum==1].reset_index()[['product_id']]#单故障sn号
+            df_temp1 = pd.DataFrame()
+            if not col1.empty:
+                for item in col1['product_id']:
+                    temp_data = df_Diag_Ram_sum.loc[df_Diag_Ram_sum['product_id'] == item]
+                    temp_data.sort_values(by = ['start_time'], axis = 0, ascending=True,inplace=True)#对故障信息按照时间进行排序
+                    df_temp1 = df_temp1.append(temp_data)
+            df_temp2 = pd.merge(col2,df_Diag_Ram_sum,on=["product_id"])#单故障码数据筛选
+            df_temp3 = pd.concat([df_temp1,df_temp2])#多故障及单故障合并
+            df_temp4 = df_temp3.reset_index(drop=True)
+            df_Diag_Ram = df_temp4
+            df_Diag_Ram_fix = df_Diag_Ram.loc[df_Diag_Ram['Batpos'] == 1]
+            df_Diag_Ram_unfix = df_Diag_Ram.loc[df_Diag_Ram['Batpos'] == 0]
+        if len(df_Diag_Ram) > 0:
+            df_Diag_Ram.to_csv(r'D:\Work\Code_write\data_analyze_platform\USER\01Screen_Problem\result.csv',index=False,encoding='GB18030')
+        if len(df_Diag_Ram_fix) > 0:
+            df_Diag_Ram_fix.to_csv(r'D:\Work\Code_write\data_analyze_platform\USER\01Screen_Problem\result_fix.csv',index=False,encoding='GB18030')
+        if len(df_Diag_Ram_unfix) > 0:
+            df_Diag_Ram_unfix.to_csv(r'D:\Work\Code_write\data_analyze_platform\USER\01Screen_Problemm\result_unfix.csv',index=False,encoding='GB18030')
+        end=time.time()
+        print(end-start)
+
+#...............................................主函数.......................................................................................................................
+if __name__ == "__main__":
+    global SNnums
+    
+    excelpath=r'D:\Work\Code_write\data_analyze_platform\USER\01Screen_Problem\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_6040 + SNnums_4840 + SNnums_U7255+ SNnums_6060
+    # SNnums=['MGMCLN750N215I005','PK504B10100004341','PK504B00100004172','MGMLXN750N2189014']
+    SNnums = ['MGMLXN750N21B5004'] #SNnums_6040
+    
+    mylog=log.Mylog('log_diag.txt','error')
+    mylog.logcfg()
+    #............................模块运行前，先读取数据库中所有结束时间为0的数据，需要从数据库中读取................
+    result=pd.read_csv(r'D:\Work\Code_write\data_analyze_platform\USER\01Screen_Problem\result.csv',encoding='gbk')
+    
+    # df_Diag_Ram=result[result['end_time']=='0000-00-00 00:00:00']
+    df_Diag_Ram=result#[result['Batpos'] == 0]#将故障依然存在的赋值
+    print('----------------输入--------')
+    print(df_Diag_Ram)
+    print('-------计算中-----------')
+    #定时任务.......................................................................................................................................................................
+    scheduler = BlockingScheduler()
+    scheduler.add_job(diag_cal, 'interval', seconds=120, id='diag_job')
+
+    try:  
+        scheduler.start()
+    except Exception as e:
+        scheduler.shutdown()
+        print(repr(e))
+        mylog.logopt(e)

LIB/FRONTEND/SaftyCenter/diagfault/run.bat

@@ -0,0 +1,4 @@
+cd /d D:\deploy\python_platform\data_analyze_platform\LIB\FRONTEND\SaftyCenter\diagfault
+title cal_saftyCenter_diagfault
+D:\env\py_pro\python.exe D:\deploy\python_platform\data_analyze_platform\LIB\FRONTEND\SaftyCenter\diagfault\deploy.py
+pause

LIB/FRONTEND/other/oss_to_iotp/deploy.py

@@ -52,7 +52,8 @@ if __name__ == "__main__":
         password='Qx@123456'
         conn = pymysql.connect(host=host, port=port, user=user, password=password, database=db)
         cursor = conn.cursor()
-        cursor.execute("select * from (select distinct(t.sn) from (select * from app_device where status = 3)t inner join (select * from app_device_log where type=3 and out_type=1) t2 on t.sn=t2.sn)t3")
+        # cursor.execute("select * from (select distinct(t.sn) from (select * from app_device where status = 3)t inner join (select * from app_device_log where type=3 and out_type=1) t2 on t.sn=t2.sn)t3")
+        cursor.execute("select distinct sn from app_device")
         res = cursor.fetchall()
         df_res = pd.DataFrame(res, columns=['sn'])
         df_res = df_res.reset_index(drop=True)