green_magic.data.commands_manager.FunctionCommandFactory.construct() - Code Metrics - Inspection of "wip; test commands" - boromir674/so-magic - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — dev ( 3c73d5...40dcc8 )

by Konstantinos

created 2020-08-16 11:33 UTC

FunctionCommandFactory.construct() A

↳ Parent: green_magic.data.commands_manager

Complexity

Conditions

Size

Total Lines	5
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

from abc import ABC, abstractmethod
import inspect
import attr
from green_magic.utils import Command
from green_magic.utils import Subject, Observer
from green_magic.data.dataset import DatapointsFactory


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:
        print("GENERIC", args)
        return Command(*args, **kwargs)

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


@BaseCommandFactory.register_as_subclass('add-attribute')
class AddAttributeCommandFactory(AbstractCommandFactory):
    def construct(self, *args, **kwargs) -> Command:
        assert len(args) > 0
        datapoints = args[0]
        values = args[1]
        new_attribute = args[2]

        return Command(args[0], '__call__', *args[1:])


class CommandFactory:
    """A factory class able to construct new command objects."""
    constructors = {k: v().construct for k, v in BaseCommandFactory.subclasses.items()}
    # datapoints_factory = DatapointsFactory()
    @classmethod
    def pick(cls, *args):
        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], dec2[decision[is_function]](args)

    @classmethod
    def create(cls, *args) -> 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("OPA 1", args)

        key, name = cls.pick(*args)
        print(f"KEY: {key}, NAME: {name}")
        if len(args) < 1:
            raise RuntimeError(args)
        print("OPA 2", args)
        return cls.constructors[key](*args), name
        # is_function = hasattr(args[0], '__code__')
        # if is_function:  # if receiver is a function; creating from function
        #     return cls.command_constructor(args[0], '__call__', *args[1:]), args[0].__code__.co_name
        # return cls.command_constructor(args[0], args[1], *args[2:]), type(args[0]) + '-' + args[1]


@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):
        assert args
        self._state, self.name = self.command_factory.create(*args)
        self.notify()
        return self._state


@attr.s
class CommandsAccumulator(Observer):
    """"""
    commands = attr.ib(init=False, default={})

    def update(self, subject: Subject) -> None:
        self.commands[getattr(subject, 'name', str(subject.state))] = subject.state


@attr.s
class CommandGetter:
    _commands_accumulator = attr.ib(init=True, default=CommandsAccumulator())

    @property
    def accumulator(self):
        return self._commands_accumulator

    def __getattr__(self, item):
        if item not in self._commands_accumulator.commands:
            raise KeyError(f"Item '{item}' not found in [{', '.join(str(_) for _ in self._commands_accumulator.commands.keys())}]")
        return self._commands_accumulator.commands[item]


@attr.s
class CommandsManager:
    """[summary]

    Args:
        prototypes (dict, optional): initial prototypes to be supplied
        command_factory (callable, optional): a callable that returns an instance of Command
    """
    _commands_getter = attr.ib(init=True, default=CommandGetter())

    @property
    def command(self):
        return self._commands_getter

    @property
    def commands_dict(self):
        return self._commands_accumulator.commands

    def __getattr__(self, item):
        if item not in self._commands_accumulator.commands:
            raise KeyError(f"Item '{item}' not found in [{', '.join(str(_) for _ in self._commands_accumulator.commands.keys())}]")
        return self._commands_accumulator.commands[item]


