Refactor plotting utilities

aiida_aurora/utils/analyzers.py

@@ -134,8 +134,7 @@ def _get_capacities(self, snapshot: dict):
  """
  try:
  data = get_data_from_raw(snapshot)
- capacities = data['Qd'] if self.is_discharge else data['Qc']
- return capacities / 3.6 # As -> mAh
+ return data['Qd'] if self.is_discharge else data['Qc']
  except KeyError as err:
  self.logger.error(f"missing '{str(err)}' in snapshot")
  return []

aiida_aurora/utils/parsers.py

@@ -1,12 +1,15 @@
 import numpy as np
+from scipy.integrate import cumtrapz
 
 from aiida.orm import ArrayData
 
 
-def get_data_from_raw(jsdata):
+def get_data_from_raw(jsdata) -> dict:
  "Extract raw data from json file."
+
  if not isinstance(jsdata, dict):
  raise TypeError('jsdata should be a dictionary')
+
  if len(jsdata["steps"]) > 1:
  raise NotImplementedError('Analysis of multiple steps is not implemented.')
 
@@ -17,37 +20,12 @@ def get_data_from_raw(jsdata):
  Ewe = np.array([ts["raw"]["Ewe"]["n"] for ts in raw_data])
  I = np.array([ts["raw"]["I"]["n"] for ts in raw_data])
 
- # find half-cycle markers
- # add last point if not already a marker
- idx = np.where(np.diff(np.sign(I)) != 0)[0]
- if (final := len(I) - 1) not in idx:
- idx = np.append(idx, final)
+ return post_process_data(t, Ewe, I)
 
- # integrate and store charge and discharge currents
- Qc, Qd = [], []
- for ii in range(len(idx) - 1):
- i0, ie = idx[ii], idx[ii + 1]
- if ie - i0 < 10:
- continue
- q = np.trapz(I[i0:ie], t[i0:ie])
- if q > 0:
- Qc.append(q)
- else:
- Qd.append(abs(q))
 
- return {
- 'time': t,
- 'Ewe': Ewe,
- 'I': I,
- 'cn': len(Qd),
- 'time-cycles': t[idx[2::2]],
- 'Qd': np.array(Qd),
- 'Qc': np.array(Qc),
- }
-
-
-def get_data_from_results(array_node):
+def get_data_from_results(array_node) -> dict:
  "Extract data from parsed ArrayData node."
+
  if not isinstance(array_node, ArrayData):
  raise TypeError('array_node should be an ArrayData')
 
@@ -57,6 +35,12 @@ def get_data_from_results(array_node):
  Ewe = array_node.get_array('step0_Ewe_n')
  I = array_node.get_array('step0_I_n')
 
+ return post_process_data(t, Ewe, I)
+
+
+def post_process_data(t: np.ndarray, Ewe: np.ndarray, I: np.ndarray) -> dict:
+ """docstring"""
+
  # find half-cycle markers
  # add last point if not already a marker
  idx = np.where(np.diff(np.sign(I)) != 0)[0]
@@ -81,6 +65,6 @@ def get_data_from_results(array_node):
  'I': I,
  'cn': len(Qd),
  'time-cycles': t[idx[2::2]],
- 'Qd': np.array(Qd),
- 'Qc': np.array(Qc),
+ 'Qd': np.array(Qd) / 3.6,
+ 'Qc': np.array(Qc) / 3.6,
  }