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-28 10:11 UTC

validate_retriever_delegate() A

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

Complexity

Conditions

Size

Total Lines	7
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	7
nop	3
dl	0
loc	7
rs	10
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))
        return delegate_ins


def validate_delegate(tabular_operator, required_members):
    members_list = list(inspect.getmembers(tabular_operator,
                                           predicate=lambda x: any([inspect.ismethod(x), inspect.isfunction(x)])))
    assert all(x in (_[0] for _ in members_list) for x in required_members)

    for member_name, required_signature in required_members:
        sig = str(inspect.signature(getattr(tabular_operator, member_name)))
        if sig != required_signature:
            raise ValueError(f"Expected signature {required_signature} for {member_name} member of object "
                             f"{tabular_operator} with type {type(tabular_operator)}. Instead got {sig}.")


def validate_retriever_delegate(_self, _attribute, value):
    validate_delegate(value, {
        'column': '(identifier, data)',
        'row': '(identifier, data)',
        'nb_columns': '(data)',
        'nb_rows': '(data)',
        'get_numerical_attributes': '(data)',
    })


# 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			return delegate_ins
82
83
84			def validate_delegate(tabular_operator, required_members):
85			members_list = list(inspect.getmembers(tabular_operator,
86			predicate=lambda x: any([inspect.ismethod(x), inspect.isfunction(x)])))
87			assert all(x in (_[0] for _ in members_list) for x in required_members)
			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...
88			for member_name, required_signature in required_members:
89			sig = str(inspect.signature(getattr(tabular_operator, member_name)))
90			if sig != required_signature:
91			raise ValueError(f"Expected signature {required_signature} for {member_name} member of object "
92			f"{tabular_operator} with type {type(tabular_operator)}. Instead got {sig}.")
93
94
95			def validate_retriever_delegate(_self, _attribute, value):
96			validate_delegate(value, {
97			'column': '(identifier, data)',
98			'row': '(identifier, data)',
99			'nb_columns': '(data)',
100			'nb_rows': '(data)',
101			'get_numerical_attributes': '(data)',
102			})
103
104
105			# CONCRETE IMPLEMENTATIONS
106
107			@attr.s
108			@EngineTabularRetriever.register_as_subclass('pd')
109			class PDTabularRetriever(EngineTabularRetriever):
110			"""The observation object is the same as the one your return from 'from_json_lines'"""
111			_delegate = attr.ib(default=attr.Factory(lambda: Delegate(PDTabularRetrieverDelegate)),
112			# validator=validate_retriever_delegate
113			)
114
115			def column(self, identifier, data):
116			return self._delegate.column(identifier, data)
117
118			def row(self, identifier, data):
119			return self._delegate.row(identifier, data)
120
121			def nb_columns(self, data):
122			return self._delegate.nb_columns(data)
123
124			def nb_rows(self, data):
125			return self._delegate.nb_rows(data)
126
127			def get_numerical_attributes(self, data):
128			return self._delegate.get_numerical_attributes(data)
129
130
131			@attr.s
132			@EngineTabularIterator.register_as_subclass('pd')
133			class PDTabularIterator(EngineTabularIterator):
134			"""The observation object is the same as the one your return from 'from_json_lines'"""
135			_delegate = attr.ib(default=attr.Factory(lambda: Delegate(PDTabularIteratorDelegate)))
136
137			def columnnames(self, data):
138			return self._delegate.columnnames(data)
139
140			def iterrows(self, data):
141			return self._delegate.iterrows(data)
142
143			def itercolumns(self, data):
144			return self._delegate.itercolumns(data)
145
146
147			@attr.s
148			@EngineTabularMutator.register_as_subclass('pd')
149			class PDTabularMutator(EngineTabularMutator):
150			_delegate = attr.ib(default=attr.Factory(lambda: Delegate(PDTabularMutatorDelegate)))
151
152			def add_column(self, datapoints, values, new_attribute, **kwargs):
153			self._delegate.add_column(datapoints, values, new_attribute, **kwargs)
154

boromir674 / so-magic

Pull Request — dev (#30)

validate_retriever_delegate() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like