Source code for sbmlsim.simulator.simulation

"""Classes for running simulations with SBML models."""
import logging

import pandas as pd
from roadrunner import RoadRunner, SelectionRecord

from sbmlsim.model import ModelChange
from sbmlsim.result import XResult
from sbmlsim.simulation import ScanSim, Timecourse, TimecourseSim

[docs]logger = logging.getLogger(__name__)
[docs]class SimulatorWorker: """Worker running simulations. Implements the timecourse simulation once which can be reused by the different simulators. """ def __init__(self): """Init placeholder.""" self.r = None
[docs] def _timecourse(self, simulation: TimecourseSim) -> pd.DataFrame: """Timecourse simulation. Requires for all timecourse definitions in the timecourse simulation to be unit normalized. The changes have no units any more for parallel simulations. You should never call this function directly! :param simulation: Simulation definition(s) :return: DataFrame with results """ if isinstance(simulation, Timecourse): simulation = TimecourseSim(timecourses=[simulation]) if simulation.reset: self.r.resetToOrigin() frames = [] t_offset = simulation.time_offset for k, tc in enumerate(simulation.timecourses): if k == 0 and tc.model_changes: # [1] apply model changes of first simulation logger.debug("Applying model changes") for key, item in tc.model_changes.items(): if key.startswith("init"): logger.error( f"Initial model changes should be provided " f"without 'init': '{key} = {item}'" ) # FIXME: implement model changes via init # init_key = f"init({key})" init_key = key try: value = item.magnitude except AttributeError: value = item try: self.r[init_key] = value except RuntimeError: logger.error(f"roadrunner RuntimeError: '{init_key} = {item}'") # boundary condition=true species, trying direct fallback # see init_key = key self.r[key] = value logger.debug(f"\t{init_key} = {item}") # [2] re-evaluate initial assignments # logger.debug("Reevaluate initial conditions") # FIXME: support initial model changes # self.r.resetAll() self.r.reset(SelectionRecord.DEPENDENT_FLOATING_AMOUNT) self.r.reset(SelectionRecord.DEPENDENT_INITIAL_GLOBAL_PARAMETER) # [3] apply model manipulations # model manipulations are applied to model if len(tc.model_manipulations) > 0: # FIXME: update to support roadrunner model changes for key, value in tc.model_changes.items(): if key == ModelChange.CLAMP_SPECIES: for sid, formula in value.items(): ModelChange.clamp_species(self.r, sid, formula) else: raise ValueError( f"Unsupported model change: " f"'{key}': {value}. Supported changes are: " f"['{ModelChange.CLAMP_SPECIES}']" ) # [4] apply changes for key, item in tc.changes.items(): logger.debug("Applying simulation changes") try: self.r[key] = float(item.magnitude) except AttributeError: self.r[key] = float(item) logger.debug(f"\t{key} = {item}") # run simulation integrator = self.r.integrator if integrator.getValue("variable_step_size"): s = self.r.simulate(start=tc.start, end=tc.end) else: s = self.r.simulate(start=tc.start, end=tc.end, steps=tc.steps) df = pd.DataFrame(s, columns=s.colnames) df.time = df.time + t_offset if not tc.discard: # discard timecourses (pre-simulation) t_offset += tc.end frames.append(df) return pd.concat(frames, sort=False)