EIS: allow other circuits, tags, param reporting

pyflowbatt/analysis.py

@@ -15,7 +15,12 @@
 from battinfoconverter_backend import convert_excel_to_jsonld
 
 from pyflowbatt import battinfo, cv, eis, gcpl, lsv, ocv
-from pyflowbatt.config import DEFAULT_AREA_CM2, PyFlowBattConfig, classify_technique_files
+from pyflowbatt.config import (
+    DEFAULT_AREA_CM2,
+    EIS_TAGS,
+    PyFlowBattConfig,
+    classify_technique_files,
+)
 from pyflowbatt.read import read_to_bdf
 
 logger = logging.getLogger(__name__)
@@ -143,6 +148,13 @@ def _generic_eis_parts(label: str) -> tuple[str, str] | None:
     return ("before" if match.group(1) == "pre" else "after", match.group(2))
 
 
+def _custom_eis_tag(label: str) -> str | None:
+    """Tag of an EIS label set in pyflowbatt.toml, e.g. "pre-20%SOC" from "eis_pre-20%SOC"."""
+    if not label.startswith("eis_") or _GENERIC_EIS_RE.match(label):
+        return None
+    return label.removeprefix("eis_")
+
+
 def raw_description(label: str, path: Path) -> str | None:
     """Describe a raw input file, naming the format only where we know what it is."""
     if label in RAW_INPUT_DESCRIPTIONS:
@@ -152,6 +164,8 @@ def raw_description(label: str, path: Path) -> str | None:
     elif parts := _generic_eis_parts(label):
         when, number = parts
         description = f"EIS measurement, {when} cycling, measurement {number}"
+    elif tag := _custom_eis_tag(label):
+        description = f"EIS measurement, tagged {tag}"
     else:
         return None
     if path.suffix.lower() == ".mpr":
@@ -169,6 +183,8 @@ def plot_description(label: str) -> str | None:
             "Nyquist plot for electrochemical impedance spectroscopy (EIS) "
             f"from {when} cycling, measurement {number}"
         )
+    if tag := _custom_eis_tag(label):
+        return f"Nyquist plot for electrochemical impedance spectroscopy (EIS), tagged {tag}"
     return None
 
 
@@ -182,6 +198,11 @@ def data_description(label: str) -> str | None:
             "Frequency-domain electrochemical impedance spectroscopy (EIS) data "
             f"from {when} cycling, measurement {number}, using the battery data format (BDF)"
         )
+    if tag := _custom_eis_tag(label):
+        return (
+            "Frequency-domain electrochemical impedance spectroscopy (EIS) data, "
+            f"tagged {tag}, using the battery data format (BDF)"
+        )
     return None
 
 
