Source code for lantz.action

# -*- coding: utf-8 -*-
"""
    lantz.action
    ~~~~~~~~~~~~

    Implements the Action class to wrap driver bound methods with Lantz's
    data handling, logging, timing.

    :copyright: 2015 by Lantz Authors, see AUTHORS for more details.
    :license: BSD, see LICENSE for more details.
"""

import time
import copy
import inspect
import functools

from weakref import WeakKeyDictionary

from .processors import (Processor, FromQuantityProcessor,
                         MapProcessor, RangeProcessor)

from .feat import MISSING


def _dget(adict, instance=MISSING):
    """Helper function to get an element by key
    from a dictionary defaulting to key=MISSING
    """

    try:
        return adict[instance]
    except KeyError:
        return adict[MISSING]

def _dset(adict, value, instance=MISSING):
    """Helper function to set an element by key
    copying taken the value from MISSING.
    """
    if instance not in adict:
        adict[instance] = copy.deepcopy(adict[MISSING])

    if isinstance(adict[instance], dict):
        adict[instance].update(value)
    else:
        adict[instance] = value


[docs]class Action(object): """Wraps a Driver method with Lantz. Can be used as a decorator. Processors can registered for each arguments to modify their values before they are passed to the body of the method. Two standard processors are defined: `values` and `units` and others can be given as callables in the `procs` parameter. If a method contains multiple arguments, use a tuple. None can be used as `do not change`. :param func: driver method to be wrapped. :param values: A dictionary to values key to values. If a list/tuple instead of a dict is given, the value is not changed but only tested to belong to the container. :param units: `Quantity` or string that can be interpreted as units. :param procs: Other callables to be applied to input arguments. """ def __init__(self, func=None, *, values=None, units=None, limits=None, procs=None): #: instance: key: value self.modifiers = WeakKeyDictionary() self.action_processors = {MISSING: ()} self.modifiers[MISSING] = {'values': values, 'units': units, 'limits': limits, 'processors': procs} self.func = func self.args = () def __call__(self, func): self.func = func self.args = inspect.getfullargspec(func).args self.__name__ = func.__name__ self.__doc__ = func.__doc__ self.rebuild(store=True) return self def __get__(self, instance, owner=None): func = functools.partial(self.call, instance) func.__wrapped__ = self.func return func @property def name(self): return self.__name__ def call(self, instance, *args, **kwargs): name = self.__name__ # This part calls to the underlying function wrapping # and timing, logging and error handling with instance._lock: if args or kwargs: instance.log_info('Calling {} with ({}, {}))', name, args, kwargs) else: instance.log_info('Calling {}', name) try: values = inspect.getcallargs(self.func, *(instance, ) + args, **kwargs) fargs = self.args values = tuple(values[farg] for farg in fargs)[1:] if len(values) == 1: t_values = (self.pre_action(values[0], instance), ) else: t_values = self.pre_action(values, instance) except Exception as e: instance.log_error('While pre-processing ({}, {}) for {}: {}', args, kwargs, name, e) raise e if args or kwargs: instance.log_debug('(raw) Calling {} with {}', name, t_values) try: tic = time.time() out = self.func(instance, *t_values) instance.timing.add(name, time.time() - tic) instance.log_info('{} returned {}', name, out) return out except Exception as e: instance.log_error('While calling {} with {}. {}', name, t_values, e) raise e def pre_action(self, value, instance=None): procs = _dget(self.action_processors, instance) for processor in procs: value = processor(value) return value def rebuild(self, instance=MISSING, build_doc=False, modifiers=None, store=False): if not modifiers: modifiers = _dget(self.modifiers, instance) procs = [] largs = len(self.args) - 1 name = self.name values = modifiers['values'] units = modifiers['units'] limits = modifiers['limits'] processors = modifiers['processors'] if values: proc = MapProcessor(values) lproc = len(proc) if isinstance(proc, Processor) else 1 if lproc != largs: raise ValueError("In {}: the number of elements in 'values' ({}) " "must match the number of arguments ({})".format(name, lproc, largs)) procs.append(proc) if units: proc = FromQuantityProcessor(units) lproc = len(proc) if isinstance(proc, Processor) else 1 if lproc != largs: raise ValueError("In {}: the number of elements in 'units' ({}) " "must match the number of arguments ({})".format(name, lproc, largs)) procs.append(proc) if limits: if isinstance(limits[0], (list, tuple)): proc = RangeProcessor(limits) else: proc = RangeProcessor((limits, )) lproc = len(proc) if isinstance(proc, Processor) else 1 if lproc != largs: raise ValueError("In {}: the number of elements in 'limits' ({}) " "must match the number of arguments ({})".format(name, lproc, largs)) procs.append(proc) if processors: for processor in processors: proc = Processor(processor) lproc = len(proc) if isinstance(proc, Processor) else 1 if lproc != largs: raise ValueError("In {}: the number of elements in 'processor' ({}) " "must match the number of arguments ({})".format(name, len(proc), largs)) procs.append(proc) if store: _dset(self.action_processors, procs, instance) return procs
class ActionProxy(object): """Proxy object for Actions that allows to store instance specific modifiers. """ def __init__(self, instance, action): super().__setattr__('instance', instance) super().__setattr__('action', action) def __getattr__(self, item): modifiers = _dget(self.action.modifiers, self.instance) if item not in modifiers: raise AttributeError() return modifiers[item] def __setattr__(self, item, value): _modifiers = _dget(self.action.modifiers, MISSING) if item not in _modifiers: raise AttributeError() _dset(self.action.modifiers, {item: value}, self.instance) self.action.rebuild(self.instance, build_doc=False, store=True)