so_magic.data.command_factories.CommandFactory.pick() - Code Metrics - Inspection of "rename package folder to so_magic" - boromir674/so-magic - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — dev ( 3b058b...d73523 )

by Konstantinos

created 2020-09-13 11:01 UTC

CommandFactory.pick() A

↳ Parent: so_magic.data.command_factories

Complexity

Conditions

Size

Total Lines	6
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	6
nop	3
dl	0
loc	6
rs	10
c	0
b	0
f	0

from abc import ABC, abstractmethod
import attr
from so_magic.utils import Command, Subject


class MyDecorator(type):
    """Metaclass that provides a decorator able to be invoked both with and without parenthesis.
    The wrapper function logic should be implemented by the client code.
    """
    @classmethod
    def magic_decorator(mcs, arg=None):
        def decorator(func):
            def wrapper(*a, **ka):
                ffunc = a[0]
                mcs._wrapper(ffunc, *a[1:], **ka)
                return ffunc
            return wrapper

        if callable(arg):
            _ = decorator(arg)
            return _  # return 'wrapper'
        else:
            _ = decorator
            return _  # ... or 'decorator'


class CommandRegistrator(MyDecorator):
    """Classes can use this class as metaclass to obtain a single registration point accessible as class attribute
    """
    def __new__(mcs, *args, **kwargs):
        class_object = super().__new__(mcs, *args, **kwargs)
        class_object.state = None
        class_object.registry = {}
        return class_object

    def __getitem__(self, item):
        if item not in self.registry:
            raise RuntimeError(f"Key '{item}' fot found in registry: [{', '.join(str(x) for x in self.registry.keys())}]")
        return self.registry[item]

    def func_decorator(cls):
        def wrapper(a_callable):
            if hasattr(a_callable, '__code__'):  # it a function (def func_name ..)
                cls.registry[a_callable.__code__.co_name] = a_callable
            else:
                raise RuntimeError(f"Expected a function to be decorated; got {type(a_callable)}")
            return a_callable
        return wrapper

class AbstractCommandFactory(ABC):
    @abstractmethod
    def construct(self, *args, **kwargs) -> Command:
        raise NotImplementedError

class BaseCommandFactory(AbstractCommandFactory, ABC):


    subclasses = {}

    @classmethod
    def register_as_subclass(cls, factory_type):
        def wrapper(subclass):
            cls.subclasses[factory_type] = subclass
            return subclass
        return wrapper

    @classmethod
    def create(cls, factory_type, *args, **kwargs):
        if factory_type not in cls.subclasses:
            raise ValueError('Bad "Factory type" \'{}\''.format(factory_type))
        return cls.subclasses[factory_type](*args, **kwargs)


@BaseCommandFactory.register_as_subclass('generic')
class GenericCommandFactory(AbstractCommandFactory):
    def construct(self, *args, **kwargs) -> Command:
        return Command(*args, **kwargs)

@BaseCommandFactory.register_as_subclass('function')
class FunctionCommandFactory(AbstractCommandFactory):
    def construct(self, *args, **kwargs) -> Command:
        if len(args) < 1:
            raise RuntimeError("Will break")
        return Command(args[0], '__call__', *args[1:])


@BaseCommandFactory.register_as_subclass('encode_nominal_subsets')
class NominalAttributeListEncodeCommandFactory(AbstractCommandFactory):

    def construct(self, *args, **kwargs) -> Command:
        from so_magic.data.features.phis import ListOfCategoricalPhi, DatapointsAttributePhi
        assert len(args) > 0
        datapoints = args[0]
        attribute = args[1]
        new_attribute = args[2]
        def _command(_datapoints, _attribute, _new_attribute):
            phi = ListOfCategoricalPhi(DatapointsAttributePhi(_datapoints))
            new_values = phi(_attribute)
            _datapoints.mutator.add_column(_datapoints, new_values, _new_attribute)
        return Command(_command, '__call__', datapoints, attribute, new_attribute)


class CommandFactory:
    """A factory class able to construct new command objects."""
    constructors = {k: v().construct for k, v in BaseCommandFactory.subclasses.items()}
    @classmethod
    def pick(cls, *args, **kwargs):
        decision = {True: 'function', False: 'generic'}
        is_function = hasattr(args[0], '__code__')
        dec2 = {'function': lambda x: x[0].__code__.co_name, 'generic': lambda x: type(x[0]).__name__ + '-' + x[1]}
        return decision[is_function], kwargs.get('name', dec2[decision[is_function]](args))

    @classmethod
    def create(cls, *args, **kwargs) -> Command:
        """Call to create a new Command object. The input arguments can be in two formats:

        1. create(an_object, method, *arguments)
        In this case the command is of the form an_object.method(*arguments)

        2. create(a_function, *arguments)
        In this case the command is of the form a_function(*arguments)

        Returns:
            Command: an instance of a command object
        """
        key, name = cls.pick(*args, **kwargs)
        if len(args) < 1:
            raise RuntimeError(args)
        return cls.constructors[key](*args), name