@@ -714,55 +735,59 @@ def analyse_sample(
     logger.info("🌈 Analysing PEIS")
     eis_res = {}
     rows = []
+    eis_units: dict[str, str] = {}
+    has_ml_guess = False
+    eis_to_fit = eis_by_tag
     if not eis_by_tag:
         logger.warning("- ☹️ No EIS files were found, skipping")
     else:
-        for tag, eis_file in eis_by_tag.items():
-            f = Path(eis_file)
-            try:
-                df = read_to_bdf(f)
-                params, Z_fit = eis.analyse(df, label=f.stem)
-                fig, _ax = eis.plot(df, Z_fit)
-                fig.savefig(results_dir / f"eis_{tag}.png")
-                plt.close(fig)
-                eis_res[tag] = params
-                df["Real Impedance Fit / ohm"] = np.real(Z_fit)
-                df["Imaginary Impedance Fit / ohm"] = np.imag(Z_fit)
-                df_save_bdf(df, results_dir / f"eis_{tag}.x", save_format=save_format)
-                tracked_outputs[f"eis_{tag}"] = [results_dir / f"eis_{tag}.png"]
-                if bdf_suffix:
-                    tracked_outputs[f"eis_{tag}"].append(
-                        (results_dir / f"eis_{tag}.x").with_suffix(bdf_suffix)
-                    )
-                tracked_mpr_inputs[f"eis_{tag}"] = [Path(eis_file)]
-                for name, values in params.items():
-                    rows.append({"file": f.stem, "tag": tag, "name": name, **values})
-            except Exception as e:
-                logger.warning("- Failed to fit %s: %s", f.stem, str(e))
-        if rows:
-            eis_df = pd.DataFrame(rows)
-            eis_df = eis_df.pivot(index=["file", "tag"], columns=["name"]).reset_index()
-            eis_df.columns = [
-                f"{name}_{field}" if name else field for field, name in eis_df.columns
-            ]
-            order = [
-                f"{elem}_{x}"
-                for elem in [
-                    "L0.l",
-                    "R0.r",
-                    "R1.r",
-                    "CPE1.q",
-                    "CPE1.alpha",
-                    "R2.r",
-                    "CPE2.q",
-                    "CPE2.alpha",
-                ]
-                for x in ["value", "err", "unit"]
-            ]
-            eis_df = eis_df[["file", "tag", *order]]
+        try:
+            eis_units, has_ml_guess = eis.check_model(config.eis_model, config.eis_summary_params)
+        except ValueError as e:
+            logger.warning("- Invalid EIS model %s, skipping EIS: %s", config.eis_model, e)
+            eis_to_fit = {}
+    for tag, eis_file in eis_to_fit.items():
+        f = Path(eis_file)
+        try:
+            df = read_to_bdf(f)
+            params, Z_fit = eis.analyse(
+                df, label=f.stem, model=config.eis_model, guess_init=has_ml_guess
+            )
+            fig, _ax = eis.plot(df, Z_fit)
+            fig.savefig(results_dir / f"eis_{tag}.png")
+            plt.close(fig)
+            eis_res[tag] = params
+            df["Real Impedance Fit / ohm"] = np.real(Z_fit)
+            df["Imaginary Impedance Fit / ohm"] = np.imag(Z_fit)
+            df_save_bdf(df, results_dir / f"eis_{tag}.x", save_format=save_format)
+            tracked_outputs[f"eis_{tag}"] = [results_dir / f"eis_{tag}.png"]
+            if bdf_suffix:
+                tracked_outputs[f"eis_{tag}"].append(
+                    (results_dir / f"eis_{tag}.x").with_suffix(bdf_suffix)
+                )
+            tracked_mpr_inputs[f"eis_{tag}"] = [Path(eis_file)]
+            for name, values in params.items():
+                rows.append({"file": f.stem, "tag": tag, "name": name, **values})
+        except Exception as e:
+            logger.warning("- Failed to fit %s: %s", f.stem, str(e))
+    if rows:
+        eis_df = pd.DataFrame(rows)
+        eis_df = eis_df.pivot(index=["file", "tag"], columns=["name"]).reset_index()
+        eis_df.columns = [f"{name}_{field}" if name else field for field, name in eis_df.columns]
+        order = [f"{name}_{x}" for name in eis_units for x in ["value", "err", "unit"]]
+        eis_df = eis_df[["file", "tag", *order]]
 
     logger.info("💪 Making sample summary")
 
+    # Standard tags always get rows so combined summaries line up, other tags follow
+    eis_tags = [*EIS_TAGS, *(tag for tag in eis_res if tag not in EIS_TAGS)]
+    eis_rows = {}
+    for param, unit in eis.summary_units(config.eis_summary_params).items():
+        for tag in eis_tags:
+            fit = eis_res.get(tag, {}).get(param, {})
+            label = eis.summary_label(param, tag, unit)
+            eis_rows[label] = {"Value": fit.get("value"), "Error": fit.get("err")}
+
     # Create keys
     cols = [
         "Av. OCV / V",
@@ -773,8 +798,7 @@ def analyse_sample(
         "F post / mF",
         "∆F / mF",
         "Assembled resistance / Ω",
-        "EIS R pre / Ω",
-        "EIS R pre-50%SOC / Ω",
+        *eis_rows,
         "1st CE / %",
         "1st EE / %",
         "1st VE / %",
@@ -805,14 +829,7 @@ def analyse_sample(
         cv_res["post"] - cv_res["pre"] if (cv_res["post"] and cv_res["pre"]) else None
     )
     summary["Assembled resistance / Ω"]["Value"] = assembled_resistance_ohm
-    summary["EIS R pre / Ω"]["Value"] = eis_res.get("pre", {}).get("R0.r", {}).get("value")
-    summary["EIS R pre / Ω"]["Error"] = eis_res.get("pre", {}).get("R0.r", {}).get("err")
-    summary["EIS R pre-50%SOC / Ω"]["Value"] = (
-        eis_res.get("pre-50%SOC", {}).get("R0.r", {}).get("value")
-    )
-    summary["EIS R pre-50%SOC / Ω"]["Error"] = (
-        eis_res.get("pre-50%SOC", {}).get("R0.r", {}).get("err")
-    )
+    summary.update(eis_rows)
 
     if cycle_df is not None:
         mask = cycle_df["Cycle Count / 1"] == 1
@@ -1062,6 +1079,18 @@ def find_all_sample_summaries(
     ]
 
 
+def _merged_row_order(indexes: list[pd.Index]) -> list[str]:
+    """All row labels, each label missing from earlier summaries placed after its predecessor."""
+    order: list[str] = []
+    for index in indexes:
+        prev = None
+        for label in index:
+            if label not in order:
+                order.insert(order.index(prev) + 1 if prev is not None else 0, label)
+            prev = label
+    return order
+
+
 def merge_summaries(
     summary_xlsxs: list[str | Path],
     sample_ids_by_sample_folder: dict[Path, str] | None = None,
@@ -1081,7 +1110,7 @@ def merge_summaries(
     dfs = [df["Value"] for df in dfs]
     for name, df in zip(names, dfs, strict=True):
         df.name = name
-    df = pd.concat(dfs, axis=1)
+    df = pd.concat(dfs, axis=1).reindex(_merged_row_order([df.index for df in dfs]))
     vals = df.copy()
 
     df["Average"] = vals.mean(axis=1)

pyflowbatt/battinfo.py

@@ -602,7 +602,7 @@ def add_data(
             "csvw:tableSchema": "https://w3id.org/battery-data-alliance/ontology/battery-data-format/schema",
             "rdfs:comment": "Time series electrochemical data using Battery Data Format (bdf) columns",
         }
-        if "eis." in rel_file_path:
+        if Path(rel_file_path).name.startswith("eis_"):
             additions["rdfs:comment"] = (
                 "Frequency-domain electrochemical data using Battery Data Format (bdf) columns"
             )
@@ -612,7 +612,7 @@ def add_data(
             "csvw:dialect": {"@type": "csvw:Dialect", "csvw:delimiter": ",", "csvw:skipRows": 0},
             "rdfs:comment": "Time series electrochemical data using Battery Data Format (bdf) columns",
         }
-        if "eis" in rel_file_path:
+        if Path(rel_file_path).name.startswith("eis_"):
             additions["rdfs:comment"] = (
                 "Frequency-domain electrochemical data using Battery Data Format (bdf) columns"
             )

pyflowbatt/cli.py

@@ -4,7 +4,7 @@
 import logging
 from pathlib import Path
 
-from pyflowbatt.config import CONFIG_FILENAME, write_template_config
+from pyflowbatt.config import CONFIG_FILENAME, ConfigError, write_template_config
 
 
 def init_config(folder: str | None = None) -> None:
@@ -43,20 +43,17 @@ def analyse(
     if not folderpath.is_dir():
         logger.error("%s is not a directory", folderpath)
         return
-    if dry:
-        dry_analyse_all_samples(
-            folderpath,
-            max_folder_searches=max_folder_searches,
-            max_search_depth=max_search_depth,
-            zip_output=zip_output,
-        )
-    else:
-        analyse_all_samples(
+    run = dry_analyse_all_samples if dry else analyse_all_samples
+    try:
+        run(
             folderpath,
             max_folder_searches=max_folder_searches,
             max_search_depth=max_search_depth,
             zip_output=zip_output,
         )
+    except ConfigError as e:
+        # The message names the file and line, a traceback would only hide it
+        logger.error("%s", e)  # noqa: TRY400
 
 
 def main() -> None: