From e5d2f70b9748e5021d1d2dfd5108ec98495184f4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Matthias=20K=C3=B6nig?= Date: Sat, 22 Jun 2024 17:36:58 +0200 Subject: [PATCH] working on fit mapping --- src/sbmlsim/experiment/runner.py | 1 - src/sbmlsim/fit/helpers.py | 115 +++++++++++++++++++++++++++++++ src/sbmlsim/fit/objects.py | 20 ++++++ src/sbmlsim/fit/runner.py | 2 +- src/sbmlsim/oven/model_state.py | 1 + 5 files changed, 137 insertions(+), 2 deletions(-) create mode 100644 src/sbmlsim/fit/helpers.py diff --git a/src/sbmlsim/experiment/runner.py b/src/sbmlsim/experiment/runner.py index 7ca1b3fa..46c94506 100644 --- a/src/sbmlsim/experiment/runner.py +++ b/src/sbmlsim/experiment/runner.py @@ -73,7 +73,6 @@ def set_simulator(self, simulator: SimulatorSerial) -> None: for experiment in self.experiments.values(): experiment.simulator = simulator - @timeit def initialize(self, experiment_classes: Union[List[Type[SimulationExperiment]], Tuple[Type[SimulationExperiment]], Set[Type[SimulationExperiment]]], **kwargs): """Initialize ExperimentRunner. diff --git a/src/sbmlsim/fit/helpers.py b/src/sbmlsim/fit/helpers.py new file mode 100644 index 00000000..1f2c4368 --- /dev/null +++ b/src/sbmlsim/fit/helpers.py @@ -0,0 +1,115 @@ +"""Helper functions for fitting.""" +from pathlib import Path + +from sbmlsim.fit import FitExperiment, FitMapping +import pandas as pd +from sbmlutils.console import console +from sbmlutils.log import get_logger + + +from sbmlsim.experiment import ExperimentRunner, SimulationExperiment +from sbmlsim.fit import FitExperiment, FitMapping, FitData + +from typing import Dict, List, Type, Union, Callable, Iterable, Tuple, Any + +from sbmlsim.fit.objects import MappingMetaData + +logger = get_logger(__name__) + + +def filtered_fit_experiments( + experiment_classes: List[Type[SimulationExperiment]], + metadata_filters: Union[Callable, Iterable[Callable]], + base_path: Path, + data_path: Path, +) -> Tuple[Dict[str, List[FitExperiment]], pd.DataFrame]: + """Fit experiments based on MappingMetaData. + + :param experiment_classes: List of SimulationExperiment class definition + :param metadata_filter: + """ + filters = [metadata_filters] if isinstance(metadata_filters, Callable) else metadata_filters + + # instantiate objects for filtering of fit mappings + runner = ExperimentRunner( + experiment_classes=experiment_classes, + base_path=base_path, + data_path=data_path, + ) + + fit_experiments: Dict[str, List[FitExperiment]] = {} + all_info: List[Dict] = [] + + for k, experiment_name in enumerate(runner.experiments): + # print(experiment_name) + experiment_class = experiment_classes[k] + experiment = runner.experiments[experiment_name] + + # filter mappings by metadata + mappings = [] + for fm_key, fit_mapping in experiment.fit_mappings().items(): + + # test all the filters + accept = True + for filter in filters: + if not filter(fm_key, fit_mapping): + accept = False + break + + if accept: + mappings.append(fm_key) + + # collect information + try: + metadata: MappingMetaData = fit_mapping.metadata + yid = "__".join(fit_mapping.observable.y.sid.split("__")[1:]) + info: Dict[str, Any] = { + "experiment": experiment_name, + "fm_key": fm_key, + "yid": yid, + **metadata.to_dict() + } + all_info.append(info) + except Exception as err: + logger.error( + f"Error in metadata for experiment '{experiment_name}', {fm_key=}" + ) + raise err + + if mappings: + # add fit experiment from filtered mappings + fit_experiments[experiment_name] = [ + FitExperiment( + experiment=experiment_class, + mappings=mappings, + weights=None, + use_mapping_weights=True, + ) + ] + + df = pd.DataFrame(all_info) + + return fit_experiments, df + +def f_fitexp( + experiment_classes: List[Type[SimulationExperiment]], + metadata_filters: Union[Callable, Iterable[Callable]], + base_path: Path, + data_path: Path, +): + """Generic function to get fit experiments for filter.""" + fit_experiments, df = filtered_fit_experiments( + experiment_classes, + metadata_filters=metadata_filters, + base_path=base_path, + data_path=data_path, + ) + console.print(df.to_string()) + + return fit_experiments + +def filter_empty(fit_mapping_key: str, fit_mapping: FitMapping) -> bool: + """Return all experiments/mappings.""" + return True + + diff --git a/src/sbmlsim/fit/objects.py b/src/sbmlsim/fit/objects.py index ca2af34a..1ff79139 100644 --- a/src/sbmlsim/fit/objects.py +++ b/src/sbmlsim/fit/objects.py @@ -1,4 +1,5 @@ """Definition of Objects used in FitProblems and optimization.""" +from __future__ import annotations import json import math from dataclasses import dataclass @@ -6,6 +7,7 @@ from typing import Any, Callable, Dict, Iterable, List, Optional, Sized, Union import numpy as np +import pandas as pd from sbmlutils import log from sbmlutils.console import console @@ -274,6 +276,19 @@ def to_json(self, path: Path = None) -> Optional[str]: """ return to_json(object=self, path=path) + def to_dict(self, path: Path = None) -> Optional[str]: + """Serialize to JSON. + + Serializes to file if path is provided, otherwise returns JSON string. + """ + return { + "pid": self.pid, + "start_value": self.start_value, + "lower_bound": self.lower_bound, + "upper_bound": self.upper_bound, + "unit": self.unit, + } + @staticmethod def from_json(json_info: Union[str, Path]) -> "FitParameter": """Load from JSON.""" @@ -284,6 +299,11 @@ def from_json(json_info: Union[str, Path]) -> "FitParameter": d = json.loads(json_info) return FitParameter(**d) + @staticmethod + def parameters_to_df(parameters: Iterable[FitParameter]) -> pd.DataFrame: + """DataFrame of parameters""" + return pd.DataFrame([p.to_dict() for p in parameters]) + class FitData: """Data used in a fit. diff --git a/src/sbmlsim/fit/runner.py b/src/sbmlsim/fit/runner.py index 34483b6a..e77e65c6 100644 --- a/src/sbmlsim/fit/runner.py +++ b/src/sbmlsim/fit/runner.py @@ -107,7 +107,7 @@ def run_optimization( n_cores = max(1, multiprocessing.cpu_count() - 1) logger.error(f"More cores then cpus requested, reducing cores to '{n_cores}'") - console.rule("START OPTIMIZATION", align="left", style="white") + console.rule("Start optimization", align="left", style="white") console.log(f"Running {n_cores} workers") if size < n_cores: logger.warning( diff --git a/src/sbmlsim/oven/model_state.py b/src/sbmlsim/oven/model_state.py index bed8e2c2..cd6ddafc 100644 --- a/src/sbmlsim/oven/model_state.py +++ b/src/sbmlsim/oven/model_state.py @@ -38,6 +38,7 @@ def get_state_path(sbml_path: Path) -> Optional[Path]: r = roadrunner.RoadRunner(str(sbml_path)) # save state r.saveState(str(state_path)) + r.saveStateS() print(f"Load from state: '{state_path}'") r.loadState(str(state_path))