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
BaseModel
BaseModel.ATTACH_SIGNAL_HANDLERS
BaseModel.USE_SYS_EXIT
BaseModel.__init__()
BaseModel.__del__()
BaseModel.name
BaseModel.log
BaseModel.rng
BaseModel.h5group
BaseModel.root_cfg
BaseModel.cfg
BaseModel.n_iterations
BaseModel.run()
BaseModel.setup()
BaseModel.should_iterate()
BaseModel.iterate()
BaseModel.should_write()
BaseModel.write_data()
BaseModel._parse_root_cfg()
BaseModel._setup_finished()
BaseModel.monitor()
BaseModel.compute_progress()
BaseModel.show_iteration_info()
BaseModel.prolog()
BaseModel.epilog()
BaseModel._pre_run()
BaseModel._post_run()
BaseModel._pre_iterate()
BaseModel._post_iterate()
BaseModel._pre_monitor()
BaseModel._post_monitor()
BaseModel._attach_signal_handlers()
BaseModel._check_signals()
BaseModel._monitor_should_emit()
BaseModel._emit_monitor()
BaseModel.trigger_monitor()
BaseModel._get_root_cfg()
BaseModel._setup_loggers()
BaseModel._setup_rng()
BaseModel._setup_output_file()
BaseModel._setup_output_group()
BaseModel._invoke_iterate()
BaseModel._invoke_write_data()
BaseModel._invoke_setup()
BaseModel._abc_impl
BaseModel._invoke_prolog()
BaseModel._invoke_epilog()
- utopya_backend.model.step module
StepwiseModel
StepwiseModel.should_iterate()
StepwiseModel.iterate()
StepwiseModel.should_write()
StepwiseModel.perform_step()
StepwiseModel._parse_root_cfg()
StepwiseModel._setup_finished()
StepwiseModel.compute_progress()
StepwiseModel.show_iteration_info()
StepwiseModel._invoke_epilog()
StepwiseModel.time
StepwiseModel.num_steps
StepwiseModel.write_start
StepwiseModel.write_every
StepwiseModel.create_ts_dset()
StepwiseModel._abc_impl
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.
- _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.
- 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()
andstop_timer()
when benchmarking is completely disabled.
- _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}')[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()
. Available keys:name
,seconds
(float),time_str
(pre-formatted usingdantro.tools.format_time()
).
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.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)[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
andSIGINT
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:
- 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 yourModel
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:
- Returns:
The imported module.
- Return type:
ModuleType
- Raises:
ImportError – If
debug
is set and import failed for whatever reasonFileNotFoundError – If
mod_dir
did not point to an existing directory