"""Serial simulator.
Executing simulations with single roadrunner instance on a single core.
"""
from __future__ import annotations
from pathlib import Path
from typing import Iterator, List, Optional
import pandas as pd
from sbmlutils import log
from sbmlsim.model.rr_model import roadrunner
from sbmlsim.simulation import TimecourseSim
from sbmlsim.simulator.rr_simulator_abstract import SimulatorAbstractRR
from sbmlsim.simulator.rr_worker import SimulationWorkerRR
[docs]logger = log.get_logger(__name__)
[docs]class SimulatorSerialRR(SimulatorAbstractRR):
"""Serial simulator using a single core."""
@staticmethod
[docs] def from_sbml(sbml_path: Path) -> SimulatorSerialRR:
"""Set model from SBML."""
print(sbml_path)
print(type(sbml_path))
rr: roadrunner.RoadRunner = roadrunner.RoadRunner(str(sbml_path))
simulator = SimulatorSerialRR()
# FIXME: implement global model cache
simulator.set_model(rr.saveStateS())
return simulator
def __init__(self):
"""Initialize serial simulator with single worker."""
self.worker = SimulationWorkerRR()
[docs] def set_model(self, model_state: bytes) -> None:
"""Set model from state."""
self.worker.set_model(model_state)
[docs] def set_timecourse_selections(
self, selections: Optional[Iterator[str]] = None
) -> None:
"""Set timecourse selections."""
self.worker.set_timecourse_selections(selections=selections)
[docs] def set_integrator_settings(self, **kwargs):
"""Set integrator settings."""
self.worker.set_integrator_settings(**kwargs)
[docs] def _run_timecourses(self, simulations: List[TimecourseSim]) -> List[pd.DataFrame]:
"""Execute timecourse simulations."""
if len(simulations) > 1:
logger.warning(
"Use of SimulatorSerial to run multiple timecourses. "
"Use SimulatorParallel instead."
)
return [self.worker._timecourse(sim) for sim in simulations]