Source code for sbmlsim.experiment.runner

"""Runner for SimulationExperiments.

The ExperimentRunner is used to execute simulation experiments.
This includes
- loading of datasets
- loading of models
- running tasks (simulation on models)
- creating outputs

from pathlib import Path
from typing import Dict, List, Optional, Set, Tuple, Type, Union

from sbmlutils import log

from sbmlsim.experiment import ExperimentResult, SimulationExperiment
from sbmlsim.model import RoadrunnerSBMLModel
from import ExperimentReport, ReportResults
from sbmlsim.simulator import SimulatorSerialRR
from sbmlsim.simulator.rr_simulator_ray import SimulatorRayRR
from sbmlsim.units import UnitRegistry, UnitsInformation
from sbmlsim.utils import timeit

[docs]logger = log.get_logger(__name__)
[docs]class ExperimentRunner(object): """Class for running simulation experiments.""" def __init__( self, experiment_classes: Union[ Type[SimulationExperiment], List[Type[SimulationExperiment]] ], base_path: Path, data_path: Path, simulator: Union[SimulatorSerialRR, SimulatorRayRR] = None, ureg: UnitRegistry = None, # FIXME: is this needed on ExperimentRunner? **kwargs, ): """Initialize the runner. FIXME: document arguments for the solver. """ # single UnitRegistry per runner if not ureg: ureg = UnitsInformation._default_ureg() self.ureg = ureg self.Q_ = ureg.Quantity # initialize experiments self.base_path = base_path self.data_path = data_path self.experiments: Dict[str, SimulationExperiment] = {} self.models = {} self.simulator: Optional[SimulatorSerialRR] = None self.initialize(experiment_classes, **kwargs) self.set_simulator(simulator)
[docs] def set_simulator(self, simulator: SimulatorSerialRR) -> None: """Set simulator on the runner and experiments.""" if simulator is None: logger.debug( "No simulator set in ExperimentRunner. This warning can be " "ignored in parameter fitting." ) else: self.simulator: SimulatorSerialRR = simulator for experiment in self.experiments.values(): experiment.simulator = simulator
[docs] def initialize( self, experiment_classes: Union[ List[Type[SimulationExperiment]], Tuple[Type[SimulationExperiment]], Set[Type[SimulationExperiment]], ], **kwargs, ): """Initialize ExperimentRunner. Initialization is required in addition to construction to allow serialization of information for parallelization. """ if not isinstance(experiment_classes, (list, tuple, set)): experiment_classes = [experiment_classes] for exp_class in experiment_classes: if not isinstance(exp_class, type): raise ValueError( f"All 'experiment_classes' must be a class definition deriving " f"from 'SimulationExperiment', but '{exp_class}' is " f"'{type(exp_class)}'." ) logger.debug(f"Initialize SimulationExperiment: {exp_class.__name__}") experiment: SimulationExperiment = exp_class( base_path=self.base_path, data_path=self.data_path, ureg=self.ureg, **kwargs, ) # resolve models for experiment _models = {} for model_id, source in experiment.models().items(): if source not in self.models: # not cashed yet, cash the model for lookup self.models[source] = RoadrunnerSBMLModel( source=source, ureg=self.ureg ) _models[model_id] = self.models[source] # set resolved models in experiment experiment._models = _models # only after model loading the unit registry is filled experiment.initialize() self.experiments[experiment.sid] = experiment
[docs] def run_experiments( self, output_path: Path, show_figures: bool = False, save_results: bool = False, figure_formats: List[str] = None, reduced_selections: bool = True, ) -> List[ExperimentResult]: """Run the experiments.""" if not output_path.exists(): output_path.mkdir(parents=True) exp_results = [] sid: str experiment: SimulationExperiment for sid, experiment in self.experiments.items():"Running SimulationExperiment: {sid}") # ExperimentResult used to create report result = simulator=self.simulator, output_path=output_path / sid, show_figures=show_figures, save_results=save_results, figure_formats=figure_formats, reduced_selections=reduced_selections, ) exp_results.append(result) return exp_results # from sbmlsim.simulator.rr_simulator_ray import SimulatorParallel
[docs]def run_experiments( experiments: Union[Type[SimulationExperiment], List[Type[SimulationExperiment]]], output_path: Path, base_path: Path = None, data_path: Union[List[Path], Tuple[Path], Optional[Path]] = None, parallel: bool = True, ) -> Path: """Run simulation experiments.""" if not isinstance(experiments, (list, tuple)): experiments = [experiments] simulator = SimulatorRayRR() if parallel else SimulatorSerialRR() runner = ExperimentRunner( experiments, simulator=simulator, data_path=data_path, base_path=base_path, ) results = runner.run_experiments( output_path=output_path, show_figures=True, ) report_results = ReportResults() for exp_result in results: report_results.add_experiment_result(exp_result=exp_result) report = ExperimentReport(report_results) report.create_report(output_path=output_path)