析锂检测算法

LIB/MIDDLE/SaftyCenter/Liplated/Li_plated.py

@@ -0,0 +1,164 @@
+import pandas as pd
+import numpy as np
+import datetime
+import matplotlib as plt
+from scipy.signal import savgol_filter
+import BatParam
+
+class Liplated_test:
+    def __init__(self,sn,celltype,df_bms):  #参数初始化
+
+        self.sn=sn
+        self.celltype=celltype
+        self.param=BatParam.BatParam(celltype)
+        self.df_bms=pd.DataFrame(df_bms)
+        self.packcrnt=df_bms['总电流[A]']*self.param.PackCrntDec
+        self.packvolt=df_bms['总电压[V]']
+        self.bms_soc=df_bms['SOC[%]']
+        self.bmstime= pd.to_datetime(df_bms['时间戳'], format='%Y-%m-%d %H:%M:%S')
+
+        self.cellvolt_name=['单体电压'+str(x) for x in range(1,self.param.CellVoltNums+1)]
+        self.celltemp_name=['单体温度'+str(x) for x in range(1,self.param.CellTempNums+1)]
+        self.bmssta = df_bms['充电状态']
+    #定义加权滤波函数..................................................................................................
+    def moving_average(interval, windowsize):
+        window = np.ones(int(windowsize)) / float(windowsize)
+        re = np.convolve(interval, window, 'same')
+        return re
+#.............................................析锂检测............................................................................
+    def liplated_detect(self):
+        #----------------------------------------筛选充电后静置数据------------------------------------------------------------
+        chrgr_rest_data_temp = self.df_bms.loc[((self.df_bms['充电状态'] == 0) & (self.df_bms['SOC[%]'] > 98) & (self.df_bms['总电流[A]'] == 0)) | 
+                                               ((self.df_bms['充电状态'] == 2) & (self.df_bms['SOC[%]'] > 98) & (self.df_bms['总电流[A]'] == 0))]#接近慢充后静置数据
+        df_lipltd_result = pd.DataFrame(columns=['sn','date','liplated','liplated_amount'])
+        if not chrgr_rest_data_temp.empty:
+            chrgr_rest_data = chrgr_rest_data_temp.reset_index(drop=True)
+            temp_rest_time = chrgr_rest_data['时间戳']
+            rest_time = pd.to_datetime(temp_rest_time)
+            delta_time = (np.diff(rest_time)/pd.Timedelta(1, 'min'))#计算时间差的分钟数
+            pos = np.where(delta_time > 30)#静置数据分段,大于30min时，认为是两个静置过程
+            splice_num = []
+            if len(pos[0]) >= 1:
+                pos_ful_tem = np.insert(pos, 0, 0)
+                pos_len = len(pos_ful_tem)
+                data_len = len(rest_time)
+                pos_ful = np.insert(pos_ful_tem, pos_len, data_len-1)
+                for item in range(0,len(pos_ful)-1):
+                    splice_num.extend(item*np.ones(pos_ful[item +1]-pos_ful[item]))
+                splice_num = np.insert(splice_num, 0, 0)
+            else:
+                splice_num = np.zeros(len(temp_rest_time))
+                pos_ful = np.array([0])
+            chrgr_rest_data['chrgr_rest'] = splice_num
+            #---------------------------判断数据点数大于30的数据段,对电压微分并画图--------------------------------------------
+            cellvolt_list = self.cellvolt_name
+            chrgr_rest_check_data = chrgr_rest_data.drop(['GSM信号','故障等级','故障代码','绝缘电阻','总电流[A]','总电压[V]','充电状态','单体压差','SOC[%]'],axis=1,inplace=False)
+            chrgr_rest_check_data.fillna(value=0)
+            df_rest_volt_diffdt = pd.DataFrame()
+            df_rest_volt_smooth = pd.DataFrame()
+            k = 0
+            for j in range(0,len(pos_ful)-1):#len(pos_ful)-1#有几段充电后静置数据
+                df_test_rest_data = chrgr_rest_check_data.loc[chrgr_rest_check_data['chrgr_rest'] == j]
+                df_rest_volt_smooth = pd.DataFrame()
+                df_test_rest_time = pd.to_datetime(df_test_rest_data['时间戳'],format='%Y-%m-%d %H:%M:%S')
+                df_test_rest_time = df_test_rest_time.reset_index(drop=True)
+                df_data_length = len(df_test_rest_time)
+                if (df_data_length > 30) & ((df_test_rest_time[df_data_length - 1] - df_test_rest_time[0])/pd.Timedelta(1, 'min') > 40):#静置时间大于40min
+                    df_test_rest_time_dif_temp = np.diff(df_test_rest_time)/pd.Timedelta(1, 'min')
+                    num_list = []
+                    data_jump_pos = np.where(df_test_rest_time_dif_temp > 3)
+                    if len(data_jump_pos[0]) > 0:
+                        if data_jump_pos[0][0] > 100:
+                            for i in range(0,data_jump_pos[0][0],15):##采样密集时每隔10行取数据
+                                num_list.append(i)
+                            df_rest_data_temp = df_test_rest_data.iloc[num_list]
+                            df_test_rest_data_choose = pd.DataFrame(df_rest_data_temp)
+                            df_test_rest_data_choose_else = df_test_rest_data.iloc[data_jump_pos[0][0]+1:len(df_test_rest_data)-1]
+                            df_rest_data_recon_temp = df_test_rest_data_choose.append(df_test_rest_data_choose_else)
+                            df_rest_data_recon =df_rest_data_recon_temp.reset_index(drop=True)
+                        else:
+                            df_rest_data_recon = df_test_rest_data
+                    else:
+                        df_rest_data_recon = df_test_rest_data
+                    df_rest_time = pd.to_datetime(df_rest_data_recon['时间戳'],format='%Y-%m-%d %H:%M:%S')
+                    df_rest_time = df_rest_time.reset_index(drop=True)
+                    df_rest_time_dif_temp = np.diff(df_rest_time)/pd.Timedelta(1, 'min')
+                    df_rest_volt = df_rest_data_recon[cellvolt_list]
+                    for item in cellvolt_list:
+                        window_temp = int(len(df_rest_volt[item])/3)
+                        if window_temp%2:#滤波函数的窗口长度需为奇数
+                            window = window_temp
+                        else:
+                            window = window_temp - 1
+                        step = min(int(window/3),5)
+                        df_volt_smooth = savgol_filter(df_rest_volt[item],window,step)
+                        df_rest_volt_smooth[item] = df_volt_smooth
+                    df_test_rest_volt_diff_temp = np.diff(df_rest_volt_smooth,axis=0)
+                    df_test_rest_time_dif = pd.DataFrame(df_rest_time_dif_temp)
+                    df_test_rest_volt_diff = pd.DataFrame(df_test_rest_volt_diff_temp)
+                    df_test_rest_volt_diffdt_temp = np.divide(df_test_rest_volt_diff,df_test_rest_time_dif)
+                    df_test_rest_volt_diffdt = pd.DataFrame(df_test_rest_volt_diffdt_temp)
+                    df_test_rest_volt_diffdt = df_test_rest_volt_diffdt.append(df_test_rest_volt_diffdt.iloc[len(df_test_rest_volt_diffdt)-1])
+                    df_test_rest_volt_diffdt.columns = cellvolt_list
+                    if len(df_test_rest_volt_diffdt) > 25:
+                        for item in cellvolt_list:
+                            df_volt_diffdt_smooth = savgol_filter(df_test_rest_volt_diffdt[item],13,3)
+                            df_test_rest_volt_diffdt[item] = df_volt_diffdt_smooth
+                        df_test_rest_volt_diffdt['chrgr_rest'] = k
+                        df_test_rest_volt_diffdt['时间戳'] = list(df_rest_time)
+                        k = k+1
+                        df_rest_volt_diffdt = df_rest_volt_diffdt.append(df_test_rest_volt_diffdt)
+                        df_rest_volt_diffdt.reset_index()
+            #--------------------------------------------------------确认是否析锂----------------------------------------------------------------------------
+            # df_lipltd_data = pd.DataFrame(columns=['sn','date','liplated'])
+            for item in range(0,k):
+                lipltd_confirm = []
+                lipltd_amount = []
+                df_check_liplated_temp = df_rest_volt_diffdt.loc[df_rest_volt_diffdt['chrgr_rest'] == item].reset_index(drop = True)
+                df_lipltd_volt_temp = df_check_liplated_temp[cellvolt_list]
+                df_lipltd_volt_len = len(df_lipltd_volt_temp)
+                df_data_temp_add = df_lipltd_volt_temp.iloc[df_lipltd_volt_len-4:df_lipltd_volt_len-1]
+                df_lipltd_volt_temp_add = df_lipltd_volt_temp.append(df_data_temp_add)
+                df_lipltd_volt_temp_dif = np.diff(df_lipltd_volt_temp_add,axis=0)#电压一次微分，计算dv/dt
+                df_lipltd_volt_temp_dif = pd.DataFrame(df_lipltd_volt_temp_dif)
+                df_lipltd_volt_temp_dif.columns = cellvolt_list
+                df_lipltd_volt_temp_difdif = np.diff(df_lipltd_volt_temp_dif,axis=0)#电压二次微分，判断升降
+                df_lipltd_volt_temp_difdif = pd.DataFrame(df_lipltd_volt_temp_difdif)
+                df_lipltd_volt_temp_difdif.columns = cellvolt_list
+                df_lipltd_volt_temp_difdif_temp = df_lipltd_volt_temp_difdif
+                df_lipltd_volt_temp_difdif_temp[df_lipltd_volt_temp_difdif_temp >= 0] = 1
+                df_lipltd_volt_temp_difdif_temp[df_lipltd_volt_temp_difdif_temp < 0] = -1
+                df_lipltd_volt_temp_difdifdif = np.diff(df_lipltd_volt_temp_difdif_temp,axis=0)#三次微分，利用-2，2判断波分和波谷
+                df_lipltd_volt_difdifdif = pd.DataFrame(df_lipltd_volt_temp_difdifdif)
+                df_lipltd_volt_difdifdif.columns = cellvolt_list
+                df_lipltd_volt_difdifdif['chrgr_rest'] = k
+                df_lipltd_volt_difdifdif['时间戳'] = list(df_check_liplated_temp['时间戳'])
+                df_lipltd_volt_difdifdif = df_lipltd_volt_difdifdif.reset_index(drop = True)
+                df_lipltd_data_temp = df_lipltd_volt_difdifdif.loc[df_lipltd_volt_difdifdif['时间戳'] < (df_check_liplated_temp['时间戳'][0] + datetime.timedelta(minutes=90))]
+                for cell_name in cellvolt_list:#对每个电芯判断
+                    df_check_plated_data = df_lipltd_data_temp[cell_name]
+                    peak_pos = np.where(df_check_plated_data == -2)
+                    bot_pos = np.where(df_check_plated_data == 2)
+                    if len(peak_pos[0]) & len(bot_pos[0]):
+                        if (peak_pos[0][0] > bot_pos[0][0]) & (df_lipltd_volt_temp_dif[cell_name][peak_pos[0][0] + 1] < 0):
+                            lipltd_confirm.append(1)#1为析锂，0为非析锂
+                            lipltd_amount.append((df_check_liplated_temp['时间戳'][bot_pos[0][0] + 2] - df_check_liplated_temp['时间戳'][0])/pd.Timedelta(1, 'min'))
+                        else:
+                            lipltd_confirm.append(0)
+                            lipltd_amount.append(0)
+                    else:
+                        lipltd_confirm.append(0)
+                        lipltd_amount.append(0)
+                if any(lipltd_confirm) & (max(lipltd_amount) > 5):
+                    df_lipltd_confir_temp = pd.DataFrame({"sn":[self.sn], "time":[df_check_liplated_temp['时间戳'][0]], "liplated":[str(lipltd_confirm)], "liplated_amount":[str(lipltd_amount)]})
+                    df_lipltd_result = df_lipltd_result.append(df_lipltd_confir_temp)
+                    df_lipltd_result = df_lipltd_result.reset_index(drop = True)
+                    df_lipltd_result.sort_values(by = ['time'], axis = 0, ascending=True,inplace=True)#对故障信息按照时间进行排序
+        # df_lipltd_data.to_csv(r'D:\Work\Code_write\data_analyze_platform\USER\lzx\01算法开发\02析锂检测\liplated\算法开发_检测\滤波后图片\析锂.csv',index=False,encoding='GB18030')
+        #返回诊断结果...........................................................................................................
+        if not df_lipltd_result.empty:
+            return df_lipltd_result
+        else:
+            return pd.DataFrame()
+
+

