From a616088e7ec8522ebf108ae360372ef6cd9c83d6 Mon Sep 17 00:00:00 2001 From: Daniel Weindl Date: Wed, 15 Oct 2025 14:33:21 +0200 Subject: [PATCH 1/7] Implement `flatten_timepoint_specific_output_overrides` for PEtab v2 Port `flatten_timepoint_specific_output_overrides`, `has_timepoint_specific_overrides`, `unflatten_simulation_df` and their respective tests to PEtab v2. --- petab/v1/core.py | 7 +- petab/v2/core.py | 260 ++++++++++++++++++++++++++++++++++- tests/v2/test_core.py | 312 ++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 573 insertions(+), 6 deletions(-) diff --git a/petab/v1/core.py b/petab/v1/core.py index 1149c67e..6a142781 100644 --- a/petab/v1/core.py +++ b/petab/v1/core.py @@ -133,7 +133,9 @@ def get_notnull_columns(df: pd.DataFrame, candidates: Iterable): ] -def get_observable_replacement_id(groupvars, groupvar) -> str: +def get_observable_replacement_id( + groupvars: list[str], groupvar: Sequence +) -> str: """Get the replacement ID for an observable. Arguments: @@ -141,7 +143,8 @@ def get_observable_replacement_id(groupvars, groupvar) -> str: The columns of a PEtab measurement table that should be unique between observables in a flattened PEtab problem. groupvar: - A specific grouping of `groupvars`. + A specific grouping of `groupvars`. Same length and order as + `groupvars`. Returns: The observable replacement ID. diff --git a/petab/v2/core.py b/petab/v2/core.py index 0e3f905e..3f28f684 100644 --- a/petab/v2/core.py +++ b/petab/v2/core.py @@ -69,8 +69,20 @@ "Parameter", "ParameterScale", "ParameterTable", + "flatten_timepoint_specific_output_overrides", + "unflatten_simulation_df", ] +_POSSIBLE_GROUPVARS_FLATTENED_PROBLEM = [ + C.MODEL_ID, + C.EXPERIMENT_ID, + C.OBSERVABLE_ID, + C.OBSERVABLE_PARAMETERS, + C.NOISE_PARAMETERS, +] + +logger = logging.getLogger(__name__) + def _is_finite_or_neg_inf(v: float, info: ValidationInfo) -> float: if not np.isfinite(v) and v != -np.inf: @@ -1143,7 +1155,11 @@ def __str__(self): f"{observables}, {measurements}, {parameters}" ) - def __getitem__(self, key): + def __getitem__( + self, key + ) -> ( + Condition | Experiment | Observable | Measurement | Parameter | Mapping + ): """Get PEtab entity by ID. This allows accessing PEtab entities such as conditions, experiments, @@ -2320,7 +2336,9 @@ def get_output_parameters( # filter out symbols that are defined in the model or mapped to # such symbols for candidate in sorted(candidates): - if self.model.symbol_allowed_in_observable_formula(candidate): + if self.model and self.model.symbol_allowed_in_observable_formula( + candidate + ): continue # does it map to a model entity? @@ -2329,8 +2347,11 @@ def get_output_parameters( mapping.petab_id == candidate and mapping.model_id is not None ): - if self.model.symbol_allowed_in_observable_formula( - mapping.model_id + if ( + self.model + and self.model.symbol_allowed_in_observable_formula( + mapping.model_id + ) ): break else: @@ -2339,6 +2360,71 @@ def get_output_parameters( return output_parameters + def has_timepoint_specific_overrides( + self, + ignore_scalar_numeric_noise_parameters: bool = False, + ignore_scalar_numeric_observable_parameters: bool = False, + ) -> bool: + """Check if the measurements have timepoint-specific observable or + noise parameter overrides. + + :param ignore_scalar_numeric_noise_parameters: + ignore scalar numeric assignments to noiseParameter placeholders + + :param ignore_scalar_numeric_observable_parameters: + ignore scalar numeric assignments to observableParameter + placeholders + + :return: True if the problem has timepoint-specific overrides, False + otherwise. + """ + if not self.measurements: + return False + + from ..v1.core import get_notnull_columns + from ..v1.lint import is_scalar_float + + measurement_df = self.measurement_df + + # mask numeric values + for col, allow_scalar_numeric in [ + ( + C.OBSERVABLE_PARAMETERS, + ignore_scalar_numeric_observable_parameters, + ), + (C.NOISE_PARAMETERS, ignore_scalar_numeric_noise_parameters), + ]: + if col not in measurement_df: + continue + + measurement_df[col] = measurement_df[col].apply(str) + + if allow_scalar_numeric: + measurement_df.loc[ + measurement_df[col].apply(is_scalar_float), col + ] = "" + + grouping_cols = get_notnull_columns( + measurement_df, + _POSSIBLE_GROUPVARS_FLATTENED_PROBLEM, + ) + grouped_df = measurement_df.groupby(grouping_cols, dropna=False) + + grouping_cols = get_notnull_columns( + measurement_df, + [ + C.MODEL_ID, + C.OBSERVABLE_ID, + C.EXPERIMENT_ID, + ], + ) + grouped_df2 = measurement_df.groupby(grouping_cols) + + # data frame has timepoint specific overrides if grouping by noise + # parameters and observable parameters in addition to observable and + # experiment id yields more groups + return len(grouped_df) != len(grouped_df2) + class ModelFile(BaseModel): """A file in the PEtab problem configuration.""" @@ -2457,3 +2543,169 @@ def format_version_tuple(self) -> tuple[int, int, int, str]: """The format version as a tuple of major/minor/patch `int`s and a suffix.""" return parse_version(self.format_version) + + +def _get_flattened_id_mappings( + petab_problem: Problem, +) -> dict[str, str]: + """Get mapping from flattened to unflattenedobservable IDs. + + :param petab_problem: + The unflattened PEtab problem. + + :returns: + A mapping from original observable ID to flattened ID. + """ + from ..v1.core import ( + get_notnull_columns, + get_observable_replacement_id, + ) + + groupvars = get_notnull_columns( + petab_problem.measurement_df, _POSSIBLE_GROUPVARS_FLATTENED_PROBLEM + ) + mappings: dict[str, str] = {} + + old_observable_ids = {obs.id for obs in petab_problem.observables} + for groupvar, _ in petab_problem.measurement_df.groupby( + groupvars, dropna=False + ): + observable_id = groupvar[groupvars.index(C.OBSERVABLE_ID)] + observable_replacement_id = get_observable_replacement_id( + groupvars, groupvar + ) + + logger.debug(f"Creating synthetic observable {observable_id}") + if ( + observable_id != observable_replacement_id + and observable_replacement_id in old_observable_ids + ): + raise RuntimeError( + "could not create synthetic observables " + f"since {observable_replacement_id} was " + "already present in observable table" + ) + + mappings[observable_replacement_id] = observable_id + + return mappings + + +def flatten_timepoint_specific_output_overrides( + petab_problem: Problem, +) -> None: + """Flatten timepoint-specific output parameter overrides. + + If the PEtab problem definition has timepoint-specific + `observableParameters` or `noiseParameters` for the same observable, + replace those by replicating the respective observable. + + This is a helper function for some tools which may not support such + timepoint-specific mappings. The observable table and measurement table + are modified in place. + + :param petab_problem: + PEtab problem to work on. Modified in place. + """ + from ..v1.core import ( + get_notnull_columns, + get_observable_replacement_id, + ) + + # Update observables + def create_new_observable(old_id, new_id) -> Observable: + if old_id not in petab_problem.observable_df.index: + raise ValueError( + f"Observable {old_id} not found in observable table." + ) + + # copy original observable and update ID + observable: Observable = copy.deepcopy(petab_problem[old_id]) + observable.id = new_id + + # update placeholders + old_obs_placeholders = observable.observable_placeholders or [] + old_noise_placeholders = observable.noise_placeholders or [] + suffix = new_id.removeprefix(old_id) + observable.observable_placeholders = [ + f"{sym.name}{suffix}" for sym in observable.observable_placeholders + ] + observable.noise_placeholders = [ + f"{sym.name}{suffix}" for sym in observable.noise_placeholders + ] + + # placeholders in formulas + subs = dict( + zip( + old_obs_placeholders, + observable.observable_placeholders, + strict=False, + ) + ) + observable.formula = observable.formula.subs(subs) + subs |= dict( + zip( + old_noise_placeholders, + observable.noise_placeholders, + strict=False, + ) + ) + observable.noise_formula = observable.noise_formula.subs(subs) + + return observable + + mappings = _get_flattened_id_mappings(petab_problem) + + petab_problem.observable_tables = [ + ObservableTable( + [ + create_new_observable(old_id, new_id) + for new_id, old_id in mappings.items() + ] + ) + ] + + # Update measurements + groupvars = get_notnull_columns( + petab_problem.measurement_df, _POSSIBLE_GROUPVARS_FLATTENED_PROBLEM + ) + for measurement_table in petab_problem.measurement_tables: + for measurement in measurement_table.measurements: + # TODO: inefficient, but ok for a start + group_vals = ( + MeasurementTable([measurement]) + .to_df() + .iloc[0][groupvars] + .tolist() + ) + new_obs_id = get_observable_replacement_id(groupvars, group_vals) + measurement.observable_id = new_obs_id + + +def unflatten_simulation_df( + simulation_df: pd.DataFrame, + petab_problem: petab.problem.Problem, +) -> pd.DataFrame: + """Unflatten simulations from a flattened PEtab problem. + + A flattened PEtab problem is the output of applying + :func:`flatten_timepoint_specific_output_overrides` to a PEtab problem. + + :param simulation_df: + The simulation dataframe. A dataframe in the same format as a PEtab + measurements table, but with the ``measurement`` column switched + with a ``simulation`` column. + :param petab_problem: + The unflattened PEtab problem. + + :returns: + The simulation dataframe for the unflattened PEtab problem. + """ + mappings = _get_flattened_id_mappings(petab_problem) + original_observable_ids = simulation_df[C.OBSERVABLE_ID].replace(mappings) + unflattened_simulation_df = simulation_df.assign( + **{ + C.OBSERVABLE_ID: original_observable_ids, + } + ) + return unflattened_simulation_df diff --git a/tests/v2/test_core.py b/tests/v2/test_core.py index 64e1ad41..7f2e943b 100644 --- a/tests/v2/test_core.py +++ b/tests/v2/test_core.py @@ -1,3 +1,4 @@ +import copy import subprocess import tempfile from pathlib import Path @@ -836,3 +837,314 @@ def test_get_output_parameters(): assert petab_problem.get_output_parameters( observable=False, noise=True ) == ["p1", "p3", "p5"] + + +def test_problem_has_timepoint_specific_overrides(): + """Test Problem.measurement_table_has_timepoint_specific_mappings""" + problem = Problem() + problem.add_measurement( + obs_id="obs1", + time=1.0, + measurement=0.1, + observable_parameters=["obsParOverride"], + ) + problem.add_measurement(obs_id="obs1", time=1.0, measurement=0.2) + assert problem.has_timepoint_specific_overrides() is True + + # both measurements different anyways + problem.measurement_tables[0].measurements[1].observable_id = "obs2" + assert problem.has_timepoint_specific_overrides() is False + + # mixed numeric string + problem.measurement_tables[0].measurements[1].observable_id = "obs1" + problem.measurement_tables[0].measurements[1].observable_parameters = [ + "obsParOverride" + ] + assert problem.has_timepoint_specific_overrides() is False + + # different numeric values + problem.measurement_tables[0].measurements[1].noise_parameters = [2.0] + assert problem.has_timepoint_specific_overrides() is True + assert ( + problem.has_timepoint_specific_overrides( + ignore_scalar_numeric_noise_parameters=True + ) + is False + ) + + +def test_flatten_timepoint_specific_output_overrides(): + """Test flatten_timepoint_specific_output_overrides""" + problem = Problem() + problem.model = SbmlModel.from_antimony("""x = 1""") + problem.add_observable( + "obs1", + formula="observableParameter1_obs1 + observableParameter2_obs1", + noise_formula=( + "(observableParameter1_obs1 + " + "observableParameter2_obs1) * noiseParameter1_obs1" + ), + observable_placeholders=[ + "observableParameter1_obs1", + "observableParameter2_obs1", + ], + noise_placeholders=["noiseParameter1_obs1"], + ) + problem.add_observable("obs2", formula="x", noise_formula="1") + + # new observable IDs + # (obs${i_obs}_${i_obsParOverride}_${i_noiseParOverride}) + obs1_1_1_1 = "obs1__obsParOverride1_1_00000000000000__noiseParOverride1" + obs1_2_1_1 = "obs1__obsParOverride2_1_00000000000000__noiseParOverride1" + obs1_2_2_1 = "obs1__obsParOverride2_1_00000000000000__noiseParOverride2" + problem_expected = Problem() + problem_expected.model = SbmlModel.from_antimony("""x = 1""") + + problem_expected.add_observable( + obs1_1_1_1, + formula=( + f"observableParameter1_{obs1_1_1_1} " + f"+ observableParameter2_{obs1_1_1_1}" + ), + noise_formula=( + f"(observableParameter1_{obs1_1_1_1} + " + f"observableParameter2_{obs1_1_1_1}) " + f"* noiseParameter1_{obs1_1_1_1}" + ), + observable_placeholders=[ + f"observableParameter1_{obs1_1_1_1}", + f"observableParameter2_{obs1_1_1_1}", + ], + noise_placeholders=[f"noiseParameter1_{obs1_1_1_1}"], + ) + problem_expected.add_observable( + obs1_2_1_1, + formula=( + f"observableParameter1_{obs1_2_1_1} " + f"+ observableParameter2_{obs1_2_1_1}" + ), + noise_formula=( + f"(observableParameter1_{obs1_2_1_1} " + f"+ observableParameter2_{obs1_2_1_1}) * " + f"noiseParameter1_{obs1_2_1_1}" + ), + observable_placeholders=[ + f"observableParameter1_{obs1_2_1_1}", + f"observableParameter2_{obs1_2_1_1}", + ], + noise_placeholders=[f"noiseParameter1_{obs1_2_1_1}"], + ) + problem_expected.add_observable( + obs1_2_2_1, + formula=( + f"observableParameter1_{obs1_2_2_1} " + f"+ observableParameter2_{obs1_2_2_1}" + ), + noise_formula=( + f"(observableParameter1_{obs1_2_2_1} " + f"+ observableParameter2_{obs1_2_2_1}) " + f"* noiseParameter1_{obs1_2_2_1}" + ), + observable_placeholders=[ + f"observableParameter1_{obs1_2_2_1}", + f"observableParameter2_{obs1_2_2_1}", + ], + noise_placeholders=[f"noiseParameter1_{obs1_2_2_1}"], + ) + problem_expected.add_observable( + "obs2", + formula="x", + noise_formula="1", + ) + + # Measurement table with timepoint-specific overrides + problem.add_measurement( + obs_id="obs1", + time=1.0, + measurement=0.1, + observable_parameters=["obsParOverride1", "1.0"], + noise_parameters=["noiseParOverride1"], + ) + problem.add_measurement( + obs_id="obs1", + time=1.0, + measurement=0.1, + observable_parameters=["obsParOverride2", "1.0"], + noise_parameters=["noiseParOverride1"], + ) + problem.add_measurement( + obs_id="obs1", + time=2.0, + measurement=0.1, + observable_parameters=["obsParOverride2", "1.0"], + noise_parameters=["noiseParOverride2"], + ) + problem.add_measurement( + obs_id="obs1", + time=2.0, + measurement=0.1, + observable_parameters=["obsParOverride2", "1.0"], + noise_parameters=["noiseParOverride2"], + ) + problem.add_measurement(obs_id="obs2", time=3.0, measurement=0.1) + + problem_expected.add_measurement( + obs_id=obs1_1_1_1, + time=1.0, + measurement=0.1, + observable_parameters=["obsParOverride1", "1.0"], + noise_parameters=["noiseParOverride1"], + ) + problem_expected.add_measurement( + obs_id=obs1_2_1_1, + time=1.0, + measurement=0.1, + observable_parameters=["obsParOverride2", "1.0"], + noise_parameters=["noiseParOverride1"], + ) + problem_expected.add_measurement( + obs_id=obs1_2_2_1, + time=2.0, + measurement=0.1, + observable_parameters=["obsParOverride2", "1.0"], + noise_parameters=["noiseParOverride2"], + ) + problem_expected.add_measurement( + obs_id=obs1_2_2_1, + time=2.0, + measurement=0.1, + observable_parameters=["obsParOverride2", "1.0"], + noise_parameters=["noiseParOverride2"], + ) + problem_expected.add_measurement(obs_id="obs2", time=3.0, measurement=0.1) + + for p in (problem, problem_expected): + p.add_parameter("noiseParOverride1", estimate=False, nominal_value=1) + p.add_parameter("noiseParOverride2", estimate=False, nominal_value=1) + p.add_parameter("obsParOverride1", estimate=False, nominal_value=1) + p.add_parameter("obsParOverride2", estimate=False, nominal_value=1) + + problem.assert_valid() + unflattened_problem = copy.deepcopy(problem) + problem_expected.assert_valid() + + # Ensure having timepoint-specific overrides + assert problem.has_timepoint_specific_overrides() is True + assert problem_expected.has_timepoint_specific_overrides() is False + + flatten_timepoint_specific_output_overrides(problem) + + # Timepoint-specific overrides should be gone now + assert problem.has_timepoint_specific_overrides() is False + + assert problem_expected.observables == problem.observables + assert problem_expected.measurements == problem.measurements + problem.assert_valid() + + simulation_df = copy.deepcopy(problem.measurement_df) + simulation_df.rename(columns={C.MEASUREMENT: C.SIMULATION}) + unflattened_simulation_df = unflatten_simulation_df( + simulation_df=simulation_df, + petab_problem=unflattened_problem, + ) + # The unflattened simulation dataframe has the original observable IDs. + assert ( + unflattened_simulation_df[OBSERVABLE_ID] == ["obs1"] * 4 + ["obs2"] + ).all() + + +def test_flatten_timepoint_specific_output_overrides_special_cases(): + """Test flatten_timepoint_specific_output_overrides + for special cases: + * no observable parameters + """ + problem = Problem() + problem.model = SbmlModel.from_antimony("""species1 = 1""") + for p in ("noiseParOverride2", "noiseParOverride1"): + problem.add_parameter(p, estimate=False, nominal_value=1) + problem_expected = copy.deepcopy(problem) + problem.add_observable( + "obs1", + formula="species1", + noise_formula="noiseParameter1_obs1", + noise_placeholders=["noiseParameter1_obs1"], + ) + + problem_expected.add_observable( + "obs1__noiseParOverride1", + formula="species1", + noise_formula="noiseParameter1_obs1__noiseParOverride1", + noise_placeholders=["noiseParameter1_obs1__noiseParOverride1"], + ) + problem_expected.add_observable( + "obs1__noiseParOverride2", + formula="species1", + noise_formula="noiseParameter1_obs1__noiseParOverride2", + noise_placeholders=["noiseParameter1_obs1__noiseParOverride2"], + ) + + # Measurement table with timepoint-specific overrides + problem.add_measurement( + "obs1", + time=1.0, + measurement=0.1, + noise_parameters=["noiseParOverride1"], + ) + problem.add_measurement( + "obs1", + time=1.0, + measurement=0.1, + noise_parameters=["noiseParOverride1"], + ) + problem.add_measurement( + "obs1", + time=2.0, + measurement=0.1, + noise_parameters=["noiseParOverride2"], + ) + problem.add_measurement( + "obs1", + time=2.0, + measurement=0.1, + noise_parameters=["noiseParOverride2"], + ) + + problem_expected.add_measurement( + "obs1__noiseParOverride1", + time=1.0, + measurement=0.1, + noise_parameters=["noiseParOverride1"], + ) + problem_expected.add_measurement( + "obs1__noiseParOverride1", + time=1.0, + measurement=0.1, + noise_parameters=["noiseParOverride1"], + ) + problem_expected.add_measurement( + "obs1__noiseParOverride2", + time=2.0, + measurement=0.1, + noise_parameters=["noiseParOverride2"], + ) + problem_expected.add_measurement( + "obs1__noiseParOverride2", + time=2.0, + measurement=0.1, + noise_parameters=["noiseParOverride2"], + ) + + problem.assert_valid() + problem_expected.assert_valid() + + # Ensure having timepoint-specific overrides + assert problem.has_timepoint_specific_overrides() is True + + flatten_timepoint_specific_output_overrides(problem) + + # Timepoint-specific overrides should be gone now + assert problem.has_timepoint_specific_overrides() is False + + assert problem_expected.observables == problem.observables + assert problem_expected.measurements == problem.measurements + problem.assert_valid() From 8da8ffad190026879dd884b7d0e85cc29b67a471 Mon Sep 17 00:00:00 2001 From: Daniel Weindl Date: Thu, 16 Oct 2025 13:23:12 +0200 Subject: [PATCH 2/7] Apply suggestions from code review Co-authored-by: Dilan Pathirana <59329744+dilpath@users.noreply.github.com> --- petab/v2/core.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/petab/v2/core.py b/petab/v2/core.py index 3f28f684..b8bb29ce 100644 --- a/petab/v2/core.py +++ b/petab/v2/core.py @@ -2548,13 +2548,13 @@ def format_version_tuple(self) -> tuple[int, int, int, str]: def _get_flattened_id_mappings( petab_problem: Problem, ) -> dict[str, str]: - """Get mapping from flattened to unflattenedobservable IDs. + """Get mapping from flattened to unflattened observable IDs. :param petab_problem: The unflattened PEtab problem. :returns: - A mapping from original observable ID to flattened ID. + A mapping from flattened ID to original observable ID. """ from ..v1.core import ( get_notnull_columns, From 2df6ab3fef59e4377f90cee67108396123dc794f Mon Sep 17 00:00:00 2001 From: Daniel Weindl Date: Thu, 16 Oct 2025 13:26:55 +0200 Subject: [PATCH 3/7] strict --- petab/v2/core.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/petab/v2/core.py b/petab/v2/core.py index b8bb29ce..6ca692c0 100644 --- a/petab/v2/core.py +++ b/petab/v2/core.py @@ -2639,7 +2639,7 @@ def create_new_observable(old_id, new_id) -> Observable: zip( old_obs_placeholders, observable.observable_placeholders, - strict=False, + strict=True, ) ) observable.formula = observable.formula.subs(subs) @@ -2647,7 +2647,7 @@ def create_new_observable(old_id, new_id) -> Observable: zip( old_noise_placeholders, observable.noise_placeholders, - strict=False, + strict=True, ) ) observable.noise_formula = observable.noise_formula.subs(subs) From 2207165ea688423db5c35a3cfc31aeabc5027502 Mon Sep 17 00:00:00 2001 From: Daniel Weindl Date: Thu, 16 Oct 2025 13:49:29 +0200 Subject: [PATCH 4/7] cleanup --- tests/v2/test_core.py | 102 +++++++++++++++--------------------------- 1 file changed, 36 insertions(+), 66 deletions(-) diff --git a/tests/v2/test_core.py b/tests/v2/test_core.py index 7f2e943b..f4cdf480 100644 --- a/tests/v2/test_core.py +++ b/tests/v2/test_core.py @@ -877,6 +877,16 @@ def test_flatten_timepoint_specific_output_overrides(): """Test flatten_timepoint_specific_output_overrides""" problem = Problem() problem.model = SbmlModel.from_antimony("""x = 1""") + for par_id in ( + "noiseParOverride1", + "noiseParOverride2", + "obsParOverride1", + "obsParOverride2", + ): + problem.add_parameter(par_id, estimate=False, nominal_value=1) + + problem_expected = copy.deepcopy(problem) + problem.add_observable( "obs1", formula="observableParameter1_obs1 + observableParameter2_obs1", @@ -894,63 +904,29 @@ def test_flatten_timepoint_specific_output_overrides(): # new observable IDs # (obs${i_obs}_${i_obsParOverride}_${i_noiseParOverride}) - obs1_1_1_1 = "obs1__obsParOverride1_1_00000000000000__noiseParOverride1" - obs1_2_1_1 = "obs1__obsParOverride2_1_00000000000000__noiseParOverride1" - obs1_2_2_1 = "obs1__obsParOverride2_1_00000000000000__noiseParOverride2" - problem_expected = Problem() - problem_expected.model = SbmlModel.from_antimony("""x = 1""") + obs1_1_1 = "obs1__obsParOverride1_1_00000000000000__noiseParOverride1" + obs1_2_1 = "obs1__obsParOverride2_1_00000000000000__noiseParOverride1" + obs1_2_2 = "obs1__obsParOverride2_1_00000000000000__noiseParOverride2" + + for obs_id in (obs1_1_1, obs1_2_1, obs1_2_2): + problem_expected.add_observable( + obs_id, + formula=( + f"observableParameter1_{obs_id} " + f"+ observableParameter2_{obs_id}" + ), + noise_formula=( + f"(observableParameter1_{obs_id} + " + f"observableParameter2_{obs_id}) " + f"* noiseParameter1_{obs_id}" + ), + observable_placeholders=[ + f"observableParameter1_{obs_id}", + f"observableParameter2_{obs_id}", + ], + noise_placeholders=[f"noiseParameter1_{obs_id}"], + ) - problem_expected.add_observable( - obs1_1_1_1, - formula=( - f"observableParameter1_{obs1_1_1_1} " - f"+ observableParameter2_{obs1_1_1_1}" - ), - noise_formula=( - f"(observableParameter1_{obs1_1_1_1} + " - f"observableParameter2_{obs1_1_1_1}) " - f"* noiseParameter1_{obs1_1_1_1}" - ), - observable_placeholders=[ - f"observableParameter1_{obs1_1_1_1}", - f"observableParameter2_{obs1_1_1_1}", - ], - noise_placeholders=[f"noiseParameter1_{obs1_1_1_1}"], - ) - problem_expected.add_observable( - obs1_2_1_1, - formula=( - f"observableParameter1_{obs1_2_1_1} " - f"+ observableParameter2_{obs1_2_1_1}" - ), - noise_formula=( - f"(observableParameter1_{obs1_2_1_1} " - f"+ observableParameter2_{obs1_2_1_1}) * " - f"noiseParameter1_{obs1_2_1_1}" - ), - observable_placeholders=[ - f"observableParameter1_{obs1_2_1_1}", - f"observableParameter2_{obs1_2_1_1}", - ], - noise_placeholders=[f"noiseParameter1_{obs1_2_1_1}"], - ) - problem_expected.add_observable( - obs1_2_2_1, - formula=( - f"observableParameter1_{obs1_2_2_1} " - f"+ observableParameter2_{obs1_2_2_1}" - ), - noise_formula=( - f"(observableParameter1_{obs1_2_2_1} " - f"+ observableParameter2_{obs1_2_2_1}) " - f"* noiseParameter1_{obs1_2_2_1}" - ), - observable_placeholders=[ - f"observableParameter1_{obs1_2_2_1}", - f"observableParameter2_{obs1_2_2_1}", - ], - noise_placeholders=[f"noiseParameter1_{obs1_2_2_1}"], - ) problem_expected.add_observable( "obs2", formula="x", @@ -989,28 +965,28 @@ def test_flatten_timepoint_specific_output_overrides(): problem.add_measurement(obs_id="obs2", time=3.0, measurement=0.1) problem_expected.add_measurement( - obs_id=obs1_1_1_1, + obs_id=obs1_1_1, time=1.0, measurement=0.1, observable_parameters=["obsParOverride1", "1.0"], noise_parameters=["noiseParOverride1"], ) problem_expected.add_measurement( - obs_id=obs1_2_1_1, + obs_id=obs1_2_1, time=1.0, measurement=0.1, observable_parameters=["obsParOverride2", "1.0"], noise_parameters=["noiseParOverride1"], ) problem_expected.add_measurement( - obs_id=obs1_2_2_1, + obs_id=obs1_2_2, time=2.0, measurement=0.1, observable_parameters=["obsParOverride2", "1.0"], noise_parameters=["noiseParOverride2"], ) problem_expected.add_measurement( - obs_id=obs1_2_2_1, + obs_id=obs1_2_2, time=2.0, measurement=0.1, observable_parameters=["obsParOverride2", "1.0"], @@ -1018,12 +994,6 @@ def test_flatten_timepoint_specific_output_overrides(): ) problem_expected.add_measurement(obs_id="obs2", time=3.0, measurement=0.1) - for p in (problem, problem_expected): - p.add_parameter("noiseParOverride1", estimate=False, nominal_value=1) - p.add_parameter("noiseParOverride2", estimate=False, nominal_value=1) - p.add_parameter("obsParOverride1", estimate=False, nominal_value=1) - p.add_parameter("obsParOverride2", estimate=False, nominal_value=1) - problem.assert_valid() unflattened_problem = copy.deepcopy(problem) problem_expected.assert_valid() From 6c4911bea94dd9ec28957c46cf077ae1e6f2f333 Mon Sep 17 00:00:00 2001 From: Daniel Weindl Date: Thu, 16 Oct 2025 14:08:23 +0200 Subject: [PATCH 5/7] .. --- petab/v2/core.py | 28 ++++++++++------------------ 1 file changed, 10 insertions(+), 18 deletions(-) diff --git a/petab/v2/core.py b/petab/v2/core.py index 6ca692c0..1724b31e 100644 --- a/petab/v2/core.py +++ b/petab/v2/core.py @@ -38,8 +38,12 @@ validate_yaml_syntax, yaml, ) +from ..v1.core import ( + get_notnull_columns, + get_observable_replacement_id, +) from ..v1.distributions import * -from ..v1.lint import is_valid_identifier +from ..v1.lint import is_scalar_float, is_valid_identifier from ..v1.math import petab_math_str, sympify_petab from ..v1.models.model import Model, model_factory from ..v1.yaml import get_path_prefix @@ -124,7 +128,7 @@ def _valid_petab_id(v: str) -> str: return v -def _valid_petab_id_or_none(v: str) -> str: +def _valid_petab_id_or_none(v: str) -> str | None: """Field validator for optional PEtab IDs.""" if not v: return None @@ -264,7 +268,7 @@ def __getitem__(self, id_: str) -> T: @classmethod @abstractmethod - def from_df(cls, df: pd.DataFrame) -> BaseTable[T]: + def from_df(cls, df: pd.DataFrame, **kwargs) -> BaseTable[T]: """Create a table from a DataFrame.""" pass @@ -2381,9 +2385,6 @@ def has_timepoint_specific_overrides( if not self.measurements: return False - from ..v1.core import get_notnull_columns - from ..v1.lint import is_scalar_float - measurement_df = self.measurement_df # mask numeric values @@ -2556,11 +2557,6 @@ def _get_flattened_id_mappings( :returns: A mapping from flattened ID to original observable ID. """ - from ..v1.core import ( - get_notnull_columns, - get_observable_replacement_id, - ) - groupvars = get_notnull_columns( petab_problem.measurement_df, _POSSIBLE_GROUPVARS_FLATTENED_PROBLEM ) @@ -2607,10 +2603,6 @@ def flatten_timepoint_specific_output_overrides( :param petab_problem: PEtab problem to work on. Modified in place. """ - from ..v1.core import ( - get_notnull_columns, - get_observable_replacement_id, - ) # Update observables def create_new_observable(old_id, new_id) -> Observable: @@ -2624,8 +2616,8 @@ def create_new_observable(old_id, new_id) -> Observable: observable.id = new_id # update placeholders - old_obs_placeholders = observable.observable_placeholders or [] - old_noise_placeholders = observable.noise_placeholders or [] + old_obs_placeholders = observable.observable_placeholders + old_noise_placeholders = observable.noise_placeholders suffix = new_id.removeprefix(old_id) observable.observable_placeholders = [ f"{sym.name}{suffix}" for sym in observable.observable_placeholders @@ -2684,7 +2676,7 @@ def create_new_observable(old_id, new_id) -> Observable: def unflatten_simulation_df( simulation_df: pd.DataFrame, - petab_problem: petab.problem.Problem, + petab_problem: Problem, ) -> pd.DataFrame: """Unflatten simulations from a flattened PEtab problem. From 4481491d63e82d2dfbd059867b735593d9876c8e Mon Sep 17 00:00:00 2001 From: Daniel Weindl Date: Thu, 16 Oct 2025 14:34:55 +0200 Subject: [PATCH 6/7] .. --- petab/v1/measurements.py | 2 -- tests/v2/test_core.py | 7 ++++++- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/petab/v1/measurements.py b/petab/v1/measurements.py index 8b23907b..f23a21c1 100644 --- a/petab/v1/measurements.py +++ b/petab/v1/measurements.py @@ -307,7 +307,6 @@ def assert_overrides_match_parameter_count( row.get(OBSERVABLE_PARAMETERS, None) ) ) - # No overrides are also allowed if actual != expected: formula = observable_df.loc[row[OBSERVABLE_ID], OBSERVABLE_FORMULA] raise AssertionError( @@ -324,7 +323,6 @@ def assert_overrides_match_parameter_count( try: expected = noise_parameters_count[row[OBSERVABLE_ID]] - # No overrides are also allowed if len(replacements) != expected: raise AssertionError( f"Mismatch of noise parameter overrides in:\n{row}\n" diff --git a/tests/v2/test_core.py b/tests/v2/test_core.py index f4cdf480..1268349f 100644 --- a/tests/v2/test_core.py +++ b/tests/v2/test_core.py @@ -848,7 +848,12 @@ def test_problem_has_timepoint_specific_overrides(): measurement=0.1, observable_parameters=["obsParOverride"], ) - problem.add_measurement(obs_id="obs1", time=1.0, measurement=0.2) + problem.add_measurement( + obs_id="obs1", + time=1.0, + measurement=0.2, + observable_parameters=["obsParOverride2"], + ) assert problem.has_timepoint_specific_overrides() is True # both measurements different anyways From 8fc20e335644e2b6b0a8e37a2b6ef54267308323 Mon Sep 17 00:00:00 2001 From: Daniel Weindl Date: Thu, 16 Oct 2025 19:23:32 +0200 Subject: [PATCH 7/7] move to amici --- petab/v2/core.py | 237 +--------------------------------- tests/v2/test_core.py | 287 ------------------------------------------ 2 files changed, 2 insertions(+), 522 deletions(-) diff --git a/petab/v2/core.py b/petab/v2/core.py index 1724b31e..4ced4b9c 100644 --- a/petab/v2/core.py +++ b/petab/v2/core.py @@ -38,12 +38,8 @@ validate_yaml_syntax, yaml, ) -from ..v1.core import ( - get_notnull_columns, - get_observable_replacement_id, -) from ..v1.distributions import * -from ..v1.lint import is_scalar_float, is_valid_identifier +from ..v1.lint import is_valid_identifier from ..v1.math import petab_math_str, sympify_petab from ..v1.models.model import Model, model_factory from ..v1.yaml import get_path_prefix @@ -73,16 +69,6 @@ "Parameter", "ParameterScale", "ParameterTable", - "flatten_timepoint_specific_output_overrides", - "unflatten_simulation_df", -] - -_POSSIBLE_GROUPVARS_FLATTENED_PROBLEM = [ - C.MODEL_ID, - C.EXPERIMENT_ID, - C.OBSERVABLE_ID, - C.OBSERVABLE_PARAMETERS, - C.NOISE_PARAMETERS, ] logger = logging.getLogger(__name__) @@ -1222,7 +1208,7 @@ def from_yaml( from .petab1to2 import petab1to2 if format_version[0] == 1 and yaml_file: - logging.debug( + logger.debug( "Auto-upgrading problem from PEtab 1.0 to PEtab 2.0" ) with TemporaryDirectory() as tmpdirname: @@ -2364,68 +2350,6 @@ def get_output_parameters( return output_parameters - def has_timepoint_specific_overrides( - self, - ignore_scalar_numeric_noise_parameters: bool = False, - ignore_scalar_numeric_observable_parameters: bool = False, - ) -> bool: - """Check if the measurements have timepoint-specific observable or - noise parameter overrides. - - :param ignore_scalar_numeric_noise_parameters: - ignore scalar numeric assignments to noiseParameter placeholders - - :param ignore_scalar_numeric_observable_parameters: - ignore scalar numeric assignments to observableParameter - placeholders - - :return: True if the problem has timepoint-specific overrides, False - otherwise. - """ - if not self.measurements: - return False - - measurement_df = self.measurement_df - - # mask numeric values - for col, allow_scalar_numeric in [ - ( - C.OBSERVABLE_PARAMETERS, - ignore_scalar_numeric_observable_parameters, - ), - (C.NOISE_PARAMETERS, ignore_scalar_numeric_noise_parameters), - ]: - if col not in measurement_df: - continue - - measurement_df[col] = measurement_df[col].apply(str) - - if allow_scalar_numeric: - measurement_df.loc[ - measurement_df[col].apply(is_scalar_float), col - ] = "" - - grouping_cols = get_notnull_columns( - measurement_df, - _POSSIBLE_GROUPVARS_FLATTENED_PROBLEM, - ) - grouped_df = measurement_df.groupby(grouping_cols, dropna=False) - - grouping_cols = get_notnull_columns( - measurement_df, - [ - C.MODEL_ID, - C.OBSERVABLE_ID, - C.EXPERIMENT_ID, - ], - ) - grouped_df2 = measurement_df.groupby(grouping_cols) - - # data frame has timepoint specific overrides if grouping by noise - # parameters and observable parameters in addition to observable and - # experiment id yields more groups - return len(grouped_df) != len(grouped_df2) - class ModelFile(BaseModel): """A file in the PEtab problem configuration.""" @@ -2544,160 +2468,3 @@ def format_version_tuple(self) -> tuple[int, int, int, str]: """The format version as a tuple of major/minor/patch `int`s and a suffix.""" return parse_version(self.format_version) - - -def _get_flattened_id_mappings( - petab_problem: Problem, -) -> dict[str, str]: - """Get mapping from flattened to unflattened observable IDs. - - :param petab_problem: - The unflattened PEtab problem. - - :returns: - A mapping from flattened ID to original observable ID. - """ - groupvars = get_notnull_columns( - petab_problem.measurement_df, _POSSIBLE_GROUPVARS_FLATTENED_PROBLEM - ) - mappings: dict[str, str] = {} - - old_observable_ids = {obs.id for obs in petab_problem.observables} - for groupvar, _ in petab_problem.measurement_df.groupby( - groupvars, dropna=False - ): - observable_id = groupvar[groupvars.index(C.OBSERVABLE_ID)] - observable_replacement_id = get_observable_replacement_id( - groupvars, groupvar - ) - - logger.debug(f"Creating synthetic observable {observable_id}") - if ( - observable_id != observable_replacement_id - and observable_replacement_id in old_observable_ids - ): - raise RuntimeError( - "could not create synthetic observables " - f"since {observable_replacement_id} was " - "already present in observable table" - ) - - mappings[observable_replacement_id] = observable_id - - return mappings - - -def flatten_timepoint_specific_output_overrides( - petab_problem: Problem, -) -> None: - """Flatten timepoint-specific output parameter overrides. - - If the PEtab problem definition has timepoint-specific - `observableParameters` or `noiseParameters` for the same observable, - replace those by replicating the respective observable. - - This is a helper function for some tools which may not support such - timepoint-specific mappings. The observable table and measurement table - are modified in place. - - :param petab_problem: - PEtab problem to work on. Modified in place. - """ - - # Update observables - def create_new_observable(old_id, new_id) -> Observable: - if old_id not in petab_problem.observable_df.index: - raise ValueError( - f"Observable {old_id} not found in observable table." - ) - - # copy original observable and update ID - observable: Observable = copy.deepcopy(petab_problem[old_id]) - observable.id = new_id - - # update placeholders - old_obs_placeholders = observable.observable_placeholders - old_noise_placeholders = observable.noise_placeholders - suffix = new_id.removeprefix(old_id) - observable.observable_placeholders = [ - f"{sym.name}{suffix}" for sym in observable.observable_placeholders - ] - observable.noise_placeholders = [ - f"{sym.name}{suffix}" for sym in observable.noise_placeholders - ] - - # placeholders in formulas - subs = dict( - zip( - old_obs_placeholders, - observable.observable_placeholders, - strict=True, - ) - ) - observable.formula = observable.formula.subs(subs) - subs |= dict( - zip( - old_noise_placeholders, - observable.noise_placeholders, - strict=True, - ) - ) - observable.noise_formula = observable.noise_formula.subs(subs) - - return observable - - mappings = _get_flattened_id_mappings(petab_problem) - - petab_problem.observable_tables = [ - ObservableTable( - [ - create_new_observable(old_id, new_id) - for new_id, old_id in mappings.items() - ] - ) - ] - - # Update measurements - groupvars = get_notnull_columns( - petab_problem.measurement_df, _POSSIBLE_GROUPVARS_FLATTENED_PROBLEM - ) - for measurement_table in petab_problem.measurement_tables: - for measurement in measurement_table.measurements: - # TODO: inefficient, but ok for a start - group_vals = ( - MeasurementTable([measurement]) - .to_df() - .iloc[0][groupvars] - .tolist() - ) - new_obs_id = get_observable_replacement_id(groupvars, group_vals) - measurement.observable_id = new_obs_id - - -def unflatten_simulation_df( - simulation_df: pd.DataFrame, - petab_problem: Problem, -) -> pd.DataFrame: - """Unflatten simulations from a flattened PEtab problem. - - A flattened PEtab problem is the output of applying - :func:`flatten_timepoint_specific_output_overrides` to a PEtab problem. - - :param simulation_df: - The simulation dataframe. A dataframe in the same format as a PEtab - measurements table, but with the ``measurement`` column switched - with a ``simulation`` column. - :param petab_problem: - The unflattened PEtab problem. - - :returns: - The simulation dataframe for the unflattened PEtab problem. - """ - mappings = _get_flattened_id_mappings(petab_problem) - original_observable_ids = simulation_df[C.OBSERVABLE_ID].replace(mappings) - unflattened_simulation_df = simulation_df.assign( - **{ - C.OBSERVABLE_ID: original_observable_ids, - } - ) - return unflattened_simulation_df diff --git a/tests/v2/test_core.py b/tests/v2/test_core.py index 1268349f..64e1ad41 100644 --- a/tests/v2/test_core.py +++ b/tests/v2/test_core.py @@ -1,4 +1,3 @@ -import copy import subprocess import tempfile from pathlib import Path @@ -837,289 +836,3 @@ def test_get_output_parameters(): assert petab_problem.get_output_parameters( observable=False, noise=True ) == ["p1", "p3", "p5"] - - -def test_problem_has_timepoint_specific_overrides(): - """Test Problem.measurement_table_has_timepoint_specific_mappings""" - problem = Problem() - problem.add_measurement( - obs_id="obs1", - time=1.0, - measurement=0.1, - observable_parameters=["obsParOverride"], - ) - problem.add_measurement( - obs_id="obs1", - time=1.0, - measurement=0.2, - observable_parameters=["obsParOverride2"], - ) - assert problem.has_timepoint_specific_overrides() is True - - # both measurements different anyways - problem.measurement_tables[0].measurements[1].observable_id = "obs2" - assert problem.has_timepoint_specific_overrides() is False - - # mixed numeric string - problem.measurement_tables[0].measurements[1].observable_id = "obs1" - problem.measurement_tables[0].measurements[1].observable_parameters = [ - "obsParOverride" - ] - assert problem.has_timepoint_specific_overrides() is False - - # different numeric values - problem.measurement_tables[0].measurements[1].noise_parameters = [2.0] - assert problem.has_timepoint_specific_overrides() is True - assert ( - problem.has_timepoint_specific_overrides( - ignore_scalar_numeric_noise_parameters=True - ) - is False - ) - - -def test_flatten_timepoint_specific_output_overrides(): - """Test flatten_timepoint_specific_output_overrides""" - problem = Problem() - problem.model = SbmlModel.from_antimony("""x = 1""") - for par_id in ( - "noiseParOverride1", - "noiseParOverride2", - "obsParOverride1", - "obsParOverride2", - ): - problem.add_parameter(par_id, estimate=False, nominal_value=1) - - problem_expected = copy.deepcopy(problem) - - problem.add_observable( - "obs1", - formula="observableParameter1_obs1 + observableParameter2_obs1", - noise_formula=( - "(observableParameter1_obs1 + " - "observableParameter2_obs1) * noiseParameter1_obs1" - ), - observable_placeholders=[ - "observableParameter1_obs1", - "observableParameter2_obs1", - ], - noise_placeholders=["noiseParameter1_obs1"], - ) - problem.add_observable("obs2", formula="x", noise_formula="1") - - # new observable IDs - # (obs${i_obs}_${i_obsParOverride}_${i_noiseParOverride}) - obs1_1_1 = "obs1__obsParOverride1_1_00000000000000__noiseParOverride1" - obs1_2_1 = "obs1__obsParOverride2_1_00000000000000__noiseParOverride1" - obs1_2_2 = "obs1__obsParOverride2_1_00000000000000__noiseParOverride2" - - for obs_id in (obs1_1_1, obs1_2_1, obs1_2_2): - problem_expected.add_observable( - obs_id, - formula=( - f"observableParameter1_{obs_id} " - f"+ observableParameter2_{obs_id}" - ), - noise_formula=( - f"(observableParameter1_{obs_id} + " - f"observableParameter2_{obs_id}) " - f"* noiseParameter1_{obs_id}" - ), - observable_placeholders=[ - f"observableParameter1_{obs_id}", - f"observableParameter2_{obs_id}", - ], - noise_placeholders=[f"noiseParameter1_{obs_id}"], - ) - - problem_expected.add_observable( - "obs2", - formula="x", - noise_formula="1", - ) - - # Measurement table with timepoint-specific overrides - problem.add_measurement( - obs_id="obs1", - time=1.0, - measurement=0.1, - observable_parameters=["obsParOverride1", "1.0"], - noise_parameters=["noiseParOverride1"], - ) - problem.add_measurement( - obs_id="obs1", - time=1.0, - measurement=0.1, - observable_parameters=["obsParOverride2", "1.0"], - noise_parameters=["noiseParOverride1"], - ) - problem.add_measurement( - obs_id="obs1", - time=2.0, - measurement=0.1, - observable_parameters=["obsParOverride2", "1.0"], - noise_parameters=["noiseParOverride2"], - ) - problem.add_measurement( - obs_id="obs1", - time=2.0, - measurement=0.1, - observable_parameters=["obsParOverride2", "1.0"], - noise_parameters=["noiseParOverride2"], - ) - problem.add_measurement(obs_id="obs2", time=3.0, measurement=0.1) - - problem_expected.add_measurement( - obs_id=obs1_1_1, - time=1.0, - measurement=0.1, - observable_parameters=["obsParOverride1", "1.0"], - noise_parameters=["noiseParOverride1"], - ) - problem_expected.add_measurement( - obs_id=obs1_2_1, - time=1.0, - measurement=0.1, - observable_parameters=["obsParOverride2", "1.0"], - noise_parameters=["noiseParOverride1"], - ) - problem_expected.add_measurement( - obs_id=obs1_2_2, - time=2.0, - measurement=0.1, - observable_parameters=["obsParOverride2", "1.0"], - noise_parameters=["noiseParOverride2"], - ) - problem_expected.add_measurement( - obs_id=obs1_2_2, - time=2.0, - measurement=0.1, - observable_parameters=["obsParOverride2", "1.0"], - noise_parameters=["noiseParOverride2"], - ) - problem_expected.add_measurement(obs_id="obs2", time=3.0, measurement=0.1) - - problem.assert_valid() - unflattened_problem = copy.deepcopy(problem) - problem_expected.assert_valid() - - # Ensure having timepoint-specific overrides - assert problem.has_timepoint_specific_overrides() is True - assert problem_expected.has_timepoint_specific_overrides() is False - - flatten_timepoint_specific_output_overrides(problem) - - # Timepoint-specific overrides should be gone now - assert problem.has_timepoint_specific_overrides() is False - - assert problem_expected.observables == problem.observables - assert problem_expected.measurements == problem.measurements - problem.assert_valid() - - simulation_df = copy.deepcopy(problem.measurement_df) - simulation_df.rename(columns={C.MEASUREMENT: C.SIMULATION}) - unflattened_simulation_df = unflatten_simulation_df( - simulation_df=simulation_df, - petab_problem=unflattened_problem, - ) - # The unflattened simulation dataframe has the original observable IDs. - assert ( - unflattened_simulation_df[OBSERVABLE_ID] == ["obs1"] * 4 + ["obs2"] - ).all() - - -def test_flatten_timepoint_specific_output_overrides_special_cases(): - """Test flatten_timepoint_specific_output_overrides - for special cases: - * no observable parameters - """ - problem = Problem() - problem.model = SbmlModel.from_antimony("""species1 = 1""") - for p in ("noiseParOverride2", "noiseParOverride1"): - problem.add_parameter(p, estimate=False, nominal_value=1) - problem_expected = copy.deepcopy(problem) - problem.add_observable( - "obs1", - formula="species1", - noise_formula="noiseParameter1_obs1", - noise_placeholders=["noiseParameter1_obs1"], - ) - - problem_expected.add_observable( - "obs1__noiseParOverride1", - formula="species1", - noise_formula="noiseParameter1_obs1__noiseParOverride1", - noise_placeholders=["noiseParameter1_obs1__noiseParOverride1"], - ) - problem_expected.add_observable( - "obs1__noiseParOverride2", - formula="species1", - noise_formula="noiseParameter1_obs1__noiseParOverride2", - noise_placeholders=["noiseParameter1_obs1__noiseParOverride2"], - ) - - # Measurement table with timepoint-specific overrides - problem.add_measurement( - "obs1", - time=1.0, - measurement=0.1, - noise_parameters=["noiseParOverride1"], - ) - problem.add_measurement( - "obs1", - time=1.0, - measurement=0.1, - noise_parameters=["noiseParOverride1"], - ) - problem.add_measurement( - "obs1", - time=2.0, - measurement=0.1, - noise_parameters=["noiseParOverride2"], - ) - problem.add_measurement( - "obs1", - time=2.0, - measurement=0.1, - noise_parameters=["noiseParOverride2"], - ) - - problem_expected.add_measurement( - "obs1__noiseParOverride1", - time=1.0, - measurement=0.1, - noise_parameters=["noiseParOverride1"], - ) - problem_expected.add_measurement( - "obs1__noiseParOverride1", - time=1.0, - measurement=0.1, - noise_parameters=["noiseParOverride1"], - ) - problem_expected.add_measurement( - "obs1__noiseParOverride2", - time=2.0, - measurement=0.1, - noise_parameters=["noiseParOverride2"], - ) - problem_expected.add_measurement( - "obs1__noiseParOverride2", - time=2.0, - measurement=0.1, - noise_parameters=["noiseParOverride2"], - ) - - problem.assert_valid() - problem_expected.assert_valid() - - # Ensure having timepoint-specific overrides - assert problem.has_timepoint_specific_overrides() is True - - flatten_timepoint_specific_output_overrides(problem) - - # Timepoint-specific overrides should be gone now - assert problem.has_timepoint_specific_overrides() is False - - assert problem_expected.observables == problem.observables - assert problem_expected.measurements == problem.measurements - problem.assert_valid()