unit-tests-ml-python/selection.py at master · sergred/unit-tests-ml-python · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
#!/usr/bin/python3.6
# -*- coding: utf-8 -*-

""""""

import numpy as np

from profilers import DataFrameProfiler
from analyzers import DataType
from messages import Message as mes


class Selector:
    def __init__(self):
        pass


class RandomSelector(Selector):
    class __RandomSelector:
        def __init__(self, column_fraction=.2, row_fraction=.1):
            # Selector.__init__(self)
            self.column_fraction = column_fraction
            self.row_fraction = row_fraction

        def on(self, data, columns=None):
            return self.run(data, columns)

        def run(self, data, columns=None):
            rows, cols = data.shape
            list_of_cols = list(data.columns)
            if columns is None:
                # random 'corrupting'
                columns = np.random.choice(range(cols),
                                           int(np.ceil(
                                               self.column_fraction*cols)),
                                           replace=False)
            elif isinstance(columns, str):
                profiles = DataFrameProfiler().on(data).profiles
                if columns == 'string':
                    columns = [list_of_cols.index(p.column_name)
                               for p in profiles if p.dtype == DataType.STRING]
                elif columns == 'numeric':
                    columns = [list_of_cols.index(p.column_name)
                               for p in profiles if p.dtype
                               in [DataType.INTEGER, DataType.FLOAT]]
                else:
                    columns = np.random.choice(range(cols),
                                               int(np.ceil(
                                                   self.column_fraction*cols)),
                                               replace=False)
            else:
                columns = [list_of_cols.index(col) for col in columns]
            tmp = dict({})
            for col in columns:
                tmp[col] = np.random.choice(
                    range(rows),
                    int(np.ceil(self.row_fraction*rows)),
                    replace=False)
            return tmp

    instance = None

    def __init__(self, column_fraction=.2, row_fraction=.1):
        if not RandomSelector.instance:
            RandomSelector.instance = (RandomSelector
                                       .__RandomSelector(column_fraction,
                                                         row_fraction))
        else:
            RandomSelector.instance.column_fraction = column_fraction
            RandomSelector.instance.row_fraction = row_fraction

    def __getattr__(self, name):
        return getattr(self.instance, name)


class PairSelector(Selector):
    class __PairSelector:
        def __init__(self, row_fraction=.1):
            # Selector.__init__(self)
            self.row_fraction = row_fraction

        def on(self, data, columns=None):
            return self.run(data, columns)

        def run(self, data, columns=None):
            rows, cols = data.shape
            list_of_cols = list(data.columns)
            if columns is None:
                # random 'corrupting'
                # columns = np.random.choice(range(cols), 2, replace=False)
                columns = 'string'
            if isinstance(columns, str):
                modes = ['string', 'numeric']
                assert columns in modes, mes().wrong_value % columns
                profiles = DataFrameProfiler().on(data).profiles
                if columns == 'string':
                    tmp = [DataType.STRING]
                elif columns == 'numeric':
                    tmp = [DataType.INTEGER, DataType.FLOAT]
                columns = [list_of_cols.index(p.column_name)
                           for p in profiles if p.dtype in tmp]
            else:
                assert isinstance(columns, list), mes().wrong_value % columns
                columns = [list_of_cols.index(col) for col in columns]
            columns = np.random.choice(columns, 2, replace=False)
            tmp = dict({})
            rows = np.random.choice(range(rows),
                                    int(np.ceil(self.row_fraction*rows)),
                                    replace=False)
            assert np.unique(columns).shape[0] > 1, "! not enough columns"
            for col in columns:
                tmp[col] = rows
            return tmp

    instance = None

    def __init__(self, column_fraction=.2, row_fraction=.1):
        if not PairSelector.instance:
            PairSelector.instance = (PairSelector
                                     .__PairSelector(row_fraction))
        else:
            PairSelector.instance.row_fraction = row_fraction

    def __getattr__(self, name):
        return getattr(self.instance, name)


class PartitionSelector(Selector):
    def __init__(self, row_fraction=.2):
        self.row_fraction = row_fraction

    def on(self, data, where):
        return self.run(data, where)

    def run(self, data, where):
        pass


def main():
    """
    """
    pass


if __name__ == "__main__":
    main()