so_magic.data.features.phi

Complexity

Total Complexity

14

Size/Duplication

Total Lines	184
Duplicated Lines	0 %

Importance

Changes

0

4 Methods

Rating	Name	Duplication	Size	Complexity
A	PhiFunctionMetaclass.__new__()	0	9	1
B	PhiFunctionRegistrator.get_name()	0	19	6
B	PhiFunctionRegistrator.register()	0	100	6
A	PhiFunctionRegistrator._register()	0	14	1

"""This module is responsible to provide a formal way of registering phi functions
at runtime. See the 'PhiFunctionRegistrator' class and its 'register' decorator method
"""
import logging
import inspect
from typing import Callable
from so_magic.utils import Subject

logger = logging.getLogger(__name__)


class PhiFunctionMetaclass(type):
    """Class type with a single broadcasting (notifies listeners) facility.

    Classes using this class as metaclass, obtain a single broadcasting facility
    as a class attribute. The class attribute is called 'subject' can be referenced
    as any class attribute.

    Example:
        class MyExampleClass(metaclass=PhiFunctionMetaclass):
            pass

        instance_object_1 = MyExampleClass()
        instance_object_2 = MyExampleClass()
        assert id(MyExampleClass.subject) == id(instance_object_1.subject) == id(instance_object_2.subject)
    """
    def __new__(mcs, *args, **kwargs):
        """Create a new class type object and set the 'subject' attribute to a new Subject instance; the broadcaster.

        Returns:
            PhiFunctionMetaclass: the new class type object
        """
        phi_function_class = super().__new__(mcs, *args, **kwargs)
        phi_function_class.subject = Subject([])
        return phi_function_class


class PhiFunctionRegistrator(metaclass=PhiFunctionMetaclass):
    """Add phi functions to the registry and notify observers/listeners.

    This class provides the 'register' decorator, that client can use to decorate either functions (defined with the
    def python special word), or classes (defined with the python class special word).
    """

    # NICE TO HAVE: make the decorator work without parenthesis
    @classmethod
    def register(cls, phi_name=''):
        r"""Add a new phi function to phi function registry and notify listeners/observers.

        Use this decorator around either a callable function (defined with the 'def' python special word) or a class
        with a takes-no-arguments (or all-optional-arguments) constructor and a __call__ magic method.

        All phi functions are expected to be registered with a __name__ and a __doc__ attribute.

        You can select your custom phi_name under which to register the phi function or default to an automatic
        determination of the phi_name to use.

        Automatic determination of phi_name is done by examining either the __name__ attribute of the function or the
        class name of the class.

        Example:

            >>> from so_magic.data.features.phi import PhiFunctionRegistrator
            >>> from so_magic.utils import Observer, ObjectRegistry

            >>> class PhiFunctionRegistry(Observer):
            ...  def __init__(self):
            ...   self.registry = ObjectRegistry()
            ...  def update(self, subject, *args, **kwargs):
            ...   self.registry.add(subject.name, subject.state)

            >>> phis = PhiFunctionRegistry()

            >>> PhiFunctionRegistrator.subject.add(phis)

            >>> @PhiFunctionRegistrator.register()
            ... def f1(x):
            ...  '''Multiply by 2.'''
            ...  return x * 2
            Registering input function f1 as phi function, at key f1.

            >>> phis.registry.get('f1').__doc__
            'Multiply by 2.'

            >>> input_value = 5
            >>> print(f"{input_value} * 2 = {phis.registry.get('f1')(input_value)}")
            5 * 2 = 10

            >>> @PhiFunctionRegistrator.register()
            ... class f2:
            ...  def __call__(self, data, **kwargs):
            ...   return data + 5
            Registering input class f2 instance as phi function, at key f2.

            >>> input_value = 1
            >>> print(f"{input_value} + 5 = {phis.registry.get('f2')(input_value)}")
            1 + 5 = 6

            >>> @PhiFunctionRegistrator.register('f3')
            ... class MyCustomClass:
            ...  def __call__(self, data, **kwargs):
            ...   return data + 1
            Registering input class MyCustomClass instance as phi function, at key f3.

            >>> input_value = 3
            >>> print(f"{input_value} + 1 = {phis.registry.get('f3')(input_value)}")
            3 + 1 = 4

        Args:
            phi_name (str, optional): custom name to register the phi function. Defaults to automatic computation.
        """
        def wrapper(a_callable):
            """Add a callable object to the phi function registry and preserve info for __name__ and __doc__ attributes.

            The callable object should either be function (defined with def) or a class (defined with class). In case of
            a class the class must have a constructor that takes no arguments (or all arguments are optional) and a
            __call__ magic method.

            Registers the callable as a phi function under the given or automatically computed name, makes sure the
            __name__ and __doc__ attributes preserve information and notifies potential listeners/observers.

            Args:
                a_callable (Callable): the object (function or class) to register as phi function
            """
            if hasattr(a_callable, '__code__'):  # it is a function (def func_name ..)
                logging.info("Registering input function %s as phi function.", a_callable.__code__.co_name)
                key = phi_name if phi_name else cls.get_name(a_callable)
                print(f"Registering input function {a_callable.__code__.co_name} as phi function, at key {key}.")
                cls._register(a_callable, key)
            else:
                if not hasattr(a_callable, '__call__'):
                    raise RuntimeError("Expected an class definition with a '__call__' instance method defined 1."
                                       f" Got {type(a_callable)}")
                members = inspect.getmembers(a_callable)
                if ('__call__', a_callable.__call__) not in members:
                    raise RuntimeError("Expected an class definition with a '__call__' instance method defined 2."
                                       f" Got {type(a_callable)}")
                instance = a_callable()
                instance.__name__ = a_callable.__name__
                instance.__doc__ = a_callable.__call__.__doc__
                key = phi_name if phi_name else cls.get_name(instance)
                print(f"Registering input class {a_callable.__name__} instance as phi function, at key {key}.")
                cls._register(instance, key)
            return a_callable
        return wrapper

    @classmethod
    def _register(cls, a_callable, key_name):
        """Register a callable as phi function and notify potential listeners/observers.

        The phi function is registered under the given key_name or in case of None the name is automatically computed
        based on the input callable.

        Args:
            a_callable (Callable): the callable that holds the business logic of the phi function
            key_name (str, optional): custom phi name. Defaults to None, which means automatic determination of the name
        """
        cls.subject.name = key_name
        cls.subject.state = a_callable
        cls.subject.notify()

    @staticmethod
    def get_name(a_callable: Callable):
        """Get the 'name' of the input callable object

        Args:
            a_callable (Callable): a callable object to get its name

        Returns:
            str: the name of the callable object
        """
        if hasattr(a_callable, 'name'):
            return a_callable.name
        if hasattr(a_callable, '__code__') and hasattr(a_callable.__code__, 'co_name'):
            return a_callable.__code__.co_name
        if hasattr(type(a_callable), 'name'):
            return type(a_callable).name
        if hasattr(type(a_callable), '__name__'):
            return type(a_callable).__name__
        raise PhiFunctionNameDeterminationError()


class PhiFunctionNameDeterminationError(Exception): pass