utopya_backend package

The utopya_backend package is a standalone package that provides tools for implementation of Python-based models that use utopya as a simulation management frontend.

Refer to the the user manual for more information.

Subpackages

• utopya_backend.io package
• utopya_backend.model package
  • Submodules
  • utopya_backend.model.base module
  • utopya_backend.model.step module

Submodules

utopya_backend.benchmark module

Benchmarking tools for Models

utopya_backend.benchmark.DEFAULT_TIMERS_ONE_SHOT: Tuple[str, ...] = ('init', 'setup', 'prolog', 'run', 'epilog', 'teardown', 'simulation')

Names of default one-shot timers in ModelBenchmarkMixin

utopya_backend.benchmark.DEFAULT_TIMERS_CUMULATIVE: Tuple[str, ...] = ('model_iteration', 'monitor', 'write_data', 'full_iteration')

Names of default cumulative timers in ModelBenchmarkMixin

class utopya_backend.benchmark.Timer(name: str, *, time_func: Optional[Callable] = None, one_shot: bool = False, start: bool = False)

Bases: object

Implements a simple timer that can be paused and continued.

_running: bool

_latest: float

_elapsed: float

_finished: bool

_name: str

_one_shot: bool

_time_func()

time() -> floating point number

Return the current time in seconds since the Epoch. Fractions of a second may be present if the system clock provides them.

_get_time() → float

_assert_not_finished()

start()

pause() → float

unpause()

_unpause()

stop() → float

reset()

property name: str

property running: bool

property finished: bool

property elapsed: float

property one_shot: bool

class utopya_backend.benchmark.ModelBenchmarkMixin(*args, **kwargs)

Bases: object

A mixin class that allows to conveniently gather information on the run time that individual parts of the model iteration require and also store it in the model’s dataset.

To use this, simply inherit it into your model class definition:

from utopya_backend import BaseModel, ModelBenchmarkMixin

class MyModel(ModelBenchmarkMixin, BaseModel):
    pass

By default, this will enable the benchmarking and will both show the result at the end of the run as well as write it to a separate benchmarking group in the default HDF5 data group. To further configure its behaviour, add a benchmark entry to your model’s configuration. For available parameters and default values, refer to _configure_benchmark().

_TIMER_FALLBACK_RV: Any = -1

The fallback value that is returned by pause_timer() and stop_timer() when benchmarking is completely disabled.

_dset_total: Optional[h5py._hl.dataset.Dataset] = None

_dset_cumulative: Optional[h5py._hl.dataset.Dataset] = None

__dgrp_name: str

__dset_dtype: str

__dset_compression: int

_dset_cumulative_invocation_times: List[int]

__enabled: bool = True

__write: bool = None

_show_on_exit: bool = True

_add_time_elapsed_to_monitor_info: bool = False

_time_elapsed_info_fstr: str

_timers: Dict[str, utopya_backend.benchmark.Timer]

_dgrp_bench: Optional[h5py._hl.group.Group] = None

_add_default_timers()

_configure_benchmark(*, enabled: bool = True, show_on_exit: bool = True, add_to_monitor: bool = False, write: bool = True, group_name: str = 'benchmark', compression: int = 3, dtype: str = 'float32', info_fstr: str = '  {name:>15s} : {time_str:s}')

Applies benchmark configuration parameters.

Parameters

• enabled (bool, optional) –

  Whether to enable benchmarking. If False, the behaviour will be exactly the same, but timer invocations will simply be ignored.

  Note

  Despite being disabled, a very minor performance hit can still be expected (a few booleans that are evaluated). Only removing the mixin altogether will alleviate that.

• show_on_exit (bool, optional) – Whether to print an info-level log message at the end of the simulation, showing elapsed times.

• add_to_monitor (bool, optional) – Whether to add elapsed times to the monitoring data.

• write (bool, optional) – Whether to write data to HDF5 dataset. The cumulative timers are stored at each invocation of write_data, while the one-shot timers are only written at the end of a simulation run.

• group_name (str, optional) – The name of the HDF5 group to nest the output datasets in.

• compression (int, optional) – HDF5 compression level.

• dtype (str, optional) – HDF5 data type for timing information. By default, this is reduced float precision, because the times given by time.time() are not as precise anyway.

• info_fstr (str, optional) – The format string to use for generation of elapsed_info(). Available keys: name, seconds (float), time_str (pre-formatted using dantro.tools.format_time()).

add_one_shot_timers(*names, **kwargs)

add_cumulative_timers(*names, **kwargs)

_add_timer(name, *, one_shot: bool, **kwargs)

property timers: Dict[str, utopya_backend.benchmark.Timer]

_get_timer(name: str) → utopya_backend.benchmark.Timer

start_timer(name: str) → None

