utopya_backend package

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#

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: Callable | None = None, one_shot: bool = False, start: bool = False)[source]#

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[source]#

_assert_not_finished()[source]#

start()[source]#

pause() → float[source]#

unpause()[source]#

_unpause()[source]#

stop() → float[source]#

reset()[source]#

property name: str#

property running: bool#

property finished: bool#

property elapsed: float#

property one_shot: bool#

class utopya_backend.benchmark.ModelBenchmarkMixin(*args, **kwargs)[source]#

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: Dataset | None = None#

_dset_cumulative: Dataset | None = 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, Timer]#

_dgrp_bench: Group | None = None#

_add_default_timers()[source]#

_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 = ' {time_str:>13s} {percent_of_max:5.1f}% {name:s}', info_kwargs: dict | None = None)[source]#

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() and as default when invoking format_elapsed_info(). Available keys: name, seconds (float), time_str (pre-formatted using dantro.tools.format_time()), w (width of longest name), max_seconds (float, largest timer value), and percent_of_max (float, relative timer value in percent compared to max_seconds).
info_kwargs (dict, optional) – additional arguments to configure how elapsed_info() formats timer information. This is not used when calling format_elapsed_info().

add_one_shot_timers(*names, **kwargs)[source]#

add_cumulative_timers(*names, **kwargs)[source]#

_add_timer(name, *, one_shot: bool, **kwargs)[source]#

property timers: Dict[str, Timer]#

_get_timer(name: str) → Timer[source]#

start_timer(name: str) → None[source]#

pause_timer(name: str) → float | Any[source]#

unpause_timer(name: str) → None[source]#

stop_timer(name: str) → float | Any[source]#

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.

Calls format_elapsed_info() with the info_fstr and info_kwargs arguments specified at initialization.

format_elapsed_info(info_fstr: str | None = None, *, sort: bool = False, width: int | None = None, ms_precision: int = 2, **fstr_kwargs) → str[source]#

Prepares a formatted string with all elapsed times.

Parameters:

info_fstr (str, optional) – The format string to use. If not given, will use the default info_fstr specified at initialization. Available keys: name, seconds (float), time_str (pre-formatted using dantro.tools.format_time()), w (width of longest name), max_seconds (float, largest timer value), and percent_of_max (float, relative timer value in percent compared to max_seconds).
sort (bool, optional) – Whether to sort timers (in descending order) before formatting them.
width (int, optional) – If given, will use this fixed value for the w format key instead of the maximum width.
ms_precision (int, optional) – How many digits precision to use for millisecond time intervals.
**fstr_kwargs – Passed on to the format operation.

_write_dset_total()[source]#

_write_dset_cumulative()[source]#

_invoke_setup()[source]#

_pre_run()[source]#

_post_run(*, finished_run: bool)[source]#

_invoke_prolog()[source]#

_pre_iterate()[source]#

_invoke_iterate()[source]#

_pre_monitor()[source]#

_emit_monitor()[source]#

_post_monitor()[source]#

_invoke_write_data()[source]#

_post_iterate()[source]#

_invoke_epilog(**kwargs)[source]#

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[source]#

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: int = Signals.SIGUSR1#: Which signal to look out for if a stop condition was fulfilled. This should match utopya.stop_conditions.SIG_STOPCOND.

Note

On Windows, without proper user-defined signals available, will hijack the floating-point-error signal for this purpose.

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)[source]#: 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)[source]#

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: str | None = None) → Any[source]#

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: str | None = None) → ModuleType[source]#

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

utopya_backend package

Contents

utopya_backend package#

Subpackages#

Submodules#

utopya_backend.benchmark module#

utopya_backend.logging module#

utopya_backend.signal module#

utopya_backend.tools module#