Merge pull request #552 from TRI-AMDD/CV_capacity

CV capacity summary statistic

beep/conversion_schemas/structured_dtypes.yaml

@@ -30,6 +30,7 @@ summary:
   paused: 'int32'
   CV_time: 'float32'
   CV_current: 'float32'
+  CV_capacity: 'float32'
 
 diagnostic_summary:
   discharge_capacity: 'float64'
@@ -46,6 +47,7 @@ diagnostic_summary:
   cycle_type: 'category'
   CV_time: 'float32'
   CV_current: 'float32'
+  CV_capacity: 'float32'
 
 diagnostic_interpolated:
   voltage: 'float32'

beep/structure/base.py

@@ -877,17 +877,22 @@ def summarize_cycles(
         summary["paused"] = self.raw_data.groupby("cycle_index").apply(
             get_max_paused_over_threshold)
 
-        # Add CV_time and CV_current summary stats
+        # Add CV_time, CV_current, CV_capacity summary stats
         CV_time = []
         CV_current = []
+        CV_capacity = []
         for cycle in summary.cycle_index:
             raw_cycle = self.raw_data.loc[self.raw_data.cycle_index == cycle]
             charge = raw_cycle.loc[raw_cycle.current > 0]
             CV = get_CV_segment_from_charge(charge)
+            if CV.empty:
+                logger.debug(f"Failed to extract CV segment for cycle {cycle}!")
             CV_time.append(get_CV_time(CV))
             CV_current.append(get_CV_current(CV))
+            CV_capacity.append(get_CV_capacity(CV))
         summary["CV_time"] = CV_time
         summary["CV_current"] = CV_current
+        summary["CV_capacity"] = CV_capacity
 
         summary = self._cast_dtypes(summary, "summary")
 
@@ -1093,20 +1098,25 @@ def summarize_diagnostic(self, diagnostic_available):
             diagnostic_available["cycle_type"] * len(starts_at)
         )
 
-        # Add CV_time and CV_current summary stats
+        # Add CV_time, CV_current, and CV_capacity summary stats
         CV_time = []
         CV_current = []
+        CV_capacity = []
         for cycle in diag_summary.cycle_index:
             raw_cycle = self.raw_data.loc[self.raw_data.cycle_index == cycle]
 
             # Charge is the very first step_index
             CCCV = raw_cycle.loc[raw_cycle.step_index == raw_cycle.step_index.min()]
             CV = get_CV_segment_from_charge(CCCV)
+            if CV.empty:
+                logger.debug(f"Failed to extract CV segment for diagnostic cycle {cycle}!")
             CV_time.append(get_CV_time(CV))
             CV_current.append(get_CV_current(CV))
+            CV_capacity.append(get_CV_capacity(CV))
 
         diag_summary["CV_time"] = CV_time
         diag_summary["CV_current"] = CV_current
+        diag_summary["CV_capacity"] = CV_capacity
 
         diag_summary = self._cast_dtypes(diag_summary, "diagnostic_summary")
 
@@ -1523,18 +1533,20 @@ def get_CV_segment_from_charge(charge, dt_tol=1, dVdt_tol=1e-5, dIdt_tol=1e-4):
         (pd.DataFrame): dataframe containing the CV segment
 
     """
-    # Compute dI and dV
-    dI = np.diff(charge.current)
-    dV = np.diff(charge.voltage)
-    dt = np.diff(charge.test_time)
-
-    # Find the first index where the CV segment begins
-    i = 0
-    while i < len(dV) and not (dt[i] > dt_tol and abs(dV[i]/dt[i]) < dVdt_tol and abs(dI[i]/dt[i]) > dIdt_tol):
-        i = i+1
-
-    # Filter for CV phase
-    if len(charge):
+    if charge.empty:
+        return(charge)
+    else:
+        # Compute dI and dV
+        dI = np.diff(charge.current)
+        dV = np.diff(charge.voltage)
+        dt = np.diff(charge.test_time)
+
+        # Find the first index where the CV segment begins
+        i = 0
+        while i < len(dV) and (dt[i] < dt_tol or abs(dV[i]/dt[i]) > dVdt_tol or abs(dI[i]/dt[i]) < dIdt_tol):
+            i = i+1
+
+        # Filter for CV phase
         return(charge.loc[charge.test_time >= charge.test_time.iat[i-1]])
 
 
@@ -1549,9 +1561,8 @@ def get_CV_time(CV):
         (float): length of the CV segment in seconds
 
     """
-    if isinstance(CV, pd.DataFrame):
-        if len(CV):
-            return(CV.test_time.iat[-1] - CV.test_time.iat[0])
+    if not CV.empty:
+        return(CV.test_time.iat[-1] - CV.test_time.iat[0])
 
 
 def get_CV_current(CV):
@@ -1565,6 +1576,20 @@ def get_CV_current(CV):
         (float): current reached at the end of the CV segment
 
     """
-    if isinstance(CV, pd.DataFrame):
-        if len(CV):
-            return(CV.current.iat[-1])
+    if not CV.empty:
+        return(CV.current.iat[-1])
+
+
+def get_CV_capacity(CV):
+    """
+    Helper function to compute CV capacity.
+
+    Args:
+        CV (pd.DataFrame): CV segement of charge
+
+    Returns:
+        (float): charge capacity during the CV segment
+
+    """
+    if not CV.empty:
+        return(CV.charge_capacity.iat[-1] - CV.charge_capacity.iat[0])

beep/structure/tests/test_base.py

@@ -345,6 +345,7 @@ def test_summarize_cycles(self):
                     "energy_efficiency",
                     "CV_time",
                     "CV_current",
+                    "CV_capacity"
                 },
                 set(summary_diag.columns),
             )
@@ -354,6 +355,7 @@ def test_summarize_cycles(self):
         self.assertEqual(summary_diag["paused"].max(), 0)
         self.assertEqual(summary_diag["CV_time"][1], np.float32(160111.796875))
         self.assertEqual(summary_diag["CV_current"][1], np.float32(0.4699016))
+        self.assertEqual(summary_diag["CV_capacity"][1], np.float32(94.090355))
         self.run_dtypes_check(summary_diag)
 
         # incorporates test_get_energy and get_charge_throughput
@@ -432,6 +434,7 @@ def test_summarize_diagnostic(self):
         self.assertEqual(diag_summary["paused"].max(), 0)
         self.assertEqual(diag_summary["CV_time"][0], np.float32(125502.578125))
         self.assertEqual(diag_summary["CV_current"][0], np.float32(2.3499656))
+        self.assertEqual(diag_summary["CV_capacity"][0], np.float32(84.70442))
 
     # based on RCRT.test_determine_paused
     def test_paused_intervals(self):