From 0a40ef884467025588c2204aa12aab0999d38815 Mon Sep 17 00:00:00 2001 From: Graham Kimbell <88666181+g-kimbell@users.noreply.github.com> Date: Wed, 14 Oct 2026 15:09:41 +0200 Subject: [PATCH 1/5] Add rich CLI output --- pyflowbatt/analysis.py | 801 ++++++++++++++++++++++------------------- pyflowbatt/cli.py | 122 +++++-- pyflowbatt/report.py | 336 +++++++++++++++++ 3 files changed, 855 insertions(+), 404 deletions(-) create mode 100644 pyflowbatt/report.py diff --git a/pyflowbatt/analysis.py b/pyflowbatt/analysis.py index 4eb06db..e65fd56 100644 --- a/pyflowbatt/analysis.py +++ b/pyflowbatt/analysis.py @@ -22,6 +22,7 @@ classify_technique_files, ) from pyflowbatt.read import read_to_bdf +from pyflowbatt.report import LOG_FILENAME, Reporter logger = logging.getLogger(__name__) @@ -456,7 +457,7 @@ def zip_folder(root_folder: Path, zip_path: Path) -> Path: files in both the RO-Crate manifest and BattINFO metadata, so a file's "@id" can be found directly inside the zip once it's extracted. - Ignores hidden EC-lab files and Excel lock files. + Ignores hidden EC-lab files, Excel lock files and the PyFlowBatt log. """ with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf: for file in sorted(root_folder.rglob("*")): @@ -464,6 +465,7 @@ def zip_folder(root_folder: Path, zip_path: Path) -> Path: file.is_file() and file.suffix not in {".mgr", ".sta", ".mpp", ".mgp"} and not file.name.startswith("~$") + and file.name != LOG_FILENAME ): zf.write(file, arcname=_rel(file, root_folder)) return zip_path @@ -492,6 +494,7 @@ def analyse_sample( config: PyFlowBattConfig | None = None, root_folder: str | Path | None = None, zenodo_package: Literal["zip", "files"] = "files", + reporter: Reporter | None = None, ) -> tuple[dict[str, list[Path]], dict[str, list[Path]], dict[str, list[Path]], str | None, str]: """Read all the files in a folder, analayse and plot everything. @@ -520,6 +523,8 @@ def analyse_sample( known: "files" (default) points directly at each file's own Zenodo download URL (matching an unzipped upload); "zip" points at one zip's Zenodo download URL with the in-archive path as a fragment (see :func:`pyflowbatt.battinfo.zenodo_download_url`). + + ``reporter`` is told when each step starts and finishes, to show progress. """ folder = Path(folder) root_folder = Path(root_folder) if root_folder is not None else folder @@ -527,8 +532,9 @@ def analyse_sample( pub_info_path = Path(pub_info_path) if pub_info_path is not None else None fcid: str | None = None config = config or PyFlowBattConfig.load(folder) + reporter = reporter or Reporter() - logger.info("\n🌊 PyFlowBatt-ing %s", folder.name) + logger.debug("Analysing %s", folder.name) classified = classify_technique_files(folder, config, warn=True) gcpl_files = classified.get("gcpl", []) gcpl_file = gcpl_files[0] if gcpl_files else None @@ -551,6 +557,25 @@ def analyse_sample( logger.debug("Reading CV after: %s", ", ".join([f.stem for f in cv_files_after])) logger.debug("Reading EIS files: %s", ", ".join([f.stem for f in eis_by_tag.values()])) logger.debug("Checking BattINFO files: %s", ", ".join([f.stem for f in battinfo_files])) + reporter.plan( + [ + "BattINFO", + "OCV", + "GCPL", + *([f"LSV {p}" for p in ("pre", "post") if classified.get(f"lsv_{p}")] or ["LSV"]), + *( + [ + f"CV {p}" + for p, files in (("pre", cv_files_before), ("post", cv_files_after)) + if files + ] + or ["CV"] + ), + *([f"EIS {tag}" for tag in eis_by_tag] or ["EIS"]), + "Summary", + "Metadata", + ] + ) cycle_df = None cv_df = None @@ -569,64 +594,65 @@ def analyse_sample( else (".csv.parquet" if save_format == "csv" else None) ) - logger.info("Battinfo-ifying") battinfo_json = None battinfo_sample_id = None raw_battinfo_json: dict | None = None if not battinfo_files: - logger.info("- ☹️ No battinfo xlsx found, skipping") + logger.debug("No BattINFO xlsx found, skipping") + reporter.skip("BattINFO", "no xlsx") else: - for file in battinfo_files: - raw_json = None - with contextlib.suppress(ValueError): - raw_json = convert_excel_to_jsonld(file) - if raw_json is None: - continue - battinfo_xlsx_path = file - fcid = raw_json["schema:productID"] - battinfo_sample_id = raw_json["schema:name"] - raw_battinfo_json = raw_json - battinfo_json = battinfo.make_test_object(raw_json) - - # Create list of everything that can be included in JSON-LD - merge_list = [ - battinfo_json, - battinfo.add_input_and_output(), - battinfo.add_software(), - ] - if pub_info: - if pub_info.get("zenodo_doi_url"): - merge_list.append(battinfo.add_zenodo_url(pub_info["zenodo_doi_url"])) - if pub_info.get("citation_string"): - merge_list.append(battinfo.add_citation(pub_info["citation_string"])) - if pub_info.get("institution"): - merge_list.append( - battinfo.add_institution( - pub_info["institution"], - pub_info.get("institutions", {}) - .get(pub_info["institution"], {}) - .get("wikidata_url"), + with reporter.task("BattINFO"): + for file in battinfo_files: + raw_json = None + with contextlib.suppress(ValueError): + raw_json = convert_excel_to_jsonld(file) + if raw_json is None: + continue + battinfo_xlsx_path = file + fcid = raw_json["schema:productID"] + battinfo_sample_id = raw_json["schema:name"] + raw_battinfo_json = raw_json + battinfo_json = battinfo.make_test_object(raw_json) + + # Create list of everything that can be included in JSON-LD + merge_list = [ + battinfo_json, + battinfo.add_input_and_output(), + battinfo.add_software(), + ] + if pub_info: + if pub_info.get("zenodo_doi_url"): + merge_list.append(battinfo.add_zenodo_url(pub_info["zenodo_doi_url"])) + if pub_info.get("citation_string"): + merge_list.append(battinfo.add_citation(pub_info["citation_string"])) + if pub_info.get("institution"): + merge_list.append( + battinfo.add_institution( + pub_info["institution"], + pub_info.get("institutions", {}) + .get(pub_info["institution"], {}) + .get("wikidata_url"), + ) + ) + if pub_info.get("authors") and pub_info.get("institutions"): + merge_list.append( + battinfo.add_authors(pub_info["authors"], pub_info["institutions"]), + ) + if pub_info.get("sample_to_fig"): + merge_list.append( + battinfo.add_associated_media( + pub_info.get("publication_doi_url"), + pub_info["sample_to_fig"], + fcid, + battinfo_sample_id, + ), ) - ) - if pub_info.get("authors") and pub_info.get("institutions"): - merge_list.append( - battinfo.add_authors(pub_info["authors"], pub_info["institutions"]), - ) - if pub_info.get("sample_to_fig"): - merge_list.append( - battinfo.add_associated_media( - pub_info.get("publication_doi_url"), - pub_info["sample_to_fig"], - fcid, - battinfo_sample_id, - ), - ) - # Merge everything - battinfo_json = battinfo.merge_jsonld_on_type(merge_list) - break - else: - logger.info("- ☹️ Couldn't convert battinfo xlsx") + # Merge everything + battinfo_json = battinfo.merge_jsonld_on_type(merge_list) + break + else: + logger.warning("Couldn't convert any xlsx to BattINFO, no metadata written") sample_id = get_sampleid_from_folderpath(folder, config, battinfo_sample_id) area_cm2 = get_area_cm2(config, raw_battinfo_json) @@ -634,142 +660,148 @@ def analyse_sample( config, raw_battinfo_json, gcpl_files[0].stem if gcpl_files else None ) - logger.info("⛓️‍💥 Analysing OCV") av_ocv = (np.nan, np.nan) if len(ocv_files) == 0: - logger.info("- ☹️ No OCV found, skipping") + logger.debug("No OCV found, skipping") + reporter.skip("OCV", "no files") else: - try: - df = read_to_bdf(ocv_files[0]) - av_ocv = ocv.analyse(df) - tracked_mpr_inputs["ocv"] = [ocv_files[0]] - except Exception: - logger.exception("Failed to analyse OCV") + with reporter.task("OCV"): + try: + df = read_to_bdf(ocv_files[0]) + av_ocv = ocv.analyse(df) + tracked_mpr_inputs["ocv"] = [ocv_files[0]] + except Exception: + logger.exception("Failed to analyse OCV") - logger.info("🔋 Analysing GCPL") if gcpl_file is None: - logger.warning("- ☹️ No GCPL files found, skipping") + logger.debug("No GCPL files found, skipping") + reporter.skip("GCPL", "no files") else: - try: - df, cycle_df = gcpl.analyse(read_to_bdf(gcpl_file)) - fig, _ax = gcpl.plot(df, cycle_df) - fig.savefig(results_dir / "gcpl.png") - plt.close(fig) - ratetest_df = gcpl.cycles_to_ratetest(cycle_df) - df_save_bdf(df, results_dir / "gcpl.x", save_format=save_format) - tracked_outputs["gcpl"] = [results_dir / "gcpl.png"] - if bdf_suffix: - tracked_outputs["gcpl"].append((results_dir / "gcpl.x").with_suffix(bdf_suffix)) - tracked_mpr_inputs["gcpl"] = [gcpl_file] - except Exception as e: - cycle_df = None - logger.warning("- Failed to analyse %s: %s", gcpl_file.stem, str(e)) - - logger.info("↗️ Analysing LSV") + with reporter.task("GCPL"): + try: + df, cycle_df = gcpl.analyse(read_to_bdf(gcpl_file)) + fig, _ax = gcpl.plot(df, cycle_df) + fig.savefig(results_dir / "gcpl.png") + plt.close(fig) + ratetest_df = gcpl.cycles_to_ratetest(cycle_df) + df_save_bdf(df, results_dir / "gcpl.x", save_format=save_format) + tracked_outputs["gcpl"] = [results_dir / "gcpl.png"] + if bdf_suffix: + tracked_outputs["gcpl"].append((results_dir / "gcpl.x").with_suffix(bdf_suffix)) + tracked_mpr_inputs["gcpl"] = [gcpl_file] + except Exception as e: + cycle_df = None + logger.error("Failed to analyse %s: %s", gcpl_file.stem, str(e)) # noqa: TRY400 + lsv_res = {"pre": np.nan, "post": np.nan} lsv_pre_files = classified.get("lsv_pre", []) lsv_post_files = classified.get("lsv_post", []) if not lsv_pre_files and not lsv_post_files: - logger.info("- ☹️ No LSV files found, skipping") + logger.debug("No LSV files found, skipping") + reporter.skip("LSV", "no files") else: - if bool(lsv_pre_files) != bool(lsv_post_files): - only = "pre" if lsv_pre_files else "post" - logger.warning("- Only one LSV file found, assuming it is %s", only) lsv_rows = [] for p, lsv_files_p in [("pre", lsv_pre_files), ("post", lsv_post_files)]: if not lsv_files_p: continue - try: - df = read_to_bdf(lsv_files_p[0]) - results = lsv.analyse(df, area_cm2=area_cm2) - results["File name"] = lsv_files_p[0].name - fig, _ax = lsv.plot(df, results) - fig.savefig(results_dir / f"lsv_{p}.png") - plt.close(fig) - df_save_bdf(df, results_dir / f"lsv_{p}.x", save_format=save_format) - lsv_res[p] = float(results["Area specific resistance / Ω cm²"]) - tracked_outputs[f"lsv_{p}"] = [results_dir / f"lsv_{p}.png"] - if bdf_suffix: - tracked_outputs[f"lsv_{p}"].append( - (results_dir / f"lsv_{p}.x").with_suffix(bdf_suffix) - ) - tracked_mpr_inputs[f"lsv_{p}"] = [lsv_files_p[0]] - lsv_rows.append({"Pre or post cycle": p, **results}) - except Exception: - logger.exception("Failed to analyse LSV file") + with reporter.task(f"LSV {p}"): + if bool(lsv_pre_files) != bool(lsv_post_files): + logger.warning("Only one LSV file found, assuming it is %s", p) + try: + df = read_to_bdf(lsv_files_p[0]) + results = lsv.analyse(df, area_cm2=area_cm2) + results["File name"] = lsv_files_p[0].name + fig, _ax = lsv.plot(df, results) + fig.savefig(results_dir / f"lsv_{p}.png") + plt.close(fig) + df_save_bdf(df, results_dir / f"lsv_{p}.x", save_format=save_format) + lsv_res[p] = float(results["Area specific resistance / Ω cm²"]) + tracked_outputs[f"lsv_{p}"] = [results_dir / f"lsv_{p}.png"] + if bdf_suffix: + tracked_outputs[f"lsv_{p}"].append( + (results_dir / f"lsv_{p}.x").with_suffix(bdf_suffix) + ) + tracked_mpr_inputs[f"lsv_{p}"] = [lsv_files_p[0]] + lsv_rows.append({"Pre or post cycle": p, **results}) + except Exception: + logger.exception("Failed to analyse LSV file") if lsv_rows: lsv_df = pd.DataFrame(lsv_rows) - logger.info("🚴 Analysing CV") cv_res = {"pre": np.nan, "post": np.nan} if len(cv_files_before) == 0 and len(cv_files_after) == 0: - logger.warning("- ☹️ No files found, skipping") + logger.debug("No CV files found, skipping") + reporter.skip("CV", "no files") else: for p, cv_files in [("pre", cv_files_before), ("post", cv_files_after)]: if not cv_files: continue - try: - df, cv_df, capacitance_mF = cv.analyse( - read_to_bdf(cv_files[0]), - v_min=config.cv_v_min, - v_max=config.cv_v_max, - v_med=config.cv_v_med, - v_range=config.cv_v_range, - min_r2=config.cv_min_r2, - ) - df_save_bdf(df, results_dir / f"cv_{p}.x", save_format=save_format) - fig, _ax = cv.plot(df, cv_df, min_r2=config.cv_min_r2) - fig.savefig(results_dir / f"cv_{p}.png") - plt.close(fig) - if capacitance_mF: - cv_res[p] = capacitance_mF - tracked_outputs[f"cv_{p}"] = [results_dir / f"cv_{p}.png"] - if bdf_suffix: - tracked_outputs[f"cv_{p}"].append( - (results_dir / f"cv_{p}.x").with_suffix(bdf_suffix) + with reporter.task(f"CV {p}"): + try: + df, cv_df, capacitance_mF = cv.analyse( + read_to_bdf(cv_files[0]), + v_min=config.cv_v_min, + v_max=config.cv_v_max, + v_med=config.cv_v_med, + v_range=config.cv_v_range, + min_r2=config.cv_min_r2, ) - tracked_mpr_inputs[f"cv_{p}"] = [cv_files[0]] - except Exception as e: - logger.warning("- Failed to analyse %s: %s", cv_files[0].stem, str(e)) + df_save_bdf(df, results_dir / f"cv_{p}.x", save_format=save_format) + fig, _ax = cv.plot(df, cv_df, min_r2=config.cv_min_r2) + fig.savefig(results_dir / f"cv_{p}.png") + plt.close(fig) + if capacitance_mF: + cv_res[p] = capacitance_mF + tracked_outputs[f"cv_{p}"] = [results_dir / f"cv_{p}.png"] + if bdf_suffix: + tracked_outputs[f"cv_{p}"].append( + (results_dir / f"cv_{p}.x").with_suffix(bdf_suffix) + ) + tracked_mpr_inputs[f"cv_{p}"] = [cv_files[0]] + except Exception as e: + logger.error("Failed to analyse %s: %s", cv_files[0].stem, str(e)) # noqa: TRY400 - 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") + logger.debug("No EIS files found, skipping") + reporter.skip("EIS", "no files") else: 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) + logger.warning("Invalid EIS model %s, skipping EIS: %s", config.eis_model, e) + for tag in eis_by_tag: + reporter.skip(f"EIS {tag}", "invalid model") 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) + with reporter.task(f"EIS {tag}"): + try: + df = read_to_bdf(f) + params, Z_fit = eis.analyse( + df, label=f.stem, model=config.eis_model, guess_init=has_ml_guess ) - 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)) + 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.error("Failed to fit %s: %s", f.stem, str(e)) # noqa: TRY400 if rows: eis_df = pd.DataFrame(rows) eis_df = eis_df.pivot(index=["file", "tag"], columns=["name"]).reset_index() @@ -777,140 +809,139 @@ def analyse_sample( 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", - "ASR pre / Ω cm²", - "ASR post / Ω cm²", - "∆ASR / Ω cm²", - "F pre / mF", - "F post / mF", - "∆F / mF", - "Assembled resistance / Ω", - *eis_rows, - "1st CE / %", - "1st EE / %", - "1st VE / %", - ] - # n cycles to include in the summary file - n_cycles = config.summary_n_cycles - for n in n_cycles: - cols.extend( - [ - f"{n} cycles avg. CE / %", - f"{n} cycles avg. EE / %", - f"{n} cycles avg. VE / %", - f"{n} cycles avg. charge capacity / mAh", - f"{n} cycles avg. discharge capacity / mAh", - ] + with reporter.task("Summary", detail="results/summary.xlsx"): + # 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", + "ASR pre / Ω cm²", + "ASR post / Ω cm²", + "∆ASR / Ω cm²", + "F pre / mF", + "F post / mF", + "∆F / mF", + "Assembled resistance / Ω", + *eis_rows, + "1st CE / %", + "1st EE / %", + "1st VE / %", + ] + # n cycles to include in the summary file + n_cycles = config.summary_n_cycles + for n in n_cycles: + cols.extend( + [ + f"{n} cycles avg. CE / %", + f"{n} cycles avg. EE / %", + f"{n} cycles avg. VE / %", + f"{n} cycles avg. charge capacity / mAh", + f"{n} cycles avg. discharge capacity / mAh", + ] + ) + summary = {col: {"Value": np.nan, "Error": np.nan} for col in cols} + + # Now fill in values + summary["Av. OCV / V"]["Value"] = av_ocv[0] + summary["Av. OCV / V"]["Error"] = av_ocv[1] + summary["ASR pre / Ω cm²"]["Value"] = lsv_res["pre"] + summary["ASR post / Ω cm²"]["Value"] = lsv_res["post"] + summary["∆ASR / Ω cm²"]["Value"] = lsv_res["post"] - lsv_res["pre"] + summary["F pre / mF"]["Value"] = cv_res["pre"] + summary["F post / mF"]["Value"] = cv_res["post"] + summary["∆F / mF"]["Value"] = ( + cv_res["post"] - cv_res["pre"] if (cv_res["post"] and cv_res["pre"]) else None ) - summary = {col: {"Value": np.nan, "Error": np.nan} for col in cols} - - # Now fill in values - summary["Av. OCV / V"]["Value"] = av_ocv[0] - summary["Av. OCV / V"]["Error"] = av_ocv[1] - summary["ASR pre / Ω cm²"]["Value"] = lsv_res["pre"] - summary["ASR post / Ω cm²"]["Value"] = lsv_res["post"] - summary["∆ASR / Ω cm²"]["Value"] = lsv_res["post"] - lsv_res["pre"] - summary["F pre / mF"]["Value"] = cv_res["pre"] - summary["F post / mF"]["Value"] = cv_res["post"] - summary["∆F / mF"]["Value"] = ( - cv_res["post"] - cv_res["pre"] if (cv_res["post"] and cv_res["pre"]) else None - ) - summary["Assembled resistance / Ω"]["Value"] = assembled_resistance_ohm - summary.update(eis_rows) + summary["Assembled resistance / Ω"]["Value"] = assembled_resistance_ohm + summary.update(eis_rows) - if cycle_df is not None: - mask = cycle_df["Cycle Count / 1"] == 1 - summary["1st CE / %"]["Value"] = float( - cycle_df.loc[mask, "Coulombic Efficiency / %"].to_numpy()[0] - ) - summary["1st EE / %"]["Value"] = float( - cycle_df.loc[mask, "Energy Efficiency / %"].to_numpy()[0] - ) - summary["1st VE / %"]["Value"] = float( - cycle_df.loc[mask, "Voltage Efficiency / %"].to_numpy()[0] - ) + if cycle_df is not None: + mask = cycle_df["Cycle Count / 1"] == 1 + summary["1st CE / %"]["Value"] = float( + cycle_df.loc[mask, "Coulombic Efficiency / %"].to_numpy()[0] + ) + summary["1st EE / %"]["Value"] = float( + cycle_df.loc[mask, "Energy Efficiency / %"].to_numpy()[0] + ) + summary["1st VE / %"]["Value"] = float( + cycle_df.loc[mask, "Voltage Efficiency / %"].to_numpy()[0] + ) - max_cycles = cycle_df["Cycle Count / 1"].max() - for n in n_cycles: - if n <= max_cycles: - mask = cycle_df["Cycle Count / 1"] <= n - summary[f"{n} cycles avg. CE / %"]["Value"] = float( - cycle_df.loc[mask, "Coulombic Efficiency / %"].mean() - ) - summary[f"{n} cycles avg. CE / %"]["Error"] = float( - cycle_df.loc[mask, "Coulombic Efficiency / %"].std() - ) - summary[f"{n} cycles avg. EE / %"]["Value"] = float( - cycle_df.loc[mask, "Energy Efficiency / %"].mean() - ) - summary[f"{n} cycles avg. EE / %"]["Error"] = float( - cycle_df.loc[mask, "Energy Efficiency / %"].std() - ) - summary[f"{n} cycles avg. VE / %"]["Value"] = float( - cycle_df.loc[mask, "Voltage Efficiency / %"].mean() - ) - summary[f"{n} cycles avg. VE / %"]["Error"] = float( - cycle_df.loc[mask, "Voltage Efficiency / %"].std() - ) - summary[f"{n} cycles avg. charge capacity / mAh"]["Value"] = float( - cycle_df.loc[mask, "Charge Capacity / mAh"].mean() - ) - summary[f"{n} cycles avg. charge capacity / mAh"]["Error"] = float( - cycle_df.loc[mask, "Charge Capacity / mAh"].std() - ) - summary[f"{n} cycles avg. discharge capacity / mAh"]["Value"] = float( - cycle_df.loc[mask, "Discharge Capacity / mAh"].mean() - ) - summary[f"{n} cycles avg. discharge capacity / mAh"]["Error"] = float( - cycle_df.loc[mask, "Discharge Capacity / mAh"].std() - ) + max_cycles = cycle_df["Cycle Count / 1"].max() + for n in n_cycles: + if n <= max_cycles: + mask = cycle_df["Cycle Count / 1"] <= n + summary[f"{n} cycles avg. CE / %"]["Value"] = float( + cycle_df.loc[mask, "Coulombic Efficiency / %"].mean() + ) + summary[f"{n} cycles avg. CE / %"]["Error"] = float( + cycle_df.loc[mask, "Coulombic Efficiency / %"].std() + ) + summary[f"{n} cycles avg. EE / %"]["Value"] = float( + cycle_df.loc[mask, "Energy Efficiency / %"].mean() + ) + summary[f"{n} cycles avg. EE / %"]["Error"] = float( + cycle_df.loc[mask, "Energy Efficiency / %"].std() + ) + summary[f"{n} cycles avg. VE / %"]["Value"] = float( + cycle_df.loc[mask, "Voltage Efficiency / %"].mean() + ) + summary[f"{n} cycles avg. VE / %"]["Error"] = float( + cycle_df.loc[mask, "Voltage Efficiency / %"].std() + ) + summary[f"{n} cycles avg. charge capacity / mAh"]["Value"] = float( + cycle_df.loc[mask, "Charge Capacity / mAh"].mean() + ) + summary[f"{n} cycles avg. charge capacity / mAh"]["Error"] = float( + cycle_df.loc[mask, "Charge Capacity / mAh"].std() + ) + summary[f"{n} cycles avg. discharge capacity / mAh"]["Value"] = float( + cycle_df.loc[mask, "Discharge Capacity / mAh"].mean() + ) + summary[f"{n} cycles avg. discharge capacity / mAh"]["Error"] = float( + cycle_df.loc[mask, "Discharge Capacity / mAh"].std() + ) - writer = pd.ExcelWriter(folder / "results" / "summary.xlsx", engine="xlsxwriter") - df = pd.DataFrame.from_dict(summary, orient="index").reset_index() - df = df.rename(columns={"index": "Quantity"}) - df.to_excel(writer, sheet_name="Summary", index=False) - workbook = writer.book - worksheet = writer.sheets["Summary"] - worksheet.autofit() - - if cycle_df is not None: - cycle_df.to_excel(writer, sheet_name="Cycling", index=False) - worksheet = writer.sheets["Cycling"] - worksheet.autofit() - if ratetest_df is not None: - ratetest_df = ratetest_df.drop("Times seen", axis=1) - ratetest_df.to_excel(writer, sheet_name="Ratetest", index=False) - worksheet = writer.sheets["Ratetest"] - worksheet.autofit() - if cv_df is not None: - cv_df = cv_df.loc[:, ["CV Cycle", *[c for c in cv_df.columns if c not in ["CV Cycle"]]]] - cv_df.to_excel(writer, sheet_name="CV", index=False) - worksheet = writer.sheets["CV"] - worksheet.autofit() - if lsv_df is not None: - lsv_df.to_excel(writer, sheet_name="LSV", index=False) - worksheet = writer.sheets["LSV"] - worksheet.autofit() - if eis_df is not None: - eis_df.to_excel(writer, sheet_name="EIS", index=False) - worksheet = writer.sheets["EIS"] + writer = pd.ExcelWriter(folder / "results" / "summary.xlsx", engine="xlsxwriter") + df = pd.DataFrame.from_dict(summary, orient="index").reset_index() + df = df.rename(columns={"index": "Quantity"}) + df.to_excel(writer, sheet_name="Summary", index=False) + workbook = writer.book + worksheet = writer.sheets["Summary"] worksheet.autofit() - workbook.close() - tracked_outputs["summary"] = [results_dir / "summary.xlsx"] + if cycle_df is not None: + cycle_df.to_excel(writer, sheet_name="Cycling", index=False) + worksheet = writer.sheets["Cycling"] + worksheet.autofit() + if ratetest_df is not None: + ratetest_df = ratetest_df.drop("Times seen", axis=1) + ratetest_df.to_excel(writer, sheet_name="Ratetest", index=False) + worksheet = writer.sheets["Ratetest"] + worksheet.autofit() + if cv_df is not None: + cv_df = cv_df.loc[:, ["CV Cycle", *[c for c in cv_df.columns if c not in ["CV Cycle"]]]] + cv_df.to_excel(writer, sheet_name="CV", index=False) + worksheet = writer.sheets["CV"] + worksheet.autofit() + if lsv_df is not None: + lsv_df.to_excel(writer, sheet_name="LSV", index=False) + worksheet = writer.sheets["LSV"] + worksheet.autofit() + if eis_df is not None: + eis_df.to_excel(writer, sheet_name="EIS", index=False) + worksheet = writer.sheets["EIS"] + worksheet.autofit() + + workbook.close() + tracked_outputs["summary"] = [results_dir / "summary.xlsx"] # Every raw file in the folder, whether or not its analysis produced anything, so the # metadata describes the data that ships rather than only what could be analysed. @@ -935,67 +966,70 @@ def analyse_sample( tracked_extra_inputs["pub_info"] = [pub_info_path] if battinfo_json is not None: - zenodo_url = pub_info.get("zenodo_doi_url") if pub_info else None - zenodo_zip_filename = pub_info.get("zenodo_zip_filename") or f"{root_folder.name}.zip" - for label, paths in raw_inputs.items(): - for path in paths: - snippet = battinfo.add_data( + with reporter.task("Metadata"): + zenodo_url = pub_info.get("zenodo_doi_url") if pub_info else None + zenodo_zip_filename = pub_info.get("zenodo_zip_filename") or f"{root_folder.name}.zip" + for label, paths in raw_inputs.items(): + for path in paths: + snippet = battinfo.add_data( + _rel(path, root_folder), + zenodo_url, + extras={"rdfs:comment": raw_description(label, path)}, + package=zenodo_package, + zip_filename=zenodo_zip_filename, + ) + battinfo_json = battinfo.merge_jsonld_on_type([battinfo_json, snippet]) + for path in mps_files: + snippet = battinfo.add_input_data( _rel(path, root_folder), zenodo_url, - extras={"rdfs:comment": raw_description(label, path)}, + EXTRA_INPUT_DESCRIPTIONS["protocol"], package=zenodo_package, zip_filename=zenodo_zip_filename, ) battinfo_json = battinfo.merge_jsonld_on_type([battinfo_json, snippet]) - for path in mps_files: - snippet = battinfo.add_input_data( - _rel(path, root_folder), - zenodo_url, - EXTRA_INPUT_DESCRIPTIONS["protocol"], - package=zenodo_package, - zip_filename=zenodo_zip_filename, - ) - battinfo_json = battinfo.merge_jsonld_on_type([battinfo_json, snippet]) - if battinfo_xlsx_path: - snippet = battinfo.add_input_data( - _rel(battinfo_xlsx_path, root_folder), - zenodo_url, - EXTRA_INPUT_DESCRIPTIONS["battinfo_xlsx"], - package=zenodo_package, - zip_filename=zenodo_zip_filename, - ) - battinfo_json = battinfo.merge_jsonld_on_type([battinfo_json, snippet]) - for path in tracked_extra_inputs.get("pub_info", []): - snippet = battinfo.add_input_data( - _rel(path, root_folder), - zenodo_url, - EXTRA_INPUT_DESCRIPTIONS["pub_info"], - package=zenodo_package, - zip_filename=zenodo_zip_filename, - ) - battinfo_json = battinfo.merge_jsonld_on_type([battinfo_json, snippet]) - for label, label_paths in tracked_outputs.items(): - for path in label_paths: - rel = _rel(path, root_folder) - if path.suffix == ".png": - comment = plot_description(label) - elif rel.endswith((".parquet", ".csv")): - comment = data_description(label) - else: - comment = OUTPUT_MISC_DESCRIPTIONS.get(label) - extras = {"rdfs:comment": comment} if comment else None - snippet = battinfo.add_data( - rel, + if battinfo_xlsx_path: + snippet = battinfo.add_input_data( + _rel(battinfo_xlsx_path, root_folder), zenodo_url, - extras=extras, + EXTRA_INPUT_DESCRIPTIONS["battinfo_xlsx"], package=zenodo_package, zip_filename=zenodo_zip_filename, ) battinfo_json = battinfo.merge_jsonld_on_type([battinfo_json, snippet]) - metadata_path = folder / f"metadata.{fcid or sample_id}.json" - with metadata_path.open("w", encoding="utf-8") as mf: - json.dump(battinfo_json, mf, indent=4, ensure_ascii=False) - tracked_outputs["metadata"] = [metadata_path] + for path in tracked_extra_inputs.get("pub_info", []): + snippet = battinfo.add_input_data( + _rel(path, root_folder), + zenodo_url, + EXTRA_INPUT_DESCRIPTIONS["pub_info"], + package=zenodo_package, + zip_filename=zenodo_zip_filename, + ) + battinfo_json = battinfo.merge_jsonld_on_type([battinfo_json, snippet]) + for label, label_paths in tracked_outputs.items(): + for path in label_paths: + rel = _rel(path, root_folder) + if path.suffix == ".png": + comment = plot_description(label) + elif rel.endswith((".parquet", ".csv")): + comment = data_description(label) + else: + comment = OUTPUT_MISC_DESCRIPTIONS.get(label) + extras = {"rdfs:comment": comment} if comment else None + snippet = battinfo.add_data( + rel, + zenodo_url, + extras=extras, + package=zenodo_package, + zip_filename=zenodo_zip_filename, + ) + battinfo_json = battinfo.merge_jsonld_on_type([battinfo_json, snippet]) + metadata_path = folder / f"metadata.{fcid or sample_id}.json" + with metadata_path.open("w", encoding="utf-8") as mf: + json.dump(battinfo_json, mf, indent=4, ensure_ascii=False) + tracked_outputs["metadata"] = [metadata_path] + else: + reporter.skip("Metadata", "no BattINFO") return tracked_outputs, tracked_extra_inputs, raw_inputs, fcid, sample_id @@ -1125,13 +1159,14 @@ def merge_summaries( return df[cols] -def analyse_all_samples( +def analyse_all_samples( # noqa: PLR0913 folder: str | Path, *, save_format: SAVE_FORMATS = "parquet", max_search_depth: int = DEFAULT_DEPTH, max_folder_searches: int = DEFAULT_SEARCH, zip_output: bool = False, + reporter: Reporter | None = None, ) -> None: """Take a folder and run all analysis. @@ -1140,8 +1175,11 @@ def analyse_all_samples( download URL with the in-archive path as a fragment. Otherwise (the default), files are left unzipped and "@id"s reference each file's own Zenodo download URL directly, for uploading every file individually. + + ``reporter`` is told when each sample and step starts and finishes, to show progress. """ folder = Path(folder).resolve() + reporter = reporter or Reporter() pub_info, pub_info_path = pub_info_from_root(folder) zenodo_package: Literal["zip", "files"] = "zip" if zip_output else "files" zip_filename = pub_info.get("zenodo_zip_filename") or f"{folder.name}.zip" @@ -1162,56 +1200,71 @@ def analyse_all_samples( raw_inputs_by_sample_folder: dict[Path, dict[str, list[Path]]] = {} fcids_by_sample_folder: dict[Path, str | None] = {} sample_ids_by_sample_folder: dict[Path, str] = {} - if len(sample_folders) > 1: - logger.info("Found %d sample folders:", len(sample_folders)) - for sample_folder in sample_folders: - ( - tracked_by_sample_folder[sample_folder], - tracked_extras_by_sample_folder[sample_folder], - raw_inputs_by_sample_folder[sample_folder], - fcids_by_sample_folder[sample_folder], - sample_ids_by_sample_folder[sample_folder], - ) = analyse_sample( - sample_folder, - save_format=save_format, - pub_info=pub_info, - pub_info_path=pub_info_path, - config=configs_by_sample_folder[sample_folder], - root_folder=folder, - zenodo_package=zenodo_package, - ) - - if len(sample_folders) > 1: - summaries = find_all_sample_summaries( - folder, max_search_depth, max_folder_searches, search_config - ) - df = merge_summaries(summaries, sample_ids_by_sample_folder) - writer = pd.ExcelWriter(folder / "combined_summary.xlsx", engine="xlsxwriter") - df.to_excel(writer, index=False, sheet_name="Summary") - workbook = writer.book - worksheet = writer.sheets["Summary"] - worksheet.autofit() - workbook.close() - logger.info("\n🎉 Combined all the results into one big summary") + reporter.start(folder, len(sample_folders)) + for i, sample_folder in enumerate(sample_folders, start=1): + name = _rel(sample_folder, folder) if sample_folder != folder else folder.name + with reporter.section(f"[{i}/{len(sample_folders)}] {name}"): + ( + tracked_by_sample_folder[sample_folder], + tracked_extras_by_sample_folder[sample_folder], + raw_inputs_by_sample_folder[sample_folder], + fcids_by_sample_folder[sample_folder], + sample_ids_by_sample_folder[sample_folder], + ) = analyse_sample( + sample_folder, + save_format=save_format, + pub_info=pub_info, + pub_info_path=pub_info_path, + config=configs_by_sample_folder[sample_folder], + root_folder=folder, + zenodo_package=zenodo_package, + reporter=reporter, + ) from pyflowbatt.rocrate_output import write_rocrate # noqa: PLC0415 - write_rocrate( - folder, - sample_folders, - tracked_by_sample_folder, - tracked_extras_by_sample_folder, - fcids_by_sample_folder, - configs_by_sample_folder, - sample_ids_by_sample_folder, - raw_inputs_by_sample_folder, - ) - logger.info("📦 Written RO-Crate metadata to %s", folder / "ro-crate-metadata.json") + with reporter.section("Finishing"): + reporter.plan( + [ + *(["Combined summary"] if len(sample_folders) > 1 else []), + "RO-Crate", + *(["Zip"] if zip_output else []), + ] + ) + if len(sample_folders) > 1: + with reporter.task("Combined summary", detail="combined_summary.xlsx"): + summaries = find_all_sample_summaries( + folder, max_search_depth, max_folder_searches, search_config + ) + df = merge_summaries(summaries, sample_ids_by_sample_folder) + writer = pd.ExcelWriter(folder / "combined_summary.xlsx", engine="xlsxwriter") + df.to_excel(writer, index=False, sheet_name="Summary") + workbook = writer.book + worksheet = writer.sheets["Summary"] + worksheet.autofit() + workbook.close() + logger.debug("Combined all the results into one big summary") + + with reporter.task("RO-Crate", detail="ro-crate-metadata.json"): + write_rocrate( + folder, + sample_folders, + tracked_by_sample_folder, + tracked_extras_by_sample_folder, + fcids_by_sample_folder, + configs_by_sample_folder, + sample_ids_by_sample_folder, + raw_inputs_by_sample_folder, + ) + logger.debug("Written RO-Crate metadata to %s", folder / "ro-crate-metadata.json") - if zip_output: - zip_path = folder.parent / zip_filename - zip_folder(folder, zip_path) - logger.info("🤐 Zipped everything into %s", zip_path) + if zip_output: + zip_path = folder.parent / zip_filename + with reporter.task("Zip", detail=str(zip_path)): + zip_folder(folder, zip_path) + logger.debug("Zipped everything into %s", zip_path) + + reporter.finish() def dry_analyse_all_samples( diff --git a/pyflowbatt/cli.py b/pyflowbatt/cli.py index 696bc32..578dd6d 100644 --- a/pyflowbatt/cli.py +++ b/pyflowbatt/cli.py @@ -1,63 +1,119 @@ """PyFlowBatt CLI.""" import argparse +import io import logging +import sys +from collections.abc import Generator +from contextlib import contextmanager from pathlib import Path from pyflowbatt.config import CONFIG_FILENAME, ConfigError, write_template_config +from pyflowbatt.report import LOG_FILENAME, RichReporter + +logger = logging.getLogger("pyflowbatt") + + +@contextmanager +def _console_logging( + reporter: RichReporter, *, verbose: bool = False, log_path: Path | None = None +) -> Generator[None, None, None]: + """Send log records, including Python warnings, to the reporter's console and a log file.""" + root = logging.getLogger() + reporter.log_handler.setLevel(logging.DEBUG if verbose else logging.INFO) + handlers: list[logging.Handler] = [reporter.log_handler] + if log_path is not None: + try: + file_handler = logging.FileHandler(log_path, mode="w", encoding="utf-8") + except OSError as e: + reporter.console.print(f"Not writing a log file, cannot open {log_path}: {e}") + else: + file_handler.setFormatter( + logging.Formatter("%(asctime)s %(levelname)-8s %(name)s: %(message)s") + ) + handlers.append(file_handler) + reporter.log_path = log_path + for handler in handlers: + root.addHandler(handler) + previous_level = logger.level + logger.setLevel(logging.DEBUG if verbose or log_path else logging.INFO) + logging.captureWarnings(True) # noqa: FBT003 + try: + yield + finally: + logging.captureWarnings(False) # noqa: FBT003 + logger.setLevel(previous_level) + for handler in handlers: + root.removeHandler(handler) + if handler is not reporter.log_handler: + handler.close() def init_config(folder: str | None = None) -> None: """Write a template pyflowbatt.toml into folder (default: current directory).""" folderpath = Path.cwd() if not folder else Path(folder) - logger = logging.getLogger("pyflowbatt") - logger.addHandler(logging.StreamHandler()) - logger.setLevel(logging.INFO) - - try: - path = write_template_config(folderpath) - except FileExistsError: - # Expected/handled condition, not a bug - a traceback here would be noisy. - logger.error("%s already exists, not overwriting", folderpath / CONFIG_FILENAME) # noqa: TRY400 - return - logger.info("Wrote template config to %s", path) + with _console_logging(RichReporter()): + try: + path = write_template_config(folderpath) + except FileExistsError: + # Expected/handled condition, not a bug - a traceback here would be noisy. + logger.error("%s already exists, not overwriting", folderpath / CONFIG_FILENAME) # noqa: TRY400 + return + logger.info("Wrote template config to %s", path) -def analyse( +def analyse( # noqa: PLR0913 folder: str | None = None, *, dry: bool = False, zip_output: bool = False, max_folder_searches: int = 10000, max_search_depth: int = 6, + verbose: bool = False, ) -> None: """Run PyFlowBatt on a folder.""" + import matplotlib as mpl # noqa: PLC0415 + + # Plots are only saved, and a GUI backend breaks when the live display's thread collects them + mpl.use("Agg") + from pyflowbatt.analysis import analyse_all_samples, dry_analyse_all_samples # noqa: PLC0415 folderpath = Path.cwd() if not folder else Path(folder) - logger = logging.getLogger("pyflowbatt") - logger.addHandler(logging.StreamHandler()) - logger.setLevel(logging.INFO) - - logger.info("Running PyFlowBatt on: %s", folderpath) + reporter = RichReporter(verbose=verbose) if not folderpath.is_dir(): - logger.error("%s is not a directory", folderpath) + with _console_logging(reporter): + logger.error("%s is not a directory", folderpath) return - 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 + log_path = None if dry else folderpath / LOG_FILENAME + with _console_logging(reporter, verbose=verbose, log_path=log_path): + try: + if dry: + logger.info("Running PyFlowBatt on: %s", folderpath) + 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( + folderpath, + max_folder_searches=max_folder_searches, + max_search_depth=max_search_depth, + zip_output=zip_output, + reporter=reporter, + ) + 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: """PyFlowBatt CLI.""" + if isinstance(sys.stdout, io.TextIOWrapper): + # File names and messages can hold characters a redirected cp1252 stdout can't encode + sys.stdout.reconfigure(errors="replace") parser = argparse.ArgumentParser(prog="pyflowbatt") subparsers = parser.add_subparsers(dest="command") @@ -87,6 +143,12 @@ def main() -> None: "every file individually)." ), ) + parser.add_argument( + "-v", + "--verbose", + action="store_true", + help="Show info and debug messages for each step, and full tracebacks", + ) parser.add_argument( "--max-folder-searches", type=int, @@ -108,7 +170,7 @@ def main() -> None: kwargs["max_folder_searches"] = args.max_folder_searches if args.max_search_depth is not None: kwargs["max_search_depth"] = args.max_search_depth - analyse(folder=args.folder, dry=args.dry, zip_output=args.zip, **kwargs) + analyse(folder=args.folder, dry=args.dry, zip_output=args.zip, verbose=args.verbose, **kwargs) if __name__ == "__main__": diff --git a/pyflowbatt/report.py b/pyflowbatt/report.py new file mode 100644 index 0000000..9f4614f --- /dev/null +++ b/pyflowbatt/report.py @@ -0,0 +1,336 @@ +"""Terminal display of analysis progress, one live line per step with warnings grouped after.""" + +import logging +import re +import time +from collections.abc import Generator +from contextlib import contextmanager +from dataclasses import dataclass, field +from pathlib import Path + +from rich.console import Console, Group +from rich.live import Live +from rich.padding import Padding +from rich.spinner import Spinner +from rich.table import Table +from rich.text import Text +from rich.traceback import Traceback + +from pyflowbatt.version import __title__, __version__ + +ICONS = { + "ok": ("✔", "green"), + "warn": ("⚠", "yellow"), + "fail": ("✖", "red"), + "skip": ("–", "dim"), # noqa: RUF001 + "info": ("·", "dim"), + "pending": ("○", "dim"), +} +# For output that can't encode the symbols above, e.g. redirected to a cp1252 file on Windows +ASCII_ICONS = { + "ok": ("+", "green"), + "warn": ("!", "yellow"), + "fail": ("x", "red"), + "skip": ("-", "dim"), + "info": ("-", "dim"), + "pending": (".", "dim"), +} + +LOG_FILENAME = "pyflowbatt.log" + +# Loggers whose warnings are informational, shown only with verbose +QUIET_LOGGERS = ("battinfoconverter_backend",) + +# First line of a warning routed through logging.captureWarnings: "path:line: Category: message" +_PY_WARNING_RE = re.compile(r"^(?P.+):(?P\d+): (?P\w+): (?P.*)$") + + +class Reporter: + """Progress hooks called during analysis, does nothing by default.""" + + def start(self, folder: Path, n_samples: int) -> None: + """Begin a run over n_samples sample folders.""" + + @contextmanager + def section(self, title: str) -> Generator[None, None, None]: # noqa: ARG002 + """Group the steps run inside under a heading.""" + yield + + def plan(self, labels: list[str]) -> None: + """Announce the steps a section will run, in order.""" + + @contextmanager + def task(self, label: str, detail: str = "") -> Generator[None, None, None]: # noqa: ARG002 + """Run one step inside a section, detail is shown next to it (e.g. an output file).""" + yield + + def skip(self, label: str, reason: str) -> None: + """Show a step that had nothing to do.""" + + def finish(self) -> None: + """End the run.""" + + +@dataclass +class _Line: + """One step and the log records emitted while it ran.""" + + label: str + status: str = "running" + detail: str = "" + error: str = "" + start: float = field(default_factory=time.perf_counter) + end: float | None = None + spinner: Spinner = field(default_factory=lambda: Spinner("dots", style="cyan")) + records: list[logging.LogRecord] = field(default_factory=list) + + def close(self, level: int) -> None: + """Finish the step, level is the worst level of its records.""" + self.end = time.perf_counter() + if self.error or level >= logging.ERROR: + self.status = "fail" + elif level >= logging.WARNING: + self.status = "warn" + else: + self.status = "ok" + + +class _CaptureHandler(logging.Handler): + """Attach every record to the step that is currently running.""" + + def __init__(self, reporter: "RichReporter") -> None: + super().__init__() + self.reporter = reporter + + def emit(self, record: logging.LogRecord) -> None: + if self.reporter.current is not None: + self.reporter.current.records.append(record) + + +class _ConsoleHandler(logging.Handler): + """Print records from outside any section, sections print their own.""" + + def __init__(self, reporter: "RichReporter") -> None: + super().__init__() + self.reporter = reporter + + def emit(self, record: logging.LogRecord) -> None: + if self.reporter.capturing: + return + self.reporter.print_record(record, indent=False) + + +class RichReporter(Reporter): + """Live spinner per step in the terminal, warnings and errors listed after each section.""" + + def __init__( + self, + console: Console | None = None, + *, + verbose: bool = False, + quiet_loggers: tuple[str, ...] = QUIET_LOGGERS, + ) -> None: + """Verbose also lists info and debug records, and prints full tracebacks.""" + self.console = console or Console() + self.verbose = verbose + self.quiet_loggers = quiet_loggers + self.log_path: Path | None = None + self.icons = ICONS if self.console.encoding.lower().startswith("utf") else ASCII_ICONS + self.capturing = False + self.current: _Line | None = None + self.log_handler = _ConsoleHandler(self) + self._lines: list[_Line] = [] + self._title = "" + self._live: Live | None = None + self._n_samples = 0 + self._warnings = 0 + self._errors = 0 + + def start(self, folder: Path, n_samples: int) -> None: + """Print the run header.""" + self._n_samples = n_samples + self.console.print( + Text.assemble( + (f"{__title__} {__version__}", "bold"), + f" {self.icons['info'][0]} {folder} {self.icons['info'][0]} ", + _plural(n_samples, "sample"), + ) + ) + + @contextmanager + def section(self, title: str) -> Generator[None, None, None]: + """Show a live block of steps, then the warnings and errors they logged.""" + loose = _Line("") + self._title = title + self._lines = [] + self.current = loose + capture = _CaptureHandler(self) + root = logging.getLogger() + root.addHandler(capture) + self.capturing = True + try: + yield + finally: + for line in self._lines: + if line.status == "pending": + line.status, line.detail = "skip", "not run" + if self._live is None: + self.console.print(self._render()) + else: + self._live.stop() + self._live = None + if not self.console.is_terminal: + self.console.line() # Live only ends its block with a newline on terminals + root.removeHandler(capture) + self.capturing = False + self.current = None + for line in [loose, *self._lines]: + self._print_line_records(line) + self._lines = [] + + def plan(self, labels: list[str]) -> None: + """Show the section's steps as waiting, so the block doesn't grow as they run.""" + self._lines.extend(_Line(label, status="pending") for label in labels) + self._start_live() + + @contextmanager + def task(self, label: str, detail: str = "") -> Generator[None, None, None]: + """Show a spinner while the step runs, then an icon for the worst level it logged.""" + line = self._claim(label) + line.status, line.detail, line.start = "running", detail, time.perf_counter() + previous, self.current = self.current, line + try: + yield + except BaseException as e: + line.error = f"{type(e).__name__}: {e}" if str(e) else type(e).__name__ + raise + finally: + self.current = previous + line.close(max(map(self._shown_level, line.records), default=logging.NOTSET)) + + def skip(self, label: str, reason: str) -> None: + """Show a dimmed line for a step with nothing to do.""" + line = self._claim(label) + line.status, line.detail = "skip", reason + + def _claim(self, label: str) -> _Line: + """Planned line waiting under this label, or a new line after the others.""" + for line in self._lines: + if line.status == "pending" and line.label == label: + return line + line = _Line(label) + self._lines.append(line) + self._start_live() + return line + + def _start_live(self) -> None: + """Start drawing the section once it has lines, so its title and steps appear together.""" + if self._live is None: + self._live = Live( + get_renderable=self._render, console=self.console, refresh_per_second=10 + ) + self._live.start(refresh=True) + + def finish(self) -> None: + """Print totals for the run.""" + parts = [_plural(self._n_samples, "sample")] + if self._warnings: + parts.append(_plural(self._warnings, "warning")) + if self._errors: + parts.append(_plural(self._errors, "error")) + style = "red" if self._errors else "yellow" if self._warnings else "green" + self.console.print() + self.console.print(Text(f"Done: {', '.join(parts)}", style=style)) + if self.log_path: + self.console.print(Text(f"Full log: {self.log_path}", style="dim")) + + def print_record(self, record: logging.LogRecord, label: str = "", *, indent: bool) -> None: + """Print one log record with an icon for its level.""" + level = self._shown_level(record) + if level < record.levelno and not self.verbose: + return + if level >= logging.ERROR: + self._errors += 1 + icon, style = self.icons["fail"] + elif level >= logging.WARNING: + self._warnings += 1 + icon, style = self.icons["warn"] + else: + icon, style = self.icons["info"] + message = record.getMessage() + if not indent and level < logging.WARNING: + self.console.print(Text(message)) + return + message = message.strip().removeprefix("- ") + if record.name == "py.warnings" and not self.verbose: + message = _short_warning(message) + exc = record.exc_info[1] if record.exc_info else None + if exc is not None and not self.verbose: + message += f": {exc}" + self._print_icon_line(icon, style, label, message, indent=indent) + if exc is not None and self.verbose and record.exc_info: + self.console.print(Traceback.from_exception(*record.exc_info)) + + def _print_line_records(self, line: _Line) -> None: + for record in line.records: + if self._shown_level(record) >= logging.WARNING or self.verbose: + self.print_record(record, line.label, indent=True) + if line.error: + self._errors += 1 + self._print_icon_line(*self.icons["fail"], line.label, line.error, indent=True) + + def _shown_level(self, record: logging.LogRecord) -> int: + """Level a record is shown at, quiet loggers' warnings count as info.""" + quiet = any( + record.name == name or record.name.startswith(f"{name}.") for name in self.quiet_loggers + ) + if quiet and record.levelno < logging.ERROR: + return min(record.levelno, logging.INFO) + return record.levelno + + def _print_icon_line( + self, icon: str, style: str, label: str, message: str, *, indent: bool + ) -> None: + """Print a message after an icon, wrapped lines aligned with the message.""" + grid = Table.grid(padding=(0, 1)) + grid.add_column(width=1) + grid.add_column(overflow="fold") + text = Text(f"{label}: ", style="bold") if label else Text() + text.append(message) + grid.add_row(Text(icon, style=style), text) + self.console.print(Padding(grid, (0, 0, 0, 2 if indent else 0), expand=False)) + + def _render(self) -> Group: + table = Table.grid(padding=(0, 1)) + table.add_column(width=1) + table.add_column(min_width=16) + table.add_column(style="dim") + now = time.perf_counter() + for line in self._lines: + if line.status == "running": + icon = line.spinner + else: + symbol, style = self.icons[line.status] + icon = Text(symbol, style=style) + if line.status == "pending": + table.add_row(icon, Text(line.label, style="dim"), "") + elif line.status == "skip": + table.add_row(icon, line.label, line.detail) + else: + elapsed = (line.end or now) - line.start + table.add_row(icon, line.label, f"{elapsed:5.1f}s {line.detail}") + return Group( + Text(), Text(self._title, style="bold"), Padding(table, (0, 0, 0, 2), expand=False) + ) + + +def _short_warning(message: str) -> str: + """Shorten a captured Python warning to "Category: message (file:line)".""" + match = _PY_WARNING_RE.match(message.splitlines()[0]) + if not match: + return message + return f"{match['category']}: {match['message']} ({Path(match['path']).name}:{match['line']})" + + +def _plural(n: int, word: str) -> str: + return f"{n} {word}" if n == 1 else f"{n} {word}s" From a96043a956510899a1b45745dfc50796675123f3 Mon Sep 17 00:00:00 2001 From: Graham Kimbell <88666181+g-kimbell@users.noreply.github.com> Date: Wed, 14 Oct 2026 15:10:04 +0200 Subject: [PATCH 2/5] Add rich to dependencies --- pyproject.toml | 1 + uv.lock | 2 ++ 2 files changed, 3 insertions(+) diff --git a/pyproject.toml b/pyproject.toml index b36e560..bee56bf 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -19,6 +19,7 @@ dependencies = [ "rocrate>=0.9", "yadg>=7.0.1,<8.0.0", "tomli>=2.0.0; python_version < '3.11'", + "rich>=13.0.0", ] [dependency-groups] diff --git a/uv.lock b/uv.lock index a26c604..f0c6af3 100644 --- a/uv.lock +++ b/uv.lock @@ -1619,6 +1619,7 @@ dependencies = [ { name = "matplotlib" }, { name = "pandas" }, { name = "pyarrow" }, + { name = "rich" }, { name = "rocrate" }, { name = "tomli", marker = "python_full_version < '3.11'" }, { name = "xlsxwriter" }, @@ -1641,6 +1642,7 @@ requires-dist = [ { name = "matplotlib", specifier = ">=3.10.6" }, { name = "pandas", specifier = ">=2.3.2" }, { name = "pyarrow", specifier = ">=21.0.0" }, + { name = "rich", specifier = ">=13.0.0" }, { name = "rocrate", specifier = ">=0.9" }, { name = "tomli", marker = "python_full_version < '3.11'", specifier = ">=2.0.0" }, { name = "xlsxwriter", specifier = ">=3.2.9" }, From df8ed332cca9ba896bf5aebd7b99f0960c3d6db0 Mon Sep 17 00:00:00 2001 From: Graham Kimbell <88666181+g-kimbell@users.noreply.github.com> Date: Wed, 14 Oct 2026 15:17:25 +0200 Subject: [PATCH 3/5] Only log info in file by default --- pyflowbatt/analysis.py | 2 +- pyflowbatt/cli.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/pyflowbatt/analysis.py b/pyflowbatt/analysis.py index e65fd56..f5f5fcb 100644 --- a/pyflowbatt/analysis.py +++ b/pyflowbatt/analysis.py @@ -534,7 +534,7 @@ def analyse_sample( config = config or PyFlowBattConfig.load(folder) reporter = reporter or Reporter() - logger.debug("Analysing %s", folder.name) + logger.info("Analysing %s", folder.name) classified = classify_technique_files(folder, config, warn=True) gcpl_files = classified.get("gcpl", []) gcpl_file = gcpl_files[0] if gcpl_files else None diff --git a/pyflowbatt/cli.py b/pyflowbatt/cli.py index 578dd6d..ccc22ae 100644 --- a/pyflowbatt/cli.py +++ b/pyflowbatt/cli.py @@ -36,7 +36,7 @@ def _console_logging( for handler in handlers: root.addHandler(handler) previous_level = logger.level - logger.setLevel(logging.DEBUG if verbose or log_path else logging.INFO) + logger.setLevel(logging.DEBUG if verbose else logging.INFO) logging.captureWarnings(True) # noqa: FBT003 try: yield From f3993b6092bf42ac945038387d0f0114c291ce0c Mon Sep 17 00:00:00 2001 From: Graham Kimbell <88666181+g-kimbell@users.noreply.github.com> Date: Wed, 14 Oct 2026 15:17:58 +0200 Subject: [PATCH 4/5] More CLI tests --- tests/test_cli.py | 48 +++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 46 insertions(+), 2 deletions(-) diff --git a/tests/test_cli.py b/tests/test_cli.py index e349681..a4bb7c7 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1,10 +1,12 @@ """Tests for the PyFlowBatt CLI, in particular `pyflowbatt init`.""" +import zipfile from pathlib import Path import pytest from pyflowbatt import cli +from pyflowbatt.analysis import zip_folder def test_init_config_writes_template(tmp_path: Path) -> None: @@ -74,7 +76,12 @@ def fake_init_config(*_args: object, **_kwargs: object) -> None: cli.main() - assert called_with == {"folder": "somefolder", "dry": True, "zip_output": False} + assert called_with == { + "folder": "somefolder", + "dry": True, + "zip_output": False, + "verbose": False, + } def test_main_zip_flag_dispatches_zip_output_true(monkeypatch: pytest.MonkeyPatch) -> None: @@ -89,7 +96,12 @@ def fake_analyse(**kwargs: object) -> None: cli.main() - assert called_with == {"folder": "somefolder", "dry": False, "zip_output": True} + assert called_with == { + "folder": "somefolder", + "dry": False, + "zip_output": True, + "verbose": False, + } def test_analyse_bad_toml_logs_error(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None: @@ -100,3 +112,35 @@ def test_analyse_bad_toml_logs_error(tmp_path: Path, caplog: pytest.LogCaptureFi cli.analyse(folder=str(tmp_path)) assert "line 1: --- Electrode area ---" in caplog.text + + +def test_analyse_writes_log_file(tmp_path: Path) -> None: + """A real run writes all logs to pyflowbatt.log.""" + cli.analyse(folder=str(tmp_path)) + log = (tmp_path / "pyflowbatt.log").read_text(encoding="utf-8") + assert "No sample folders found" in log + + +def test_dry_run_writes_no_log_file(tmp_path: Path) -> None: + """Dry runs do not change files.""" + cli.analyse(folder=str(tmp_path), dry=True) + assert not (tmp_path / "pyflowbatt.log").exists() + + +def test_zip_skips_log_file(tmp_path: Path) -> None: + """The log is not included with the zip.""" + root = tmp_path / "root" + root.mkdir() + (root / "data.csv").write_text("x") + (root / "pyflowbatt.log").write_text("log") + with zipfile.ZipFile(zip_folder(root, tmp_path / "out.zip")) as zf: + assert zf.namelist() == ["data.csv"] + + +@pytest.mark.parametrize("verbose", [False, True]) +def test_log_file_has_debug_only_when_verbose(tmp_path: Path, *, verbose: bool) -> None: + """Debug records reach pyflowbatt.log only with --verbose.""" + (tmp_path / "pyflowbatt.toml").write_text("") + cli.analyse(folder=str(tmp_path), verbose=verbose) + log = (tmp_path / "pyflowbatt.log").read_text(encoding="utf-8") + assert ("Loading PyFlowBatt config from" in log) is verbose From c5837a5f60cace371d7446a66d181c19ea66e960 Mon Sep 17 00:00:00 2001 From: Graham Kimbell <88666181+g-kimbell@users.noreply.github.com> Date: Wed, 14 Oct 2026 15:18:14 +0200 Subject: [PATCH 5/5] Tests for rich output --- tests/test_report.py | 202 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 202 insertions(+) create mode 100644 tests/test_report.py diff --git a/tests/test_report.py b/tests/test_report.py new file mode 100644 index 0000000..56deac9 --- /dev/null +++ b/tests/test_report.py @@ -0,0 +1,202 @@ +"""Tests for the terminal progress reporter.""" + +import io +import logging +from pathlib import Path + +import pytest +from rich.console import Console + +from pyflowbatt.report import Reporter, RichReporter + +logger = logging.getLogger("pyflowbatt.test") + + +def make_reporter(*, verbose: bool = False) -> tuple[RichReporter, io.StringIO]: + """Make a reporter writing plain text to a buffer.""" + out = io.StringIO() + console = Console(file=out, width=120, color_system=None) + return RichReporter(console, verbose=verbose), out + + +@pytest.fixture(autouse=True) +def _debug_level() -> None: + logger.setLevel(logging.DEBUG) + + +def test_task_icons_follow_worst_level() -> None: + """Clean steps get a tick, warnings a warning sign, errors a cross.""" + reporter, out = make_reporter() + with reporter.section("[1/1] sample"): + with reporter.task("OCV"): + logger.info("fine") + with reporter.task("LSV post"): + logger.warning("- Only one LSV file found") + with reporter.task("EIS pre"): + logger.error("Failed to fit eis_pre") + reporter.skip("CV", "no files") + + text = out.getvalue() + assert "✔ OCV" in text + assert "⚠ LSV post" in text + assert "✖ EIS pre" in text + assert "– CV" in text # noqa: RUF001 + assert "no files" in text + assert "⚠ LSV post: Only one LSV file found" in text + assert "✖ EIS pre: Failed to fit eis_pre" in text + assert "fine" not in text + + +def test_verbose_lists_info_records() -> None: + """With verbose, info records from a step are listed after the section.""" + reporter, out = make_reporter(verbose=True) + with reporter.section("s"), reporter.task("OCV"): + logger.info("read 10 rows") + assert "OCV: read 10 rows" in out.getvalue() + + +def test_records_outside_tasks_grouped_without_label() -> None: + """A warning inside a section but outside any step is listed unlabelled.""" + reporter, out = make_reporter() + with reporter.section("s"): + logger.warning("Could not find sample ID") + assert "⚠ Could not find sample ID" in out.getvalue() + + +def test_exception_marks_task_failed_and_propagates() -> None: + """An exception escaping a step shows a cross and is re-raised.""" + reporter, out = make_reporter() + + def fail() -> None: + with reporter.section("s"), reporter.task("BattINFO"): + msg = "mismatch" + raise ValueError(msg) + + with pytest.raises(ValueError, match="mismatch"): + fail() + assert "✖ BattINFO: ValueError: mismatch" in out.getvalue() + + +def test_console_handler_quiet_inside_sections() -> None: + """The console handler only prints records from outside sections.""" + reporter, out = make_reporter() + root = logging.getLogger() + root.addHandler(reporter.log_handler) + try: + logger.warning("before") + with reporter.section("s"), reporter.task("OCV"): + logger.warning("inside") + finally: + root.removeHandler(reporter.log_handler) + text = out.getvalue() + assert text.count("before") == 1 + assert text.count("inside") == 1 + + +def test_records_still_propagate(caplog: pytest.LogCaptureFixture) -> None: + """Capturing for display does not hide records from other handlers.""" + reporter, _ = make_reporter() + with caplog.at_level("WARNING"), reporter.section("s"), reporter.task("OCV"): + logger.warning("visible") + assert "visible" in caplog.text + + +def test_finish_counts_warnings_and_errors() -> None: + """The final line totals warnings and errors across sections.""" + reporter, out = make_reporter() + reporter.start(Path("root"), 2) + for name in ("a", "b"): + with reporter.section(name), reporter.task("OCV"): + logger.warning("w") + with reporter.section("c"), reporter.task("EIS"): + logger.error("e") + reporter.finish() + assert "Done: 2 samples, 2 warnings, 1 error" in out.getvalue() + + +def test_base_reporter_is_silent() -> None: + """The default reporter runs steps without output or capture.""" + reporter = Reporter() + with reporter.section("s"), reporter.task("OCV", detail="x"): + reporter.skip("CV", "no files") + reporter.finish() + + +def test_ascii_icons_when_output_cannot_encode_symbols() -> None: + """Output in a non-UTF encoding gets ASCII icons instead of failing to encode.""" + raw = io.BytesIO() + out = io.TextIOWrapper(raw, encoding="cp1252") + reporter = RichReporter(Console(file=out, width=120, color_system=None)) + with reporter.section("s"), reporter.task("OCV"): + logger.warning("w") + out.flush() + text = raw.getvalue().decode("cp1252") + assert "! OCV" in text + assert "! OCV: w" in text + + +def test_quiet_logger_warnings_hidden_and_do_not_mark_task() -> None: + """BattINFO converter warnings are informational, its critical messages still show.""" + reporter, out = make_reporter() + quiet = logging.getLogger("battinfoconverter_backend.validate") + with reporter.section("s"): + with reporter.task("BattINFO"): + quiet.warning("Term already in the default context") + with reporter.task("Metadata"): + quiet.critical("Missing required value") + text = out.getvalue() + assert "✔ BattINFO" in text + assert "default context" not in text + assert "✖ Metadata: Missing required value" in text + + +def test_quiet_logger_warnings_shown_when_verbose() -> None: + """With verbose, quiet loggers' warnings are listed as info.""" + reporter, out = make_reporter(verbose=True) + with reporter.section("s"), reporter.task("BattINFO"): + logging.getLogger("battinfoconverter_backend").warning("Term already in context") + text = out.getvalue() + assert "✔ BattINFO" in text + assert "· BattINFO: Term already in context" in text + + +def test_planned_steps_keep_their_order() -> None: + """Steps fill the planned lines in place, whatever order they run in.""" + reporter, out = make_reporter() + with reporter.section("s"): + reporter.plan(["BattINFO", "OCV", "GCPL"]) + with reporter.task("GCPL"): + pass + reporter.skip("BattINFO", "no xlsx") + with reporter.task("OCV"): + pass + text = out.getvalue() + assert text.index("BattINFO") < text.index("OCV") < text.index("GCPL") + assert text.count("GCPL") == 1 + + +def test_planned_steps_never_run_marked_not_run() -> None: + """A planned step left waiting when the section ends, e.g. after an error, says so.""" + reporter, out = make_reporter() + + def fail() -> None: + with reporter.section("s"): + reporter.plan(["BattINFO", "Summary"]) + with reporter.task("BattINFO"): + msg = "mismatch" + raise ValueError(msg) + + with pytest.raises(ValueError, match="mismatch"): + fail() + assert "– Summary" in out.getvalue() # noqa: RUF001 + assert "not run" in out.getvalue() + + +def test_pending_lines_rendered_before_steps_run() -> None: + """The live block lists every planned step as waiting before any of them starts.""" + reporter, _ = make_reporter() + reporter.plan(["OCV", "GCPL"]) + out = io.StringIO() + Console(file=out, width=80, color_system=None).print(reporter._render()) # noqa: SLF001 + assert "○ OCV" in out.getvalue() + assert "○ GCPL" in out.getvalue()