Passed
Push — master ( e85313...dd7f59 )
by Simon
01:21
created

hyperactive.hyperactive.Hyperactive.add_search()   A

Complexity

Conditions 4

Size

Total Lines 38
Code Lines 33

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
cc 4
eloc 33
nop 12
dl 0
loc 38
rs 9.0879
c 0
b 0
f 0

How to fix   Many Parameters   

Many Parameters

Methods with many parameters are not only hard to understand, but their parameters also often become inconsistent when you need more, or different data.

There are several approaches to avoid long parameter lists:

1
# Author: Simon Blanke
2
# Email: [email protected]
3
# License: MIT License
4
5
6
import multiprocessing
7
from tqdm import tqdm
8
9
from .optimizers import RandomSearchOptimizer
10
from .run_search import run_search
11
12
13
def set_n_jobs(n_jobs):
14
    """Sets the number of jobs to run in parallel"""
15
    num_cores = multiprocessing.cpu_count()
16
    if n_jobs == -1 or n_jobs > num_cores:
17
        return num_cores
18
    else:
19
        return n_jobs
20
21
22
class HyperactiveResults:
23
    def __init__(*args, **kwargs):
24
        pass
25
26
    def _sort_results_objFunc(self, objective_function):
27
        import numpy as np
28
        import pandas as pd
29
30
        best_score = -np.inf
31
        best_para = None
32
        results_list = []
33
34
        for results in self.results_list:
35
            nth_process = results["nth_process"]
36
37
            process_infos = self.process_infos[nth_process]
38
            objective_function_ = process_infos["objective_function"]
39
40
            if objective_function_ != objective_function:
41
                continue
42
43
            if results["best_score"] > best_score:
44
                best_score = results["best_score"]
45
                best_para = results["best_para"]
46
47
            results_list.append(results["results"])
48
49
        results = pd.concat(results_list)
50
51
        self.objFunc2results[objective_function] = {
52
            "best_para": best_para,
53
            "best_score": best_score,
54
            "results": results,
55
        }
56
57
    def _sort_results_search_id(self, search_id):
58
        for results in self.results_list:
59
            nth_process = results["nth_process"]
60
            search_id_ = self.process_infos[nth_process]["search_id"]
61
62
            if search_id_ != search_id:
63
                continue
64
65
            best_score = results["best_score"]
66
            best_para = results["best_para"]
67
            results = results["results"]
68
69
            self.search_id2results[search_id] = {
70
                "best_para": best_para,
71
                "best_score": best_score,
72
                "results": results,
73
            }
74
75
    def _get_one_result(self, id_, result_name):
76
        if isinstance(id_, str):
77
            if id_ not in self.search_id2results:
78
                self._sort_results_search_id(id_)
79
80
            return self.search_id2results[id_][result_name]
81
82
        else:
83
            if id_ not in self.objFunc2results:
84
                self._sort_results_objFunc(id_)
85
86
            return self.objFunc2results[id_][result_name]
87
88
    def best_para(self, id_):
89
        return self._get_one_result(id_, "best_para")
90
91
    def best_score(self, id_):
92
        return self._get_one_result(id_, "best_score")
93
94
    def results(self, id_):
95
        return self._get_one_result(id_, "results")
96
97
98
class Hyperactive(HyperactiveResults):
99
    def __init__(
100
        self,
101
        verbosity=["progress_bar", "print_results", "print_times"],
102
        distribution={
103
            "multiprocessing": {
104
                "initializer": tqdm.set_lock,
105
                "initargs": (tqdm.get_lock(),),
106
            }
107
        },
108
    ):
109
        super().__init__()
110
        if verbosity is False:
111
            verbosity = []
112
113
        self.verbosity = verbosity
114
        self.distribution = distribution
115
        self.search_ids = []
116
117
        self.process_infos = {}
118
119
        self.objFunc2results = {}
120
        self.search_id2results = {}
121
122
    def _add_search_processes(
123
        self,
124
        random_state,
125
        objective_function,
126
        search_space,
127
        optimizer,
128
        n_iter,
129
        n_jobs,
130
        max_score,
131
        memory,
132
        memory_warm_start,
133
        search_id,
134
    ):
135
        for nth_job in range(set_n_jobs(n_jobs)):
136
            nth_process = len(self.process_infos)
137
138
            self.process_infos[nth_process] = {
139
                "random_state": random_state,
140
                "verbosity": self.verbosity,
141
                "nth_process": nth_process,
142
                "objective_function": objective_function,
143
                "search_space": search_space,
144
                "optimizer": optimizer,
145
                "n_iter": n_iter,
146
                "max_score": max_score,
147
                "memory": memory,
148
                "memory_warm_start": memory_warm_start,
149
                "search_id": search_id,
150
            }
151
152
    def add_search(
153
        self,
154
        objective_function,
155
        search_space,
156
        n_iter,
157
        search_id=None,
158
        optimizer="default",
159
        n_jobs=1,
160
        initialize={"grid": 4, "random": 2, "vertices": 4},
161
        max_score=None,
162
        random_state=None,
163
        memory=True,
164
        memory_warm_start=None,
165
    ):
166
167
        if isinstance(optimizer, str):
168
            if optimizer == "default":
169
                optimizer = RandomSearchOptimizer()
170
        optimizer.init(search_space, initialize)
171
172
        if search_id is not None:
173
            search_id = search_id
174
            self.search_ids.append(search_id)
175
        else:
176
            search_id = str(len(self.search_ids))
177
            self.search_ids.append(search_id)
178
179
        self._add_search_processes(
180
            random_state,
181
            objective_function,
182
            search_space,
183
            optimizer,
184
            n_iter,
185
            n_jobs,
186
            max_score,
187
            memory,
188
            memory_warm_start,
189
            search_id,
190
        )
191
192
    def run(
193
        self,
194
        max_time=None,
195
    ):
196
        for nth_process in self.process_infos.keys():
197
            self.process_infos[nth_process]["max_time"] = max_time
198
199
        self.results_list = run_search(self.process_infos, self.distribution)
200