|
1
|
|
|
import pytest |
|
2
|
|
|
|
|
3
|
|
|
|
|
4
|
|
|
@pytest.fixture |
|
5
|
|
|
def define_command(): |
|
6
|
|
|
def _define_engine_command(decorator, command_function): |
|
7
|
|
|
decorator(command_function) |
|
8
|
|
|
return command_function.__name__ |
|
9
|
|
|
return _define_engine_command |
|
10
|
|
|
|
|
11
|
|
|
|
|
12
|
|
|
@pytest.fixture |
|
13
|
|
|
def get_command(somagic): |
|
14
|
|
|
def _get_command(command_name: str): |
|
15
|
|
|
return getattr(somagic.command, command_name) |
|
16
|
|
|
return _get_command |
|
17
|
|
|
|
|
18
|
|
|
|
|
19
|
|
|
@pytest.fixture |
|
20
|
|
|
def test_discretizer(): |
|
21
|
|
|
from so_magic.data.discretization import Discretizer, BinningAlgorithm |
|
22
|
|
|
|
|
23
|
|
|
alg = BinningAlgorithm.from_built_in('pd.cut') |
|
24
|
|
|
|
|
25
|
|
|
discretizer = Discretizer.from_algorithm(alg) |
|
26
|
|
|
return discretizer |
|
27
|
|
|
|
|
28
|
|
|
|
|
29
|
|
|
@pytest.fixture |
|
30
|
|
|
def discretize_command(): |
|
31
|
|
|
def get_discretize_command(discretizer): |
|
32
|
|
|
def test_discretize_command(data_manager, datapoints, attribute, nb_bins, new_column_name): |
|
33
|
|
|
output = discretizer.discretize(datapoints, attribute, nb_bins) |
|
34
|
|
|
data_manager.datapoints.add_column(output['result'], new_column_name) |
|
35
|
|
|
return test_discretize_command |
|
36
|
|
|
return get_discretize_command |
|
37
|
|
|
|
|
38
|
|
|
|
|
39
|
|
|
@pytest.fixture |
|
40
|
|
|
def validate_discretization_operation_behaviour(): |
|
41
|
|
|
def _validate_discretization_operation(cmd, algorithm): |
|
42
|
|
|
datapoints = cmd.args[0] |
|
43
|
|
|
target_column = cmd.args[1] |
|
44
|
|
|
nb_bins = cmd.args[2] |
|
45
|
|
|
computed_bins = algorithm.output['settings']['used_bins'] |
|
46
|
|
|
assert [_ for _ in computed_bins] == [-0.1, 25.0, 50.0, 75.0, 100.0] |
|
47
|
|
|
|
|
48
|
|
|
input_arguments = algorithm.output['settings']['arguments'] |
|
49
|
|
|
to_check = [len(input_arguments[0]), input_arguments[1]] |
|
50
|
|
|
assert to_check == [len(datapoints), nb_bins] |
|
51
|
|
|
assert type(datapoints.column(target_column)) == type(input_arguments[0]) |
|
52
|
|
|
assert list(datapoints.column(target_column)) == list(input_arguments[0]) |
|
53
|
|
|
# assert algorithm.output['settings']['parameters'] == [] |
|
54
|
|
|
return _validate_discretization_operation |
|
55
|
|
|
|
|
56
|
|
|
|
|
57
|
|
|
@pytest.fixture |
|
58
|
|
|
def discretization_cmd(somagic, test_datapoints, define_command, get_command, discretize_command, test_discretizer): |
|
59
|
|
|
"""Get a discretization command after some 'pre-processing' done on the test datapoints.""" |
|
60
|
|
|
series = somagic.dataset.datapoints.column('Creative').replace('', 0.0, inplace=False) |
|
61
|
|
|
assert all(type(x) == float for x in series) |
|
62
|
|
|
|
|
63
|
|
|
somagic.datapoints.add_column(list(series), 'Creative') |
|
64
|
|
|
|
|
65
|
|
|
assert all(type(x) == float for x in somagic.datapoints.observations['Creative']) |
|
66
|
|
|
|
|
67
|
|
|
test_discretize_command_name: str = define_command(somagic.commands_decorators.data_manager_command(), |
|
68
|
|
|
discretize_command(test_discretizer)) |
|
69
|
|
|
return get_command(test_discretize_command_name) |
|
70
|
|
|
|
|
71
|
|
|
|
|
72
|
|
|
@pytest.fixture(params=[ |
|
73
|
|
|
['Creative'], |
|
74
|
|
|
# [], # add more columns when we know the discretization command will succeed for them |
|
75
|
|
|
]) |
|
76
|
|
|
def cmd_to_succeed(request, test_datapoints, discretization_cmd): |
|
77
|
|
|
discretization_cmd.args = [test_datapoints, request.param[0], 4, f'binned_{request.param[0]}'] |
|
78
|
|
|
return discretization_cmd |
|
79
|
|
|
|
|
80
|
|
|
|
|
81
|
|
|
def test_discretization_operation(cmd_to_succeed, test_discretizer, validate_discretization_operation_behaviour): |
|
82
|
|
|
cmd_to_succeed.execute() |
|
83
|
|
|
validate_discretization_operation_behaviour(cmd_to_succeed, test_discretizer.algorithm) |
|
84
|
|
|
|
|
85
|
|
|
|
|
86
|
|
|
@pytest.fixture(params=[ |
|
87
|
|
|
['Energetic'], |
|
88
|
|
|
# [], # add more columns when we know the discretization command will fail for them |
|
89
|
|
|
]) |
|
90
|
|
|
def cmd_to_fail(request, test_datapoints, discretization_cmd): |
|
91
|
|
|
discretization_cmd.args = [test_datapoints, request.param[0], 4, f'binned_{request.param[0]}'] |
|
92
|
|
|
return discretization_cmd |
|
93
|
|
|
|
|
94
|
|
|
|
|
95
|
|
|
def test_discretization_on_non_preprocessed_attribute(cmd_to_fail): |
|
96
|
|
|
with pytest.raises(TypeError): |
|
97
|
|
|
cmd_to_fail.execute() |
|
98
|
|
|
|
boromir674 /
so-magic
