Source code for pycopancore.runners.hooks

"""Register hooks for the run inside a runner."""

# This file is part of pycopancore.
#
# Copyright (C) 2016-2017 by COPAN team at Potsdam Institute for Climate
# Impact Research
#
# URL: <http://www.pik-potsdam.de/copan/software>
# Contact: core@pik-potsdam.de
# License: BSD 2-clause license

#
# Imports
#
from enum import Enum, unique

#
# Specific Error Definition
#


[docs] class HookRegistrationError(BaseException): """Dummy docstring""" # TODO: missing class docstring pass
[docs] class HooksError(BaseException): """Dummy docstring""" # TODO: missing class docstring pass
# # Definition of class _AbstractRunner #
[docs] class Hooks(object): """Class managing the hooks that are called before, during and after a run.""" # class variables _pre_hooks = {} """dict of hooks that are executed before the run keys: None or Entity class or Taxon class values: list of functions """ _mid_hooks = {} """dict of hooks that are executed every time the run is halted for a step or an event keys: None or Entity class or Taxon class values: list of functions """ _post_hooks = {} """dict of hooks that are executed after the run keys: None or Entity class or Taxon class values: list of functions """ tmp = [] # TODO: to be deleted again, just for testing purposes
[docs] @unique class Types(Enum): """Defines the three hook types: pre, mid, post""" pre = 1 mid = 2 post = 3
[docs] @classmethod def register_hook(cls, type, hook, theclass=None): """Register a hook. Parameters ========== type: HookTypes member Specifies the type of the hook. hook: function The function to be called. arguments: instance of the corresponding enitity or taxon, time when called """ assert type in cls.Types, "please give a type from {}.HookTypes".format( cls.__qualname__) cls.tmp.append(theclass) if type is cls.Types.pre: hooks = cls._pre_hooks elif type is cls.Types.mid: hooks = cls._mid_hooks elif type is cls.Types.post: hooks = cls._post_hooks else: # if the Code ends up here, there is an error in the implementation because # cls.HookTypes has been extended but there is no registration done # here raise HooksError("unknown hook type") if theclass in hooks: if hook in hooks[theclass]: raise HookRegistrationError( "already registered hook: {!r}".format(hook)) hooks[theclass].append(hook) else: hooks[theclass] = [hook]
[docs] @classmethod def unregister_hook(cls, type, hook, theclass=None, error_if_not_registered=True): """Register a hook. Parameters ========== type: HookTypes member Specifies the type of the hook. hook: function The function to be removed. Errors ====== raises HookRegistrationError when hook is not listed as registered """ assert type in cls.Types, "please give a type from {}.HookTypes".format( cls.__qualname__) cls.tmp.append(theclass) if type is cls.Types.pre: hooks = cls._pre_hooks elif type is cls.Types.mid: hooks = cls._mid_hooks elif type is cls.Types.post: hooks = cls._post_hooks else: # if the Code ends up here, there is an error in the implementation because # cls.HookTypes has been extended but there is no registration done # here raise HooksError("unknown hook type") try: if theclass not in hooks: raise ValueError("class has no hook registered") hooks[theclass].remove(hook) except ValueError: if error_if_not_registered: raise HookRegistrationError( "hook is not listed as registered, so it cannot be unregistered")
# else: ignore quietly
[docs] @classmethod def execute_hooks(cls, type, model, t): """Dummy docstring""" # TODO: Missing method docstring assert type in cls.Types, "please give a type from {}.HookTypes".format( cls.__qualname__) if type is cls.Types.pre: hooks = cls._pre_hooks elif type is cls.Types.mid: hooks = cls._mid_hooks elif type is cls.Types.post: hooks = cls._post_hooks else: # if the Code ends up here, there is an error in the implementation because # cls.HookTypes has been extended but there is no registration done # here raise HooksError("unknown hook type") if hooks: if None in hooks: # go throught all hooks that don't have an entity or taxon # associated with it for hook in hooks[None]: hook(t) # look for the correct components fitting to each hook_class # and then run the hook for each instance for component in model.entity_types + model.process_taxa: for hook_class in hooks: if issubclass(component, hook_class): for instance in component.instances: for hook in hooks[hook_class]: hook(instance, t)
[docs] def __new__(cls, *args, **kwargs): """raises an error because this class should not be instantiated""" raise HooksError("This class should not be instantiated.")