pause_timer(name: str) → Union[float, Any]

unpause_timer(name: str) → None

stop_timer(name: str) → Union[float, Any]

property elapsed: Dict[str, float]

property elapsed_cumulative: Dict[str, float]

property elapsed_one_shot: Dict[str, float]

property elapsed_info: str

Prepares a formatted string with all elapsed times

_write_dset_total()

_write_dset_cumulative()

_invoke_setup()

_pre_run()

_post_run(*, finished_run: bool)

_invoke_prolog()

_pre_iterate()

_invoke_iterate()

_pre_monitor()

_emit_monitor()

_post_monitor()

_invoke_write_data()

_post_iterate()

_invoke_epilog(**kwargs)

utopya_backend.logging module

Implements logging-related infrastructure

utopya_backend.logging.LOG_LEVELS: Dict[str, int] = {'critical': 50, 'debug': 10, 'error': 40, 'fatal': 50, 'info': 20, 'none': 0, 'not_set': 0, 'notset': 0, 'trace': 5, 'warn': 30, 'warning': 30}

A map of log level names to actual level values

utopya_backend.logging.DEFAULT_LOG_FORMAT = '%(levelname)-7s %(message)s'

The default logging format to use; can also include %(name)-14s here to show the logger’s name.

utopya_backend.logging.backend_logger = <RootLogger root (INFO)>

A backend-wide logger instance which is the same as the root logger.

Note

The BaseModel may adjust the level of this logger.

utopya_backend.logging.get_level(s: str) → int

Returns the integer log level from a string, looking it up in LOG_LEVELS.

Parameters

s (str) – Name of the log level, not case-sensitive.

Returns

The desired log level

Return type

int

utopya_backend.signal module

Signal handling when using the utopya backend, e.g. for catching stop conditions and handling them gracefully.

utopya_backend.signal.SIG_STOPCOND = Signals.SIGUSR1

Which signal to look out for if a stop condition was fulfilled. This should match utopya.stop_conditions.SIG_STOPCOND.

utopya_backend.signal.SIGNAL_INFO = {'at_time': None, 'frame': None, 'got_signal': False, 'signum': None}

A dict that holds information on whether any kind of signal was received and at which time. This can be analysed by other modules to determine which action to take.

utopya_backend.signal._handle_signal(signum, frame)

A signal handler function that writes information into the SIGNAL_INFO dict.

utopya_backend.signal.attach_signal_handlers(*, for_stop_conds: bool = True, for_interrupts: bool = True)

A function that can be invoked to attach signal handlers for use within utopya. There are two kinds of signals:

• stop conditions, as implemented in utopya.stop_conditions

• interrupt signals SIGTERM and SIGINT

Parameters

• for_stop_conds (bool, optional) – Whether to attach signal handlers for stop conditions.

• for_interrupts (bool, optional) – Whether to attach signal handlers for interrupts.

utopya_backend.tools module

This module implements various generic tools

utopya_backend.tools.load_cfg_file(fpath: str, *, loader: Optional[str] = None) → Any

Loads a configuration file from the given file. Allows to automatically determine which kind of loading function to use.

Currently supported loading functions: YAML

Parameters

• fpath (str) – Path to the configuration file to load

• loader (str, optional) – Name of the loader to use. If not given, will determine it from the file extension of fpath.

Returns

The return value of the load function.

Return type

Any

Raises

ValueError – On invalid loader argument or a file extension that does not map to a supported loader.

utopya_backend.tools.import_package_from_dir(mod_dir: str, *, mod_str: Optional[str] = None) → module

Helper function to import a package-like module that is importable only when adding the module’s parent directory to sys.path.

The mod_dir directory needs to contain an __init__.py file. If that is not the case, you cannot use this function, because the directory does not represent a package.

Hint

This function is very useful to get access to a local package that is not installed, as might be the case for your model implementation. Assuming you have an impl package right beside the current __file__ and that package includes your Model class implementation:

- run_model.py         # Current __file__
- impl/                # Implementation package
  |-- __init__.py      # Exposes impl.model.Model
  |-- model.py         # Implements Model class
  |-- ...

You can get access to it like this from within run_model.py:

import os
from utopya_backend import import_package_from_dir

impl = import_package_from_dir(
    os.path.join(os.path.dirname(__file__), "impl")
)
Model = impl.Model

Parameters

• mod_dir (str) – Path to the module’s root directory, ~ expanded. For robustness, relative paths are not allowed.

• mod_str (str, optional) – Name under which the module can be imported with the parent of mod_dir being in sys.path. If not given, will assume it is equal to the last segment of mod_dir.

Returns

The imported module.

Return type

ModuleType

Raises

• ImportError – If debug is set and import failed for whatever reason

• FileNotFoundError – If mod_dir did not point to an existing directory