LIB/MIDDLE/SaftyCenter/Liplated/main.py

@@ -0,0 +1,97 @@
+import datetime
+import pandas as pd
+from LIB.BACKEND import DBManager, Log
+import time, datetime
+from apscheduler.schedulers.blocking import BlockingScheduler
+import log
+from pandas.core.frame import DataFrame
+import Li_plated
+
+#...................................电池包电芯安全诊断函数......................................................................................................................
+def cell_platd_test():
+    global SNnums
+    global df_Diag_lipltd
+    start=time.time()
+    now_time=datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
+    now_time=datetime.datetime.strptime(now_time,'%Y-%m-%d %H:%M:%S')
+    start_time=now_time-datetime.timedelta(days=3)
+    end_time=str(now_time)
+    start_time=str(start_time)
+    k = 1
+    for sn in SNnums:
+        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()
+        print('计算的第' + str(k) + '个:' + sn)
+        k = k + 1
+        #读取原始数据库数据........................................................................................................................................................
+        start_time = '2021-11-15 12:00:00'
+        end_time = '2021-11-18 12:00:00'
+        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']
+        df_Diag_lipltd_add = pd.DataFrame(columns = ['sn','time','liplated', 'liplated_amount'])
+
+        #析锂诊断................................................................................................................................................................
+        if not df_bms.empty:
+            Diag_lipltd_temp = Li_plated.Liplated_test(sn,celltype,df_bms)#析锂检测
+            df_Diag_lipltd_add = Diag_lipltd_temp.liplated_detect()        
+        if not df_Diag_lipltd_add.empty:
+            df_Diag_lipltd_temp = df_Diag_lipltd.append(df_Diag_lipltd_add)
+            df_Diag_lipltd = df_Diag_lipltd_temp.drop_duplicates(subset = ['sn','time'], keep = 'first', inplace = False)
+            df_Diag_lipltd.sort_values(by = ['sn'], axis = 0, ascending=True,inplace=True)#对故障信息按照时间进行排序
+            df_Diag_lipltd.to_csv(r'D:\Work\Code_write\data_analyze_platform\USER\lzx\01算法开发\02析锂检测\01下载数据\格林美-力信7255\SNnums_6040_liplated_sn.csv',index=False,encoding='GB18030')
+        end=time.time()
+        print(end-start)
+
+#...............................................主函数.......................................................................................................................
+if __name__ == "__main__":
+    global SNnums
+    global df_Diag_lipltd
+    
+    excelpath=r'D:\Work\Code_write\data_analyze_platform\USER\lzx\01算法开发\04故障诊断\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 = SNnums_6040 #SNnums_C7255 #SNnums_6040['MGMCLN750N215N049'] 
+    # SNnums = pd.read_csv(r'D:\Work\Code_write\data_analyze_platform\USER\lzx\01算法开发\02析锂检测\liplated\疑似析锂电池sn.csv',encoding='GB18030')
+    
+    mylog=log.Mylog('log_diag.txt','error')
+    mylog.logcfg()
+    #............................模块运行前，先读取数据库中所有结束时间为0的数据，需要从数据库中读取................
+    df_Diag_lipltd=pd.read_csv(r'D:\Work\Code_write\data_analyze_platform\USER\lzx\01算法开发\02析锂检测\01下载数据\格林美-力信7255\析锂.csv',encoding='GB18030')
+
+    print('----------------输入--------')
+    print('-------计算中-----------')
+    #定时任务.......................................................................................................................................................................
+    scheduler = BlockingScheduler()
+    scheduler.add_job(cell_platd_test, 'interval', seconds=10, id='diag_job')
+
+    try:  
+        scheduler.start()
+    except Exception as e:
+        scheduler.shutdown()
+        print(repr(e))
+        mylog.logopt(e)