@attr.s
class MagicCommandFactory(Subject):
    """Instances of this class act as callable command factories that notify,
    subscribed observers/listeners upon new command object creation.

    Args:
        command_factory (CommandFactory, optional): an instance of a CommandFActory
    """
    command_factory = attr.ib(init=True, default=CommandFactory())

    def __call__(self, *args, **kwargs):
        self._state, self.name = self.command_factory.create(*args, **kwargs)
        self.notify()
        return self._state


class DataManagerCommandFactory(AbstractCommandFactory, ABC):


    subclasses = {}

    @classmethod
    def register_as_subclass(cls, factory_type):
        def wrapper(subclass):
            cls.subclasses[factory_type] = subclass
            return subclass
        return wrapper

    @classmethod
    def create(cls, factory_type, *args, **kwargs):
        if factory_type not in cls.subclasses:
            raise ValueError('Bad "Factory type" \'{}\''.format(factory_type))
        return cls.subclasses[factory_type](*args, **kwargs)

#### Register command factories
@DataManagerCommandFactory.register_as_subclass('select_variables')
class SelectVariablesCommandFactory(DataManagerCommandFactory):

    def construct(self, *args, **kwargs) -> Command:
        def command(variables):
            args[0].feature_manager.feature_configuration = variables
        return Command(command, '__call__', *args[1:])


@DataManagerCommandFactory.register_as_subclass('select_variables')
class SelectVariablesCommandFactory(DataManagerCommandFactory):

    def construct(self, *args, **kwargs) -> Command:
        def command(variables):
            args[0].feature_manager.feature_configuration = variables
        return Command(command, '__call__', *args[1:])

#################

@attr.s
class MegaCommandFactory(Subject):
    _data_manager = attr.ib(init=True)
    command_factory = attr.ib(init=True, default=DataManagerCommandFactory)

    def __call__(self, command_type, *args, **kwargs):
        self._state, self.name = self.command_factory.create(command_type).construct(self._data_manager, *args, **kwargs), command_type
        self.notify()
        return self._state


1		from abc import ABC, abstractmethod
2		import attr
3		from so_magic.utils import Command, Subject
4
5
6		class MyDecorator(type):
7		"""Metaclass that provides a decorator able to be invoked both with and without parenthesis.
8		The wrapper function logic should be implemented by the client code.
9		"""
10		@classmethod
11		def magic_decorator(mcs, arg=None):
12		def decorator(func):
13		def wrapper(a, *ka):
14		ffunc = a[0]
15		mcs._wrapper(ffunc, a[1:], *ka)
16		return ffunc
17		return wrapper
18
19		if callable(arg):
20		_ = decorator(arg)
21		return _ # return 'wrapper'
22		else:
23		_ = decorator
24		return _ # ... or 'decorator'
25
26
27		class CommandRegistrator(MyDecorator):
28		"""Classes can use this class as metaclass to obtain a single registration point accessible as class attribute
29		"""
30		def __new__(mcs, args, *kwargs):
31		class_object = super().__new__(mcs, args, *kwargs)
32		class_object.state = None
33		class_object.registry = {}
34		return class_object
35
36		def __getitem__(self, item):
37		if item not in self.registry:
38		raise RuntimeError(f"Key '{item}' fot found in registry: [{', '.join(str(x) for x in self.registry.keys())}]")
39		return self.registry[item]
40
41		def func_decorator(cls):
42		def wrapper(a_callable):
43		if hasattr(a_callable, '__code__'): # it a function (def func_name ..)
44		cls.registry[a_callable.__code__.co_name] = a_callable
45		else:
46		raise RuntimeError(f"Expected a function to be decorated; got {type(a_callable)}")
47		return a_callable
48		return wrapper
49
50		class AbstractCommandFactory(ABC):
51		@abstractmethod
52		def construct(self, args, *kwargs) -> Command:
53		raise NotImplementedError
54
55	View Code Duplication	class BaseCommandFactory(AbstractCommandFactory, ABC):
		0 ignored issues – show Duplication introduced 2020-08-29 10:47 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
