feat(batch): aggregate + qc + outputs (#701)

cellpy/batch/__init__.py

@@ -38,8 +38,14 @@
 )
 from cellpy.batch.runner import load_cell, run
 from cellpy.batch.store import CellStore
+from cellpy.batch import aggregate, outputs, qc
+from cellpy.batch.aggregate import combine_summaries
 
 __all__ = [
+    "aggregate",
+    "qc",
+    "outputs",
+    "combine_summaries",
     "Journal",
     "read_journal",
     "write_journal",

cellpy/batch/aggregate.py

@@ -0,0 +1,74 @@
+"""Batch aggregation (batch v3, #701).
+
+Turns a set of loaded cells into one tidy, long-format frame with ``cell`` /
+``group`` / ``sub_group`` key columns -- replacing the legacy wide/multiindex
+``join_summaries`` machinery. This is the frame the collectors redesign
+(Epic B) builds on, so it lands here as ``batch.aggregate.combine_summaries``.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+import polars as pl
+
+from cellpy.batch.journal import FILENAME, Journal
+
+
+def _group_lookup(journal: Journal | None) -> dict[str, tuple[Any, Any]]:
+    """label -> (group, sub_group) from the journal pages."""
+    lookup: dict[str, tuple[Any, Any]] = {}
+    if journal is None:
+        return lookup
+    pages = journal.pages
+    if FILENAME not in pages.columns:
+        return lookup
+    has_group = "group" in pages.columns
+    has_sub = "sub_group" in pages.columns
+    for row in pages.iter_rows(named=True):
+        lookup[row[FILENAME]] = (
+            row.get("group") if has_group else None,
+            row.get("sub_group") if has_sub else None,
+        )
+    return lookup
+
+
+def _summary_of(cell: Any) -> pl.DataFrame | None:
+    summary = getattr(getattr(cell, "data", None), "summary", None)
+    if summary is None:
+        return None
+    if isinstance(summary, pl.DataFrame):
+        return summary
+    try:
+        return pl.from_pandas(summary)
+    except (TypeError, ValueError):
+        return None
+
+
+def combine_summaries(
+    cells: Mapping[str, Any], journal: Journal | None = None
+) -> pl.DataFrame:
+    """Concatenate per-cell summaries into one tidy long-format frame.
+
+    Each row keeps its cell's summary columns plus ``cell``/``group``/
+    ``sub_group`` keys. Cells without a summary are skipped. Returns an empty
+    frame when nothing has a summary.
+    """
+    lookup = _group_lookup(journal)
+    frames: list[pl.DataFrame] = []
+    for label, cell in cells.items():
+        summary = _summary_of(cell)
+        if summary is None or summary.height == 0:
+            continue
+        group, sub_group = lookup.get(label, (None, None))
+        frames.append(
+            summary.with_columns(
+                pl.lit(label).alias("cell"),
+                pl.lit(group).alias("group"),
+                pl.lit(sub_group).alias("sub_group"),
+            )
+        )
+    if not frames:
+        return pl.DataFrame()
+    return pl.concat(frames, how="diagonal_relaxed")

cellpy/batch/outputs.py

@@ -0,0 +1,33 @@
+"""Pure output writers (batch v3, #701).
+
+Each writer takes a frame and an explicit path and writes it -- nothing more.
+Exporting never creates directory trees implicitly (that is
+``layout.ensure_dirs``); the parent directory must already exist.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import polars as pl
+
+
+def write_csv(frame: pl.DataFrame, path: Path | str) -> Path:
+    """Write ``frame`` to a CSV file at ``path``."""
+    path = Path(path)
+    frame.write_csv(path)
+    return path
+
+
+def write_parquet(frame: pl.DataFrame, path: Path | str) -> Path:
+    """Write ``frame`` to a Parquet file at ``path``."""
+    path = Path(path)
+    frame.write_parquet(path)
+    return path
+
+
+def write_excel(frame: pl.DataFrame, path: Path | str) -> Path:
+    """Write ``frame`` to an ``.xlsx`` file at ``path`` (via openpyxl)."""
+    path = Path(path)
+    frame.to_pandas().to_excel(path, index=False, engine="openpyxl")
+    return path

cellpy/batch/qc.py

@@ -0,0 +1,92 @@
+"""Batch quality control (batch v3, #701).
+
+The legacy ``_check_cell_*`` family (batch.py:367-456) as one function that
+returns a tidy per-cell pass/fail frame, instead of ten methods feeding a
+styled report. The facade's ``report()`` renders this frame.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+import polars as pl
+
+#: Preferred charge-capacity column for the cap statistics (native first).
+_CAP_COLS = ("charge_capacity_gravimetric", "charge_capacity")
+
+
+def _frame_len(frame: Any) -> int | None:
+    if frame is None:
+        return None
+    height = getattr(frame, "height", None)
+    if height is not None:
+        return int(height)
+    try:
+        return int(len(frame))
+    except TypeError:
+        return None
+
+
+def _cap_col(summary: Any) -> str | None:
+    columns = getattr(summary, "columns", [])
+    for candidate in _CAP_COLS:
+        if candidate in columns:
+            return candidate
+    return None
+
+
+def _check_one(label: str, cell: Any) -> dict:
+    row: dict[str, Any] = {
+        "cell": label,
+        "empty": None,
+        "n_raw": None,
+        "n_steps": None,
+        "n_summary": None,
+        "n_cycles": None,
+        "max_cap": None,
+        "min_cap": None,
+        "avg_cap": None,
+        "std_cap": None,
+        "pass": False,
+    }
+    try:
+        row["empty"] = bool(cell.empty)
+    except Exception:  # noqa: BLE001 - QC never raises on a single cell
+        pass
+
+    data = getattr(cell, "data", None)
+    row["n_raw"] = _frame_len(getattr(data, "raw", None))
+    row["n_steps"] = _frame_len(getattr(data, "steps", None))
+    summary = getattr(data, "summary", None)
+    row["n_summary"] = _frame_len(summary)
+
+    steps = getattr(data, "steps", None)
+    if steps is not None and "cycle_num" in getattr(steps, "columns", []):
+        try:
+            row["n_cycles"] = int(steps["cycle_num"].max())
+        except Exception:  # noqa: BLE001
+            pass
+
+    if summary is not None and _frame_len(summary):
+        col = _cap_col(summary)
+        if col is not None:
+            series = summary[col]
+            try:
+                row["max_cap"] = float(series.max())
+                row["min_cap"] = float(series.min())
+                row["avg_cap"] = float(series.mean())
+                row["std_cap"] = float(series.std())
+            except Exception:  # noqa: BLE001
+                pass
+
+    row["pass"] = (row["empty"] is False) and bool(row["n_summary"])
+    return row
+
+
+def check(cells: Mapping[str, Any], journal: Any | None = None) -> pl.DataFrame:
+    """Return a tidy per-cell QC frame (one row per cell)."""
+    rows = [_check_one(label, cell) for label, cell in cells.items()]
+    if not rows:
+        return pl.DataFrame()
+    return pl.DataFrame(rows)

tests/test_batch_v3_aggregate.py

@@ -0,0 +1,81 @@
+"""Tests for batch v3 aggregate/qc/outputs (#701)."""
+
+import polars as pl
+import pytest
+
+import cellpy
+from cellpy.batch import combine_summaries, outputs, qc
+from cellpy.batch.journal import FILENAME, Journal
+from tests import fdv
+
+
+@pytest.fixture(scope="module")
+def loaded_cells():
+    cell = cellpy.get(cellpy_file=fdv.cellpy_file_path, testing=True)
+    return {"c45": cell}
+
+
+# ---- aggregate.combine_summaries ----------------------------------------
+
+
+def test_combine_summaries_tidy(loaded_cells):
+    frame = combine_summaries(loaded_cells)
+    assert isinstance(frame, pl.DataFrame)
+    assert frame.height > 0
+    for key in ("cell", "group", "sub_group"):
+        assert key in frame.columns
+    assert set(frame["cell"].unique().to_list()) == {"c45"}
+
+
+def test_combine_summaries_uses_journal_groups(loaded_cells):
+    journal = Journal(
+        pages=pl.DataFrame(
+            {FILENAME: ["c45"], "group": [2], "sub_group": [5]}
+        )
+    )
+    frame = combine_summaries(loaded_cells, journal)
+    assert frame["group"].unique().to_list() == [2]
+    assert frame["sub_group"].unique().to_list() == [5]
+
+
+def test_combine_summaries_empty():
+    assert combine_summaries({}).height == 0
+
+
+# ---- qc.check -----------------------------------------------------------
+
+
+def test_qc_check(loaded_cells):
+    frame = qc.check(loaded_cells)
+    assert isinstance(frame, pl.DataFrame)
+    assert frame.height == 1
+    row = frame.row(0, named=True)
+    assert row["cell"] == "c45"
+    assert row["empty"] is False
+    assert row["n_summary"] and row["n_summary"] > 0
+    assert row["pass"] is True
+
+
+# ---- outputs (pure writers) ---------------------------------------------
+
+
+def test_outputs_roundtrip(tmp_path):
+    frame = pl.DataFrame({"a": [1, 2], "b": ["x", "y"]})
+
+    csv_path = outputs.write_csv(frame, tmp_path / "f.csv")
+    assert csv_path.is_file()
+    assert pl.read_csv(csv_path).equals(frame)
+
+    pq_path = outputs.write_parquet(frame, tmp_path / "f.parquet")
+    assert pq_path.is_file()
+    assert pl.read_parquet(pq_path).equals(frame)
+
+    xlsx_path = outputs.write_excel(frame, tmp_path / "f.xlsx")
+    assert xlsx_path.is_file()
+
+
+def test_outputs_do_not_create_dirs(tmp_path):
+    frame = pl.DataFrame({"a": [1]})
+    missing = tmp_path / "does_not_exist" / "f.csv"
+    with pytest.raises((FileNotFoundError, OSError)):
+        outputs.write_csv(frame, missing)