so_magic.data.backend.panda_handling.df_backend.with_self() - 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:03 UTC

with_self() A

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

Complexity

Conditions

Size

Total Lines	7
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	4
nop	1
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_signature(tabular_operator, function_name, signature):
    sig = str(inspect.signature(getattr(tabular_operator, function_name)))
    if sig != signature:
        raise ValueError(f"Expected signature {signature} for {function_name} member of object {tabular_operator} with "
                         f"type {type(tabular_operator)}. Instead got {sig}.")


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:
        _validate_signature(tabular_operator, member_name, required_signature)


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

boromir674 / so-magic

Pull Request — dev (#30)

with_self() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like