Merge pull request #107 from echemdata/ml-evs/analog-in-fix

Add `Analog IN <n>/V` columns to map

.github/workflows/ci.yml

@@ -17,8 +17,7 @@ jobs:
 
   pytest:
     name: Run Python unit tests
-    # Note that 20.04 is currently required until galvani supports mdbtools>=1.0.
+    runs-on: ubuntu-22.04
-    runs-on: ubuntu-20.04
 
     strategy:
       fail-fast: false
@@ -27,13 +26,14 @@ jobs:
         python-version: ['3.8', '3.9', '3.10', '3.11']
 
     steps:
-      - uses: actions/checkout@v3
+      - name: Cached LFS checkout
+        uses: nschloe/action-cached-lfs-checkout@v1.2.2
         with:
           fetch-depth: 0
-          lfs: true
+          enableCrossOsArchive: true
 
       - name: Set up Python ${{ matrix.python-version }}
-        uses: actions/setup-python@v4
+        uses: actions/setup-python@v5
         with:
           python-version: ${{ matrix.python-version }}
 

README.md

@@ -7,21 +7,60 @@ SPDX-FileCopyrightText: 2013-2020 Christopher Kerr, Peter Attia
 SPDX-License-Identifier: GPL-3.0-or-later
 -->
 
-Read proprietary file formats from electrochemical test stations
+Read proprietary file formats from electrochemical test stations.
 
-## Bio-Logic .mpr files ##
+# Usage
+
+## Bio-Logic .mpr files
 
 Use the `MPRfile` class from BioLogic.py (exported in the main package)
 
-````
+```python
 from galvani import BioLogic
 import pandas as pd
 
 mpr_file = BioLogic.MPRfile('test.mpr')
 df = pd.DataFrame(mpr_file.data)
-````
+```
 
-## Arbin .res files ##
+## Arbin .res files
 