56
57		subclasses = {}
58
59		@classmethod
60		def register_as_subclass(cls, factory_type):
61		def wrapper(subclass):
62		cls.subclasses[factory_type] = subclass
63		return subclass
64		return wrapper
65
66		@classmethod
67		def create(cls, factory_type, args, *kwargs):
68		if factory_type not in cls.subclasses:
69		raise ValueError('Bad "Factory type" \'{}\''.format(factory_type))
70		return cls.subclasses[factory_type](args, *kwargs)
71
72
73		@BaseCommandFactory.register_as_subclass('generic')
74		class GenericCommandFactory(AbstractCommandFactory):
75		def construct(self, args, *kwargs) -> Command:
76		return Command(args, *kwargs)
77
78		@BaseCommandFactory.register_as_subclass('function')
79		class FunctionCommandFactory(AbstractCommandFactory):
80		def construct(self, args, *kwargs) -> Command:
81		if len(args) < 1:
82		raise RuntimeError("Will break")
83		return Command(args[0], '__call__', *args[1:])
84
85
86		@BaseCommandFactory.register_as_subclass('encode_nominal_subsets')
87		class NominalAttributeListEncodeCommandFactory(AbstractCommandFactory):
88
89		def construct(self, args, *kwargs) -> Command:
90		from so_magic.data.features.phis import ListOfCategoricalPhi, DatapointsAttributePhi
91		assert len(args) > 0
92		datapoints = args[0]
93		attribute = args[1]
94		new_attribute = args[2]
95		def _command(_datapoints, _attribute, _new_attribute):
96		phi = ListOfCategoricalPhi(DatapointsAttributePhi(_datapoints))
97		new_values = phi(_attribute)
98		_datapoints.mutator.add_column(_datapoints, new_values, _new_attribute)
99		return Command(_command, '__call__', datapoints, attribute, new_attribute)
100
101
102		class CommandFactory:
103		"""A factory class able to construct new command objects."""
104		constructors = {k: v().construct for k, v in BaseCommandFactory.subclasses.items()}
105		@classmethod
106		def pick(cls, args, *kwargs):
107		decision = {True: 'function', False: 'generic'}
108		is_function = hasattr(args[0], '__code__')
109		dec2 = {'function': lambda x: x[0].__code__.co_name, 'generic': lambda x: type(x[0]).__name__ + '-' + x[1]}
110		return decision[is_function], kwargs.get('name', dec2[decision[is_function]](args))
111
112		@classmethod
113		def create(cls, args, *kwargs) -> Command:
114		"""Call to create a new Command object. The input arguments can be in two formats:
115
116		1. create(an_object, method, *arguments)
117		In this case the command is of the form an_object.method(*arguments)
118
119		2. create(a_function, *arguments)
120		In this case the command is of the form a_function(*arguments)
121
122		Returns:
123		Command: an instance of a command object
124		"""
125		key, name = cls.pick(args, *kwargs)
126		if len(args) < 1:
127		raise RuntimeError(args)
128		return cls.constructors[key](*args), name
129
130
131		@attr.s
132		class MagicCommandFactory(Subject):
133		"""Instances of this class act as callable command factories that notify,
134		subscribed observers/listeners upon new command object creation.
135
136		Args:
137		command_factory (CommandFactory, optional): an instance of a CommandFActory
138		"""
139		command_factory = attr.ib(init=True, default=CommandFactory())
140
141		def __call__(self, args, *kwargs):
142		self._state, self.name = self.command_factory.create(args, *kwargs)
143		self.notify()
144		return self._state
145
146
147	View Code Duplication	class DataManagerCommandFactory(AbstractCommandFactory, ABC):
		0 ignored issues – show Duplication introduced 2020-08-29 10:47 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
148
149		subclasses = {}
150
151		@classmethod
152		def register_as_subclass(cls, factory_type):
153		def wrapper(subclass):
154		cls.subclasses[factory_type] = subclass
155		return subclass
156		return wrapper
157
158		@classmethod
159		def create(cls, factory_type, args, *kwargs):
160		if factory_type not in cls.subclasses:
161		raise ValueError('Bad "Factory type" \'{}\''.format(factory_type))
162		return cls.subclasses[factory_type](args, *kwargs)
163
164		#### Register command factories
165		@DataManagerCommandFactory.register_as_subclass('select_variables')
166		class SelectVariablesCommandFactory(DataManagerCommandFactory):
167
168		def construct(self, args, *kwargs) -> Command:
169		def command(variables):
170		args[0].feature_manager.feature_configuration = variables
171		return Command(command, '__call__', *args[1:])
172
173
174		@DataManagerCommandFactory.register_as_subclass('select_variables')
175		class SelectVariablesCommandFactory(DataManagerCommandFactory):
176
177		def construct(self, args, *kwargs) -> Command:
178		def command(variables):
179		args[0].feature_manager.feature_configuration = variables
180		return Command(command, '__call__', *args[1:])
181
182		#################
183
184		@attr.s
185		class MegaCommandFactory(Subject):
186		_data_manager = attr.ib(init=True)
187		command_factory = attr.ib(init=True, default=DataManagerCommandFactory)
188
189		def __call__(self, command_type, args, *kwargs):
190		self._state, self.name = self.command_factory.create(command_type).construct(self._data_manager, args, *kwargs), command_type
191		self.notify()
192		return self._state
193

boromir674 / so-magic

Push — dev ( 3b058b...d73523 )

CommandFactory.pick() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like