diff --git a/petab/v2/core.py b/petab/v2/core.py index a7e54e52..2efba572 100644 --- a/petab/v2/core.py +++ b/petab/v2/core.py @@ -1180,7 +1180,9 @@ def __repr__(self): def __str__(self): pid = repr(self.id) if self.id else "without ID" - model = f"with models {self.models}" if self.model else "without model" + model = ( + f"with models {self.models}" if self.models else "without model" + ) ne = len(self.experiments) experiments = f"{ne} experiments" @@ -1990,20 +1992,33 @@ def validate( ) ) + validation_tasks = validation_tasks or self.validation_tasks if len(self.models) > 1: - # TODO https://github.com/PEtab-dev/libpetab-python/issues/392 - # We might just want to split the problem into multiple - # problems, one for each model, and then validate each - # problem separately. - validation_results.append( - ValidationIssue( - ValidationIssueSeverity.WARNING, - "Problem contains multiple models. " - "Validation is not yet fully supported.", - ) + validation_results.extend( + self._validate_multi_model(validation_tasks) + ) + else: + validation_results.extend( + self._run_validation_tasks(validation_tasks) ) - for task in validation_tasks or self.validation_tasks: + return validation_results + + def _run_validation_tasks( + self, validation_tasks: list[ValidationTask] + ) -> ValidationResultList: + """Run the given validation tasks on this problem. + + Stops at the first critical issue. + """ + from ..v2.lint import ( + ValidationIssue, + ValidationIssueSeverity, + ValidationResultList, + ) + + validation_results = ValidationResultList() + for task in validation_tasks: try: cur_result = task.run(self) except Exception as e: # noqa BLE001 @@ -2021,6 +2036,192 @@ def validate( return validation_results + def _validate_multi_model( + self, validation_tasks: list[ValidationTask] + ) -> ValidationResultList: + """Validate a problem with multiple models. + + Runs the validation tasks on each part of + :meth:`Problem.split_by_model`. Checks of the shared tables + (model IDs of measurements, unique IDs, the mapping table, unused + experiments, conditions and parameters) run on the full problem. + See https://github.com/PEtab-dev/libpetab-python/issues/392. + """ + from ..v2.lint import ( + CheckAllParametersPresentInParameterTable, + CheckMappingTable, + CheckMeasurementModelId, + CheckUniquePrimaryKeys, + CheckUnusedConditions, + CheckUnusedExperiments, + ValidationError, + ValidationResultList, + ) + + validation_results = ValidationResultList() + + model_ids = [model.model_id for model in self.models] + if duplicates := sorted( + {id_ for id_ in model_ids if model_ids.count(id_) > 1} + ): + validation_results.append( + ValidationError(f"Duplicate model IDs: {duplicates}") + ) + # always required, otherwise the problem can't be split + validation_results.extend( + self._run_validation_tasks([CheckMeasurementModelId()]) + ) + if validation_results.has_errors(): + return validation_results + + # the split drops unused or other models' table entries, + # so checks of the shared tables have to run on the full problem + full_problem_task_types = ( + CheckUniquePrimaryKeys, + CheckMappingTable, + CheckUnusedExperiments, + CheckUnusedConditions, + ) + validation_results.extend( + self._run_validation_tasks( + [ + task + for task in validation_tasks + if isinstance(task, full_problem_task_types) + ] + ) + ) + + sub_problems = self.split_by_model() + for model_id, sub_problem in sub_problems.items(): + for issue in sub_problem._run_validation_tasks( + [ + task + for task in validation_tasks + if not isinstance(task, full_problem_task_types) + ] + ): + issue.message = f"Model `{model_id}`: {issue.message}" + validation_results.append(issue) + + # extraneous parameters are dropped by the split + if not any( + isinstance(task, CheckAllParametersPresentInParameterTable) + for task in validation_tasks + ): + return validation_results + used_parameter_ids = set().union( + *(sub_problem.x_ids for sub_problem in sub_problems.values()) + ) + if unused := [x for x in self.x_ids if x not in used_parameter_ids]: + validation_results.append( + ValidationError( + "Extraneous parameter(s) in parameter table, not used " + f"by any model: {unused}" + ) + ) + + return validation_results + + def split_by_model(self) -> dict[str, Problem]: + """Split the problem into one single-model problem per model. + + Each part contains the model, its measurements, and the experiments, + conditions, observables, mappings and parameters these require. + Mappings to entities of other models are dropped, all other mappings + (e.g., to extension entities) are kept. + The parts share their models and table entries with this problem. + + See https://github.com/PEtab-dev/libpetab-python/issues/392. + + :returns: The single-model problems, keyed by model ID. + :raises ValueError: If model IDs are not unique, or if any + measurement is not assigned to a model. + """ + from ..v2.lint import get_valid_parameters_for_parameter_table + + model_ids = {model.model_id for model in self.models} + if len(model_ids) != len(self.models): + raise ValueError("Model IDs must be unique.") + + def get_model_id(measurement: Measurement) -> str | None: + # with a single model, the model ID is optional + if measurement.model_id is None and len(self.models) == 1: + return self.models[0].model_id + return measurement.model_id + + if unassigned := [ + m for m in self.measurements if get_model_id(m) not in model_ids + ]: + raise ValueError( + f"Measurement not assigned to any model: {unassigned[0]}" + ) + + sub_problems = {} + for model in self.models: + measurements = [ + m + for m in self.measurements + if get_model_id(m) == model.model_id + ] + experiment_ids = {m.experiment_id for m in measurements} + experiments = [ + e for e in self.experiments if e.id in experiment_ids + ] + condition_ids = { + condition_id + for e in experiments + for period in e.periods + for condition_id in period.condition_ids + } + observable_ids = {m.observable_id for m in measurements} + + sub_problem = Problem( + models=[model], + condition_tables=[ + ConditionTable( + [c for c in self.conditions if c.id in condition_ids] + ) + ], + experiment_tables=[ExperimentTable(experiments)], + observable_tables=[ + ObservableTable( + [o for o in self.observables if o.id in observable_ids] + ) + ], + measurement_tables=[MeasurementTable(measurements)], + mapping_tables=[ + MappingTable( + [ + m + for m in self.mappings + if m.model_id is None + or model.has_entity_with_id(m.model_id) + or not any( + other.has_entity_with_id(m.model_id) + for other in self.models + ) + ] + ) + ], + extensions=self.extensions, + config=self.config, + ) + sub_problem.validation_tasks = self.validation_tasks.copy() + # get_valid_parameters_for_parameter_table doesn't depend on the + # parameter table, so we can use it before adding the parameters + valid_parameter_ids = get_valid_parameters_for_parameter_table( + sub_problem + ) + sub_problem.parameter_tables = [ + ParameterTable( + [p for p in self.parameters if p.id in valid_parameter_ids] + ) + ] + sub_problems[model.model_id] = sub_problem + + return sub_problems + def assert_valid(self, **kwargs) -> None: """Assert that the PEtab problem is valid. @@ -2188,6 +2389,7 @@ def add_measurement( | float | None = None, noise_parameters: Sequence[str | float] | str | float | None = None, + model_id: str | None = None, ): """Add a measurement to the problem. @@ -2201,6 +2403,7 @@ def add_measurement( measurement: The measurement value observable_parameters: The observable parameters noise_parameters: The noise parameters + model_id: The ID of the model the measurement belongs to """ if observable_parameters is not None and not isinstance( observable_parameters, Sequence @@ -2222,6 +2425,7 @@ def add_measurement( measurement=measurement, observable_parameters=observable_parameters, noise_parameters=noise_parameters, + model_id=model_id, ) ) diff --git a/tests/v2/test_core.py b/tests/v2/test_core.py index 6f7c3845..e33378e9 100644 --- a/tests/v2/test_core.py +++ b/tests/v2/test_core.py @@ -30,7 +30,7 @@ ) from petab.v2.core import * from petab.v2.core import ExtensionConfig -from petab.v2.lint import ValidationIssueSeverity +from petab.v2.lint import CheckUnusedConditions, ValidationIssueSeverity from petab.v2.models.sbml_model import SbmlModel from petab.v2.petab1to2 import petab1to2 @@ -980,3 +980,165 @@ def test_objective_type(): ) assert problem.has_map_objective is True assert problem.has_ml_objective is False + + +def _two_model_problem() -> Problem: + """A valid problem with two models that share some parameters.""" + problem = Problem() + problem.models = [ + SbmlModel.from_antimony( + "x_A' = k_A - k_deg * x_A; x_A = 0; k_A = 1; k_deg = 0.1", + model_id="model_A", + ), + SbmlModel.from_antimony( + "x_B' = k_B - k_deg * x_B; x_B = 0; k_B = 1; k_deg = 0.1", + model_id="model_B", + ), + ] + problem.add_mapping("k_B", "k_B", name="production rate B") + problem.add_condition("cA", x_A=1) + problem.add_condition("cB", x_B=2) + problem.add_experiment("eA", 0, "cA") + problem.add_experiment("eB", 0, "cB") + problem.add_observable("obs_A", "x_A", "sigma") + problem.add_observable("obs_B", "x_B", "sigma") + for t in (1, 2): + problem.add_measurement( + "obs_A", + experiment_id="eA", + time=t, + measurement=t, + model_id="model_A", + ) + problem.add_measurement( + "obs_B", + experiment_id="eB", + time=t, + measurement=t, + model_id="model_B", + ) + for parameter_id in ("k_A", "k_B", "k_deg", "sigma"): + problem.add_parameter(parameter_id, nominal_value=1, lb=0.1, ub=10) + return problem + + +def test_split_by_model(): + problem = _two_model_problem() + sub_problems = problem.split_by_model() + + assert list(sub_problems) == ["model_A", "model_B"] + for suffix in ("A", "B"): + sub_problem = sub_problems[f"model_{suffix}"] + assert sub_problem.models == [problem.models["AB".index(suffix)]] + assert [c.id for c in sub_problem.conditions] == [f"c{suffix}"] + assert [e.id for e in sub_problem.experiments] == [f"e{suffix}"] + assert [o.id for o in sub_problem.observables] == [f"obs_{suffix}"] + assert {m.observable_id for m in sub_problem.measurements} == { + f"obs_{suffix}" + } + assert len(sub_problem.measurements) == 2 + sub_problem.assert_valid() + + assert sub_problems["model_A"].x_ids == ["k_A", "k_deg", "sigma"] + assert sub_problems["model_B"].x_ids == ["k_B", "k_deg", "sigma"] + assert [m.petab_id for m in sub_problems["model_A"].mappings] == [] + assert [m.petab_id for m in sub_problems["model_B"].mappings] == ["k_B"] + + # mappings to non-model entities, e.g., of extensions, are kept + problem.add_mapping("net1_ps", "net1.parameters") + for sub_problem in problem.split_by_model().values(): + assert "net1_ps" in {m.petab_id for m in sub_problem.mappings} + + # a measurement without model + problem.measurements[0].model_id = None + with pytest.raises(ValueError, match="not assigned"): + problem.split_by_model() + + # duplicate model IDs + problem = _two_model_problem() + problem.models[1].model_id = "model_A" + with pytest.raises(ValueError, match="unique"): + problem.split_by_model() + + +def test_split_by_model_single_model(): + """With one model, measurements don't need a model ID.""" + problem = _two_model_problem().split_by_model()["model_A"] + for m in problem.measurements: + m.model_id = None + sub_problems = problem.split_by_model() + assert list(sub_problems) == ["model_A"] + assert len(sub_problems["model_A"].measurements) == 2 + + +def test_validate_multi_model(): + problem = _two_model_problem() + assert (result := problem.validate()) == [], result + problem.assert_valid() + str(problem) + + # parameter not used by any model + problem.add_parameter("unused", nominal_value=1, lb=0.1, ub=10) + result = problem.validate() + assert result.has_errors() + assert len(result) == 1 + assert "unused" in result[0].message + assert "not used by any model" in result[0].message + + # ... unless parameter table checks are not requested + assert problem.validate([CheckUnusedConditions()]) == [] + + # missing model ID + problem = _two_model_problem() + problem.measurements[0].model_id = None + result = problem.validate() + assert result.has_errors() + assert "does not have a model ID" in result[0].message + + # duplicate model IDs + problem = _two_model_problem() + problem.models[1].model_id = "model_A" + result = problem.validate() + assert result.has_errors() + assert "Duplicate model IDs" in result[0].message + + # unknown model ID + problem = _two_model_problem() + problem.measurements[0].model_id = "model_C" + result = problem.validate() + assert result.has_errors() + assert "does not match any of the available models" in result[0].message + + # issues found in a single model are labelled with the model ID + problem = _two_model_problem() + problem.add_measurement( + "obs_undefined", + experiment_id="eB", + time=1, + measurement=1, + model_id="model_B", + ) + result = problem.validate() + assert result.has_errors() + assert any( + r.message.startswith("Model `model_B`: ") + and "obs_undefined" in r.message + for r in result + ), result + + # unused experiments and conditions are reported + problem = _two_model_problem() + problem.add_condition("c_unused", x_A=3) + problem.add_experiment("e_unused", 0, "cA") + result = problem.validate() + assert not result.has_errors() + assert any("e_unused" in r.message for r in result), result + assert any("c_unused" in r.message for r in result), result + + # duplicate IDs of entries that the split drops are reported + problem = _two_model_problem() + problem.add_condition("c_dup", x_A=3) + problem.add_condition("c_dup", x_A=4) + result = problem.validate() + assert result.has_errors() + assert any("duplicate IDs: ['c_dup']" in r.message for r in result), result