-Use the res2sqlite.py script to convert the .res file to a sqlite3 database
+Use the `./galvani/res2sqlite.py` script to convert the .res file to a sqlite3 database with the same schema, which can then be interrogated with external tools or directly in Python.
-with the same schema.
+For example, to extract the data into a pandas DataFrame (will need to be installed separately):
+
+```python
+import sqlite3
+import pandas as pd
+from galvani.res2sqlite import convert_arbin_to_sqlite
+convert_arbin_to_sqlite("input.res", "output.sqlite")
+with sqlite3.connect("output.sqlite") as db:
+    df = pd.read_sql(sql="select * from Channel_Normal_Table", con=db)
+```
+
+This functionality requires [MDBTools](https://github.com/mdbtools/mdbtools) to be installed on the local system.
+
+# Installation
+
+The latest galvani releases can be installed from [PyPI](https://pypi.org/project/galvani/) via
+
+```shell
+pip install galvani
+```
+
+The latest development version can be installed with `pip` directly from GitHub:
+
+```shell
+pip install git+https://github.com/echemdata/galvani
+```
+
+## Development installation and contributing 
+
+If you wish to contribute to galvani, please clone the repository and install the testing dependencies:
+
+```shell
+git clone git@github.com:echemdata/galvani
+cd galvani
+pip install -e .\[tests\]
+```
+
+Code can be contributed back via [GitHub pull requests](https://github.com/echemdata/galvani/pulls) and new features or bugs can be discussed in the [issue tracker](https://github.com/echemdata/galvani/issues).

galvani/BioLogic.py

@@ -48,6 +48,13 @@ def fieldname_to_dtype(fieldname):
         "|Z|/Ohm",
         "Re(Z)/Ohm",
         "-Im(Z)/Ohm",
+        "Re(M)",
+        "Im(M)",
+        "|M|",
+        "Re(Permittivity)",
+        "Im(Permittivity)",
+        "|Permittivity|",
+        "Tan(Delta)",
     ):
         return (fieldname, np.float_)
     elif fieldname in (
@@ -60,13 +67,13 @@ def fieldname_to_dtype(fieldname):
         "Capacity/mA.h",
     ):
         return (fieldname, np.float_)
-    elif fieldname in ("cycle number", "I Range", "Ns", "half cycle"):
+    elif fieldname in ("cycle number", "I Range", "Ns", "half cycle", "z cycle"):
         return (fieldname, np.int_)
     elif fieldname in ("dq/mA.h", "dQ/mA.h"):
         return ("dQ/mA.h", np.float_)
     elif fieldname in ("I/mA", "<I>/mA"):
         return ("I/mA", np.float_)
-    elif fieldname in ("Ewe/V", "<Ewe>/V", "Ecell/V"):
+    elif fieldname in ("Ewe/V", "<Ewe>/V", "Ecell/V", "<Ewe/V>"):
         return ("Ewe/V", np.float_)
     elif fieldname.endswith(
         (
@@ -86,8 +93,14 @@ def fieldname_to_dtype(fieldname):
             "/F",
             "/mF",
             "/uF",
+            "/µF",
+            "/nF",
             "/C",
             "/Ohm",
+            "/Ohm-1",
+            "/Ohm.cm",
+            "/mS/cm",
+            "/%",
         )
     ):
         return (fieldname, np.float_)
@@ -230,7 +243,7 @@ def MPTfileCSV(file_or_path):
     return mpt_csv, comments
 
 
-VMPmodule_hdr = np.dtype(
+VMPmodule_hdr_v1 = np.dtype(
     [
         ("shortname", "S10"),
         ("longname", "S25"),
@@ -240,17 +253,30 @@ VMPmodule_hdr = np.dtype(
     ]
 )
 
+VMPmodule_hdr_v2 = np.dtype(
+    [
+        ("shortname", "S10"),
+        ("longname", "S25"),
+        ("max length", "<u4"),
+        ("length", "<u4"),
+        ("version", "<u4"),
+        ("unknown2", "<u4"),  # 10 for set, log and loop, 11 for data
+        ("date", "S8"),
+    ]
+)
+
 # Maps from colID to a tuple defining a numpy dtype
 VMPdata_colID_dtype_map = {
     4: ("time/s", "<f8"),
     5: ("control/V/mA", "<f4"),
     6: ("Ewe/V", "<f4"),
-    7: ("dQ/mA.h", "<f8"),
+    7: ("dq/mA.h", "<f8"),
     8: ("I/mA", "<f4"),  # 8 is either I or <I> ??
     9: ("Ece/V", "<f4"),
-    11: ("I/mA", "<f8"),
+    11: ("<I>/mA", "<f8"),
     13: ("(Q-Qo)/mA.h", "<f8"),
     16: ("Analog IN 1/V", "<f4"),
+    17: ("Analog IN 2/V", "<f4"),  # Probably column 18 is Analog IN 3/V, if anyone hits this error in the future
     19: ("control/V", "<f4"),
     20: ("control/mA", "<f4"),
     23: ("dQ/mA.h", "<f8"),  # Same as 7?
@@ -267,7 +293,7 @@ VMPdata_colID_dtype_map = {
     39: ("I Range", "<u2"),
     69: ("R/Ohm", "<f4"),
     70: ("P/W", "<f4"),
-    74: ("Energy/W.h", "<f8"),
+    74: ("|Energy|/W.h", "<f8"),
     75: ("Analog OUT/V", "<f4"),
     76: ("<I>/mA", "<f4"),
     77: ("<Ewe>/V", "<f4"),
@@ -287,8 +313,30 @@ VMPdata_colID_dtype_map = {
     169: ("Cs/µF", "<f4"),
     172: ("Cp/µF", "<f4"),
     173: ("Cp-2/µF-2", "<f4"),
-    174: ("Ewe/V", "<f4"),
+    174: ("<Ewe>/V", "<f4"),
-    241: ("|E1|/V", "<f4"),
+    178: ("(Q-Qo)/C", "<f4"),
+    179: ("dQ/C", "<f4"),
+    211: ("Q charge/discharge/mA.h", "<f8"),
+    212: ("half cycle", "<u4"),
+    213: ("z cycle", "<u4"),
+    217: ("THD Ewe/%", "<f4"),
+    218: ("THD I/%", "<f4"),
+    220: ("NSD Ewe/%", "<f4"),
+    221: ("NSD I/%", "<f4"),
+    223: ("NSR Ewe/%", "<f4"),
+    224: ("NSR I/%", "<f4"),
+    230: ("|Ewe h2|/V", "<f4"),
+    231: ("|Ewe h3|/V", "<f4"),
+    232: ("|Ewe h4|/V", "<f4"),
+    233: ("|Ewe h5|/V", "<f4"),
+    234: ("|Ewe h6|/V", "<f4"),
+    235: ("|Ewe h7|/V", "<f4"),
+    236: ("|I h2|/A", "<f4"),
+    237: ("|I h3|/A", "<f4"),
+    238: ("|I h4|/A", "<f4"),
+    239: ("|I h5|/A", "<f4"),
+    240: ("|I h6|/A", "<f4"),
+    241: ("|I h7|/A", "<f4"),
     242: ("|E2|/V", "<f4"),
     271: ("Phase(Z1) / deg", "<f4"),
     272: ("Phase(Z2) / deg", "<f4"),
@@ -441,11 +489,18 @@ def read_VMP_modules(fileobj, read_module_data=True):
             raise ValueError(
                 "Found %r, expecting start of new VMP MODULE" % module_magic
             )
+        VMPmodule_hdr = VMPmodule_hdr_v1
 
+        # Reading headers binary information
         hdr_bytes = fileobj.read(VMPmodule_hdr.itemsize)
         if len(hdr_bytes) < VMPmodule_hdr.itemsize:
             raise IOError("Unexpected end of file while reading module header")
 
+        # Checking if EC-Lab version is >= 11.50
+        if hdr_bytes[35:39] == b"\xff\xff\xff\xff":
+            VMPmodule_hdr = VMPmodule_hdr_v2
+            hdr_bytes += fileobj.read(VMPmodule_hdr_v2.itemsize - VMPmodule_hdr_v1.itemsize)
+
         hdr = np.frombuffer(hdr_bytes, dtype=VMPmodule_hdr, count=1)
         hdr_dict = dict(((n, hdr[n][0]) for n in VMPmodule_hdr.names))
         hdr_dict["offset"] = fileobj.tell()
@@ -457,7 +512,11 @@ def read_VMP_modules(fileobj, read_module_data=True):
                     current module: %s
                     length read: %d
                     length expected: %d"""
-                    % (hdr_dict["longname"], len(hdr_dict["data"]), hdr_dict["length"])
+                    % (
+                        hdr_dict["longname"],
+                        len(hdr_dict["data"]),
+                        hdr_dict["length"],
+                    )
                 )
             yield hdr_dict
         else:
@@ -495,6 +554,7 @@ class MPRfile:
             raise ValueError("Invalid magic for .mpr file: %s" % magic)
 
         modules = list(read_VMP_modules(mpr_file))
+
         self.modules = modules
         (settings_mod,) = (m for m in modules if m["shortname"] == b"VMP Set   ")
         (data_module,) = (m for m in modules if m["shortname"] == b"VMP data  ")
@@ -505,15 +565,22 @@ class MPRfile:
         n_columns = np.frombuffer(data_module["data"][4:5], dtype="u1").item()
 
         if data_module["version"] == 0:
-            column_types = np.frombuffer(
+            # If EC-Lab version >= 11.50, column_types is [0 1 0 3 0 174...] instead of [1 3 174...]
-                data_module["data"][5:], dtype="u1", count=n_columns
+            if np.frombuffer(data_module["data"][5:6], dtype="u1").item():
-            )
+                column_types = np.frombuffer(data_module["data"][5:], dtype="u1", count=n_columns)
                 remaining_headers = data_module["data"][5 + n_columns:100]
                 main_data = data_module["data"][100:]
-        elif data_module["version"] in [2, 3]:
+            else:
                 column_types = np.frombuffer(
-                data_module["data"][5:], dtype="<u2", count=n_columns
+                    data_module["data"][5:], dtype="u1", count=n_columns * 2
                 )
+                column_types = column_types[1::2]  # suppressing zeros in column types array
+                # remaining headers should be empty except for bytes 5 + n_columns * 2
+                # and 1006 which are sometimes == 1
+                remaining_headers = data_module["data"][6 + n_columns * 2:1006]
+                main_data = data_module["data"][1007:]
+        elif data_module["version"] in [2, 3]:
+            column_types = np.frombuffer(data_module["data"][5:], dtype="<u2", count=n_columns)
             # There are bytes of data before the main array starts
             if data_module["version"] == 3:
                 num_bytes_before = 406  # version 3 added `\x01` to the start
@@ -542,7 +609,7 @@ class MPRfile:
         if maybe_loop_module:
             (loop_module,) = maybe_loop_module
             if loop_module["version"] == 0:
-                self.loop_index = np.fromstring(loop_module["data"][4:], dtype="<u4")
+                self.loop_index = np.frombuffer(loop_module["data"][4:], dtype="<u4")
                 self.loop_index = np.trim_zeros(self.loop_index, "b")
             else:
                 raise ValueError(

galvani/res2sqlite.py

@@ -439,7 +439,8 @@ CREATE VIEW IF NOT EXISTS Capacity_View
 def mdb_get_data_text(s3db, filename, table):
     print("Reading %s..." % table)
     insert_pattern = re.compile(
-        r'INSERT INTO "\w+" \([^)]+?\) VALUES \(("[^"]*"|[^")])+?\);\n', re.IGNORECASE
+        r"""INSERT INTO "\w+" \([^)]+?\) VALUES (\((('[^']*')|"[^"]*"|[^')])+?\),?\s*)+;\n""",
+        re.IGNORECASE,
     )
     try:
         # Initialize values to avoid NameError in except clause

tests/test_BioLogic.py

@@ -9,7 +9,7 @@ import re
 from datetime import date, datetime
 
 import numpy as np
-from numpy.testing import assert_array_almost_equal, assert_array_equal
+from numpy.testing import assert_array_almost_equal, assert_array_equal, assert_allclose
 import pytest
 
 from galvani import BioLogic, MPTfile, MPRfile
@@ -210,6 +210,95 @@ def assert_MPR_matches_MPT(mpr, mpt, comments):
         pass
 
 
+def assert_MPR_matches_MPT_v2(mpr, mpt, comments):
+    """
+    Asserts that the fields in the MPR.data ar the same as in the MPT.
+
+    Modified from assert_MPR_matches_MPT. Automatically converts dtype from MPT data
+    to dtype from MPR data before comparing the columns.
+
+    Special case for EIS_indicators: these fields are valid only at f<100kHz so their
+    values are replaced by -1 or 0 at high frequency in the MPT file, this is not the
+    case in the MPR data.
+
+    Parameters
+    ----------
+    mpr : MPRfile
+        Data extracted with the MPRfile class.
+    mpt : np.array
+        Data extracted with MPTfile method.
+
+    Returns
+    -------
+    None.
+
+    """
+
+    def assert_field_matches(fieldname):
+        EIS_quality_indicators = [
+            "THD Ewe/%",
+            "NSD Ewe/%",
+            "NSR Ewe/%",
+            "|Ewe h2|/V",
+            "|Ewe h3|/V",
+            "|Ewe h4|/V",
+            "|Ewe h5|/V",
+            "|Ewe h6|/V",
+            "|Ewe h7|/V",
+            "THD I/%",
+            "NSD I/%",
+            "NSR I/%",
+            "|I h2|/A",
+            "|I h3|/A",
+            "|I h4|/A",
+            "|I h5|/A",
+            "|I h6|/A",
+            "|I h7|/A",
+        ]
+
+        if fieldname in EIS_quality_indicators:  # EIS quality indicators only valid for f < 100kHz
+            index_inf_100k = np.where(mpr.data["freq/Hz"] < 100000)[0]
+            assert_allclose(
+                mpr.data[index_inf_100k][fieldname],
+                mpt[index_inf_100k][fieldname].astype(mpr.data[fieldname].dtype),
+            )
+        elif fieldname == "<Ewe>/V":
+            assert_allclose(
+                mpr.data[fieldname],
+                mpt["Ewe/V"].astype(mpr.data[fieldname].dtype),
+            )
+        elif fieldname == "<I>/mA":
+            assert_allclose(
+                mpr.data[fieldname],
+                mpt["I/mA"].astype(mpr.data[fieldname].dtype),
+            )
+        elif fieldname == "dq/mA.h":
+            assert_allclose(
+                mpr.data[fieldname],
+                mpt["dQ/mA.h"].astype(mpr.data[fieldname].dtype),
+            )
+        else:
+            assert_allclose(
+                mpr.data[fieldname],
+                mpt[fieldname].astype(mpr.data[fieldname].dtype),
+            )
+
+    def assert_field_exact(fieldname):
+        if fieldname in mpr.dtype.fields:
+            assert_array_equal(mpr.data[fieldname], mpt[fieldname])
+
+    for key in mpr.flags_dict.keys():
+        assert_array_equal(mpr.get_flag(key), mpt[key])
+
+    for d in mpr.dtype.descr[1:]:
+        assert_field_matches(d[0])
+
+    try:
+        assert timestamp_from_comments(comments) == mpr.timestamp.replace(microsecond=0)
+    except AttributeError:
+        pass
+
+
 @pytest.mark.parametrize(
     "basename",
     [
@@ -252,3 +341,20 @@ def test_MPR6_matches_MPT6(testdata_dir):
     mpt, comments = MPTfile(os.path.join(testdata_dir, "bio_logic6.mpt"))
     mpr.data = mpr.data[:958]  # .mpt file is incomplete
     assert_MPR_matches_MPT(mpr, mpt, comments)
+
+
+@pytest.mark.parametrize(
+    "basename_v1150",
+    ["v1150_CA", "v1150_CP", "v1150_GCPL", "v1150_GEIS", "v1150_MB", "v1150_OCV", "v1150_PEIS"],
+)
+def test_MPR_matches_MPT_v1150(testdata_dir, basename_v1150):
+    """Check the MPR parser against the MPT parser.
+
+    Load a binary .mpr file and a text .mpt file which should contain
+    exactly the same data. Check that the loaded data actually match.
+    """
+    binpath = os.path.join(testdata_dir, "v1150", basename_v1150 + ".mpr")
+    txtpath = os.path.join(testdata_dir, "v1150", basename_v1150 + ".mpt")
+    mpr = MPRfile(binpath)
+    mpt, comments = MPTfile(txtpath, encoding="latin1")
+    assert_MPR_matches_MPT_v2(mpr, mpt, comments)