Source code for osp.core.session.sim_wrapper_session

"""Abstract class that contains important method of a simulation session."""

from abc import abstractmethod
from .wrapper_session import WrapperSession, consumes_buffers
from osp.core.session.buffers import BufferContext


[docs]class SimWrapperSession(WrapperSession): """Abstract class used for simulation sessions. Contains methods necessary for all simulation sessions. """ def __init__(self, engine, **kwargs): """Initialize the simulation session. Args: engine (Any): The simulation engine object. """ super().__init__(engine, **kwargs) self._ran = False @consumes_buffers def run(self): """Run the simulation.""" self.log_buffer_status(BufferContext.USER) self._check_cardinalities() root_obj = self._registry.get(self.root) added, updated, deleted = self._buffers[BufferContext.USER] if not self._ran: self._initialize(root_obj, added) else: self._apply_added(root_obj, added) self._apply_updated(root_obj, updated) self._apply_deleted(root_obj, deleted) self._reset_buffers(BufferContext.USER) self._run(root_obj) self._ran = True self.expire_all() @abstractmethod def _run(self, root_cuds_object): """Run the engine. Args: root_cuds_object (Cuds): The wrapper cuds object """ @abstractmethod def _apply_added(self, root_obj, buffer): """Add the added cuds_objects to the engine. Args: root_obj (Cuds): The wrapper cuds object buffer (Dict[UUID, Cuds]): All Cuds objects that have been added """ @abstractmethod def _apply_updated(self, root_obj, buffer): """Update the updated cuds_objects in the engine. Args: root_obj (Cuds): The wrapper cuds object buffer (Dict[UUID, Cuds]): All Cuds objects that have been updated """ @abstractmethod def _apply_deleted(self, root_obj, buffer): """Delete the deleted cuds_objects from the engine. Args: root_obj (Cuds): The wrapper cuds object. buffer (Dict[UUID, Cuds]): All Cuds objects that have been deleted """ def _initialize(self, root_obj, buffer): """Initialize the session. This method is executed before the first run. Args: root_obj (Cuds): The wrapper cuds object. buffer (Dict[UUID, Cuds]): All Cuds objects that have been added """ self._apply_added(root_obj, buffer)