diff --git a/HISTORY.md b/HISTORY.md index 09b030d0..5b733e03 100644 --- a/HISTORY.md +++ b/HISTORY.md @@ -2,6 +2,19 @@ ## [Unreleased] +* Campaign merge (#507, v2 Phase 1-2, themes V2-03/V2-07): `CellpyCell.merge` + is rewritten (old signature was dead code) — `merge(cells, mode="campaign")` + folds different tests into one multi-test object: distinct compact `test_id` + per source stamped on raw (overwrites tester-assigned ids; provenance stays + in `meta_test_dependent.test_ID`), per-test metadata records in `Data.tests`, + globally renumbered cycles, offset data points, unshifted timelines, and no + cumulative carry-forward (summaries window per test on recompute). + `mode="continuation"` keeps the classic fold (identical to + `from_raw([f1, f2])`, which is untouched). New non-mutating `merge_cells()` + helper and `test_meta.cycle_ranges_per_test()`. Step tables of campaign + objects carry a `test_id` column so the engine groups and windows per test + end-to-end. Also fixes latent bugs in `_append`'s `merge_step_table` branch. + * Per-test metadata API (#506, v2 Phase 1, themes V2-01/02/04): `Data.tests` exposes a `cellpycore.metadata.TestMetaCollection` keyed by `test_id` (active record derived from the legacy meta boxes, which stay authoritative; diff --git a/cellpy/readers/cellreader.py b/cellpy/readers/cellreader.py index b7f08cdc..68716db3 100644 --- a/cellpy/readers/cellreader.py +++ b/cellpy/readers/cellreader.py @@ -1638,18 +1638,83 @@ def _convert2fid_list(tbl): # -------------------- cellpy file handling end ---------------------- - def merge(self, datasets: list, **kwargs): - """This function merges datasets into one set.""" - logging.info("Merging") - self.data = datasets.pop(0) - for data in datasets: - self.data = self._append(self.data, data, **kwargs) - for raw_data_file, file_size in zip( - data.raw_data_files, - data.raw_data_files_length, - ): - self.data.raw_data_files.append(raw_data_file) - self.data.raw_data_files_length.append(file_size) + def merge(self, cells, mode="campaign", renumber_cycles=True, **kwargs): + """Merge other cells/datasets into this one. + + Two distinct semantics (issue #507, epic #402 V2-03/V2-07): + + - ``mode="campaign"`` (default): the sources are *different tests* + (possibly different cells or programs). Each source keeps its + identity via a distinct compact ``test_id`` stamped on the raw + frame (this **overwrites** tester-assigned ids such as Arbin's + ``Test_ID`` — those remain as provenance in + ``meta_test_dependent.test_ID``), and its metadata becomes a + record in ``self.data.tests``. Cycle numbers are renumbered to be + globally unique; data points are offset; ``test_time`` / + ``date_time`` are *not* shifted (independent timelines). Sources + are never mutated. Mixing different ``cycle_mode`` values is + allowed here, but computing steps/summary on the merged object + then raises ``MixedCycleModesError``. + - ``mode="continuation"``: the sources are the *same physical test* + resumed across files — the classic fold, numerically identical to + ``from_raw([file1, file2])`` (cycles, data points and test time + are renumbered into one continuous run; source metadata is + dropped). Extra ``**kwargs`` (e.g. ``recalc``) are forwarded. + + Args: + cells: a CellpyCell or Data instance, or a sequence of them. + mode (str): "campaign" (default) or "continuation". + renumber_cycles (bool): campaign mode only. v1 supports only + ``True`` (the legacy summary format cannot represent + duplicate cycle numbers); ``False`` raises + ``NotImplementedError`` (tracked for the native-schema path). + **kwargs: forwarded to the continuation fold. + + Returns: + self (chainable), with ``self.data`` holding the merged object. + """ + from cellpy.readers import merger + + if cells is None: + raise TypeError( + "merge() requires the cells/datasets to merge into this one " + "(a CellpyCell/Data or a sequence of them)" + ) + if isinstance(cells, (CellpyCell, ds.Data)): + cells = [cells] + datas = [c.data if isinstance(c, CellpyCell) else c for c in cells] + + if mode == "continuation": + logging.info("Merging (continuation)") + for other in datas: + self.data = self._append(self.data, other, **kwargs) + for raw_data_file, file_size in zip( + other.raw_data_files, + other.raw_data_files_length, + ): + self.data.raw_data_files.append(raw_data_file) + self.data.raw_data_files_length.append(file_size) + return self + + if mode != "campaign": + raise ValueError(f"unknown merge mode: {mode!r}") + if not renumber_cycles: + raise NotImplementedError( + "campaign merge with original (overlapping) cycle numbers " + "requires per-test summary support through the core bridge; " + "tracked for the native-schema path (#511)" + ) + + logging.info("Merging (campaign)") + for other in datas: + merger.campaign_fold(self.data, other, **kwargs) + modes = test_meta.cycle_modes_in_data(self.data) + if len(modes) > 1: + logging.warning( + f"merged object stores tests with different cycle_modes " + f"{sorted(modes)}; computing steps/summary will raise " + f"MixedCycleModesError until per-test engine polarity lands" + ) return self def _append(self, t1, t2, merge_summary=False, merge_step_table=False, recalc=True): @@ -1718,7 +1783,6 @@ def _append(self, t1, t2, merge_summary=False, merge_step_table=False, recalc=Tr raw = externals.pandas.concat([t1.raw, t2.raw], ignore_index=True) data.raw = raw data.loaded_from.append(t2.loaded_from) - step_table_made = False if merge_summary: # checking if we already have made a summary file of these datasets @@ -1739,12 +1803,6 @@ def _append(self, t1, t2, merge_summary=False, merge_step_table=False, recalc=Tr summary_made = False logging.info("The summary is not complete - run make_summary()") - # checking if we already have made step tables for these datasets - if t1.has_steps and t2.has_steps: - step_table_made = True - else: - step_table_made = False - if summary_made: # check if (self-made) summary exists. logging.debug("merge summaries") @@ -1777,12 +1835,14 @@ def _append(self, t1, t2, merge_summary=False, merge_step_table=False, recalc=Tr ) if merge_step_table: - if step_table_made: - cycle_index_header = self.headers_normal.cycle_index_txt - t2.steps[self.headers_step_table.cycle] = ( - t2.raw[self.headers_step_table.cycle] + last_cycle - ) - + # (fixed in #507: was gated on a flag only set inside the + # merge_summary branch, offset the wrong frame, and NameError'd + # on last_cycle when recalc=False) + if t1.has_steps and t2.has_steps: + if recalc: + t2.steps[self.headers_step_table.cycle] = ( + t2.steps[self.headers_step_table.cycle] + last_cycle + ) steps2 = externals.pandas.concat( [t1.steps, t2.steps], ignore_index=True ) @@ -2172,6 +2232,24 @@ def make_step_table( if from_data_point is not None: return result + + # Campaign objects only (issue #507): re-stamp test_id onto the step + # table. The core bridge groups steps by (test_id, cycle, step) using + # the raw test_id column but strips the column from the returned + # legacy frame; putting it back makes the subsequent summary use + # per-test windowing (use_tid) instead of cycle-only grouping. Gated + # on _extra_tests so single-test and continuation objects (including + # multi-tester-id Arbin raw) are untouched. + if self.data._extra_tests and not self.data.steps.empty: + from cellpy.readers import merger + + test_id_hdr = self.headers_normal.test_id_txt + if test_id_hdr in self.data.raw.columns: + point_first = f"{self.headers_step_table.point}_first" + self.data.steps[test_id_hdr] = merger._steps_test_id_from_raw( + self.data, self.data.steps, point_first + ) + return self def select_steps(self, step_dict, append_df=False): @@ -5233,6 +5311,35 @@ def _dev_update_from_raw(self, file_names=None, data_points=None, **kwargs): return self +def merge_cells(cells, mode="campaign", **kwargs) -> "CellpyCell": + """Merge several cells into a new CellpyCell without mutating any of them. + + Convenience wrapper around :meth:`CellpyCell.merge` (issue #507): the + first cell is deep-copied and the rest are folded in. See the method + docstring for the "campaign" vs "continuation" semantics. + + Args: + cells: sequence of CellpyCell (or Data) instances; order matters. + mode (str): "campaign" (default) or "continuation". + **kwargs: forwarded to :meth:`CellpyCell.merge`. + + Returns: + A new CellpyCell holding the merged object. + """ + cells = list(cells) + if not cells: + raise ValueError("merge_cells() needs at least one cell") + first, rest = cells[0], cells[1:] + if isinstance(first, CellpyCell): + merged = copy.deepcopy(first) + else: + merged = CellpyCell(initialize=True) + merged.data = copy.deepcopy(first) + if rest: + merged.merge(rest, mode=mode, **kwargs) + return merged + + def get( filename=None, instrument=None, diff --git a/cellpy/readers/merger.py b/cellpy/readers/merger.py new file mode 100644 index 00000000..69d935ae --- /dev/null +++ b/cellpy/readers/merger.py @@ -0,0 +1,206 @@ +"""Campaign merge: fold several tests into one multi-test ``Data`` object. + +Implements the v2 "campaign" merge (issue #507, epic #402 themes V2-03/V2-07): +each source keeps its identity via a distinct compact ``test_id`` stamped on the +raw frame, and its metadata becomes a record in the per-test collection +(``Data.tests``, issue #506). Cycle numbers are renumbered to be globally +unique (v1 — the legacy summary format cannot represent duplicate cycle +numbers; keeping original per-test numbering is deferred to the native-schema +path). ``test_time`` / ``date_time`` are *not* shifted — campaign sources have +independent timelines (mirrors ``cellpycore.merge.merge_data``). + +This intentionally mirrors the semantics of ``cellpycore.merge.merge_data`` + +``cellpycore.metadata.io.merge_test_meta``; those cannot be called directly at +the legacy seam (they read native schema attribute names that the legacy +header objects do not have), and the metadata helper does not return the +``test_id`` remap needed to keep frames and records consistent. + +Note the compact-key policy: the raw ``test_id`` column is overwritten with +compact keys (0, 1, 2, ...) at merge time — tester-assigned ids (e.g. Arbin's +``Test_ID``) remain available as provenance in +``meta_test_dependent.test_ID`` / ``TestMeta.cell``-level metadata. +""" + +import dataclasses +import logging +from typing import Dict + +from cellpy.parameters.internal_settings import ( + get_headers_normal, + get_headers_step_table, + get_headers_summary, +) + +from . import externals as externals + +logger = logging.getLogger(__name__) + + +def _next_free(used) -> int: + candidate = 0 + while candidate in used: + candidate += 1 + return candidate + + +def normalize_test_id_column(data) -> None: + """Stamp the compact ``test_id`` key onto ``data.raw`` (in place). + + For a non-campaign object (no extra test records) the column is created or + *overwritten* with ``data.active_test_id`` — tester-assigned ids that some + loaders (e.g. arbin_res) put in this column are provenance, not the + grouping key. Campaign objects already carry compact ids; left untouched. + """ + hdr = get_headers_normal().test_id_txt + if data._extra_tests: + if hdr not in data.raw.columns: + raise ValueError( + "campaign object without a test_id column in raw - cannot merge" + ) + return + data.raw[hdr] = data.active_test_id + + +def fold_test_metadata(left, right) -> Dict[int, int]: + """Fold ``right``'s per-test records into ``left`` and return the id map. + + Mirrors ``cellpycore.metadata.io.merge_test_meta`` priority semantics + (earlier records keep their ids; colliding later ids get the next free + one), but returns the ``{old_id: new_id}`` map so the raw frames can be + remapped consistently. ``right`` is never mutated (records are copied via + ``dataclasses.replace``). + """ + used = set(left.tests.test_ids) + id_map: Dict[int, int] = {} + for record in right.tests: + target = record.test_id if record.test_id not in used else _next_free(used) + id_map[record.test_id] = target + left.set_test_meta(dataclasses.replace(record, test_id=target)) + used.add(target) + return id_map + + +def campaign_fold(left, right, *, merge_steps=True, merge_summary=True) -> None: + """Fold ``right`` (a ``Data`` object) into ``left`` as a distinct test. + + Mutates ``left`` in place; never mutates ``right``. See the module + docstring for the semantics (compact test_id stamping, global cycle + renumbering, no time shifting, per-test metadata records). + """ + hn = get_headers_normal() + hs = get_headers_summary() + hst = get_headers_step_table() + test_id_hdr = hn.test_id_txt + + if right.raw.empty: + logger.warning("campaign merge: skipping a source with empty raw data") + return + if left.raw.empty: + raise ValueError("campaign merge: the target object has empty raw data") + + if dict(left.raw_units) != dict(right.raw_units): + raise ValueError( + f"campaign merge: raw_units differ between sources " + f"({left.raw_units} vs {right.raw_units}); convert first" + ) + if dict(left.raw_limits) != dict(right.raw_limits): + logger.warning( + "campaign merge: raw_limits differ between sources; keeping the " + "target's limits" + ) + + normalize_test_id_column(left) + id_map = fold_test_metadata(left, right) + + right_raw = right.raw.copy() + if right._extra_tests: + if test_id_hdr not in right_raw.columns: + raise ValueError( + "campaign source without a test_id column in raw - cannot merge" + ) + right_raw[test_id_hdr] = right_raw[test_id_hdr].map( + lambda v: id_map[int(v)] + ) + else: + right_raw[test_id_hdr] = id_map[right.active_test_id] + + dp_hdr = hn.data_point_txt + cyc_raw_hdr = hn.cycle_index_txt + dp_offset = int(left.raw[dp_hdr].max()) + cycle_offset = int(left.raw[cyc_raw_hdr].max()) + right_raw[dp_hdr] = right_raw[dp_hdr] + dp_offset + right_raw[cyc_raw_hdr] = right_raw[cyc_raw_hdr] + cycle_offset + + left.raw = externals.pandas.concat([left.raw, right_raw], ignore_index=True) + + # steps: offset + stamp test_id when both sides carry a step table, + # else clear so the user re-runs make_step_table on the merged object. + if merge_steps and not left.steps.empty and not right.steps.empty: + point_first = f"{hst.point}_first" + point_last = f"{hst.point}_last" + left_steps = left.steps + if test_id_hdr not in left_steps.columns: + left_steps = left_steps.copy() + left_steps[test_id_hdr] = _steps_test_id_from_raw( + left, left_steps, point_first + ) + right_steps = right.steps.copy() + right_steps[hst.cycle] = right_steps[hst.cycle] + cycle_offset + for col in (point_first, point_last): + if col in right_steps.columns: + right_steps[col] = right_steps[col] + dp_offset + right_steps[test_id_hdr] = _steps_test_id_from_raw( + left, right_steps, point_first + ) + left.steps = externals.pandas.concat( + [left_steps, right_steps], ignore_index=True + ) + else: + if merge_steps and not (left.steps.empty and right.steps.empty): + logger.info( + "campaign merge: step tables incomplete - run make_step_table()" + ) + left.steps = externals.pandas.DataFrame() + + # summary: offset cycle/data_point; deliberately NO cumulative + # carry-forward - per-test resets are the correct campaign semantics. + summary_ok = ( + merge_summary + and not left.summary.empty + and not right.summary.empty + and hs.cycle_index in left.summary.columns + and hs.cycle_index in right.summary.columns # arbin stats-frame guard + ) + if summary_ok: + right_summary = right.summary.copy() + right_summary[hs.cycle_index] = right_summary[hs.cycle_index] + cycle_offset + if hs.data_point in right_summary.columns: + right_summary[hs.data_point] = right_summary[hs.data_point] + dp_offset + left.summary = externals.pandas.concat( + [left.summary, right_summary], ignore_index=True + ) + else: + if merge_summary and not (left.summary.empty and right.summary.empty): + logger.info( + "campaign merge: summaries incomplete - run make_summary()" + ) + left.summary = externals.pandas.DataFrame() + + # provenance bookkeeping + left.raw_data_files.extend(right.raw_data_files) + left.raw_data_files_length.extend(right.raw_data_files_length) + if not isinstance(left.loaded_from, list): + left.loaded_from = [left.loaded_from] + left.loaded_from.append(right.loaded_from) + + +def _steps_test_id_from_raw(data, steps, point_first_col): + """Look up each step row's ``test_id`` from the (merged) raw frame via the + step's first data point.""" + hn = get_headers_normal() + mapping = ( + data.raw[[hn.data_point_txt, hn.test_id_txt]] + .drop_duplicates(subset=hn.data_point_txt) + .set_index(hn.data_point_txt)[hn.test_id_txt] + ) + return steps[point_first_col].map(mapping) diff --git a/cellpy/readers/test_meta.py b/cellpy/readers/test_meta.py index ce02b267..27291c13 100644 --- a/cellpy/readers/test_meta.py +++ b/cellpy/readers/test_meta.py @@ -151,3 +151,26 @@ def cycle_modes_in_data(data) -> Set[str]: if mode is not None: modes.add(mode) return modes + + +def cycle_ranges_per_test(data) -> dict: + """Per-test cycle ranges ``{test_id: (cycle_min, cycle_max)}``. + + Derived from the raw frame's ``test_id`` column (campaign-merged objects, + issue #507), so it survives save/load of raw without extra state. A raw + frame without the column reports the active test spanning all cycles. + """ + from cellpy.parameters.internal_settings import get_headers_normal + + hn = get_headers_normal() + raw = data.raw + if raw.empty: + return {} + if hn.test_id_txt not in raw.columns: + cycles = raw[hn.cycle_index_txt] + return {data.active_test_id: (int(cycles.min()), int(cycles.max()))} + grouped = raw.groupby(hn.test_id_txt)[hn.cycle_index_txt].agg(["min", "max"]) + return { + int(test_id): (int(row["min"]), int(row["max"])) + for test_id, row in grouped.iterrows() + } diff --git a/tests/test_cell_readers.py b/tests/test_cell_readers.py index 9f3722c1..5c7f0d63 100644 --- a/tests/test_cell_readers.py +++ b/tests/test_cell_readers.py @@ -11,7 +11,7 @@ import cellpy.readers.data_structures from cellpy import log, prms from cellpy.exceptions import DeprecatedFeature, WrongFileVersion -from cellpy.parameters.internal_settings import get_headers_summary +from cellpy.parameters.internal_settings import get_headers_normal, get_headers_summary from cellpy.internals.connections import OtherPath log.setup_logging(default_level="DEBUG", testing=True) @@ -114,31 +114,30 @@ def test_cellpy_version_5(cellpy_data_instance, parameters): # print(f"cellpy version: {v}") -def test_merge(cellpy_data_instance, parameters): - # TODO @jepe: refactor and use col names directly from HeadersNormal instead +def test_merge_continuation_matches_from_raw_list(parameters): + """merge(mode='continuation') is numerically identical to from_raw([f1, f2]).""" + from cellpy import cellreader + f1 = parameters.res_file_path f2 = parameters.res_file_path2 assert os.path.isfile(f1) assert os.path.isfile(f2) - cellpy_data_instance.from_raw(f1) - cellpy_data_instance.from_raw(f2) - assert len(cellpy_data_instance.datasets) == 2 + via_list = cellreader.CellpyCell() + via_list.from_raw([f1, f2]) - table_first = cellpy_data_instance.data.raw.describe() - count_first = table_first.loc["count", "data_point"] + left = cellreader.CellpyCell() + left.from_raw(f1) + right = cellreader.CellpyCell() + right.from_raw(f2) + left.merge(right, mode="continuation") - table_second = cellpy_data_instance.datasets[1].raw.describe() - count_second = table_second.loc["count", "data_point"] - - cellpy_data_instance.merge() - assert len(cellpy_data_instance.datasets) == 2 - - table_all = cellpy_data_instance.datasets[0].raw.describe() - count_all = table_all.loc["count", "data_point"] - assert len(cellpy_data_instance.datasets) == 1 - - assert pytest.approx(count_all, 0.001) == (count_first + count_second) + hn = get_headers_normal() + for col in (hn.data_point_txt, hn.cycle_index_txt, hn.test_time_txt): + assert left.data.raw[col].reset_index(drop=True).equals( + via_list.data.raw[col].reset_index(drop=True) + ), f"continuation merge differs from from_raw list on {col}" + assert len(left.data.raw) == len(via_list.data.raw) def test_merge_auto_from_list(parameters): @@ -361,11 +360,6 @@ def test_set_logger(dataset): print("missing test") -def test_merge(dataset): - print("missing test") - print("maybe deprecated") - - def test_fid(cellpy_data_instance, parameters): cellpy_data_instance.from_raw(parameters.res_file_path) my_test = cellpy_data_instance.data diff --git a/tests/test_merge_campaign.py b/tests/test_merge_campaign.py new file mode 100644 index 00000000..e1f4dcaf --- /dev/null +++ b/tests/test_merge_campaign.py @@ -0,0 +1,282 @@ +"""Tests for the campaign merge (issue #507, V2-03/V2-07). + +Campaign merge folds several *different tests* into one multi-test object: +distinct compact ``test_id`` per source in raw, per-test metadata records +(``Data.tests``), globally renumbered cycles, and per-test summary windowing +via the re-stamped step-table ``test_id``. +""" + +from __future__ import annotations + +import logging + +import pandas as pd +import pytest + +from cellpycore.metadata import io as core_meta_io +from cellpycore.metadata.models import TestMeta, TestMetaCollection + +from cellpy import cellreader +from cellpy.exceptions import MixedCycleModesError +from cellpy.parameters.internal_settings import ( + get_headers_normal, + get_headers_summary, +) +from cellpy.readers import merger, test_meta +from cellpy.readers.cellreader import merge_cells +from cellpy.readers.data_structures import Data + +log_setup_done = False + +HN = get_headers_normal() +HS = get_headers_summary() + + +def _mini_data(cycles=(1, 2), dp_start=1, mass=1.0, cycle_mode="anode", name="a"): + """Small synthetic Data with a raw frame (2 rows per cycle).""" + data = Data() + rows = [] + dp = dp_start + for cyc in cycles: + for k in range(2): + rows.append( + { + HN.data_point_txt: dp, + HN.cycle_index_txt: cyc, + HN.test_time_txt: float(dp), + HN.current_txt: 1.0 if k == 0 else -1.0, + HN.voltage_txt: 3.5, + } + ) + dp += 1 + data.raw = pd.DataFrame(rows) + data.meta_common.mass = mass + data.meta_common.cell_name = name + data.meta_test_dependent.cycle_mode = cycle_mode + data.loaded_from = f"{name}.raw" + return data + + +def _cell(data) -> cellreader.CellpyCell: + c = cellreader.CellpyCell(initialize=True) + c.data = data + return c + + +# ------------------------------------------------------------- fold basics ---- + + +def test_campaign_two_sources_ids_and_offsets(): + left = _cell(_mini_data(cycles=(1, 2, 3), mass=1.0, name="one")) + right = _cell(_mini_data(cycles=(1, 2), mass=2.0, name="two")) + n_left, n_right = len(left.data.raw), len(right.data.raw) + + left.merge(right) + + raw = left.data.raw + assert len(raw) == n_left + n_right + assert set(raw[HN.test_id_txt].unique()) == {0, 1} + assert left.data.tests.test_ids == [0, 1] + # cycles globally unique, contiguous blocks + assert raw[HN.cycle_index_txt].max() == 5 + assert test_meta.cycle_ranges_per_test(left.data) == {0: (1, 3), 1: (4, 5)} + # datapoints strictly increasing across the boundary + dp = raw[HN.data_point_txt] + assert dp.iloc[n_left] > dp.iloc[:n_left].max() + + +def test_campaign_metadata_fold_and_sources_unmutated(): + left = _cell(_mini_data(mass=1.0, cycle_mode="anode", name="one")) + right_data = _mini_data(mass=2.0, cycle_mode="anode", name="two") + right_raw_before = right_data.raw.copy() + + left.merge(right_data) + + rec0, rec1 = left.data.tests.get(0), left.data.tests.get(1) + assert rec0.cell.mass == 1.0 and rec0.cell_name == "one" + assert rec1.cell.mass == 2.0 and rec1.cell_name == "two" + # active legacy boxes untouched by the fold + assert left.data.meta_common.mass == 1.0 + # source not mutated + pd.testing.assert_frame_equal(right_data.raw, right_raw_before) + assert right_data._extra_tests == {} + + +def test_campaign_three_way_and_remerge_id_remap(): + a = _cell(_mini_data(name="a")) + a.merge([_mini_data(name="b"), _mini_data(name="c")]) + assert a.data.tests.test_ids == [0, 1, 2] + + # merging an already-merged object remaps its colliding ids + d = _cell(_mini_data(name="d")) + d.merge(a) + assert d.data.tests.test_ids == [0, 1, 2, 3] + names = {tid: d.data.tests.get(tid).cell_name for tid in d.data.tests.test_ids} + assert names[0] == "d" and set(names.values()) == {"a", "b", "c", "d"} + + +def test_campaign_metadata_renumbering_matches_core_merge_test_meta(): + """Our id assignment mirrors cellpycore's merge_test_meta priority rule.""" + left = _cell(_mini_data(name="one")) + right = _mini_data(name="two") + left.merge(right) + ours = {tid: left.data.tests.get(tid).cell_name for tid in left.data.tests.test_ids} + + c_left = TestMetaCollection() + c_left.add(TestMeta(test_id=0, cell_name="one")) + c_right = TestMetaCollection() + c_right.add(TestMeta(test_id=0, cell_name="two")) + core_merged = core_meta_io.merge_test_meta(c_left, c_right, renumber=True) + core_names = {tid: core_merged.get(tid).cell_name for tid in core_merged.test_ids} + assert ours == core_names + + +def test_campaign_guards(): + left = _cell(_mini_data(name="one")) + empty = Data() + n = len(left.data.raw) + left.merge(empty) # empty source skipped with warning + assert len(left.data.raw) == n + assert left.data.tests.test_ids == [0] + + other = _mini_data(name="two") + other.raw_units["current"] = "mA" + with pytest.raises(ValueError, match="raw_units differ"): + left.merge(other) + + with pytest.raises(NotImplementedError, match="original"): + left.merge(_mini_data(name="three"), renumber_cycles=False) + + with pytest.raises(ValueError, match="unknown merge mode"): + left.merge(_mini_data(name="four"), mode="bogus") + + with pytest.raises(TypeError, match="requires the cells"): + left.merge(None) + + +def test_merge_cells_convenience_does_not_mutate(): + c1 = _cell(_mini_data(name="one")) + c2 = _cell(_mini_data(name="two")) + n1 = len(c1.data.raw) + + merged = merge_cells([c1, c2]) + + assert merged is not c1 + assert merged.data.tests.test_ids == [0, 1] + assert len(c1.data.raw) == n1 + assert c1.data._extra_tests == {} + + +# ---------------------------------------------------------------- mixed modes ---- + + +def test_campaign_mixed_modes_stored_but_compute_raises(caplog): + left = _cell(_mini_data(cycle_mode="anode", name="one")) + with caplog.at_level(logging.WARNING): + left.merge(_mini_data(cycle_mode="cathode", name="two")) + assert any("different cycle_modes" in r.message for r in caplog.records) + assert test_meta.cycle_modes_in_data(left.data) == {"anode", "cathode"} + + with pytest.raises(MixedCycleModesError): + left.make_step_table() + with pytest.raises(MixedCycleModesError): + left.make_summary() + + +# ------------------------------------------------- real data: compute pins ---- + + +@pytest.fixture +def campaign_cell(parameters): + """Two real res files campaign-merged as distinct tests (same cycle_mode).""" + left = cellreader.CellpyCell() + left.from_raw(parameters.res_file_path) + right = cellreader.CellpyCell() + right.from_raw(parameters.res_file_path2) + left.merge(right) + return left + + +def test_campaign_real_data_structure(campaign_cell): + raw = campaign_cell.data.raw + assert set(raw[HN.test_id_txt].unique()) == {0, 1} + ranges = test_meta.cycle_ranges_per_test(campaign_cell.data) + assert set(ranges) == {0, 1} + assert ranges[1][0] > ranges[0][1] # no cycle overlap + + +def test_campaign_recompute_stamps_steps_and_windows_summary(campaign_cell): + """The bridge tripwire: steps carry test_id; cumulatives reset per test.""" + campaign_cell.make_step_table() + steps = campaign_cell.data.steps + assert HN.test_id_txt in steps.columns + assert set(steps[HN.test_id_txt].unique()) == {0, 1} + + campaign_cell.make_summary(find_ir=False, find_end_voltage=False) + summary = campaign_cell.data.summary + ranges = test_meta.cycle_ranges_per_test(campaign_cell.data) + first_cycle_t1 = ranges[1][0] + cum_col = "cumulated_charge_capacity" + assert cum_col in summary.columns + block1 = summary[summary[HS.cycle_index] >= first_cycle_t1] + first_row = block1.iloc[0] + # per-test windowing: the cumulation restarts at the test boundary, so the + # first cumulated value of test 1 equals its own first charge capacity + # (not the carried total of test 0). + assert first_row[cum_col] == pytest.approx( + first_row["charge_capacity"], rel=1e-6 + ) + + +def test_single_test_steps_have_no_test_id_column(parameters): + """Safety pin: the re-stamp gate stays off for non-campaign objects.""" + c = cellreader.CellpyCell() + c.from_raw(parameters.res_file_path) + c.make_step_table() + assert HN.test_id_txt not in c.data.steps.columns + + +def test_campaign_save_warns_and_reloads_single(campaign_cell, tmp_path, caplog): + campaign_cell.make_step_table() + campaign_cell.make_summary(find_ir=False, find_end_voltage=False) + outfile = tmp_path / "campaign.h5" + with caplog.at_level(logging.WARNING): + campaign_cell.save(outfile) + assert any("not persist" in r.message or "persists" in r.message for r in caplog.records) + + reloaded = cellreader.CellpyCell() + reloaded.load(outfile) + assert reloaded.data.tests.test_ids == [0] + + +# ------------------------------------------------------- steps/summary concat ---- + + +def test_campaign_merges_precomputed_steps_and_summary(parameters): + left = cellreader.CellpyCell() + left.from_raw(parameters.res_file_path) + left.make_step_table() + left.make_summary(find_ir=False, find_end_voltage=False) + right = cellreader.CellpyCell() + right.from_raw(parameters.res_file_path2) + right.make_step_table() + right.make_summary(find_ir=False, find_end_voltage=False) + + n_steps = len(left.data.steps) + len(right.data.steps) + n_summary = len(left.data.summary) + len(right.data.summary) + right_first_cum = right.data.summary["cumulated_charge_capacity"].iloc[0] + + left.merge(right) + + assert len(left.data.steps) == n_steps + assert HN.test_id_txt in left.data.steps.columns + assert set(left.data.steps[HN.test_id_txt].unique()) == {0, 1} + assert len(left.data.summary) == n_summary + # no cumulative carry-forward: test 1's first cumulated value unchanged + ranges = test_meta.cycle_ranges_per_test(left.data) + block1 = left.data.summary[ + left.data.summary[HS.cycle_index] >= ranges[1][0] + ] + assert block1["cumulated_charge_capacity"].iloc[0] == pytest.approx( + right_first_cum, rel=1e-9 + )