green_magic.data.command_factories.GenericCommandFactory.construct() - Code Metrics - Inspection of "wip" - boromir674/so-magic - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — dev ( 3ac626...71cb7d )

by Konstantinos

created 2020-08-29 10:46 UTC

GenericCommandFactory.construct() A

↳ Parent: green_magic.data.command_factories

Complexity

Conditions

Size

Total Lines	2
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

from abc import ABC, abstractmethod
import attr
from green_magic.utils import Command
from green_magic.utils import 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):
        print("DEBUG magic_decorator 1, args:", str(arg))
        def decorator(func):
            print("DEBUG magic_decorator 2, args:", str(func))
            def wrapper(*a, **ka):
                print("DEBUG magic_decorator 3, args: [{}]".format(', '.join(str(x) for x in a)))
                ffunc = a[0]
                mcs._wrapper(ffunc, *a[1:], **ka)
                return ffunc
            print("DEBUG magic_decorator 4, func: {}".format(str(func)))
            return wrapper

        if callable(arg):
            print("Decoration invokation WITHOUT parenthesis")
            _ = decorator(arg)
            print("OUT: {}".format(str(_)))
            print(type(_))
            return _  # return 'wrapper'
        else:
            print("Decoration invokation WITH parenthesis")
            _ = decorator
            print(f"OUT: {str(_)}")
            print(type(_))
            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 ..)
                print(f"Registering input function {a_callable.__code__.co_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 green_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__')
        print(f'is function?: {is_function}')
        print(f'PICK: {args}')
        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
        """
        print("CMD FCT")
        print("args: ", args)
        print("kwargs", kwargs)

        key, name = cls.pick(*args, **kwargs)
        print(f"KEY: {key}, NAME: {name}")
        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)


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

boromir674 / so-magic

Push — dev ( 3ac626...71cb7d )

GenericCommandFactory.construct() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like