diff --git a/pyflowbatt/analysis.py b/pyflowbatt/analysis.py index 1abfe4d..4eb06db 100644 --- a/pyflowbatt/analysis.py +++ b/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) diff --git a/pyflowbatt/battinfo.py b/pyflowbatt/battinfo.py index 8cb9895..318a2ff 100644 --- a/pyflowbatt/battinfo.py +++ b/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" ) diff --git a/pyflowbatt/cli.py b/pyflowbatt/cli.py index 111c389..696bc32 100644 --- a/pyflowbatt/cli.py +++ b/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: diff --git a/pyflowbatt/config.py b/pyflowbatt/config.py index 15dee8b..224f16c 100644 --- a/pyflowbatt/config.py +++ b/pyflowbatt/config.py @@ -2,6 +2,7 @@ from __future__ import annotations +import difflib import logging import re from dataclasses import dataclass, field @@ -16,9 +17,96 @@ CONFIG_FILENAME = "pyflowbatt.toml" EIS_TAGS = ["pre", "pre-50%SOC", "post-50%SOC", "post"] +# Standard protocol layouts, keyed by (EIS count before, after cycling), in time order +_EIS_LAYOUTS: dict[tuple[int, int], tuple[list[str], list[str]]] = { + (2, 2): (["pre", "pre-50%SOC"], ["post-50%SOC", "post"]), + (1, 1): (["pre"], ["post"]), +} DEFAULT_AREA_CM2 = 5.0 # fallback electrode area (cm^2) when no toml or BattINFO value exists _LSV_NUMBER_RE = re.compile(r"_(\d+)_LSV_") _STEP_NUMBER_RE = re.compile(r"_(\d+)_[A-Za-z]") +_EIS_TAG_RE = re.compile(r"^[A-Za-z0-9%.+-]+$") # tags become part of output filenames + +# ------------ Warning/error messages for reading in settings files ---------- # + +_TOML_ERROR_LINE_RE = re.compile(r"at line (\d+)") +_TOML_TABLE_RE = re.compile(r"^\s*\[\s*([^\]]+?)\s*\]") +_TOML_KEY_RE = re.compile(r"""^\s*["']?([^"'=\s]+)["']?\s*=""") + + +class ConfigError(ValueError): + """A pyflowbatt.toml file that cannot be read.""" + + +def _read_toml(path: Path) -> tuple[dict, str]: + """Parse a config file into data and text, raising ConfigError naming the bad line.""" + try: + text = path.read_text(encoding="utf-8-sig") + except UnicodeDecodeError as e: + msg = f"Cannot read {path}: not saved as UTF-8 text ({e})" + raise ConfigError(msg) from e + try: + return tomllib.loads(text), text + except tomllib.TOMLDecodeError as e: + msg = f"Cannot read {path}: {e}" + lines = text.splitlines() + line = None + if (match := _TOML_ERROR_LINE_RE.search(str(e))) and int(match.group(1)) <= len(lines): + lineno = int(match.group(1)) + line = lines[lineno - 1].strip() + msg += f"\n line {lineno}: {line}" + msg += f"\n{_toml_error_hint(str(e), line)}" + raise ConfigError(msg) from e + + +# tomllib error message fragment -> likely cause, checked in order +_TOML_ERROR_HINTS: list[tuple[str, str]] = [ + ( + "twice", + "A [section] appears twice, or a setting with the same name sits above it. " + "Check that you deleted the # in front of the [section] line above each setting you use, " + "and that each [section] appears once.", + ), + ("Cannot overwrite a value", "This setting is set twice in this file: keep only one."), + ( + "Unescaped '\\'", + "A \\ inside double quotes starts an escape code. Use single quotes for text with " + "backslashes, such as regexes and Windows paths, e.g. '^\\d+_.+$'.", + ), + ("Illegal character '\\n'", "A text value is missing its closing quote."), + ("Unclosed array", "A list is missing its closing ], or a comma between items."), + ( + "Invalid value", + 'Text values need quotes, e.g. model = "R0-(R1,CPE1)". Lists use [ ], ' + "true/false are lowercase, and numbers have no units.", + ), + ( + "Expected newline or end of document", + "Only put one value after the =, with no units, e.g. area_cm2 = 5 not 5 cm2. " + "A note after a value needs a # in front of it.", + ), +] + + +def _toml_error_hint(error: str, line: str | None) -> str: + """Likely cause of a TOML parse error, from the error message and the offending line.""" + for fragment, hint in _TOML_ERROR_HINTS: + if fragment in error: + return hint + if line is not None and line.startswith("["): + return "Section lines look like [eis], with nothing else on the line." + if line is not None and "=" not in line: + return ( + "This looks like an explanation line (## in the template). Explanations must " + "stay commented: put the # back at the start of the line." + ) + if "Expected '=' after a key" in error: + return "Setting names before the = can't contain spaces, e.g. area_cm2 = 5." + return ( + "Only delete the single # in front of a setting or [section] line, and check its " + 'value: text in quotes, lists in [ ], numbers bare, e.g. model = "R0-(R1,CPE1)".' + ) + _TECHNIQUE_KEYS: dict[str, str] = { "gcpl": "gcpl_patterns", @@ -38,11 +126,91 @@ } +# Settings allowed in each section, "" for the top level; [eis_tags] takes any tag name +_SECTION_KEYS: dict[str, set[str]] = { + "": { + "sample_name", + "sample_name_pattern", + "extensions", + "extra_extensions", + "lsv_threshold", + "area_cm2", + "assembled_resistance_ohm", + "summary_n_cycles", + }, + "extra_patterns": set(_TECHNIQUE_KEYS), + "eis_tags": set(), + "eis": {"model", "summary_params"}, + "cv": set(_CV_KEYS), +} + + def _is_number(value: object) -> bool: """Check if value is an int or float.""" return isinstance(value, (int, float)) and not isinstance(value, bool) +def _section(data: dict, name: str) -> dict: + """Get the [name] table of a parsed config, empty if missing or not a table.""" + table = data.get(name, {}) + return table if isinstance(table, dict) else {} + + +def _setting_section(key: str) -> str | None: + """Section a setting belongs in, "" for the top level, None if it isn't a setting.""" + return next((section for section, keys in _SECTION_KEYS.items() if key in keys), None) + + +def _misplaced_hint(key: str, home: str | None, *, is_table: bool) -> str: + """Explain where a setting found in the wrong place belongs.""" + if is_table: + sections = [name for name in _SECTION_KEYS if name] + close = difflib.get_close_matches(key, sections, n=1) + return f"unknown section, did you mean [{close[0]}]?" if close else "unknown section" + if home is None and key in _SECTION_KEYS: + return f"{key} is a section, so write it as a [{key}] line with its settings below" + if home is None: + close = difflib.get_close_matches(key, set().union(*_SECTION_KEYS.values()), n=1) + return f"unknown setting, did you mean {close[0]}?" if close else "unknown setting" + if home == "": + return "it is a top-level setting, so move it above the first [section] line" + return ( + f"it belongs under [{home}], so delete the # in front of the [{home}] line " + f"above it, or move it there" + ) + + +def _key_line(text: str, section: str, key: str) -> int | None: + """Line number of a key in a [section] ("" for the top level), or of a [key] line.""" + current = "" + for lineno, line in enumerate(text.splitlines(), start=1): + if match := _TOML_TABLE_RE.match(line): + current = match.group(1).strip("\"'") + if section == "" and current == key: + return lineno + elif current == section and (match := _TOML_KEY_RE.match(line)) and match.group(1) == key: + return lineno + return None + + +def _warn_misplaced_settings(data: dict, text: str, path: Path) -> None: + """Warn about unknown settings and settings in the wrong section, saying where they belong.""" + tables = {"": data} | {name: _section(data, name) for name in _SECTION_KEYS if name} + for section, table in tables.items(): + where = f"in [{section}]" if section else "at the top level" + for key, value in table.items(): + home = _setting_section(key) + is_table = section == "" and isinstance(value, dict) + if (is_table and key in _SECTION_KEYS) or home == section: + continue + if section == "eis_tags" and home is None: + continue + hint = _misplaced_hint(key, home, is_table=is_table) + lineno = _key_line(text, section, key) + location = f"{path}, line {lineno}" if lineno else str(path) + logger.warning("Ignoring %s %s (%s): %s", key, where, location, hint) + + @dataclass class PyFlowBattConfig: r"""Configuration controlling how PyFlowBatt detects technique files and sample IDs. @@ -66,10 +234,15 @@ class PyFlowBattConfig: eis = ["*_EIS_*"] # adds to the built-in *_PEIS_* gcpl = ["*_GCD_*"] - [eis_tags] # pin EIS files to tags instead of filename order + [eis_tags] # pin EIS files to tags, any name, instead of by layout "pre" = ["*_02_PEIS_*"] + "pre-20%SOC" = ["*_03_PEIS_*"] "post" = ["*_06_PEIS_*"] + [eis] + model = "L0-R0-(R1,CPE1)-(R2,CPE2)" # any fasteis circuit string + summary_params = ["R0.r", "R1.r"] # fit parameters reported in the summary + [cv] v_min = 0.4 # voltage window used to plot v_max = 0.6 @@ -96,8 +269,10 @@ class PyFlowBattConfig: cv_post_patterns: list[str] = field(default_factory=lambda: ["*_CVApost*", "*_CVpost*"]) eis_patterns: list[str] = field(default_factory=lambda: ["*_PEIS_*"]) extensions: list[str] = field(default_factory=lambda: [".mpr"]) - # EIS tag -> glob patterns; tags not listed are assigned in filename order + # EIS tag -> glob patterns; unpinned files are tagged by protocol layout eis_tag_patterns: dict[str, list[str]] = field(default_factory=dict) + eis_model: str = "L0-R0-(R1,CPE1)-(R2,CPE2)" # fasteis circuit string fitted to EIS + eis_summary_params: list[str] = field(default_factory=lambda: ["R0.r"]) sample_name_pattern: str = r"^\d+_.+_.+$" sample_name: str | None = None # overrides pattern entirely lsv_threshold: int = 8 # numeric cutoff for pre/post when only one LSV file is found @@ -159,10 +334,10 @@ def all_patterns(self) -> list[str]: def _apply_toml(self, path: Path) -> None: """Merge a single TOML config file into this config (mutates in place).""" - with path.open("rb") as f: - data = tomllib.load(f) + data, text = _read_toml(path) + _warn_misplaced_settings(data, text, path) - extra = data.get("extra_patterns", {}) + extra = _section(data, "extra_patterns") for toml_key, attr in _TECHNIQUE_KEYS.items(): if toml_key in extra: values = extra[toml_key] @@ -178,11 +353,15 @@ def _apply_toml(self, path: Path) -> None: if v not in current: current.append(v) - eis_tags = data.get("eis_tags", {}) + eis_tags = _section(data, "eis_tags") for tag, values in eis_tags.items(): - if tag not in EIS_TAGS: + if _setting_section(tag) is not None: + continue # a misplaced setting, already warned about + if not _EIS_TAG_RE.match(tag): logger.warning( - "Ignoring unknown eis_tags.%s in %s (expected one of %s)", tag, path, EIS_TAGS + "Ignoring eis_tags.%s in %s (tags may only use letters, digits and %%.+-)", + tag, + path, ) continue if not isinstance(values, list) or not all(isinstance(v, str) for v in values): @@ -192,6 +371,21 @@ def _apply_toml(self, path: Path) -> None: continue self.eis_tag_patterns[tag] = list(values) + eis = _section(data, "eis") + if "model" in eis: + if isinstance(eis["model"], str): + self.eis_model = eis["model"] + else: + logger.warning("Ignoring bad eis.model in %s (expected string)", path) + if "summary_params" in eis: + values = eis["summary_params"] + if isinstance(values, list) and all(isinstance(v, str) for v in values): + self.eis_summary_params = list(values) + else: + logger.warning( + "Ignoring bad eis.summary_params in %s (expected list of strings)", path + ) + for key in ("extensions", "extra_extensions"): if key in data: values = data[key] @@ -234,7 +428,7 @@ def _apply_toml(self, path: Path) -> None: "Ignoring bad summary_n_cycles in %s (expected list of numbers)", path ) - cv = data.get("cv", {}) + cv = _section(data, "cv") for toml_key, attr in _CV_KEYS.items(): if toml_key in cv: if _is_number(cv[toml_key]): @@ -256,39 +450,42 @@ def _apply_toml(self, path: Path) -> None: TEMPLATE_TOML = """\ -# pyflowbatt.toml — optional PyFlowBatt configuration. -# -# Every setting below is commented out and shows its built-in default. -# Uncomment and edit only the settings you want to override. -# -# You can put config files in: -# The home folder (~/pyflowbatt.toml): will apply EVERYWHERE -# The parent folder: applies to every sample folder inside -# The sample folder itself: only applies to that sample -# On conflicts, sample takes priority, then parent, then home. -# -# --- Sample name (default: reads from battinfo, otherwise uses folder name) --- +## pyflowbatt.toml: optional PyFlowBatt configuration. +## +## HOW TO EDIT THIS FILE +## A '#' at the start of a line comments it out +## Lines starting with ## are explanations: leave them as comments +## Lines starting with # are settings: you can remove the # and edit the value +## Some settings are under a [section] line, like [eis] +## You need to uncomment the [section] to use those settings +## +## You can put config files in: +## The home folder (~/pyflowbatt.toml): will apply EVERYWHERE +## The parent folder: applies to every sample folder inside +## The sample folder itself: only applies to that sample +## On conflicts, sample takes priority, then parent, then home. + +## --- Sample name (default: reads from battinfo, otherwise uses folder name) --- # sample_name = "daves-new-sample-01" # hard-code an explicit sample name # sample_name_pattern = "^[A-Z]+-\\\\d+$" # or a custom regex, if not hard-coded -# --- File extensions searched for technique files --- - +## --- File extensions searched for technique files --- # extra_extensions = [".mgr"] # add extra extensions to the default [".mpr"] # extensions = [".mpr", ".mgr"] # or fully replace the extension list instead of adding to it -# --- LSV pre/post split: file numbers below this are "pre", at/above are "post" --- +## --- LSV pre/post split: file numbers below this are "pre", at/above are "post" --- # lsv_threshold = 8 -# --- Electrode area (cm^2), used to normalise LSV resistance --- +## --- Electrode area (cm^2), used to normalise LSV resistance --- # area_cm2 = 5 -# --- Assembled/external resistance (ohms) --- +## --- Assembled/external resistance (ohms) --- # assembled_resistance_ohm = 25000 -# --- Cycle counts reported in the summary sheet --- +## --- Cycle counts reported in the summary sheet --- # summary_n_cycles = [10, 20, 30, 40, 50] -# --- Extra filename glob patterns per technique (added to the built-in patterns below) --- +## --- Extra filename glob patterns per technique (added to the built-in patterns below) --- # [extra_patterns] # gcpl = ["*_cycling_*"] # built-in: *_GCPL_* # ocv = ["*_rest_*"] # built-in: *_OCV_* @@ -297,14 +494,32 @@ def _apply_toml(self, path: Path) -> None: # cv_post = ["*_cyclic-voltammetry-post_*"] # built-in: *_CVApost*, *_CVpost* # eis = ["*_impedance_*"] # built-in: *_PEIS_* -# --- Pin EIS files to specific tags (default: tagged in filename order) --- -# Tags: "pre", "pre-50%SOC", "post-50%SOC", "post". Unlisted tags are filled in -# filename order from the remaining EIS files. +## --- Pin EIS files to specific tags (default: tagged by protocol layout) --- +## Standard tags: "pre", "pre-50%SOC", "post-50%SOC", "post". +## Any other name made of letters, digits and %.+- also works, e.g. "pre-20%SOC". +## Each tag gets its own row in the summary sheet. +## Unpinned EIS files are split before/after cycling and fill the unused standard +## tags of a standard layout (2 before + 2 after, or 1 + 1). Pinned standard tags +## count towards the layout, custom tags don't. If the unpinned files then fit no +## standard layout, they all get numbered tags instead: pre_1, post_1, post_2, ... +## e.g. pinning "pre-20%SOC" to one of four standard files leaves 1 + 2, so the +## other three become pre_1, post_1, post_2. # [eis_tags] # "pre" = ["*_02_PEIS_*"] +# "pre-20%SOC" = ["*_03_PEIS_*"] # "post" = ["*_06_PEIS_*"] -# --- CV analysis parameters --- +## --- EIS equivalent circuit model --- +## Any fasteis circuit string e.g. "R0-(R1,CPE1)", or a built-in name like "randles". +## Only models in the fasteis ML library get automatic initial fit parameters. +## See https://empaeconversion.github.io/fasteis/models/ +## summary_params lists the fit parameters reported in the summary for each EIS tag, +## named as in the EIS sheet, e.g. ["R0.r", "R1.r", "CPE1.alpha"]. +# [eis] +# model = "L0-R0-(R1,CPE1)-(R2,CPE2)" +# summary_params = ["R0.r"] + +## --- CV analysis parameters --- # [cv] # v_min = 0.4004 # voltage window used to compute scan rate # v_max = 0.6 @@ -425,13 +640,12 @@ def _classify_eis( ) -> dict[str, list[Path]]: """Tag EIS files, splitting them around the main cycling step. - Files pinned by `eis_tag_patterns` win; any remaining files are tagged in - filename order if pins exist, otherwise by :func:`_eis_order_and_tags`. + Files pinned by `eis_tag_patterns` win; any remaining files are tagged by + :func:`_eis_order_and_tags`, with pinned standard tags counted in the layout. """ result: dict[str, list[Path]] = {} pinned: set[Path] = set() - for tag in EIS_TAGS: - patterns = config.eis_tag_patterns.get(tag) + for tag, patterns in config.eis_tag_patterns.items(): if not patterns: continue matches = sorted(_glob_many(folder, patterns, config.extensions)) @@ -451,12 +665,8 @@ def _classify_eis( if not free_files: return result - # With pins in play, the remaining files just fill whatever tags are left over. - if pinned: - ordered = free_files - tags = [tag for tag in EIS_TAGS if f"eis_{tag}" not in result] - else: - ordered, tags = _eis_order_and_tags(free_files, gcpl_file, warn=warn) + taken = {tag for tag in EIS_TAGS if f"eis_{tag}" in result} + ordered, tags = _eis_order_and_tags(free_files, gcpl_file, taken=taken, warn=warn) for tag, f in zip(tags, ordered, strict=False): result[f"eis_{tag}"] = [f] return result @@ -510,31 +720,42 @@ def _eis_sort_keys( def _eis_order_and_tags( - eis_files: list[Path], gcpl_file: Path | None, *, warn: bool + eis_files: list[Path], + gcpl_file: Path | None, + *, + taken: set[str] | None = None, + warn: bool, ) -> tuple[list[Path], list[str]]: """Order EIS files by when they ran and tag them by the protocol's layout. Two before and two after cycling is the standard protocol (0% and 50% SOC, each - side); one before and one after is the early protocol (0% SOC either side). Any - other layout can't be mapped onto the SOC-specific tags, so files get numbered - `pre_N`/`post_N` tags instead. Falls back to filename order when neither step - numbers nor timestamps are readable. + side); one before and one after is the early protocol (0% SOC either side). + Standard tags in `taken` are already pinned to other files: they count towards + the layout, and `eis_files` fill the rest. Any other layout can't be mapped onto + the SOC-specific tags, so files get numbered `pre_N`/`post_N` tags instead. + Falls back to filename order when neither step numbers nor timestamps are readable. """ + taken = taken or set() keys, gcpl_key = _eis_sort_keys(eis_files, gcpl_file) if keys is None or gcpl_key is None: if warn: logger.warning( "Cannot tell which EIS files are before/after cycling, tagging in filename order" ) - return eis_files, EIS_TAGS + return eis_files, [tag for tag in EIS_TAGS if tag not in taken] ordered = [f for _, f in sorted(zip(keys, eis_files, strict=True))] before = [key for key in keys if key < gcpl_key] after = [key for key in keys if key >= gcpl_key] - if len(before) == 2 and len(after) == 2: - return ordered, EIS_TAGS - if len(before) == 1 and len(after) == 1: - return ordered, ["pre", "post"] + n_pre_taken = sum(tag.startswith("pre") for tag in taken) + n_post_taken = len(taken) - n_pre_taken + layout = _EIS_LAYOUTS.get((len(before) + n_pre_taken, len(after) + n_post_taken)) + if layout: + pre_tags = [tag for tag in layout[0] if tag not in taken] + post_tags = [tag for tag in layout[1] if tag not in taken] + # A pinned tag outside the layout, e.g. pre-50%SOC in the early protocol, won't fit + if len(pre_tags) == len(before) and len(post_tags) == len(after): + return ordered, pre_tags + post_tags if warn: logger.warning( diff --git a/pyflowbatt/eis.py b/pyflowbatt/eis.py index bc662ff..ff8f966 100644 --- a/pyflowbatt/eis.py +++ b/pyflowbatt/eis.py @@ -14,18 +14,105 @@ logger = logging.getLogger(__name__) +DEFAULT_MODEL = "L0-R0-(R1,CPE1)-(R2,CPE2)" +MODELS_URL = "https://empaeconversion.github.io/fasteis/models/" + +# fasteis unit strings -> units for summary row labels +PRETTY_UNITS: dict[str, str] = { + "ohm": "Ω", + "F": "F", + "H": "H", + "H*s": "H s", + "s": "s", + "-": "1", + "ohm^-1*s^alpha": "Ω⁻¹ s^α", # noqa: RUF001 + "ohm*s^-0.5": "Ω s^-0.5", + "F*s^(gamma-1)": "F s^(γ-1)", # noqa: RUF001 + "ohm*m^2": "Ω m²", +} + + +def summary_label(param: str, tag: str, unit: str) -> str: + """Summary sheet row label for one fit parameter of one tagged EIS measurement.""" + return f"EIS {param} {tag} / {PRETTY_UNITS.get(unit, unit)}" + + +def summary_units(summary_params: list[str]) -> dict[str, str]: + """Units of each summary parameter. + + Looks up element in fasteis, so summary rows exist even if the full model is invalid or there + are no EIS files. Names that aren't a fasteis element parameter are warned about and dropped. + """ + import fasteis # noqa: PLC0415 + + units = {} + for param in summary_params: + try: + circuit = fasteis.Circuit(param.split(".")[0]) + element_units = dict(zip(circuit.param_names(), circuit.param_units(), strict=True)) + except ValueError: + element_units = {} + if param in element_units: + units[param] = element_units[param] + else: + logger.warning( + "- EIS summary parameter %s is not a fasteis parameter name, so it is left out " + "of the summary. Use names like R0.r or CPE1.alpha, see %s", + param, + MODELS_URL, + ) + return units + + +def check_model(model: str, summary_params: list[str]) -> tuple[dict[str, str], bool]: + """Validate an EIS circuit string, warning about missing summary params or ML model. + + Returns the parameter units keyed by name, and whether fasteis can guess initial + parameters. Raises ValueError if the string is not a valid fasteis circuit. + """ + import fasteis # noqa: PLC0415 + + circuit = fasteis.Circuit(model) + units = dict(zip(circuit.param_names(), circuit.param_units(), strict=True)) + for param in summary_params: + if param not in units: + logger.warning( + "- EIS summary parameter %s is not in model %s (parameters: %s), so it will " + "be empty in the summary. See %s for how to write circuits", + param, + model, + ", ".join(units), + MODELS_URL, + ) + # A synthetic spectrum is enough to find out if a trained model exists + f = np.logspace(-1, 5, 10) + try: + circuit.guess(f, circuit.impedance(f)) + except ValueError: + logger.warning( + "- EIS model %s is not in the fasteis ML model library, so initial fit " + "parameters cannot be guessed and the fit may be poor. See %s for available models", + model, + MODELS_URL, + ) + return units, False + return units, True + def analyse( df: pd.DataFrame, label: str = "", + model: str = DEFAULT_MODEL, + *, + guess_init: bool | None = None, ) -> tuple[dict[str, dict], np.ndarray]: - """Fit EIS to R-(R,CPE)-(R,CPE) model, warning if the fit looks unreliable.""" + """Fit EIS to an equivalent circuit model, warning if the fit looks unreliable.""" import fasteis # noqa: PLC0415 f = df["Frequency / Hz"] Z = df["Real Impedance / ohm"] + 1j * df["Imaginary Impedance / ohm"] - circuit = fasteis.Circuit("L0-R0-(R1,CPE1)-(R2,CPE2)") - res = circuit.fit(f, Z) + circuit = fasteis.Circuit(model) + res = circuit.fit(f, Z, guess_init=guess_init) Z_fit = res.circuit.impedance(f) errs = res.stderr or {} params = { diff --git a/pyflowbatt/rocrate_output.py b/pyflowbatt/rocrate_output.py index e000463..316854f 100644 --- a/pyflowbatt/rocrate_output.py +++ b/pyflowbatt/rocrate_output.py @@ -12,6 +12,7 @@ EXTRA_INPUT_DESCRIPTIONS, OTHER_RAW_LABEL, OUTPUT_MISC_DESCRIPTIONS, + _custom_eis_tag, _generic_eis_parts, _rel, data_description, @@ -44,6 +45,8 @@ def _measurement_technique(label: str) -> str: if parts := _generic_eis_parts(label): when, number = parts return f"Electrochemical Impedance Spectroscopy ({when} cycling, measurement {number})" + if tag := _custom_eis_tag(label): + return f"Electrochemical Impedance Spectroscopy ({tag})" return label diff --git a/tests/test_cli.py b/tests/test_cli.py index 5c7f072..e349681 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -90,3 +90,13 @@ def fake_analyse(**kwargs: object) -> None: cli.main() assert called_with == {"folder": "somefolder", "dry": False, "zip_output": True} + + +def test_analyse_bad_toml_logs_error(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None: + """A broken pyflowbatt.toml is reported as one error message, not a traceback.""" + (tmp_path / "pyflowbatt.toml").write_text("--- Electrode area ---\n") + + with caplog.at_level("ERROR", logger="pyflowbatt"): + cli.analyse(folder=str(tmp_path)) + + assert "line 1: --- Electrode area ---" in caplog.text diff --git a/tests/test_config.py b/tests/test_config.py index 1ab6031..d83d6ef 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -1,6 +1,7 @@ """Tests for pyflowbatt.toml configuration and technique file classification.""" import json +import re import zipfile from pathlib import Path @@ -236,8 +237,94 @@ def test_eis_tags_pin_files_to_tags(tmp_path: Path) -> None: assert eis == {"eis_pre": "s_02_PEIS_A.mpr", "eis_post": "s_06_PEIS_A.mpr"} +def test_eis_tags_custom_names(tmp_path: Path) -> None: + """Any filename-safe tag can be pinned, and unpinned files fill the standard tags.""" + sample = _write_two_eis_sample(tmp_path / "sample") + (sample / "s_07_PEIS_A.mpr").write_text("x") + (sample / "pyflowbatt.toml").write_text( + '[eis_tags]\n"pre-20%SOC" = ["*_02_PEIS_*"]\n"post-80%SOC" = ["*_07_PEIS_*"]\n' + ) + config = PyFlowBattConfig.load(sample, home=tmp_path / "home") + + classified = classify_technique_files(sample, config) + + eis = {k: v[0].name for k, v in classified.items() if k.startswith("eis_")} + # Only one unpinned file is left, after cycling, which matches no standard layout + assert eis == { + "eis_pre-20%SOC": "s_02_PEIS_A.mpr", + "eis_post-80%SOC": "s_07_PEIS_A.mpr", + "eis_post_1": "s_06_PEIS_A.mpr", + } + + +def test_eis_tags_custom_pin_keeps_standard_layout(tmp_path: Path) -> None: + """Pinning an extra custom tag leaves the standard protocol tagging unchanged.""" + sample = _write_synthetic_sample(tmp_path / "sample") + (sample / "sample_03_PEIS_X.mpr").write_text("x") + (sample / "pyflowbatt.toml").write_text('[eis_tags]\n"pre-20%SOC" = ["*_03_PEIS_*"]\n') + config = PyFlowBattConfig.load(sample, home=tmp_path / "home") + + classified = classify_technique_files(sample, config) + + eis = {k: v[0].name for k, v in classified.items() if k.startswith("eis_")} + assert eis == { + "eis_pre-20%SOC": "sample_03_PEIS_X.mpr", + "eis_pre": "sample_02_PEIS_A.mpr", + "eis_pre-50%SOC": "sample_05_PEIS_B.mpr", + "eis_post-50%SOC": "sample_10_PEIS_C.mpr", + "eis_post": "sample_12_PEIS_D.mpr", + } + + +def test_eis_tags_custom_pin_splits_unpinned_around_cycling(tmp_path: Path) -> None: + """Unpinned files are split before/after cycling, not filled in filename order.""" + sample = _write_synthetic_sample(tmp_path / "sample") + (sample / "pyflowbatt.toml").write_text('[eis_tags]\n"pre-20%SOC" = ["*_05_PEIS_*"]\n') + config = PyFlowBattConfig.load(sample, home=tmp_path / "home") + + classified = classify_technique_files(sample, config) + + # One before and two after cycling is not a standard layout + eis = {k: v[0].name for k, v in classified.items() if k.startswith("eis_")} + assert eis == { + "eis_pre-20%SOC": "sample_05_PEIS_B.mpr", + "eis_pre_1": "sample_02_PEIS_A.mpr", + "eis_post_1": "sample_10_PEIS_C.mpr", + "eis_post_2": "sample_12_PEIS_D.mpr", + } + + +def test_eis_tags_pinned_standard_tag_fills_its_side(tmp_path: Path) -> None: + """A pinned post tag leaves the other post tag for the unpinned file after cycling.""" + sample = _write_synthetic_sample(tmp_path / "sample") + (sample / "pyflowbatt.toml").write_text('[eis_tags]\n"post" = ["*_12_PEIS_*"]\n') + config = PyFlowBattConfig.load(sample, home=tmp_path / "home") + + classified = classify_technique_files(sample, config) + + eis = {k: v[0].name for k, v in classified.items() if k.startswith("eis_")} + assert eis == { + "eis_post": "sample_12_PEIS_D.mpr", + "eis_pre": "sample_02_PEIS_A.mpr", + "eis_pre-50%SOC": "sample_05_PEIS_B.mpr", + "eis_post-50%SOC": "sample_10_PEIS_C.mpr", + } + + +def test_eis_tags_pinned_tag_outside_layout_numbers_the_rest(tmp_path: Path) -> None: + """A pinned 50% SOC tag doesn't fit the early protocol, so the rest are numbered.""" + sample = _write_two_eis_sample(tmp_path / "sample") + (sample / "pyflowbatt.toml").write_text('[eis_tags]\n"pre-50%SOC" = ["*_02_PEIS_*"]\n') + config = PyFlowBattConfig.load(sample, home=tmp_path / "home") + + classified = classify_technique_files(sample, config) + + eis = {k: v[0].name for k, v in classified.items() if k.startswith("eis_")} + assert eis == {"eis_pre-50%SOC": "s_02_PEIS_A.mpr", "eis_post_1": "s_06_PEIS_A.mpr"} + + def test_eis_tags_partial_fills_remaining_tags_in_order(tmp_path: Path) -> None: - """Tags not listed in eis_tags are filled in filename order from the unpinned files.""" + """A pinned standard tag counts towards the layout, unpinned files fill the rest.""" sample = _write_two_eis_sample(tmp_path / "sample") (sample / "pyflowbatt.toml").write_text('[eis_tags]\n"post" = ["*_06_PEIS_*"]\n') config = PyFlowBattConfig.load(sample, home=tmp_path / "home") @@ -261,10 +348,11 @@ def test_eis_tags_sample_folder_overrides_parent(tmp_path: Path) -> None: def test_eis_tags_bad_entries_ignored(tmp_path: Path) -> None: - """Unknown tags and non-list values in eis_tags are ignored.""" + """Tags with unsafe filename characters and non-list values in eis_tags are ignored.""" sample = _write_two_eis_sample(tmp_path / "sample") (sample / "pyflowbatt.toml").write_text( - '[eis_tags]\n"during" = ["*_06_PEIS_*"]\n"post" = "*_06_PEIS_*"\n' + '[eis_tags]\n"pre/20" = ["*_06_PEIS_*"]\n"mid 1" = ["*_06_PEIS_*"]\n' + '"post" = "*_06_PEIS_*"\n' ) config = PyFlowBattConfig.load(sample, home=tmp_path / "home") @@ -1398,3 +1486,425 @@ def test_cascade_priority(tmp_path: Path) -> None: "*_FROM_PARENT_*", "*_FROM_SAMPLE_*", ] + + +def test_eis_model_default_matches_eis_module() -> None: + """Without a pyflowbatt.toml, the EIS model is eis.DEFAULT_MODEL.""" + from pyflowbatt import eis + + assert PyFlowBattConfig().eis_model == eis.DEFAULT_MODEL + + +def test_eis_model_toml_override(tmp_path: Path) -> None: + """[eis] model in pyflowbatt.toml accepts any string.""" + sample_dir = tmp_path / "sample" + sample_dir.mkdir() + (sample_dir / "pyflowbatt.toml").write_text('[eis]\nmodel = "R0-(R1,CPE1)-Wo1"\n') + + config = PyFlowBattConfig.load(sample_dir, home=tmp_path / "home") + assert config.eis_model == "R0-(R1,CPE1)-Wo1" + + +def test_eis_model_bad_value_ignored(tmp_path: Path) -> None: + """A non-string [eis] model is ignored, keeping the default.""" + sample_dir = tmp_path / "sample" + sample_dir.mkdir() + (sample_dir / "pyflowbatt.toml").write_text("[eis]\nmodel = 5\n") + + config = PyFlowBattConfig.load(sample_dir, home=tmp_path / "home") + assert config.eis_model == "L0-R0-(R1,CPE1)-(R2,CPE2)" + + +@pytest.mark.parametrize( + ("model", "summary_params", "has_ml", "warnings"), + [ + ("L0-R0-(R1,CPE1)-(R2,CPE2)", ["R0.r"], True, []), + ("two_rq_l", ["R0.r", "CPE2.alpha"], True, []), + ("(R1,CPE1)-R0", ["R0.r"], True, []), + ("R1-(R2,CPE2)", ["R0.r"], True, ["R0.r is not in model"]), + ("R0-(R1,CPE1)-W2-Wo3", ["R0.r"], False, ["not in the fasteis ML model library"]), + ( + "R1-(R2,CPE2)-W3-Wo4", + ["R0.r", "Wo4.tau"], + False, + ["R0.r is not in model", "not in the fasteis ML model library"], + ), + ], +) +def test_eis_check_model_warnings( + model: str, + summary_params: list[str], + has_ml: bool, + warnings: list[str], + caplog: pytest.LogCaptureFixture, +) -> None: + """check_model warns about missing summary params and a missing ML model.""" + from pyflowbatt import eis + + with caplog.at_level("WARNING", logger="pyflowbatt.eis"): + units, ml = eis.check_model(model, summary_params) + assert ml is has_ml + assert len(units) > 0 + messages = [r.getMessage() for r in caplog.records] + assert len(messages) == len(warnings) + for message, expected in zip(messages, warnings, strict=True): + assert expected in message + assert eis.MODELS_URL in message + + +def test_eis_check_model_invalid_raises() -> None: + """An invalid circuit string raises ValueError.""" + from pyflowbatt import eis + + with pytest.raises(ValueError): + eis.check_model("R0-(R1", ["R0.r"]) + + +def test_eis_analyse_custom_model_without_ml_guess() -> None: + """A model outside the ML library still fits, with columns named by the circuit.""" + import fasteis + import numpy as np + import pandas as pd + + from pyflowbatt import eis + + model = "R0-(R1,CPE1)-Wo1" + f = np.logspace(5, -1, 40) + Z = fasteis.Circuit(model).with_values([1.0, 2.0, 1e-3, 0.9, 3.0, 10.0]).impedance(f) + df = pd.DataFrame( + {"Frequency / Hz": f, "Real Impedance / ohm": Z.real, "Imaginary Impedance / ohm": Z.imag} + ) + params, Z_fit = eis.analyse(df, model=model, guess_init=False) + assert list(params) == ["R0.r", "R1.r", "CPE1.q", "CPE1.alpha", "Wo1.z0", "Wo1.tau"] + assert len(Z_fit) == len(f) + + +def test_eis_summary_params_toml_override(tmp_path: Path) -> None: + """[eis] summary_params replaces the default list, and bad values are ignored.""" + sample_dir = tmp_path / "sample" + sample_dir.mkdir() + home = tmp_path / "home" + toml = sample_dir / "pyflowbatt.toml" + assert PyFlowBattConfig.load(sample_dir, home=home).eis_summary_params == ["R0.r"] + + toml.write_text('[eis]\nsummary_params = ["R0.r", "R1.r"]\n') + assert PyFlowBattConfig.load(sample_dir, home=home).eis_summary_params == ["R0.r", "R1.r"] + + toml.write_text('[eis]\nsummary_params = "R1.r"\n') + assert PyFlowBattConfig.load(sample_dir, home=home).eis_summary_params == ["R0.r"] + + +def test_eis_summary_label_pretty_units() -> None: + """Summary labels use display units, passing unknown unit strings through.""" + from pyflowbatt import eis + + assert eis.summary_label("R0.r", "pre", "ohm") == "EIS R0.r pre / Ω" + assert eis.summary_label("CPE1.alpha", "post", "-") == "EIS CPE1.alpha post / 1" + assert eis.summary_label("X1.y", "pre", "odd") == "EIS X1.y pre / odd" + + +def test_analyse_sample_summary_reports_every_eis_tag( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Every summary param gets a row per standard tag plus each extra fitted tag.""" + import numpy as np + import pandas as pd + + from pyflowbatt import analysis as analysis_module + + sample = _write_two_eis_sample(tmp_path / "sample") + (sample / "s_07_PEIS_A.mpr").write_text("x") + (sample / "pyflowbatt.toml").write_text( + '[eis_tags]\n"pre-20%SOC" = ["*_07_PEIS_*"]\n' + '[eis]\nsummary_params = ["R0.r", "CPE1.alpha", "Missing.x"]\n' + ) + fitted = {"s_02_PEIS_A": 1.0, "s_06_PEIS_A": 2.0, "s_07_PEIS_A": 3.0} + + def fake_analyse(_df: object, label: str, **_kwargs: object) -> tuple[dict, np.ndarray]: + r0 = fitted[label] + params = { + "R0.r": {"value": r0, "err": r0 / 10, "unit": "ohm"}, + "CPE1.alpha": {"value": 0.9, "err": None, "unit": "-"}, + } + return params, np.array([1 + 1j]) + + def fake_plot(_df: object, _z_fit: np.ndarray) -> tuple: + import matplotlib.pyplot as plt + + return plt.subplots() + + def fake_check_model(_model: str, _params: list[str]) -> tuple[dict[str, str], bool]: + return {"R0.r": "ohm", "CPE1.alpha": "-"}, True + + monkeypatch.setattr(analysis_module, "read_to_bdf", lambda _f: pd.DataFrame({"x": [0]})) + monkeypatch.setattr(analysis_module, "df_save_bdf", lambda *_a, **_k: None) + monkeypatch.setattr(analysis_module.eis, "analyse", fake_analyse) + monkeypatch.setattr(analysis_module.eis, "plot", fake_plot) + monkeypatch.setattr(analysis_module.eis, "check_model", fake_check_model) + + config = analysis_module.PyFlowBattConfig.load(sample, home=tmp_path / "home") + analysis_module.analyse_sample(sample, save_format=None, config=config) + + df = pd.read_excel(sample / "results" / "summary.xlsx", sheet_name="Summary", index_col=0) + eis_rows = [q for q in df.index if q.startswith("EIS")] + assert eis_rows == [ + "EIS R0.r pre / Ω", + "EIS R0.r pre-50%SOC / Ω", + "EIS R0.r post-50%SOC / Ω", + "EIS R0.r post / Ω", + "EIS R0.r pre-20%SOC / Ω", + "EIS CPE1.alpha pre / 1", + "EIS CPE1.alpha pre-50%SOC / 1", + "EIS CPE1.alpha post-50%SOC / 1", + "EIS CPE1.alpha post / 1", + "EIS CPE1.alpha pre-20%SOC / 1", + ] + assert df.loc["EIS R0.r pre / Ω", "Value"] == 1.0 + assert df.loc["EIS R0.r pre / Ω", "Error"] == 0.1 + assert df.loc["EIS R0.r pre-20%SOC / Ω", "Value"] == 3.0 + assert df.loc["EIS R0.r post / Ω", "Value"] == 2.0 + assert np.isnan(df.loc["EIS R0.r pre-50%SOC / Ω", "Value"]) + + +def test_custom_eis_tag_descriptions() -> None: + """Custom EIS tags get descriptions naming the tag, standard ones keep theirs.""" + from pyflowbatt import analysis as analysis_module + from pyflowbatt.rocrate_output import _measurement_technique + + label = "eis_pre-20%SOC" + assert "tagged pre-20%SOC" in analysis_module.raw_description(label, Path("a.mpr")) + assert "tagged pre-20%SOC" in analysis_module.plot_description(label) + assert "tagged pre-20%SOC" in analysis_module.data_description(label) + assert _measurement_technique(label) == "Electrochemical Impedance Spectroscopy (pre-20%SOC)" + standard = analysis_module.OUTPUT_PLOT_DESCRIPTIONS["eis_pre"] + assert analysis_module.plot_description("eis_pre") == standard + assert "measurement 2" in analysis_module.plot_description("eis_post_2") + + +def test_eis_summary_units_from_element(caplog: pytest.LogCaptureFixture) -> None: + """Summary units come from each parameter's own element, dropping unknown names.""" + from pyflowbatt import eis + + with caplog.at_level("WARNING", logger="pyflowbatt.eis"): + units = eis.summary_units(["R0.r", "CPE1.alpha", "Wo3.tau", "Missing.x", "R0.x"]) + assert units == {"R0.r": "ohm", "CPE1.alpha": "-", "Wo3.tau": "s"} + messages = [r.getMessage() for r in caplog.records] + assert len(messages) == 2 + assert "Missing.x" in messages[0] + assert "R0.x" in messages[1] + + +def test_analyse_sample_summary_has_eis_rows_without_eis( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Without EIS files or a valid model, the standard EIS rows are still in the summary.""" + import pandas as pd + + from pyflowbatt import analysis as analysis_module + + sample = _write_two_eis_sample(tmp_path / "sample") + for f in sample.glob("*_PEIS_*"): + f.unlink() + (sample / "pyflowbatt.toml").write_text('[eis]\nmodel = "R0-(R1"\n') + monkeypatch.setattr(analysis_module, "read_to_bdf", lambda _f: pd.DataFrame({"x": [0]})) + monkeypatch.setattr(analysis_module, "df_save_bdf", lambda *_a, **_k: None) + + config = analysis_module.PyFlowBattConfig.load(sample, home=tmp_path / "home") + analysis_module.analyse_sample(sample, save_format=None, config=config) + + df = pd.read_excel(sample / "results" / "summary.xlsx", sheet_name="Summary", index_col=0) + eis_rows = [q for q in df.index if q.startswith("EIS")] + assert eis_rows == [ + "EIS R0.r pre / Ω", + "EIS R0.r pre-50%SOC / Ω", + "EIS R0.r post-50%SOC / Ω", + "EIS R0.r post / Ω", + ] + assert df.loc[eis_rows, "Value"].isna().all() + + +def test_merge_summaries_keeps_missing_rows_in_place(tmp_path: Path) -> None: + """Rows only some samples have go after their neighbour, not at the bottom.""" + import pandas as pd + + from pyflowbatt.analysis import merge_summaries + + rows_by_sample = { + "a": ["OCV", "EIS R0.r pre", "EIS R0.r post", "1st CE"], + "b": ["OCV", "EIS R0.r pre", "EIS R0.r post", "EIS R0.r pre-20%SOC", "1st CE", "Extra"], + "c": ["New first", "OCV", "1st CE"], + } + paths = [] + for name, rows in rows_by_sample.items(): + path = tmp_path / name / "results" / "summary.xlsx" + path.parent.mkdir(parents=True) + df = pd.DataFrame({"Value": range(len(rows)), "Error": None}, index=rows) + df.to_excel(path, sheet_name="Summary") + paths.append(path) + + df = merge_summaries(paths, {tmp_path / name: name for name in rows_by_sample}) + + assert list(df["Quantity"]) == [ + "New first", + "OCV", + "EIS R0.r pre", + "EIS R0.r post", + "EIS R0.r pre-20%SOC", + "1st CE", + "Extra", + ] + assert df.set_index("Quantity").loc["EIS R0.r pre-20%SOC", "b"] == 3 + assert pd.isna(df.set_index("Quantity").loc["EIS R0.r pre-20%SOC", "a"]) + + +@pytest.mark.parametrize( + ("rel_path", "frequency_domain"), + [ + ("s/results/eis_pre.bdf.parquet", True), + ("s/results/eis_pre-20%SOC.bdf.csv", True), + ("s/results/s.bdf.parquet", False), + ("weis_sample/results/weis_sample.bdf.csv", False), + ], +) +def test_battinfo_eis_data_comment(rel_path: str, frequency_domain: bool) -> None: + """BDF outputs from EIS are described as frequency-domain, others as time series.""" + from pyflowbatt.battinfo import add_data + + comment = add_data(rel_path, None)["hasOutput"]["dcat:distribution"]["rdfs:comment"] + assert comment.startswith("Frequency-domain" if frequency_domain else "Time series") + + +def test_template_every_single_hash_line_uncomments_to_valid_toml( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """Explanations use ##, and deleting the # from every other line gives a valid config.""" + from pyflowbatt.config import TEMPLATE_TOML + + lines = TEMPLATE_TOML.splitlines() + assert all(not line or line.startswith(("##", "# ")) for line in lines) + uncommented = [line.removeprefix("# ") for line in lines] + sample_dir = tmp_path / "sample" + sample_dir.mkdir() + (sample_dir / "pyflowbatt.toml").write_text("\n".join(uncommented), encoding="utf-8") + + with caplog.at_level("WARNING", logger="pyflowbatt"): + config = PyFlowBattConfig.load(sample_dir, home=tmp_path / "home") + assert caplog.records == [] + assert config.eis_model == "L0-R0-(R1,CPE1)-(R2,CPE2)" + assert config.cv_v_max == 0.6 + + +def test_bad_toml_error_names_file_and_line(tmp_path: Path) -> None: + """An uncommented explanation line raises ConfigError with the file, line and a hint.""" + from pyflowbatt.config import ConfigError + + sample_dir = tmp_path / "sample" + sample_dir.mkdir() + toml = sample_dir / "pyflowbatt.toml" + toml.write_text( + "## --- Electrode area ---\narea_cm2 = 3\n\n" + "--- LSV pre/post split: file numbers below this are pre ---\nlsv_threshold = 8\n" + ) + + with pytest.raises(ConfigError) as excinfo: + PyFlowBattConfig.load(sample_dir, home=tmp_path / "home") + message = str(excinfo.value) + assert str(toml) in message + assert "line 4: --- LSV pre/post split" in message + assert "put the # back" in message + + +@pytest.mark.parametrize( + ("toml_text", "hint"), + [ + ('## Standard tags:\nStandard tags: "pre", "post".\n', "explanation line"), + ('model = "R0"\n# [cv]\neis = ["*_x_*"]\n[eis]\n', "deleted the # in front"), + ("area_cm2 = 5\narea_cm2 = 6\n", "set twice"), + ('sample_name_pattern = "^\\d+$"\n', "Use single quotes"), + ('model = "R0\n', "missing its closing quote"), + ('summary_params = ["R0.r", "R1.r"\n', "missing its closing ]"), + ("model = L0-R0\n", "Text values need quotes"), + ("lsv_threshold = True\n", "true/false are lowercase"), + ("area_cm2 = 5 cm2\n", "no units"), + ('[eis\nmodel = "R0"\n', "Section lines look like [eis]"), + ("area cm2 = 5\n", "can't contain spaces"), + ("x = {a = 1\n", "Only delete the single #"), + ], +) +def test_bad_toml_hint_matches_mistake(tmp_path: Path, toml_text: str, hint: str) -> None: + """Each kind of TOML mistake gets a hint about its likely cause.""" + from pyflowbatt.config import ConfigError + + sample_dir = tmp_path / "sample" + sample_dir.mkdir() + (sample_dir / "pyflowbatt.toml").write_text(toml_text) + + with pytest.raises(ConfigError, match=re.escape(hint)): + PyFlowBattConfig.load(sample_dir, home=tmp_path / "home") + + +def test_toml_with_bom_loads(tmp_path: Path) -> None: + """A UTF-8 file with a byte order mark, as some Windows editors save, still loads.""" + sample_dir = tmp_path / "sample" + sample_dir.mkdir() + (sample_dir / "pyflowbatt.toml").write_bytes(b"\xef\xbb\xbfarea_cm2 = 3\n") + + assert PyFlowBattConfig.load(sample_dir, home=tmp_path / "home").area_cm2 == 3 + + +@pytest.mark.parametrize( + ("toml_text", "expected"), + [ + ( + '# [eis]\nmodel = "R0"\n', + "Ignoring model at the top level ({path}, line 2): it belongs under [eis]", + ), + ( + '[eis]\nmodel = "R0"\n\n# [cv]\nv_min = 0.1\n', + "Ignoring v_min in [eis] ({path}, line 5): it belongs under [cv]", + ), + ( + "[cv]\nmin_r2 = 0.5\narea_cm2 = 3\n", + "Ignoring area_cm2 in [cv] ({path}, line 3): it is a top-level setting", + ), + ( + '[eis_tags]\n"pre" = ["*_02_*"]\n# [eis]\nsummary_params = ["R1.r"]\n', + "Ignoring summary_params in [eis_tags] ({path}, line 4): it belongs under [eis]", + ), + ("area_cm = 3\n", "({path}, line 1): unknown setting, did you mean area_cm2?"), + ('[eiss]\nmodel = "R0"\n', "({path}, line 1): unknown section, did you mean [eis]?"), + ("cv = 5\n", "({path}, line 1): cv is a section, so write it as a [cv] line"), + ], +) +def test_misplaced_setting_warns_with_line( + tmp_path: Path, caplog: pytest.LogCaptureFixture, toml_text: str, expected: str +) -> None: + """A setting outside its section is ignored with a warning naming its line and section.""" + sample_dir = tmp_path / "sample" + sample_dir.mkdir() + toml = sample_dir / "pyflowbatt.toml" + toml.write_text(toml_text) + + with caplog.at_level("WARNING", logger="pyflowbatt"): + config = PyFlowBattConfig.load(sample_dir, home=tmp_path / "home") + + assert [r.getMessage() for r in caplog.records if expected.format(path=toml) in r.getMessage()] + assert config.area_cm2 is None + assert config.cv_v_min == 0.4004 + assert config.eis_summary_params == ["R0.r"] + assert config.eis_tag_patterns in ({}, {"pre": ["*_02_*"]}) + + +def test_eis_tags_any_name_not_warned_as_misplaced( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """Custom tag names under [eis_tags] are not reported as unknown settings.""" + sample_dir = tmp_path / "sample" + sample_dir.mkdir() + (sample_dir / "pyflowbatt.toml").write_text('[eis_tags]\n"pre-20%SOC" = ["*_03_*"]\n') + + with caplog.at_level("WARNING", logger="pyflowbatt"): + config = PyFlowBattConfig.load(sample_dir, home=tmp_path / "home") + + assert caplog.records == [] + assert config.eis_tag_patterns == {"pre-20%SOC": ["*_03_*"]}