working on fit mapping

Author: matthiaskoenig

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.
 

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
+
+

src/sbmlsim/fit/objects.py

@@ -1,11 +1,13 @@
 """Definition of Objects used in FitProblems and optimization."""
+from __future__ import annotations
 import json
 import math
 from dataclasses import dataclass
 from pathlib import Path
 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.

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(

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))