Estimating Lithium-ion Battery Health¶

import verticapy as vp
vp.__version__

vp.connect("VerticaDSN")

vp.drop("battery_data", method="schema")
vp.create_schema("battery_data", True)

battery5 = vp.read_csv("data/battery5_data.csv")

display(battery5.head(5))

battery5.describe()

battery5['start_time'].describe()

battery5.analytic(func="first_value",
                  columns="Voltage_measured",
                  by="start_time",
                  order_by={"Time":"asc"},
                  name="first_voltage_measured")

battery5.analytic(func="first_value",
                  columns="Voltage_measured",
                  by="start_time",
                  order_by={"Time":"desc"},
                  name="last_voltage_measured")

cycling_info = battery5.groupby(columns = ['start_time',
                                           'type',
                                           'first_voltage_measured',
                                           'last_voltage_measured'], 
                                expr = ["COUNT(*) AS nr_of_measurements",
                                        "MAX(Time) AS cycle_duration"]).sort('start_time')
cycling_info['cycle_id'] = "ROW_NUMBER() OVER(ORDER BY start_time)"
cycling_info

import warnings
warnings.filterwarnings('ignore')
cycling_info.animated(ts="start_time",
                      columns= ["type","cycle_duration"], 
                      by="type", 
                      kind="bar",)

cycling_info.bar(["type"], 
                 method = "avg", 
                 of = "cycle_duration")

cycling_info.groupby('type',['MIN(first_voltage_measured) AS min_first_voltage',
                             'AVG(first_voltage_measured) AS avg_first_voltage',
                             'MAX(first_voltage_measured) AS max_first_voltage',
                             'MIN(last_voltage_measured)  AS min_last_voltage',
                             'AVG(last_voltage_measured)  AS avg_last_voltage',
                             'MAX(last_voltage_measured)  AS max_last_voltage'])

# Visualize the capacity degradation curve along with its smoothed version
discharging_data = battery5[battery5['type'] == 'discharge']
d_cap = discharging_data[['start_time', 'Capacity']].groupby(['start_time', 'Capacity'])
d_cap["discharge_id"] = "ROW_NUMBER() OVER(ORDER BY start_time, Capacity)"
d_cap.rolling(func = 'mean',
              columns = 'capacity',
              window = (-100, -1),
              name = 'smooth_capacity')


import matplotlib.pyplot as plt
from matplotlib.pyplot import axhline

fig = plt.figure()
ax = d_cap.plot(ts = 'discharge_id', columns = ['Capacity', 'smooth_capacity'])
ax.axhline(y=1.4, label='End-of-life criteria')
ax.set_title('Capacity degradation curve of the battery, its smoothed version and its end-of-life threshold')
ax.legend() 
plt.show()

battery5.outliers(columns = ["Voltage_measured","Current_measured","Temperature_measured","Capacity"],
                  name = "global_outlier",
                  threshold = 4.0)
battery5.filter("global_outlier = 0").drop('global_outlier')

sample_cycle = battery5[battery5['Capacity'] == '1.83514614292266']
sample_cycle["Voltage_measured"].plot(ts = "Time")

sample_cycle["Temperature_measured"].plot(ts = "Time")

# filter for discharge cycles
discharging_data = battery5[battery5['type'] == 'discharge']


# define new features
discharge_cycle_metrics = discharging_data.groupby(columns = ['start_time'], 
                                                   expr = ['MIN(Temperature_measured) AS min_temp',
                                                           'MAX(Temperature_measured) AS max_temp',
                                                           'MIN(Voltage_measured) AS min_volt']).join(
                                                           discharging_data, 
                                                           how = "left",
                                                           on = {"min_volt":"voltage_measured"},
                                                           expr1 = ["*"],
                                                           expr2 = ["Time AS time_to_reach_minvolt"]).join(
                                                           discharging_data, 
                                                           how = "left",
                                                           on = {"max_temp":"temperature_measured"},
                                                           expr1 = ["*"],
                                                           expr2 = ["Time AS time_to_reach_maxtemp"])

# calculate values of SOH
discharging_data = discharging_data.groupby(['start_time','Capacity'])
discharging_data['SOH'] = discharging_data['Capacity'] * 0.5

# define the final dataset and save it to db
final_df = discharge_cycle_metrics.join(discharging_data,
                     on_interpolate = {"start_time":"start_time"},
                     how = "left",
                     expr1 = ["*"],
                     expr2 = ["SOH AS SOH"])

# normalize the features
final_df.normalize(method = "minmax",
                   columns = ["min_temp",
                              "max_temp",
                              "min_volt",
                              "time_to_reach_minvolt",
                              "time_to_reach_maxtemp"])

# save it to db
final_df.to_db(name = "battery_data.finaldata_battery_5")

from verticapy.learn.delphi import AutoML

model = AutoML("battery_data.battery_autoML", 
               estimator = "native")
model.fit("battery_data.finaldata_battery_5", 
          X = ["min_temp",
               "max_temp",
               "min_volt",
               "time_to_reach_minvolt",
               "time_to_reach_maxtemp"],
          y = "SOH")

model.plot()

# take the best model and its parameters
best_model = model.best_model_
params = best_model.get_params()
print(best_model.type)

from verticapy.learn.ensemble import RandomForestRegressor

# define a regression model based on the selected parameters
model_rf = RandomForestRegressor(name = "btr_rf1", **params)
model_rf.fit(final_df,
             X = ["min_temp",
                  "max_temp",
                  "min_volt",
                  "time_to_reach_minvolt",
                  "time_to_reach_maxtemp"],
             y = "SOH")
model_rf.regression_report()

# take the predicted values and the plot them along the true ones
result = model_rf.predict(final_df, 
                          name = "SOH_estimates")
result.plot(ts = 'start_time', 
            columns = ['SOH', 'SOH_estimates'])