Source code for pyretis.simulation.simulation_task

# -*- coding: utf-8 -*-
# Copyright (c) 2023, PyRETIS Development Team.
# Distributed under the LGPLv2.1+ License. See LICENSE for more info.
"""Definition of a class for simulation tasks.

Important classes defined here
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

SimulationTask (:py:class:`.SimulationTask`)
    A class representing a simulation task.

SimulationTaskList (:py:class:`.SimulationTaskList`)
    A class for representing a list of simulation tasks. This class
    defines functionality for adding tasks from a dictionary description.

"""
import logging
from pyretis.core.common import inspect_function
from pyretis.inout.simulationio import Task
logger = logging.getLogger(__name__)  # pylint: disable=invalid-name
logger.addHandler(logging.NullHandler())


[docs]def _check_args(function, given_args=None, given_kwargs=None): """Check consistency for function and the given (keyword) arguments. Here we assume that the arguments are given in a list and that the keyword arguments are given as a dictionary. The function `inspect.getargspec` is used to check the input function. Parameters ---------- function : callable The function we will inspect. given_args : list, optional A list of the arguments to pass to the function. 'self' will not be considered here since it passed implicitly. given_kwargs : dict, optional A dictionary with keyword arguments. Returns ------- out : boolean False if there is some inconsistencies, i.e. when the calling of the given `function` will probably fail. True otherwise. """ arguments = inspect_function(function) args = [arg for arg in arguments['args'] if arg != 'self'] defaults = list(arguments['kwargs']) # first test, do we give correct number of required arguments? if given_args is not None: given = len(given_args) else: given = 0 if len(args) != given: msgtxt = 'Wrong number of arguments given' logger.warning(msgtxt) return False # Check kwargs but only check in case some kwargs are given here. # If they are not given, we assume that the user knows what's happening # and that the default kwargs will be used. if given_kwargs is not None: if defaults: extra = [key for key in given_kwargs if key not in defaults] if extra: msg = [f'Task Keyword arguments: {defaults}'] msg += [f'Unexpected keyword argument: {extra}'] msgtxt = '\n'.join(msg) logger.warning(msgtxt) return False else: msgtxt = 'Unexpected keyword argument!' logger.warning(msgtxt) return False return True
[docs]class SimulationTask(Task): """Representation of simulation tasks. This class defines a task object. A task is executed at specific points, at regular intervals etc. in a simulation. A task will typically provide a result, but it does not need to. It can simply just alter the state of the passed argument(s). Attributes ---------- function : function The function to execute. when : dict Determines when the task should be executed. args : list List of arguments to the function. kwargs : dict The keyword arguments to the function. first : boolean True if this task should be executed before the first step of the simulation. result : string This is a label for the result created by the task. """
[docs] def __init__(self, function, args=None, kwargs=None, when=None, result=None, first=False): """Initialise the task. Parameters ---------- function : callable The function to execute. args : list, optional List of arguments to the function. kwargs : dict, optional The keyword arguments to the function. when : dict, optional Determines if the task should be executed. result : string, optional This is a label for the result created by the task. first : boolean, optional True if this task should be executed before the first step of the simulation. """ if not callable(function): msg = 'The given function for the task is not callable!' raise AssertionError(msg) ok_to_add = _check_args(function, given_args=args, given_kwargs=kwargs) if not ok_to_add: msg = 'Wrong arguments or keyword arguments!' raise AssertionError(msg) super().__init__(when) self.function = function self.args = args self.kwargs = kwargs self._result = result self.first = first
[docs] def execute(self, step): """Execute the task. Parameters ---------- step : dict of ints The keys are: * 'step': the current cycle number. * 'startcycle': the cycle number at the start. * 'stepno': the number of cycles we have performed so far. Returns ------- out : unknown type The result of running `self.function`. """ args = self.args kwargs = self.kwargs if self.execute_now(step): if args is None: if kwargs is None: return self.function() return self.function(**kwargs) if kwargs is None: return self.function(*args) return self.function(*args, **kwargs) return None
@property def result(self): """Return the result label.""" return self._result
[docs] def run_first(self): """Return True if task should be executed before first step.""" return self.first
[docs] def task_dict(self): """Return a dict representing the task.""" return {'func': self.function, 'args': self.args, 'kwargs': self.kwargs, 'when': self.when, 'result': self.result, 'first': self.first, 'func-name': self.function.__name__}
[docs] def __call__(self, step): """Execute the task. Parameters ---------- step : dict of ints The keys are: * 'step': the current cycle number. * 'startcycle': the cycle number at the start. * 'stepno': the number of cycles we have performed so far. Returns ------- out : unknown type The result of `self.execute(step)`. """ return self.execute(step)
[docs] def __str__(self): """Output info about the task.""" msg = ['Task:'] msg += [f' -> Function name: {self.function.__name__}'] msg += [f' -> Function args: {self.args}'] msg += [f' -> Function kwargs: {self.kwargs}'] msg += [f' -> Execute when: {self.when}'] msg += [f' -> Execute at first: {self.first}'] msg += [f' -> Result: {self._result}'] return '\n'.join(msg)