Support for Neware (#9)

* Initial stubs

* Add NewareNDA dep

* Add neware test and bolster arbin res test

* Adjust neware interface

* Add test NDAx file

* Temporarily bump to my fork of NewareNDA

* Remap NewareNDA format to expected navani columns

* Update lockfile

* Catch neware reader warnings

* Update lockfile

* Disable LFS in CI so that galvani can be installed

* Update lockfile

.github/workflows/ci.yml

@@ -5,6 +5,11 @@ on:
     branches:
       - main
 
+env:
+  # Need to set this to avoid pipenv syncing an entire LFS repo and failing with bandwidth quota (see
+  # https://github.com/the-grey-group/datalab/issues/603)
+  GIT_LFS_SKIP_SMUDGE: 1
+
 jobs:
 
   pytest:

Pipfile

@@ -8,7 +8,8 @@ requests = "*"
 numpy = "*"
 pandas = "*"
 openpyxl = "*"
-galvani = { git = "git+https://github.com/chatcannon/galvani@JhonFlash-master" }
+galvani = { git = "git+https://github.com/echemdata/galvani@master" }
+NewareNDA = { git = "git+https://github.com/ml-evs/NewareNDA@ml-evs/development" }
 scipy = "*"
 matplotlib = "*"
 

navani/echem.py

@@ -1,15 +1,18 @@
 from galvani import MPRfile
 from galvani import res2sqlite as r2s
+from NewareNDA.NewareNDA import read
 import pandas as pd
 import numpy as np
+import warnings
 from scipy.signal import savgol_filter
 import sqlite3
 import os
 import matplotlib.pyplot as plt
-# Version 0.1.0
+from typing import Union
+from pathlib import Path
+
 # Different cyclers name their columns slightly differently 
 # These dictionaries are guides for the main things you want to plot
-
 res_col_dict = {'Voltage': 'Voltage',
                 'Capacity': 'Capacity'}
 
@@ -70,13 +73,17 @@ def echem_file_loader(filepath):
             df = new_land_processing(df)
         elif "Record" in names[0]:
             df_list = [xlsx.parse(0)]
+            if not isinstance(df_list, list) or not isinstance(df_list[0], pd.DataFrame):
+                raise RuntimeError("First sheet is not a dataframe; cannot continue parsing {filepath=}")
             col_names = df_list[0].columns
 
             for sheet_name in names[1:]:
                 if "Record" in sheet_name:
                     if len(xlsx.parse(sheet_name, header=None)) != 0:
                         df_list.append(xlsx.parse(sheet_name, header=None))
             for sheet in df_list:
+                if not isinstance(sheet, pd.DataFrame):
+                    raise RuntimeError("Sheet is not a dataframe; cannot continue parsing {filepath=}")
                 sheet.columns = col_names
             df = pd.concat(df_list)
             df.set_index('Index', inplace=True)
@@ -93,11 +100,16 @@ def echem_file_loader(filepath):
                 df = arbin_excel(df)
             else:
                 raise ValueError('Names of sheets not recognised')
+    elif extension in (".nda", ".ndax"):
+        df = neware_reader(filepath)
     else:
         print(extension)
+        raise RuntimeError("Filetype {extension=} not recognised.")
 
     # Adding a full cycle column
-    df['full cycle'] = (df['half cycle']/2).apply(np.ceil)
+    if "half cycle" in df.columns:
+        df['full cycle'] = (df['half cycle']/2).apply(np.ceil)
+
     return df
 
 def arbin_res(df):
@@ -330,6 +342,31 @@ def arbin_state(x):
 
     return df
 
+def neware_reader(filename: Union[str, Path]) -> pd.DataFrame:
+    """Reads a Neware NDA/NDAX file into a pandas DataFrame using NewareNDA."""
+    filename = str(filename)
+    # capture any warnings from NewareNDA and handle the ones about autoscaling
+    with warnings.catch_warnings(record=True) as w:
+        df = read(filename, error_on_missing=False)
+    for warning in w:
+        warnings.warn(f"{warning.message}", category=warning.category)
+
+    # remap to expected navani columns and units
+    df.set_index("Index", inplace=True)
+    df.index.rename("index", inplace=True)
+    df["Capacity"] = 1000 * df["Discharge_Capacity(mAh)"] + df["Charge_Capacity(mAh)"]
+    df["Current"] = 1000 * df["Current(mA)"]
+    df["state"] = pd.Categorical(values=["unknown"] * len(df["Status"]), categories=["R", 1, 0, "unknown"])
+    df["state"][df["Status"] == "Rest"] = "R"
+    df["state"][df["Status"] == "CC_Chg"] = 1
+    df["state"][df["Status"] == "CC_DChg"] = 0
+    df["half cycle"] = df["Cycle"]
+    df['cycle change'] = False
+    not_rest_idx = df[df['state'] != 'R'].index
+    df.loc[not_rest_idx, 'cycle change'] = df.loc[not_rest_idx, 'state'].ne(df.loc[not_rest_idx, 'state'].shift())
+    return df
+
+
 def dqdv_single_cycle(capacity, voltage, 
                     polynomial_spline=3, s_spline=1e-5,
                     polyorder_1 = 5, window_size_1=101,

tests/test_echem.py

@@ -166,6 +166,58 @@ def test_arbin_res():
     )
     df = ec.echem_file_loader(test_path)
 
+    cols = (
+        "state",
+        "cycle change",
+        "half cycle",
+        "Capacity",
+        "Voltage",
+        "Current",
+        "full cycle",
+    )
+
+    assert all(c in df for c in cols)
+
+def test_nda():
+    import navani.echem as ec
+
+    test_path = pathlib.Path(__file__).parent.parent / "Example_data" / "test.nda"
+
+    with pytest.warns(RuntimeWarning, match="scaling") as record:
+        df = ec.echem_file_loader(test_path)
+
+    # Filter out any other warning messages
+    record = [r for r in record if r.category is RuntimeWarning and "scaling" in str(r.message)]
+    assert len(record) == 1
+    cols = (
+        "state",
+        "cycle change",
+        "half cycle",
+        "Capacity",
+        "Voltage",
+        "Current",
+        "full cycle",
+    )
+    assert all(c in df for c in cols), f"Some columns from {cols} were missing in {df.columns}: {set(cols) - set(df.columns)}"
+
+def test_ndax():
+    import navani.echem as ec 
+
+    test_path = pathlib.Path(__file__).parent.parent / "Example_data" / "test.ndax"
+
+    df = ec.echem_file_loader(test_path)
+    cols = (
+        "state",
+        "cycle change",
+        "half cycle",
+        "Capacity",
+        "Voltage",
+        "Current",
+        "full cycle",
+    )
+
+    assert all(c in df for c in cols), f"Some columns from {cols} were missing in {df.columns}"
+
 @pytest.mark.parametrize("test_path", [
     "00_test_01_OCV_C01.mpr",
     "00_test_02_MB_C01.mpr",