so_magic.data.backend.panda_handling.df_backend.validate_retriever_delegate() - Code Metrics - Inspection of "Mpeta" - boromir674/so-magic - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — dev (#30)

by Konstantinos

created 2021-05-27 18:45 UTC

validate_retriever_delegate() B

↳ Parent: so_magic.data.backend.panda_handling.df_backend

Complexity

Conditions

Size

Total Lines	15
Code Lines	14

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	7
eloc	14
nop	3
dl	0
loc	15
rs	8
c	0
b	0
f	0

import inspect
import types
import attr
from so_magic.data.backend.engine_specs import EngineTabularRetriever, EngineTabularIterator, EngineTabularMutator

__all__ = ['PDTabularRetriever', 'PDTabularIterator', 'PDTabularMutator']


# DELEGATES
# User defined (engine dependent implementations of tabular operations)

class PDTabularRetrieverDelegate(EngineTabularRetriever):
    """The observation object is the same as the one your return from 'from_json_lines'"""

    @classmethod
    def column(cls, identifier, data):
        return data.observations[identifier]

    @classmethod
    def row(cls, identifier, data):
        return data.observations.loc(identifier)

    @classmethod
    def nb_columns(cls, data):
        return len(data.observations.columns)

    @classmethod
    def nb_rows(cls, data):
        print('\n------ DEBUG NB ROWS PDTabularRetrieverDelegate DATA TYPE', type(data), ' ------\n')
        return len(data.observations)

    @classmethod
    def get_numerical_attributes(cls, data):
        return data.observations._get_numeric_data().columns.values


class PDTabularIteratorDelegate(EngineTabularIterator):
    """The observation object is the same as the one your return from 'from_json_lines'"""

    @classmethod
    def columnnames(cls, data):
        return list(data.observations.columns)

    @classmethod
    def iterrows(cls, data):
        return iter(data.observations.iterrows())

    @classmethod
    def itercolumns(cls, data):
        return iter(data.observations[column] for column in data.observations.columns)


class PDTabularMutatorDelegate(EngineTabularMutator):

    @classmethod
    def add_column(cls, datapoints, values, new_attribute, **kwargs):
        datapoints.observations[new_attribute] = values


# INFRASTRUCTURE

def with_self(function):
    # foo_code = compile(f'def {f_name}({arg_string}): return f({params_str})', "<string>", "exec")
    # foo_func1 = types.FunctionType(foo_code.co_consts[0], globals(), f_name)

    def _function(_self, *args, **kwargs):
        return function(*args, **kwargs)
    return _function


class Delegate:
    def __new__(cls, *args, **kwargs):
        delegate_ins = super().__new__(cls)
        tabular_operator = args[0]
        for _member_name, member in inspect.getmembers(
                tabular_operator, predicate=lambda x: any([inspect.ismethod(x), inspect.isfunction(x)])):
            if isinstance(member, types.FunctionType):  # if no decorator is used
                setattr(delegate_ins, member.__name__, types.MethodType(member, delegate_ins))
            if isinstance(member, types.MethodType):  # if @classmethod is used
                setattr(delegate_ins, member.__name__, types.MethodType(with_self(member), delegate_ins))
            print()
        return delegate_ins


def validate_retriever_delegate(_self, _attribute, value):
    members_list = list(inspect.getmembers(value,
                                           predicate=lambda x: any([inspect.ismethod(x), inspect.isfunction(x)])))
    assert all(x in (_[0] for _ in members_list) for x in ('row', 'column', 'nb_rows', 'nb_columns',

                                                            'get_numerical_attributes'))
    for member_name, _member in members_list:
        if member_name in ['row', 'column']:
            sig = str(inspect.signature(getattr(value, member_name)))
            if sig != '(identifier, data)':
                raise ValueError(f"Expected signature (identifier, data) for {member_name} of retriever {value}. "
                                 f"Instead got {sig}.")
        if member_name in ['nb_rows', 'nb_columns', 'get_numerical_attributes']:
            sig = str(inspect.signature(getattr(value, member_name)))
            if sig != '(data)':
                raise ValueError(f"Expected signature (data) for {member_name} of retriever {value}. "
                                 f"Instead got {sig}.")


# CONCRETE IMPLEMENTATIONS

@attr.s
@EngineTabularRetriever.register_as_subclass('pd')
class PDTabularRetriever(EngineTabularRetriever):
    """The observation object is the same as the one your return from 'from_json_lines'"""
    _delegate = attr.ib(default=attr.Factory(lambda: Delegate(PDTabularRetrieverDelegate)),
    # validator=validate_retriever_delegate
    )

    def column(self, identifier, data):
        return self._delegate.column(identifier, data)

    def row(self, identifier, data):
        return self._delegate.row(identifier, data)

    def nb_columns(self, data):
        return self._delegate.nb_columns(data)

    def nb_rows(self, data):
        return self._delegate.nb_rows(data)

    def get_numerical_attributes(self, data):
        return self._delegate.get_numerical_attributes(data)


@attr.s
@EngineTabularIterator.register_as_subclass('pd')
class PDTabularIterator(EngineTabularIterator):
    """The observation object is the same as the one your return from 'from_json_lines'"""
    _delegate = attr.ib(default=attr.Factory(lambda: Delegate(PDTabularIteratorDelegate)))

    def columnnames(self, data):
        return self._delegate.columnnames(data)

    def iterrows(self, data):
        return self._delegate.iterrows(data)

    def itercolumns(self, data):
        return self._delegate.itercolumns(data)


@attr.s
@EngineTabularMutator.register_as_subclass('pd')
class PDTabularMutator(EngineTabularMutator):
    _delegate = attr.ib(default=attr.Factory(lambda: Delegate(PDTabularMutatorDelegate)))

    def add_column(self, datapoints, values, new_attribute, **kwargs):
        self._delegate.add_column(datapoints, values, new_attribute, **kwargs)


1			import inspect
2			import types
3			import attr
4			from so_magic.data.backend.engine_specs import EngineTabularRetriever, EngineTabularIterator, EngineTabularMutator
5
6			__all__ = ['PDTabularRetriever', 'PDTabularIterator', 'PDTabularMutator']
7
8
9			# DELEGATES
10			# User defined (engine dependent implementations of tabular operations)
11
12			class PDTabularRetrieverDelegate(EngineTabularRetriever):
13			"""The observation object is the same as the one your return from 'from_json_lines'"""
14
15			@classmethod
16			def column(cls, identifier, data):
17			return data.observations[identifier]
18
19			@classmethod
20			def row(cls, identifier, data):
21			return data.observations.loc(identifier)
22
23			@classmethod
24			def nb_columns(cls, data):
25			return len(data.observations.columns)
26
27			@classmethod
28			def nb_rows(cls, data):
29			print('\n------ DEBUG NB ROWS PDTabularRetrieverDelegate DATA TYPE', type(data), ' ------\n')
30			return len(data.observations)
31
32			@classmethod
33			def get_numerical_attributes(cls, data):
34			return data.observations._get_numeric_data().columns.values
35
36
37			class PDTabularIteratorDelegate(EngineTabularIterator):
38			"""The observation object is the same as the one your return from 'from_json_lines'"""
39
40			@classmethod
41			def columnnames(cls, data):
42			return list(data.observations.columns)
43
44			@classmethod
45			def iterrows(cls, data):
46			return iter(data.observations.iterrows())
47
48			@classmethod
49			def itercolumns(cls, data):
50			return iter(data.observations[column] for column in data.observations.columns)
51
52
53			class PDTabularMutatorDelegate(EngineTabularMutator):
54
55			@classmethod
56			def add_column(cls, datapoints, values, new_attribute, **kwargs):
57			datapoints.observations[new_attribute] = values
58
59
60			# INFRASTRUCTURE
61
62			def with_self(function):
63			# foo_code = compile(f'def {f_name}({arg_string}): return f({params_str})', "<string>", "exec")
64			# foo_func1 = types.FunctionType(foo_code.co_consts[0], globals(), f_name)
65
66			def _function(_self, args, *kwargs):
67			return function(args, *kwargs)
68			return _function
69
70
71			class Delegate:
72			def __new__(cls, args, *kwargs):
73			delegate_ins = super().__new__(cls)
74			tabular_operator = args[0]
75			for _member_name, member in inspect.getmembers(
76			tabular_operator, predicate=lambda x: any([inspect.ismethod(x), inspect.isfunction(x)])):
77			if isinstance(member, types.FunctionType): # if no decorator is used
78			setattr(delegate_ins, member.__name__, types.MethodType(member, delegate_ins))
79			if isinstance(member, types.MethodType): # if @classmethod is used
80			setattr(delegate_ins, member.__name__, types.MethodType(with_self(member), delegate_ins))
81			print()
82			return delegate_ins
83
84
85			def validate_retriever_delegate(_self, _attribute, value):
86			members_list = list(inspect.getmembers(value,
87			predicate=lambda x: any([inspect.ismethod(x), inspect.isfunction(x)])))
88			assert all(x in (_[0] for _ in members_list) for x in ('row', 'column', 'nb_rows', 'nb_columns',
			0 ignored issues – show Comprehensibility Best Practice introduced 2021-05-27 18:41 UTC by Report Bug Copy Issue Report The variable `_` does not seem to be defined. Loading history...
89			'get_numerical_attributes'))
90			for member_name, _member in members_list:
91			if member_name in ['row', 'column']:
92			sig = str(inspect.signature(getattr(value, member_name)))
93			if sig != '(identifier, data)':
94			raise ValueError(f"Expected signature (identifier, data) for {member_name} of retriever {value}. "
95			f"Instead got {sig}.")
96			if member_name in ['nb_rows', 'nb_columns', 'get_numerical_attributes']:
97			sig = str(inspect.signature(getattr(value, member_name)))
98			if sig != '(data)':
99			raise ValueError(f"Expected signature (data) for {member_name} of retriever {value}. "
100			f"Instead got {sig}.")
101
102
103			# CONCRETE IMPLEMENTATIONS
104
105			@attr.s
106			@EngineTabularRetriever.register_as_subclass('pd')
107			class PDTabularRetriever(EngineTabularRetriever):
108			"""The observation object is the same as the one your return from 'from_json_lines'"""
109			_delegate = attr.ib(default=attr.Factory(lambda: Delegate(PDTabularRetrieverDelegate)),
110			# validator=validate_retriever_delegate
111			)
112
113			def column(self, identifier, data):
114			return self._delegate.column(identifier, data)
115
116			def row(self, identifier, data):
117			return self._delegate.row(identifier, data)
118
119			def nb_columns(self, data):
120			return self._delegate.nb_columns(data)
121
122			def nb_rows(self, data):
123			return self._delegate.nb_rows(data)
124
125			def get_numerical_attributes(self, data):
126			return self._delegate.get_numerical_attributes(data)
127
128
129			@attr.s
130			@EngineTabularIterator.register_as_subclass('pd')
131			class PDTabularIterator(EngineTabularIterator):
132			"""The observation object is the same as the one your return from 'from_json_lines'"""
133			_delegate = attr.ib(default=attr.Factory(lambda: Delegate(PDTabularIteratorDelegate)))
134
135			def columnnames(self, data):
136			return self._delegate.columnnames(data)
137
138			def iterrows(self, data):
139			return self._delegate.iterrows(data)
140
141			def itercolumns(self, data):
142			return self._delegate.itercolumns(data)
143
144
145			@attr.s
146			@EngineTabularMutator.register_as_subclass('pd')
147			class PDTabularMutator(EngineTabularMutator):
148			_delegate = attr.ib(default=attr.Factory(lambda: Delegate(PDTabularMutatorDelegate)))
149
150			def add_column(self, datapoints, values, new_attribute, **kwargs):
151			self._delegate.add_column(datapoints, values, new_attribute, **kwargs)
152

boromir674 / so-magic

Pull Request — dev (#30)

validate_retriever_delegate() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like