Source code for utopya_backend.model.base

"""This module implements the :py:class:`.BaseModel` class which can be
inherited from for implementing a utopya-controlled model.

Its main aim is to provide shared simulation infrastructure and does not make
further assumptions about the abstraction a model makes, like the step-wise
iteration done in :py:class:`~utopya_backend.model.step.StepwiseModel`."""

import abc
import random
import signal
import sys
import time
from typing import Optional, Union

import h5py as h5
import numpy as np

from ..logging import backend_logger as _backend_logger
from ..logging import get_level as _get_level
from ..signal import SIG_STOPCOND, SIGNAL_INFO, attach_signal_handlers
from import load_cfg_file as _load_cfg_file

# -----------------------------------------------------------------------------

[docs]class BaseModel(abc.ABC): """An abstract base model class that can be inherited from to implement a model. This class provides basic simulation infrastructure in a way that it couples to utopya as a frontend: - A shared RNG instance and a logger. - includes logic to evaluate ``num_steps`` and ``write_{start,every}`` - emits monitor information to the frontend, informing the utopya frontend about simulation progress For more specific purposes, there are specialized classes that are built around a shared modelling paradigm like step-wise iteration of the model, see :py:class:`~utopya_backend.model.step.StepwiseModel`. """ ATTACH_SIGNAL_HANDLERS: bool = True """If true, calls :py:func:`~utopya_backend.signal.attach_signal_handlers` to attach signal handlers for stop conditions and interrupts. .. hint:: You may want to disable this in a subclass in case you cannot handle a case where a signal is meant to stop your simulation gracefully. """ USE_SYS_EXIT: bool = True """If false, will not call sys.exit upon handled signal, but just return the error code.""" # .. Initialization and Teardown ..........................................
[docs] def __init__( self, *, cfg_file_path: str, _log: "logging.Logger" = None, ): """Initialize the model instance, constructing an RNG and HDF5 group to write the output data to. .. todo:: Allow initializing from a "parent model" such that hierarchical nesting of models becomes possible. Args: cfg_file_path (str): The path to the config file. _log (logging.Logger, optional): The logger instance from which to create a child logger for this model. If not given, will use the backend logger instance. """ if _log is None: _log = _backend_logger # Get the root configuration and get the instance name from there self._root_cfg = self._get_root_cfg(cfg_file_path, _log=_log) self._name = self._root_cfg["root_model_name"] _bases = (b.__name__ for b in type(self).__bases__)"Setting up model infrastructure ...")" Class name: %s", type(self).__name__)" Parent class: %s", ", ".join(_bases))" Instance name: %s", # Can now set up the loggers self._log = None self._setup_loggers(_log) # Optionally attach signal handlers for stop conditions and interrupts if self.ATTACH_SIGNAL_HANDLERS: self._signal_info = SIGNAL_INFO self._attach_signal_handlers() # Monitoring"Extracting monitoring settings ...") self._last_emit = 0 self._monitor_info = dict() self._monitor_emit_interval = self.root_cfg["monitor_emit_interval"] # Keep track of number of iterations self._n_iterations = 0 # RNG # In order to ensure that a simulation is deterministic even when not # using the model-specific RNG instance, some singleton-like default # RNGs are (by default) seeded additionally. To avoid equal random # number sequences, they are salted. self._rng = self._setup_rng( seed=self.root_cfg["seed"], seed_numpy_rng=self.root_cfg.get("seed_numpy_rng", True), seed_system_rng=self.root_cfg.get("seed_system_rng", True), ) # Create the output file self._h5file = self._setup_output_file() self._h5group = self._setup_output_group() # Allow subclasses to parse root config parameters, potentiaally adding # more attributes self._parse_root_cfg(**self.root_cfg) # Actual model initialization self._cfg = self.root_cfg[] self._invoke_setup() # First monitoring self.trigger_monitor(force=True) # Done. "Fully initialized %s named '%s'.\n", type(self).__name__,, )
[docs] def __del__(self): """Takes care of tearing down the model""" self.log.debug("Tearing down model instance ...") try: self._h5file.close() except Exception as exc: self.log.error( "Closing HDF5 file failed, got %s: %s", type(exc).__name__, exc ) self.log.debug("Teardown complete.")
# .. Properties ........................................................... @property def name(self) -> str: """Returns the name of this model instance""" return self._name @property def log(self) -> "logging.Logger": """Returns the model's logger instance""" return self._log @property def rng(self) -> "numpy.random.Generator": """Returns the shared random number generator instance""" return self._rng @property def h5group(self) -> "h5py.Group": """The HDF5 group this model should write to""" return self._h5group @property def root_cfg(self) -> dict: """Returns the root configuration of the simulation run""" return self._root_cfg @property def cfg(self) -> dict: """Returns the model configuration, ``self.root_cfg[]``""" return self._cfg @property def n_iterations(self) -> int: """Returns the number of iterations performed by this base class, i.e. the number of times :py:meth:`.iterate` was called. .. note:: This may not correspond to potentially existing other measures that specialized base classes implement. For instance, :py:meth:`utopya_backend.model.step.StepwiseModel.time` is *not* the same as the number of iterations. """ return self._n_iterations # .. Simulation control ...................................................
[docs] def run(self) -> int: """Performs a simulation run for this model, calling the :py:meth:`.iterate` method while :py:meth:`.should_iterate` is true. In addition, it takes care to invoke data writing and monitoring. Returns: int: exit code, non-zero upon handled signals """ self._pre_run() self._invoke_prolog()"Commencing model run ...") while self.should_iterate(): self._pre_iterate() self._invoke_iterate() self._n_iterations += 1 # Allow to monitor simulation progress self.trigger_monitor() # Allow writing data if self.should_write(): self._invoke_write_data() # Inform about the iteration self.show_iteration_info() # Handle signals, which may lead to a sys.exit if (exit_code := self._check_signals()) is not None: self._invoke_epilog(finished_run=False) self._post_run(finished_run=False) if self.USE_SYS_EXIT:"Now exiting (code: %d) ...", exit_code) sys.exit(exit_code) else:"Now returning (code: %d) ...", exit_code) return exit_code self._post_iterate() self._invoke_epilog(finished_run=True) self._post_run(finished_run=True)"Simulation run finished.\n") return 0
# .. Abstract methods .....................................................
[docs] @abc.abstractmethod def setup(self) -> None: """Called upon initialization of the model"""
[docs] @abc.abstractmethod def should_iterate(self) -> bool: """A method that determines whether :py:meth:`.iterate` should be called or not."""
[docs] @abc.abstractmethod def iterate(self) -> None: """Called repeatedly until the end of the simulation, which can be either due to"""
[docs] @abc.abstractmethod def should_write(self) -> bool: """A method that determines whether :py:meth:`.write_data` should be called after an iteration or not."""
[docs] @abc.abstractmethod def write_data(self) -> None: """Performs data writing if :py:meth:`.should_write` returned true."""
# .. Optionally subclassable methods ......................................
[docs] def _parse_root_cfg(self, **_) -> None: """Invoked from within :py:meth:`.__init__`, parses and handles configuration parameters. .. hint:: This method can be specialized in a subclass. """ pass
[docs] def _setup_finished(self) -> None: """Invoked from within :py:meth:`.__init__` after the call to the :py:meth:`.setup` method has finished. .. hint:: This method can be specialized in a subclass. """ pass
[docs] def monitor(self, monitor_info: dict) -> dict: """Called when a monitor emission is imminent; should be used to update the (model-specific) monitoring information passed here as arguments. .. hint:: This method can be specialized in a subclass. """ return monitor_info
[docs] def compute_progress(self) -> float: """Computes the progress of the simulation run. Should return a float between 0 and 1 and should *always* be monotonic. .. hint:: This method can be specialized in a subclass. """ return 0.0
[docs] def show_iteration_info(self) -> None: """A method that informs about the current iteration""" self.log.debug("Finished iteration %d.", self.n_iterations)
[docs] def prolog(self) -> None: """Invoked at the beginning of :py:meth:`.run`, before the first call to :py:meth:`.iterate`. .. hint:: This method can be specialized in a subclass. """ pass
[docs] def epilog(self, *, finished_run: bool) -> None: """Always invoked at the end of :py:meth:`.run`. This may happen either after :py:meth:`.should_iterate` returned False or any time before that, e.g. due to an interrupt signal or a stop condition. In the latter case, ``finished_run`` will be False. .. hint:: This method can be specialized in a subclass. """ pass
[docs] def _pre_run(self): """Invoked at beginning of :py:meth:`.run`""" pass
[docs] def _post_run(self, *, finished_run: bool) -> None: """Invoked at end of :py:meth:`.run`""" pass
[docs] def _pre_iterate(self): """Invoked at beginning of a *full* iteration""" pass
[docs] def _post_iterate(self): """Invoked at end of a *full* iteration (including monitoring, data writing etc.)""" pass
[docs] def _pre_monitor(self): """Invoked before monitor emission""" pass
[docs] def _post_monitor(self): """Invoked after monitor emission""" pass
# .. Signalling ...........................................................
[docs] def _attach_signal_handlers(self): """Invoked from :py:meth:`.__init__`, attaches a signal handler for the stop condition signal and other interrupts. .. note:: This should only be overwritten if you want or need to do your own signal handling. """"Attaching signal handlers ...") attach_signal_handlers()
[docs] def _check_signals(self) -> Union[None, int]: """Evaluates whether the iteration should stop due to an (expected) signal, e.g. from a stop condition or an interrupt. If it should stop, will return an integer, which can then be passed into :py:func:`sys.exit`. Exit codes will be ``128 + abs(signum)``, as is convention. This is also expected by :py:class:`~utopya.workermanager.WorkerManager` and is used to behave differently on a stop-condition-related signal than on an interrupt signal. Returns: Union[None, int]: An integer if the signal denoted that there should be a system exit; None otherwise. """ signal_info = self._signal_info if not signal_info["got_signal"]: return None # Received a signal signum = signal_info["signum"] if signum == SIG_STOPCOND: self.log.warning("A stop condition was fulfilled.") elif signum in (signal.SIGINT, signal.SIGTERM): self.log.warning("Was told to stop.") else: self.log.warning( "Got an unexpected signal: %d. Stopping ...", signum ) return 128 + abs(signum)
# .. Monitoring ...........................................................
[docs] def _monitor_should_emit(self, *, t: float = None) -> bool: """Evaluates whether the monitor should emit. This method will only return True once a monitor emit interval has passed since the last time the monitor was emitted. Args: t (None, optional): If given, uses this time, otherwise calls :py:func:`time.time`. Returns: bool: Whether to emit or not. """ t = t if t is not None else time.time() if t > self._last_emit + self._monitor_emit_interval: return True return False
[docs] def _emit_monitor(self): """Actually emits the monitoring information using :py:func:`print`.""" # TODO Consider using YAML for creating the monitor string def parse_val(v) -> str: if isinstance(v, float): return f"{v:6g}" return repr(v) progress = str(self.compute_progress()) monitor_info = ", ".join( f"{k}: {parse_val(v)}" for k, v in self._monitor_info.items() ) # Now emit ... # fmt: off print( "!!map { " + "progress: " + progress + ", " + "n_iter: " + str(self.n_iterations) + ", " + + ": {" + monitor_info + "}" + "}", flush=True, )
# fmt: on
[docs] def trigger_monitor(self, *, force: bool = False): """Invokes the monitoring procedure: #. Checks whether :py:meth:`._monitor_should_emit`. #. If so, calls :py:meth:`.monitor` to update monitoring information. #. Then calls :py:meth:`._emit_monitor` to emit that information. If ``force`` is given, will always emit. .. hint:: This method should not be subclassed, but it can be invoked from within the subclass at any desired point. """ t = time.time() if force or self._monitor_should_emit(t=t): self._pre_monitor() self._monitor_info = self.monitor(self._monitor_info) self._emit_monitor() self._last_emit = t self._post_monitor()
# .. Other helpers ........................................................
[docs] def _get_root_cfg( self, cfg_file_path: str, *, _log: "logging.Logger" ) -> dict: """Retrieves the root configuration for this simulation run by loading it from the given file path. """"Loading configuration file ...\n %s", cfg_file_path) return _load_cfg_file(cfg_file_path)
[docs] def _setup_loggers(self, _log: "logging.Logger"): """Sets up the model logger and configures the backend logger according to the ``log_levels`` entry set in the root configuration. .. todo:: Allow setting the logging *format* as well. Args: _log (logging.Logger): The logger to initialize the model logger from, typically the """"Setting up loggers ...") # TODO Allow changing log formatters globally # Model logger and its level self._log = _log.getChild( log_level = self.root_cfg["log_levels"]["model"] self._log.setLevel(_get_level(log_level))" Model logger initialized with '%s' level.", log_level) # May want to adjust the backend logger backend_log_level = self.root_cfg["log_levels"].get("backend") if backend_log_level is not None: _backend_logger.setLevel(_get_level(backend_log_level)) " Set backend logger's level to '%s'.", backend_log_level )
[docs] def _setup_rng( self, *, seed: int, seed_numpy_rng: Optional[Union[bool, int]] = None, seed_system_rng: Optional[Union[bool, int]] = None, **rng_kwargs, ) -> "numpy.random.Generator": """Sets up the shared RNG. .. note:: If also seeding the other RNGs, make sure to use different seeds for them, such that random number sequences are ensured to be different even if the underlying generator may be the same. Args: seed (int): The seed for the new, model-specific RNG, constructed via :py:func:`numpy.random.default_rng` seed_numpy_rng (Optional[Union[bool, int]], optional): If not False or None, will also seed numpy's singleton (i.e. *default*) RNG by calling :py:func:`numpy.random.seed`. If True, will use ``seed + 1`` for that. seed_system_rng (Optional[Union[bool, int]], optional): If not False or None, will also seed the system's default RNG by calling :py:func:`random.seed`. If True, will use ``seed + 2`` for that. **rng_kwargs: Passed on to :py:func:`numpy.random.default_rng` """"Creating shared RNG (seed: %s) ...", seed) rng = np.random.default_rng(seed, **rng_kwargs) if seed_numpy_rng not in (None, False): if seed_numpy_rng is True: seed_numpy_rng = seed + 1 np.random.seed(seed_numpy_rng) self.log.debug( " Default numpy RNG seeded with: %s", seed_numpy_rng ) if seed_system_rng not in (None, False): if seed_system_rng is True: seed_system_rng = seed + 2 random.seed(seed_system_rng) self.log.debug( " Default system RNG seeded with: %s", seed_system_rng ) return rng
[docs] def _setup_output_file(self) -> "h5py.File": """Creates the output file for this model; by default, it is a HDF5 file that is managed by a :py:class:`h5py.File` object. .. note:: This method can be subclassed to implement different output file formats. In that case, consider not using the ``_h5file`` and ``_h5group`` attributes but something else. """ "Creating HDF5 output file at:\n %s\n", self.root_cfg["output_path"], ) return h5.File(self.root_cfg["output_path"], mode="x")
[docs] def _setup_output_group(self, h5file: "h5py.File" = None) -> "h5py.Group": """Creates the group that this model's output is written to""" if h5file is None: h5file = self._h5file return h5file.create_group(
# .. Invocation wrappers ..................................................
[docs] def _invoke_iterate(self): self.iterate()
[docs] def _invoke_write_data(self): self.write_data()
[docs] def _invoke_setup(self):"Invoking model setup ...") self.setup(**self.cfg)"Model setup finished.\n") self._setup_finished()
[docs] def _invoke_prolog(self): """Helps invoking the :py:meth:`.prolog`""" self.log.debug("Invoking prolog ...") self.prolog() self.log.debug("Prolog finished.\n")
[docs] def _invoke_epilog(self, **kwargs): """Helps invoking the :py:meth:`.epilog`""" self.log.debug("Invoking epilog ...") self.epilog(**kwargs) self.log.debug("Epilog finished.\n")