1			from abc import ABC, abstractmethod
2			import inspect
3			import attr
4			from green_magic.utils import Command
5			from green_magic.utils import Subject, Observer
6			from green_magic.data.dataset import DatapointsFactory
7
8
9			class MyDecorator(type):
10			"""Metaclass that provides a decorator able to be invoked both with and without parenthesis.
11			The wrapper function logic should be implemented by the client code.
12			"""
13			@classmethod
14			def magic_decorator(mcs, arg=None):
15			print("DEBUG magic_decorator 1, args:", str(arg))
16			def decorator(func):
17			print("DEBUG magic_decorator 2, args:", str(func))
18			def wrapper(a, *ka):
19			print("DEBUG magic_decorator 3, args: [{}]".format(', '.join(str(x) for x in a)))
20			ffunc = a[0]
21			mcs._wrapper(ffunc, a[1:], *ka)
22			return ffunc
23			print("DEBUG magic_decorator 4, func: {}".format(str(func)))
24			return wrapper
25
26			if callable(arg):
27			print("Decoration invokation WITHOUT parenthesis")
28			_ = decorator(arg)
29			print("OUT: {}".format(str(_)))
30			print(type(_))
31			return _ # return 'wrapper'
32			else:
33			print("Decoration invokation WITH parenthesis")
34			_ = decorator
35			print(f"OUT: {str(_)}")
36			print(type(_))
37			return _ # ... or 'decorator'
38
39			class CommandRegistrator(MyDecorator):
40			"""Classes can use this class as metaclass to obtain a single registration point accessible as class attribute
41			"""
42			def __new__(mcs, args, *kwargs):
43			class_object = super().__new__(mcs, args, *kwargs)
44			class_object.state = None
45			class_object.registry = {}
46			return class_object
47
48			def __getitem__(self, item):
49			if item not in self.registry:
50			raise RuntimeError(f"Key '{item}' fot found in registry: [{', '.join(str(x) for x in self.registry.keys())}]")
51			return self.registry[item]
52
53			def func_decorator(cls):
54			def wrapper(a_callable):
55			if hasattr(a_callable, '__code__'): # it a function (def func_name ..)
56			print(f"Registering input function {a_callable.__code__.co_name}")
57			cls.registry[a_callable.__code__.co_name] = a_callable
58			else:
59			raise RuntimeError(f"Expected a function to be decorated; got {type(a_callable)}")
60			return a_callable
61			return wrapper
62
63			class AbstractCommandFactory(ABC):
64			@abstractmethod
65			def construct(self, args, *kwargs) -> Command:
66			raise NotImplementedError
67
68			class BaseCommandFactory(AbstractCommandFactory, ABC):
69
70			subclasses = {}
71
72			@classmethod
73			def register_as_subclass(cls, factory_type):
74			def wrapper(subclass):
75			cls.subclasses[factory_type] = subclass
76			return subclass
77			return wrapper
78
79			@classmethod
80			def create(cls, factory_type, args, *kwargs):
81			if factory_type not in cls.subclasses:
82			raise ValueError('Bad "Factory type" \'{}\''.format(factory_type))
83			return cls.subclasses[factory_type](args, *kwargs)
84
85
86			@BaseCommandFactory.register_as_subclass('generic')
87			class GenericCommandFactory(AbstractCommandFactory):
88			def construct(self, args, *kwargs) -> Command:
89			print("GENERIC", args)
90			return Command(args, *kwargs)
91
92			@BaseCommandFactory.register_as_subclass('function')
93			class FunctionCommandFactory(AbstractCommandFactory):
94			def construct(self, args, *kwargs) -> Command:
95			print("FUNCTION", args)
96			if len(args) < 1:
97			raise RuntimeError("Will break")
98			return Command(args[0], '__call__', *args[1:])
99
100
101			@BaseCommandFactory.register_as_subclass('add-attribute')
102			class AddAttributeCommandFactory(AbstractCommandFactory):
103			def construct(self, args, *kwargs) -> Command:
104			assert len(args) > 0
105			datapoints = args[0]
106			values = args[1]
107			new_attribute = args[2]
108
109			return Command(args[0], '__call__', *args[1:])
110
111
112			class CommandFactory:
113			"""A factory class able to construct new command objects."""
114			constructors = {k: v().construct for k, v in BaseCommandFactory.subclasses.items()}
115			# datapoints_factory = DatapointsFactory()
116			@classmethod
117			def pick(cls, *args):
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], dec2[decision[is_function]](args)
124
125			@classmethod
126			def create(cls, *args) -> 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("OPA 1", args)
139
140			key, name = cls.pick(*args)
141			print(f"KEY: {key}, NAME: {name}")
142			if len(args) < 1:
143			raise RuntimeError(args)
144			print("OPA 2", args)
145			return cls.constructors[key](*args), name
146			# is_function = hasattr(args[0], '__code__')
147			# if is_function: # if receiver is a function; creating from function
148			# return cls.command_constructor(args[0], '__call__', *args[1:]), args[0].__code__.co_name
149			# return cls.command_constructor(args[0], args[1], *args[2:]), type(args[0]) + '-' + args[1]
150
151
152			@attr.s
153			class MagicCommandFactory(Subject):
154			"""Instances of this class act as callable command factories that notify,
155			subscribed observers/listeners upon new command object creation.
156
157			Args:
158			command_factory (CommandFactory, optional): an instance of a CommandFActory
159			"""
160			command_factory = attr.ib(init=True, default=CommandFactory())
161
162			def __call__(self, args, *kwargs):
163			assert args
164			self._state, self.name = self.command_factory.create(*args)
165			self.notify()
166			return self._state
167
168
169			@attr.s
170			class CommandsAccumulator(Observer):
171			""""""
172			commands = attr.ib(init=False, default={})
173
174			def update(self, subject: Subject) -> None:
175			self.commands[getattr(subject, 'name', str(subject.state))] = subject.state
			0 ignored issues – show Comprehensibility Best Practice introduced 2020-08-09 18:51 UTC by Report Bug Copy Issue Report The variable `getattr` does not seem to be defined. Loading history... Comprehensibility Best Practice introduced 2020-08-09 18:51 UTC by Report Bug Copy Issue Report The variable `str` does not seem to be defined. Loading history...
176
177			@attr.s
178			class CommandGetter:
179			_commands_accumulator = attr.ib(init=True, default=CommandsAccumulator())
180
181			@property
182			def accumulator(self):
183			return self._commands_accumulator
184
185			def __getattr__(self, item):
186			if item not in self._commands_accumulator.commands:
187			raise KeyError(f"Item '{item}' not found in [{', '.join(str(_) for _ in self._commands_accumulator.commands.keys())}]")
188			return self._commands_accumulator.commands[item]
189
190
191			@attr.s
192			class CommandsManager:
193			"""[summary]
194
195			Args:
196			prototypes (dict, optional): initial prototypes to be supplied
197			command_factory (callable, optional): a callable that returns an instance of Command
198			"""
199			_commands_getter = attr.ib(init=True, default=CommandGetter())
200
201			@property
202			def command(self):
203			return self._commands_getter
204
205			@property
206			def commands_dict(self):
207			return self._commands_accumulator.commands
208
209			def __getattr__(self, item):
210			if item not in self._commands_accumulator.commands:
211			raise KeyError(f"Item '{item}' not found in [{', '.join(str(_) for _ in self._commands_accumulator.commands.keys())}]")
212			return self._commands_accumulator.commands[item]
213

boromir674 / so-magic

Push — dev ( 3c73d5...40dcc8 )

FunctionCommandFactory.construct() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like