import pandas as pd
import numpy as np
import datetime
import matplotlib as plt
from scipy.signal import savgol_filter
from LIB.MIDDLE.CellStateEstimation.Common.V1_0_1 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','time','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-3:df_lipltd_volt_len-1]
df_lipltd_volt_temp_dif = df_lipltd_volt_temp.append(df_data_temp_add)
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]):
ini_dvdt = df_check_liplated_temp[cell_name][0]
peak_dvdt = df_check_liplated_temp[cell_name][peak_pos[0][0] + 1]
bot_dvdt = df_check_liplated_temp[cell_name][bot_pos[0][0] + 1]
peak_hight = peak_dvdt - ini_dvdt
peak_bot_hight = peak_dvdt - bot_dvdt
liplted_amount_temp = (df_check_liplated_temp['时间戳'][bot_pos[0][0] + 1] - df_check_liplated_temp['时间戳'][0])/pd.Timedelta(1, 'min')
if ((bot_pos[0][0] - peak_pos[0][0]) > 3) & (df_check_liplated_temp[cell_name][peak_pos[0][0] + 1] < 0) & (peak_bot_hight > 0.05*peak_hight) & (liplted_amount_temp > 15):
lipltd_confirm.append(1)#1为析锂，0为非析锂
lipltd_amount.append(liplted_amount_temp)
else:
lipltd_confirm.append(0)
lipltd_amount.append(0)
else:
lipltd_confirm.append(0)
lipltd_amount.append(0)
if any(lipltd_confirm):
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()