diff --git a/.issueflows/01-current-issues/issue510_status.md b/.issueflows/01-current-issues/issue510_status.md index 0bf362dc..0f99404f 100644 --- a/.issueflows/01-current-issues/issue510_status.md +++ b/.issueflows/01-current-issues/issue510_status.md @@ -5,16 +5,12 @@ ## What's done - Plan confirmed (2026-07-17): v9 zip-of-parquet + `.cellpy`, default write v9, native cols; three milestones A→B→C. -- **Milestone A (V2-13)** — implemented: - - `CELLPY_FILE_VERSION = 9`, `HDF5_FILE_VERSION = 8` - - `cellpy/readers/cellpy_file/v9.py` — zip-of-parquet writer/reader + `meta.json` (full `TestMetaCollection`, units/limits, schema stamps) - - On-disk native headers via `translate.to_native` / `to_legacy` I/O adapter - - `load` sniffs zip magic vs HDF5; `.h5` suffix still writes v8 - - Default `CellpyCell.save` → `.cellpy` / v9 - - Essential tests in `tests/test_cellpy_file_v9.py` (v8→v9→read, extras persist, `.h5` escape) - - `uv run pytest -m essential` green +- **Milestone A (V2-13)** — merged via PR #521. +- **Milestone B (V2-14)** — implemented: + - `cellpy/readers/cellpy_file/meta_archive.py` — `save_meta_archive` / `load_meta_archive` / `apply_meta_document` (core stubs stay stubs) + - v9 uses shared meta document; preserves campaign `test_id` columns on steps/summary + - Essential tests: `tests/test_meta_archive.py` + campaign v9 round-trip in `test_merge_campaign.py` ## Remaining work -- **B (V2-14):** cellpy-owned meta archive helpers (`save_meta_archive` / `load_meta_archive`) + merged two-test campaign round-trip fixture - **C (V2-15):** exact `cellpycore==` pin + migration guide + release procedure diff --git a/cellpy/readers/cellpy_file/__init__.py b/cellpy/readers/cellpy_file/__init__.py index 95521c99..a7c6f9f7 100644 --- a/cellpy/readers/cellpy_file/__init__.py +++ b/cellpy/readers/cellpy_file/__init__.py @@ -29,9 +29,11 @@ "LoadSelector", "get_format", "load", + "load_meta_archive", "read_fid_table", "read_table", "save", + "save_meta_archive", ] @@ -52,4 +54,12 @@ def __getattr__(name: str): from cellpy.readers.cellpy_file.fids import read_fid_table return read_fid_table + if name == "save_meta_archive": + from cellpy.readers.cellpy_file.meta_archive import save_meta_archive + + return save_meta_archive + if name == "load_meta_archive": + from cellpy.readers.cellpy_file.meta_archive import load_meta_archive + + return load_meta_archive raise AttributeError(f"module {__name__!r} has no attribute {name!r}") diff --git a/cellpy/readers/cellpy_file/meta_archive.py b/cellpy/readers/cellpy_file/meta_archive.py new file mode 100644 index 00000000..b750c048 --- /dev/null +++ b/cellpy/readers/cellpy_file/meta_archive.py @@ -0,0 +1,257 @@ +"""Cellpy-owned metadata archive helpers (issue #510, V2-14). + +``cellpycore.metadata.io.save_archive`` / ``load_archive`` stay deliberate +stubs — real persistence lives here. Use these helpers (or the v9 ``.cellpy`` +path which embeds the same document as ``meta.json``). +""" + +from __future__ import annotations + +import enum +import json +import logging +from datetime import date, datetime +from pathlib import Path +from typing import TYPE_CHECKING, Any, Mapping, Optional, Union + +from cellpycore.config import Cols, default_schema +from cellpycore.metadata.io import from_dict, to_dict +from cellpycore.metadata.models import CellMeta, TestMeta, TestMetaCollection +from cellpycore.units import CellpyUnits + +from cellpy.exceptions import WrongFileVersion +from cellpy.parameters.internal_settings import CellpyLimits +from cellpy.readers import test_meta as test_meta_helpers +from cellpy.readers.cellpy_file.format import CELLPY_FILE_VERSION + +if TYPE_CHECKING: + from cellpy.readers.data_structures import Data + +_module_logger = logging.getLogger(__name__) + +PathLike = Union[str, Path] +MetaArchiveSource = Union["Data", TestMetaCollection, TestMeta] + + +def _json_default(obj: Any) -> Any: + if isinstance(obj, enum.Enum): + return obj.value + if isinstance(obj, (datetime, date)): + return obj.isoformat() + if isinstance(obj, Path): + return obj.as_posix() + # OtherPath / path-like without being a pathlib.Path + if hasattr(obj, "as_posix"): + try: + return obj.as_posix() + except Exception: + pass + if hasattr(obj, "__fspath__"): + return Path(obj).as_posix() + if hasattr(obj, "item"): + return obj.item() + return str(obj) + + +def build_meta_document( + data: "Data", + *, + cellpy_units: Optional[Mapping[str, Any]] = None, + frames_had_test_id: Optional[Mapping[str, bool]] = None, +) -> dict: + """Build the archive / v9 ``meta.json`` document from a ``Data`` object.""" + tests_doc: dict[str, dict] = {} + cell_doc: Optional[dict] = None + active_tester_id = test_meta_helpers._unwrap( + getattr(data.meta_test_dependent, "test_ID", None) + ) + for record in data.tests: + payload = to_dict(record) + if cell_doc is None and isinstance(payload.get("cell"), dict): + cell_doc = payload["cell"] + if int(record.test_id) == int(data.active_test_id): + payload["tester_assigned_test_id"] = active_tester_id + tests_doc[str(record.test_id)] = payload + + if cell_doc is None: + cell_doc = to_dict(CellMeta()) + + schema = default_schema() + doc = { + "cellpy_file_version": CELLPY_FILE_VERSION, + "schema_version": Cols.__version__, + "raw_schema_version": schema.raw.__version__, + "step_schema_version": schema.step.__version__, + "cycle_schema_version": schema.cycle.__version__, + "cell": cell_doc, + "tests": tests_doc, + "raw_units": dict(data.raw_units), + "cellpy_units": dict(cellpy_units) if cellpy_units else {}, + "limits": dict(data.raw_limits), + "active_test_id": int(data.active_test_id), + "loaded_from": getattr(data, "loaded_from", None), + } + if frames_had_test_id is not None: + doc["frames_had_test_id"] = dict(frames_had_test_id) + return doc + + +def build_meta_document_from_collection( + collection: TestMetaCollection, + *, + cell: Optional[CellMeta] = None, + active_test_id: Optional[int] = None, +) -> dict: + """Build an archive document from a standalone ``TestMetaCollection``.""" + tests_doc = {str(record.test_id): to_dict(record) for record in collection} + cell_doc = to_dict(cell) if cell is not None else to_dict(CellMeta()) + if active_test_id is None: + active_test_id = ( + collection.test_ids[0] if collection.test_ids else 0 + ) + schema = default_schema() + return { + "cellpy_file_version": CELLPY_FILE_VERSION, + "schema_version": Cols.__version__, + "raw_schema_version": schema.raw.__version__, + "step_schema_version": schema.step.__version__, + "cycle_schema_version": schema.cycle.__version__, + "cell": cell_doc, + "tests": tests_doc, + "raw_units": {}, + "cellpy_units": {}, + "limits": {}, + "active_test_id": int(active_test_id), + "loaded_from": None, + } + + +def collection_from_meta_document(meta_doc: Mapping[str, Any]) -> TestMetaCollection: + """Rebuild a ``TestMetaCollection`` from an archive document.""" + collection = TestMetaCollection() + cell_fallback = meta_doc.get("cell") or {} + for payload in (meta_doc.get("tests") or {}).values(): + if not isinstance(payload, dict): + continue + record = from_dict(TestMeta, payload) + # Only fill shared cell when the record omitted ``cell`` entirely — + # explicit ``null`` means no linked CellMeta. + if record.cell is None and cell_fallback and "cell" not in payload: + record.cell = from_dict(CellMeta, cell_fallback) + collection.add(record) + return collection + + +def apply_meta_document(data: "Data", meta_doc: Mapping[str, Any]) -> None: + """Apply an archive / v9 ``meta.json`` document onto a ``Data`` object.""" + version = int(meta_doc.get("cellpy_file_version", CELLPY_FILE_VERSION)) + if version != CELLPY_FILE_VERSION: + raise WrongFileVersion( + f"meta archive expected cellpy_file_version={CELLPY_FILE_VERSION}, " + f"got {version}" + ) + + units_payload = dict(meta_doc.get("raw_units") or {}) + limits_payload = dict(meta_doc.get("limits") or {}) + data.raw_units = CellpyUnits( + **{k: v for k, v in units_payload.items() if k in CellpyUnits()} + ) + limits = CellpyLimits() + for key, value in limits_payload.items(): + if key in limits: + limits[key] = value + data.raw_limits = limits + if meta_doc.get("loaded_from") is not None: + data.loaded_from = meta_doc["loaded_from"] + data.meta_common.cellpy_file_version = version + + active_id = int(meta_doc.get("active_test_id", 0)) + data._active_test_id = active_id + + tests_doc = meta_doc.get("tests") or {} + cell_fallback = meta_doc.get("cell") or {} + + records: dict[int, TestMeta] = {} + tester_assigned: dict[int, Any] = {} + for payload in tests_doc.values(): + if not isinstance(payload, dict): + continue + tid_key = int(payload.get("test_id", 0)) + if "tester_assigned_test_id" in payload: + tester_assigned[tid_key] = payload.get("tester_assigned_test_id") + record = from_dict(TestMeta, payload) + if record.cell is None and cell_fallback and "cell" not in payload: + record.cell = from_dict(CellMeta, cell_fallback) + records[int(record.test_id)] = record + + if active_id not in records and records: + active_id = next(iter(records)) + data._active_test_id = active_id + + active_record = records.get(active_id) + data._extra_tests = {tid: rec for tid, rec in records.items() if tid != active_id} + + if active_record is not None: + test_meta_helpers.apply_test_meta_to_legacy( + active_record, data.meta_common, data.meta_test_dependent + ) + if active_id in tester_assigned: + data.meta_test_dependent.test_ID = tester_assigned[active_id] + provenance = {} + for key in test_meta_helpers._CORE_ONLY_TEST_FIELDS - {"cell", "comment"}: + value = getattr(active_record, key, None) + if value is not None: + provenance[key] = value + data._provenance = provenance + + +def save_meta_archive( + source: MetaArchiveSource, + path: PathLike, + *, + cellpy_units: Optional[Mapping[str, Any]] = None, + cell: Optional[CellMeta] = None, + active_test_id: Optional[int] = None, +) -> None: + """Save a metadata archive JSON file (cellpy-owned; core stubs stay stubs). + + Args: + source: A ``Data`` object, ``TestMetaCollection``, or single ``TestMeta``. + path: Destination path (typically ``*.meta.json``). + cellpy_units: Optional units mapping when ``source`` is ``Data``. + cell: Optional shared ``CellMeta`` when saving a collection alone. + active_test_id: Active test when saving a collection alone. + """ + if isinstance(source, TestMeta): + collection = TestMetaCollection() + collection.add(source) + doc = build_meta_document_from_collection( + collection, cell=cell, active_test_id=active_test_id + ) + elif isinstance(source, TestMetaCollection): + doc = build_meta_document_from_collection( + source, cell=cell, active_test_id=active_test_id + ) + else: + doc = build_meta_document(source, cellpy_units=cellpy_units) + + path = Path(path) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(doc, indent=2, default=_json_default), encoding="utf-8") + _module_logger.debug("wrote meta archive %s", path) + + +def load_meta_archive(path: PathLike) -> dict: + """Load a metadata archive JSON document. + + Returns: + The archive document (same shape as v9 ``meta.json``). + """ + path = Path(path) + doc = json.loads(path.read_text(encoding="utf-8")) + version = int(doc.get("cellpy_file_version", 0)) + if version != CELLPY_FILE_VERSION: + raise WrongFileVersion( + f"Unsupported meta archive version {version} in {path}" + ) + return doc diff --git a/cellpy/readers/cellpy_file/v9.py b/cellpy/readers/cellpy_file/v9.py index b8b01e42..03746b51 100644 --- a/cellpy/readers/cellpy_file/v9.py +++ b/cellpy/readers/cellpy_file/v9.py @@ -4,30 +4,25 @@ runtime still speaks legacy names, so ``save`` translates legacy → native before writing and ``load`` translates native → legacy after reading (I/O boundary adapter; full native runtime is #511). + +Metadata document shape is owned by ``meta_archive`` (cellpy policy; core +``save_archive`` / ``load_archive`` stubs stay stubs). """ from __future__ import annotations -import enum import io import json import logging import zipfile -from datetime import date, datetime from pathlib import Path from typing import TYPE_CHECKING, Any, Mapping, Optional, Union -from cellpycore.config import Cols, default_schema -from cellpycore.metadata.io import from_dict, to_dict -from cellpycore.metadata.models import CellMeta, TestMeta -from cellpycore.units import CellpyUnits - from cellpy.exceptions import CorruptCellpyFile, WrongFileVersion -from cellpy.parameters.internal_settings import CellpyLimits from cellpy.readers import data_structures as ds from cellpy.readers import externals -from cellpy.readers import test_meta as test_meta_helpers from cellpy.readers.cellpy_file import fids as cellpy_file_fids +from cellpy.readers.cellpy_file import meta_archive from cellpy.readers.cellpy_file import translate as cellpy_file_translate from cellpy.readers.cellpy_file.format import ( CELLPY_FILE_VERSION, @@ -47,6 +42,7 @@ _module_logger = logging.getLogger(__name__) PathLike = Union[str, Path] +_TEST_ID = "test_id" def is_zip_cellpy(path: PathLike) -> bool: @@ -58,60 +54,44 @@ def is_zip_cellpy(path: PathLike) -> bool: return False -def _json_default(obj: Any) -> Any: - if isinstance(obj, enum.Enum): - return obj.value - if isinstance(obj, (datetime, date)): - return obj.isoformat() - if hasattr(obj, "item"): # numpy scalar - return obj.item() - raise TypeError(f"Object of type {type(obj)!r} is not JSON serializable") - - -def build_meta_document( - data: "Data", - *, - cellpy_units: Optional[Mapping[str, Any]] = None, -) -> dict: - """Build the v9 ``meta.json`` document from a ``Data`` object.""" - tests_doc: dict[str, dict] = {} - cell_doc: Optional[dict] = None - # Compact ``test_id`` (grouping key) is not the tester-assigned ``test_ID``. - # Preserve the latter so apply_test_meta_to_legacy does not clobber it. - active_tester_id = test_meta_helpers._unwrap( - getattr(data.meta_test_dependent, "test_ID", None) - ) - for record in data.tests: - payload = to_dict(record) - if cell_doc is None and isinstance(payload.get("cell"), dict): - cell_doc = payload["cell"] - if int(record.test_id) == int(data.active_test_id): - payload["tester_assigned_test_id"] = active_tester_id - tests_doc[str(record.test_id)] = payload - - if cell_doc is None: - cell_doc = to_dict(CellMeta()) - - schema = default_schema() +def _frames_had_test_id(data: "Data") -> dict[str, bool]: + """Record which frames already carried ``test_id`` before native inject.""" return { - "cellpy_file_version": CELLPY_FILE_VERSION, - "schema_version": Cols.__version__, - "raw_schema_version": schema.raw.__version__, - "step_schema_version": schema.step.__version__, - "cycle_schema_version": schema.cycle.__version__, - "cell": cell_doc, - "tests": tests_doc, - "raw_units": dict(data.raw_units), - "cellpy_units": dict(cellpy_units) if cellpy_units else {}, - "limits": dict(data.raw_limits), - "active_test_id": int(data.active_test_id), - "loaded_from": getattr(data, "loaded_from", None), + "raw": bool( + getattr(data, "raw", None) is not None + and len(data.raw.columns) + and _TEST_ID in data.raw.columns + ), + "steps": bool( + getattr(data, "steps", None) is not None + and len(data.steps.columns) + and _TEST_ID in data.steps.columns + ), + "summary": bool( + getattr(data, "summary", None) is not None + and len(data.summary.columns) + and _TEST_ID in data.summary.columns + ), } +def _strip_injected_test_id(data: "Data", had: Mapping[str, bool]) -> None: + """Drop ``test_id`` from steps/summary when it was only injected at save.""" + if not had.get("steps") and getattr(data, "steps", None) is not None: + if _TEST_ID in data.steps.columns: + data.steps = data.steps.drop(columns=[_TEST_ID]) + if not had.get("summary") and getattr(data, "summary", None) is not None: + if _TEST_ID in data.summary.columns: + data.summary = data.summary.drop(columns=[_TEST_ID]) + + +def _normalize_frame_nulls(frame): + """Align parquet nulls with pandas float-NaN convention used by HDF5 loads.""" + return frame.replace({None: externals.numpy.nan}) + + def _frame_to_parquet_bytes(frame) -> bytes: buf = io.BytesIO() - # Reset index so keys live in columns (native / polars convention). to_write = frame if getattr(frame, "index", None) is not None and frame.index.name is not None: name = frame.index.name @@ -123,12 +103,6 @@ def _frame_to_parquet_bytes(frame) -> bytes: return buf.getvalue() -def _normalize_frame_nulls(frame): - """Align parquet nulls with pandas float-NaN convention used by HDF5 loads.""" - # pyarrow writes pandas NA/NaN in object columns as Python None on read. - return frame.replace({None: externals.numpy.nan}) - - def _read_parquet_member(zf: zipfile.ZipFile, name: str): try: raw = zf.read(name) @@ -150,7 +124,8 @@ def save( ``data`` object is **not** mutated (work is done on copies). """ path = Path(path) - # Work on a shallow copy of frames so translate.to_native does not mutate caller. + had_test_id = _frames_had_test_id(data) + scratch = ds.Data() scratch.raw = data.raw.copy() if data.raw is not None else externals.pandas.DataFrame() scratch.steps = ( @@ -174,8 +149,11 @@ def save( cellpy_file_translate.to_native(scratch) - meta_doc = build_meta_document(data, cellpy_units=cellpy_units) - # Prefer native frames' test_id story from the scratch object for consistency. + meta_doc = meta_archive.build_meta_document( + data, + cellpy_units=cellpy_units, + frames_had_test_id=had_test_id, + ) meta_doc["active_test_id"] = int(data.active_test_id) fid_table = cellpy_file_fids.convert2fid_table(data) @@ -185,7 +163,7 @@ def save( with zipfile.ZipFile(path, mode="w", compression=zipfile.ZIP_DEFLATED) as zf: zf.writestr( META_JSON_NAME, - json.dumps(meta_doc, indent=2, default=_json_default), + json.dumps(meta_doc, indent=2, default=meta_archive._json_default), ) zf.writestr(V9_RAW_PARQUET, _frame_to_parquet_bytes(scratch.raw)) zf.writestr(V9_STEPS_PARQUET, _frame_to_parquet_bytes(scratch.steps)) @@ -196,71 +174,9 @@ def save( _module_logger.debug("wrote v9 cellpy-file %s", path) -def _apply_meta_document(data: "Data", meta_doc: dict) -> None: - version = int(meta_doc.get("cellpy_file_version", CELLPY_FILE_VERSION)) - if version != CELLPY_FILE_VERSION: - raise WrongFileVersion( - f"v9 reader expected cellpy_file_version={CELLPY_FILE_VERSION}, got {version}" - ) - - units_payload = dict(meta_doc.get("raw_units") or {}) - limits_payload = dict(meta_doc.get("limits") or {}) - data.raw_units = CellpyUnits(**{k: v for k, v in units_payload.items() if k in CellpyUnits()}) - # Overlay onto defaults so missing keys keep CellpyLimits defaults. - limits = CellpyLimits() - for key, value in limits_payload.items(): - if key in limits: - limits[key] = value - data.raw_limits = limits - if meta_doc.get("loaded_from") is not None: - data.loaded_from = meta_doc["loaded_from"] - data.meta_common.cellpy_file_version = version - - active_id = int(meta_doc.get("active_test_id", 0)) - data._active_test_id = active_id - - tests_doc = meta_doc.get("tests") or {} - cell_fallback = meta_doc.get("cell") or {} - - records: dict[int, TestMeta] = {} - tester_assigned: dict[int, Any] = {} - for payload in tests_doc.values(): - if not isinstance(payload, dict): - continue - tid_key = int(payload.get("test_id", 0)) - if "tester_assigned_test_id" in payload: - tester_assigned[tid_key] = payload.get("tester_assigned_test_id") - record = from_dict(TestMeta, payload) - if record.cell is None and cell_fallback: - record.cell = from_dict(CellMeta, cell_fallback) - records[int(record.test_id)] = record - - if active_id not in records and records: - active_id = next(iter(records)) - data._active_test_id = active_id - - active_record = records.get(active_id) - data._extra_tests = {tid: rec for tid, rec in records.items() if tid != active_id} - - if active_record is not None: - test_meta_helpers.apply_test_meta_to_legacy( - active_record, data.meta_common, data.meta_test_dependent - ) - # apply maps TestMeta.test_id → legacy test_ID; restore tester-assigned. - if active_id in tester_assigned: - data.meta_test_dependent.test_ID = tester_assigned[active_id] - # Core-only fields have no legacy home — keep them in _provenance. - provenance = {} - for key in test_meta_helpers._CORE_ONLY_TEST_FIELDS - {"cell", "comment"}: - value = getattr(active_record, key, None) - if value is not None: - provenance[key] = value - data._provenance = provenance - - def load(filename: PathLike, *, selector=None) -> LoadResult: """Load a v9 ``.cellpy`` zip into a legacy-named ``Data`` object.""" - del selector # max_cycle selection not implemented for v9 in Milestone A + del selector # max_cycle selection not implemented for v9 yet path = Path(filename) if not path.is_file(): raise IOError(f"File does not exist: {filename}") @@ -293,9 +209,15 @@ def load(filename: PathLike, *, selector=None) -> LoadResult: data.raw_data_files = [] data.raw_data_files_length = [] - _apply_meta_document(data, meta_doc) - # Runtime still expects legacy headers. - cellpy_file_translate.to_legacy(data, injected_test_id=True) + meta_archive.apply_meta_document(data, meta_doc) + # Keep real campaign test_id columns; strip only injected ones. + cellpy_file_translate.to_legacy(data, injected_test_id=False) + had = meta_doc.get("frames_had_test_id") or { + "raw": True, + "steps": False, + "summary": False, + } + _strip_injected_test_id(data, had) data.loaded_from = str(path) limits = LoadLimits() @@ -312,3 +234,8 @@ def get_version_from_zip(path: PathLike) -> int: def suggest_extension() -> str: """Default filename extension for v9 (without the leading dot).""" return V9_EXTENSION.lstrip(".") + + +# Re-export for callers that imported these from v9 during Milestone A. +build_meta_document = meta_archive.build_meta_document +apply_meta_document = meta_archive.apply_meta_document diff --git a/tests/test_merge_campaign.py b/tests/test_merge_campaign.py index e1f4dcaf..35dba52e 100644 --- a/tests/test_merge_campaign.py +++ b/tests/test_merge_campaign.py @@ -237,6 +237,7 @@ def test_single_test_steps_have_no_test_id_column(parameters): def test_campaign_save_warns_and_reloads_single(campaign_cell, tmp_path, caplog): + """v8 HDF5 escape still drops non-active TestMeta (legacy limitation).""" campaign_cell.make_step_table() campaign_cell.make_summary(find_ir=False, find_end_voltage=False) outfile = tmp_path / "campaign.h5" @@ -249,6 +250,35 @@ def test_campaign_save_warns_and_reloads_single(campaign_cell, tmp_path, caplog) assert reloaded.data.tests.test_ids == [0] +@pytest.mark.essential +def test_campaign_v9_roundtrip_preserves_tests_and_test_id(campaign_cell, tmp_path): + """Milestone B: v9 save/load keeps both TestMeta rows and frame test_id.""" + campaign_cell.make_step_table() + campaign_cell.make_summary(find_ir=False, find_end_voltage=False) + before_meta = { + tid: core_meta_io.to_dict(campaign_cell.data.tests.get(tid)) + for tid in campaign_cell.data.tests.test_ids + } + before_raw_ids = set(campaign_cell.data.raw[HN.test_id_txt].unique()) + before_step_ids = set(campaign_cell.data.steps[HN.test_id_txt].unique()) + + outfile = tmp_path / "campaign.cellpy" + campaign_cell.save(outfile) + + reloaded = cellreader.CellpyCell() + reloaded.load(outfile) + + assert sorted(reloaded.data.tests.test_ids) == [0, 1] + after_meta = { + tid: core_meta_io.to_dict(reloaded.data.tests.get(tid)) + for tid in reloaded.data.tests.test_ids + } + assert after_meta == before_meta + assert set(reloaded.data.raw[HN.test_id_txt].unique()) == before_raw_ids + assert HN.test_id_txt in reloaded.data.steps.columns + assert set(reloaded.data.steps[HN.test_id_txt].unique()) == before_step_ids + + # ------------------------------------------------------- steps/summary concat ---- diff --git a/tests/test_meta_archive.py b/tests/test_meta_archive.py new file mode 100644 index 00000000..dd125819 --- /dev/null +++ b/tests/test_meta_archive.py @@ -0,0 +1,66 @@ +"""Cellpy-owned metadata archive helpers (issue #510 Milestone B / V2-14).""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +from cellpycore.metadata import io as core_meta_io +from cellpycore.metadata.models import TestMeta, TestMetaCollection + +from cellpy.readers.cellpy_file import meta_archive +from tests.cellpy_file_support import load_cellpy_file + +HDF5_DIR = Path(__file__).resolve().parents[1] / "testdata" / "hdf5" +V8_WITH_FIDS = HDF5_DIR / "20160805_test001_45_cc_v8_with_fids.h5" + + +def _require_v8() -> Path: + if not V8_WITH_FIDS.is_file(): + pytest.skip(f"missing fixture: {V8_WITH_FIDS}") + return V8_WITH_FIDS + + +@pytest.mark.essential +def test_save_load_meta_archive_roundtrip_data(tmp_path): + cell = load_cellpy_file(_require_v8()) + cell.data.set_test_meta(TestMeta(test_id=1, cycle_mode="full", cell_name="extra")) + before = { + tid: core_meta_io.to_dict(cell.data.tests.get(tid)) + for tid in cell.data.tests.test_ids + } + + path = tmp_path / "campaign.meta.json" + meta_archive.save_meta_archive(cell.data, path) + doc = meta_archive.load_meta_archive(path) + collection = meta_archive.collection_from_meta_document(doc) + + assert sorted(collection.test_ids) == [0, 1] + assert collection.get(1).cycle_mode == "full" + assert { + tid: core_meta_io.to_dict(collection.get(tid)) for tid in collection.test_ids + } == before + + +@pytest.mark.essential +def test_save_load_meta_archive_from_collection(tmp_path): + collection = TestMetaCollection() + collection.add(TestMeta(test_id=0, cycle_mode="anode", cell_name="a")) + collection.add(TestMeta(test_id=1, cycle_mode="cathode", cell_name="b")) + + path = tmp_path / "tests.meta.json" + meta_archive.save_meta_archive(collection, path, active_test_id=0) + loaded = meta_archive.collection_from_meta_document( + meta_archive.load_meta_archive(path) + ) + assert loaded.test_ids == [0, 1] + assert loaded.get(1).cycle_mode == "cathode" + + +@pytest.mark.essential +def test_core_archive_stubs_still_raise(): + """Boundary: persistence stays in cellpy; core stubs remain stubs.""" + with pytest.raises(NotImplementedError): + core_meta_io.save_archive(TestMeta(test_id=0), "unused.h5") + with pytest.raises(NotImplementedError): + core_meta_io.load_archive("unused.h5")