"""Definition of timecourses and timecourse simulations."""
import json
from copy import deepcopy
from pathlib import Path
from typing import Any, Dict, List, Optional, Union
import numpy as np
from pint import Quantity
from sbmlutils import log
from sbmlsim.serialization import ObjectJSONEncoder
from sbmlsim.simulation import AbstractSim, Dimension
from sbmlsim.units import Units, UnitsInformation
[docs]logger = log.get_logger(__name__)
[docs]class Timecourse(ObjectJSONEncoder):
"""Simulation definition.
Definition of all information necessary to run a single timecourse simulation.
A single simulation consists of multiple changes which are applied,
all simulations are performed and collected.
Changesets and selections are deep copied for persistence.
"""
def __init__(
self,
start: float,
end: float,
steps: int,
changes: Dict[str, Quantity] = None,
model_changes: Dict[str, Quantity] = None,
model_manipulations: dict = None,
discard: bool = False,
):
"""Create a time course definition for simulation.
Discarded simulations do not add time shifts, i.e. a pre-simulation
does not increase the time of simulation.
:param start: start time
:param end: end time
:param steps: simulation steps
:param changes: parameter and initial condition changes
:param model_changes: model parameter and initial condition changes
:param model_manipulations: model structure changes
:param discard: discards results from simulations (e.g. pre-simulations)
"""
# Create empty changes and model changes for serialization
if changes is None:
changes = {}
if model_changes is None:
model_changes = {}
if model_manipulations is None:
model_manipulations = {}
self.start = start
self.end = end
self.steps = steps
self.changes = deepcopy(changes)
self.model_changes = deepcopy(model_changes)
self.model_manipulations = deepcopy(model_manipulations)
self.discard = discard
[docs] def __repr__(self) -> str:
"""Get string representation."""
return f"Timecourse([{self.start}:{self.end}])"
[docs] def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary."""
d = dict()
for key in self.__dict__:
d[key] = self.__dict__[key]
return d
[docs] def add_change(self, sid: str, value: float) -> None:
"""Add change."""
self.changes[sid] = value
[docs] def remove_change(self, sid: str) -> None:
"""Remove change for given id."""
del self.changes[sid]
[docs] def add_model_change(self, sid: str, change) -> None:
"""Add model change."""
self.model_changes[sid] = change
[docs] def add_model_changes(self, model_changes: Dict[str, str]) -> None:
"""Add model changes."""
self.model_changes.update(model_changes)
[docs] def remove_model_change(self, sid: str) -> None:
"""Remove model change for id."""
del self.model_changes[sid]
[docs] def normalize(self, uinfo: UnitsInformation) -> None:
"""Normalize values to model units for all changes."""
self.model_changes = UnitsInformation.normalize_changes(
changes=self.model_changes, uinfo=uinfo
)
self.changes = UnitsInformation.normalize_changes(
changes=self.changes, uinfo=uinfo
)
[docs] def strip_units(self) -> None:
"""Strip units from changes for parallel simulation.
All changes must be normalized before stripping !.
"""
self.model_changes = {k: v.magnitude for k, v in self.model_changes.items()}
self.changes = {k: v.magnitude for k, v in self.changes.items()}
[docs]class TimecourseSim(AbstractSim):
"""Timecourse simulation consisting of multiple concatenated timecourses.
In case of a single timecourse, only the single timecourse is executed.
"""
def __init__(
self,
timecourses: Union[List[Timecourse], Timecourse],
selections: Optional[List[str]] = None,
reset: bool = True,
time_offset: float = 0.0,
):
"""Initialize timecourse sim.
:param timecourses:
:param selections:
:param reset: complete reset of model
:param time_offset: time shift of simulation
"""
if isinstance(timecourses, Timecourse):
timecourses = [timecourses]
self.timecourses = []
for tc in timecourses:
if not tc:
# remove empty elements (allows for cleaner syntax)
continue
if isinstance(tc, dict):
# construct from dict
tc = Timecourse(**tc)
# make a copy to ensure independence of instances
self.timecourses.append(deepcopy(tc))
if len(self.timecourses) == 0:
logger.error("No timecourses in simulation")
else:
for k, tc in enumerate(self.timecourses):
if k > 0 and tc.model_changes:
logger.error(
f"'model_changes' only allowed on first timecourse: {tc}"
)
self.selections = deepcopy(selections)
self.reset = reset
self.time_offset = time_offset
self.time = self._time()
[docs] def __repr__(self) -> str:
"""Get representation."""
return f"TimecourseSim({[tc for tc in self.timecourses]})"
[docs] def _time(self) -> np.ndarray:
"""Calculate the time vector complete simulation."""
t_offset = self.time_offset
time_vecs = []
for tc in self.timecourses:
time_vecs.append(np.linspace(tc.start, tc.end, num=tc.steps + 1) + t_offset)
t_offset += tc.end
res: np.ndarray = np.concatenate(time_vecs)
return res
[docs] def dimensions(self) -> List[Dimension]:
"""Get dimensions."""
return [Dimension(dimension="time", index=self.time)]
[docs] def add_model_changes(self, model_changes: Dict) -> None:
"""Add model changes to given simulation."""
if self.timecourses:
tc = self.timecourses[0] # type: Timecourse
tc.add_model_changes(model_changes)
[docs] def normalize(self, uinfo: UnitsInformation) -> None:
"""Normalize timecourse simulation."""
# logger.error("NORMALIZING TIMECOURSESIM")
for tc in self.timecourses:
tc.normalize(uinfo=uinfo)
[docs] def strip_units(self) -> None:
"""Strip units from simulation."""
for tc in self.timecourses:
tc.strip_units()
[docs] def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary."""
d = {
"type": self.__class__.__name__,
"selections": self.selections,
"reset": self.reset,
"time_offset": self.time_offset,
"timecourses": [tc.to_dict() for tc in self.timecourses],
}
return d
[docs] def to_json(self, path: Path = None) -> str:
"""Convert definition to JSON."""
if path is None:
return json.dumps(self, cls=ObjectJSONEncoder, indent=2)
else:
with open(path, "w") as f_json:
json.dump(self, fp=f_json, cls=ObjectJSONEncoder, indent=2)
@staticmethod
[docs] def from_json(json_info: Union[str, Path]) -> "TimecourseSim":
"""Load from JSON."""
if isinstance(json_info, Path):
with open(json_info, "r") as f_json:
d = json.load(f_json)
else:
d = json.loads(json_info)
if "type" in d:
d.pop("type") # serialized property
return TimecourseSim(**d)
[docs] def __str__(self) -> str:
"""Get string representation."""
lines = ["-" * 40, f"{self.__class__.__name__}", "-" * 40, self.to_json()]
return "\n".join(lines)