|
1
|
|
|
import abc |
|
2
|
|
|
import inspect |
|
3
|
|
|
import types |
|
4
|
|
|
from functools import wraps |
|
5
|
|
|
|
|
6
|
|
|
__all__ = ['Transformer'] |
|
7
|
|
|
|
|
8
|
|
|
class TransformerInterface(abc.ABC): |
|
9
|
|
|
"""The interface defining a method to transform structured data. Anyone, implementing this has the ability to receive |
|
10
|
|
|
some kind of data and return some kind of transformed version of them. |
|
11
|
|
|
""" |
|
12
|
|
|
@abc.abstractmethod |
|
13
|
|
|
def transform(self, data, **kwargs): |
|
14
|
|
|
"""Takes data and optional keyword arguments and transforms them. |
|
15
|
|
|
Input data can represent either a single variable of an observations (scalar) |
|
16
|
|
|
or a vector of observations of the same variable. |
|
17
|
|
|
|
|
18
|
|
|
Example 1: |
|
19
|
|
|
obs1 = [x1, y1, z1] |
|
20
|
|
|
fa = f_a(x) |
|
21
|
|
|
fb = f_b(x) |
|
22
|
|
|
fc = f_c(x) |
|
23
|
|
|
feature_vector1 = [fa(x1), fb(y1), fc(z1)] |
|
24
|
|
|
|
|
25
|
|
|
So, each of fa, fb and fc can implement the Transformer interface. |
|
26
|
|
|
|
|
27
|
|
|
Example 2: |
|
28
|
|
|
obs1 = [x1, y1, z1] |
|
29
|
|
|
obs2 = [x2, y2, z2] |
|
30
|
|
|
obs3 = [x3, y3, z3] |
|
31
|
|
|
obs4 = [x4, y4, z4] |
|
32
|
|
|
data = [obs1; |
|
33
|
|
|
obs2; |
|
34
|
|
|
obs3; |
|
35
|
|
|
obs4] shape = (4,3) |
|
36
|
|
|
fa = f_a(x) |
|
37
|
|
|
fb = f_b(x) |
|
38
|
|
|
fc = f_c(x) |
|
39
|
|
|
feature_vectors = [fa(data[:0], fb(data[:1], fc(data[:2])] - shape = (4,3) |
|
40
|
|
|
|
|
41
|
|
|
Again each of fa, fb and fc can implement the Transformer interface. |
|
42
|
|
|
|
|
43
|
|
|
Args: |
|
44
|
|
|
data (object): the input data to transform; the x in an f(x) invocation |
|
45
|
|
|
|
|
46
|
|
|
Raises: |
|
47
|
|
|
NotImplementedError: [description] |
|
48
|
|
|
""" |
|
49
|
|
|
raise NotImplementedError |
|
50
|
|
|
|
|
51
|
|
|
|
|
52
|
|
|
class RuntimeTransformer(TransformerInterface, abc.ABC): |
|
53
|
|
|
"""Examines whether the input object is callable, if it can receive at least one input argument and also |
|
54
|
|
|
whether it can accept kwargs. Depending on the kwargs check, "configures" the '_transform' method to process |
|
55
|
|
|
any kwargs at runtime or to ignore them. |
|
56
|
|
|
|
|
57
|
|
|
Args: |
|
58
|
|
|
a_callable (callable): a callable object that can be potentially used to delegate the transformation operation |
|
59
|
|
|
""" |
|
60
|
|
|
def __new__(cls, *args, **kwargs): |
|
61
|
|
|
x = super().__new__(cls) |
|
62
|
|
|
a_callable = args[0] |
|
63
|
|
|
if not callable(a_callable): |
|
64
|
|
|
raise ValueError(f"Expected a callable as argument; instead got '{type(a_callable)}'") |
|
65
|
|
|
nb_mandatory_arguments = a_callable.__code__.co_argcount # this counts sums both *args and **kwargs |
|
66
|
|
|
# use syntax like 'def a(b, *, c=1, d=2): .. to separate pos args from kwargs and to inform 'inspect' lib about it |
|
67
|
|
|
if nb_mandatory_arguments < 1: |
|
68
|
|
|
raise ValueError(f"Expected a callable that receives at least one positional argument; instead got a callable that " |
|
69
|
|
|
f"receives '{nb_mandatory_arguments}'") |
|
70
|
|
|
signature = inspect.signature(a_callable) |
|
71
|
|
|
parameters = [param for param in signature.parameters.values()] |
|
72
|
|
|
|
|
73
|
|
|
if 1 < nb_mandatory_arguments: |
|
74
|
|
|
def _transform(self, data, **keyword_args): |
|
75
|
|
|
return a_callable(data, **keyword_args) |
|
76
|
|
|
x._transform = types.MethodType(_transform, x) |
|
77
|
|
|
elif nb_mandatory_arguments == len(parameters): |
|
78
|
|
|
def _transform(self, data, **keyword_args): |
|
79
|
|
|
return a_callable(data) |
|
80
|
|
|
x._transform = types.MethodType(_transform, x) |
|
81
|
|
|
else: |
|
82
|
|
|
raise Exception("Something went really bad. Check code above.") |
|
83
|
|
|
x._callable = a_callable |
|
84
|
|
|
return x |
|
85
|
|
|
|
|
86
|
|
|
|
|
87
|
|
|
class Transformer(RuntimeTransformer): |
|
88
|
|
|
"""Delegates all the transformation operation to its '_transform' method provided by its '_callable' field.""" |
|
89
|
|
|
def transform(self, data, **kwargs): |
|
90
|
|
|
return self._transform(data, **kwargs) |
|
91
|
|
|
|
|
92
|
|
|
|
|
93
|
|
|
def my_decorator(a_callable): |
|
94
|
|
|
@wraps(a_callable) |
|
95
|
|
|
def wrapper(*args, **kwds): |
|
96
|
|
|
print('Calling decorated function') |
|
97
|
|
|
return a_callable(*args, **kwds) |
|
98
|
|
|
return wrapper |
|
99
|
|
|
|
|
100
|
|
|
|
|
101
|
|
|
if __name__ == '__main__': |
|
102
|
|
|
tr1 = Transformer(lambda x: x + 1) |
|
103
|
|
|
inp = 10 |
|
104
|
|
|
out = tr1.transform(inp) |
|
105
|
|
|
assert 11 == out |
|
106
|
|
|
|
|
107
|
|
|
def gg(a, b=2): |
|
108
|
|
|
return a*b + 1 |
|
109
|
|
|
tr1 = Transformer(gg) |
|
110
|
|
|
inp = 10 |
|
111
|
|
|
out = tr1.transform(inp) |
|
112
|
|
|
assert 21 == out |
|
113
|
|
|
out = tr1.transform(inp, b=3) |
|
114
|
|
|
assert 31 == out |
|
115
|
|
|
|
boromir674 /
so-magic
