Passed
Push — master ( 7dc8c7...0f59c2 )
by Simon
03:25
created

Search._init_verb_dict()   A

Complexity

Conditions 4

Size

Total Lines 19
Code Lines 14

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
cc 4
eloc 14
nop 2
dl 0
loc 19
rs 9.7
c 0
b 0
f 0
1
# Author: Simon Blanke
2
# Email: [email protected]
3
# License: MIT License
4
5
import time
6
import random
7
8
import numpy as np
9
import pandas as pd
10
11
from .init_positions import Initializer
12
from .progress_bar import ProgressBarLVL0, ProgressBarLVL1
13
from .times_tracker import TimesTracker
14
from .memory import Memory
15
from .print_info import print_info
16
17
18
def time_exceeded(start_time, max_time):
19
    run_time = time.time() - start_time
20
    return max_time and run_time > max_time
21
22
23
def score_exceeded(score_best, max_score):
24
    return max_score and score_best >= max_score
25
26
27
def set_random_seed(nth_process, random_state):
28
    """
29
    Sets the random seed separately for each thread
30
    (to avoid getting the same results in each thread)
31
    """
32
    if nth_process is None:
33
        nth_process = 0
34
35
    if random_state is None:
36
        random_state = np.random.randint(0, high=2 ** 32 - 2)
37
38
    random.seed(random_state + nth_process)
39
    np.random.seed(random_state + nth_process)
40
41
42
class Search(TimesTracker):
43
    def __init__(self):
44
        super().__init__()
45
46
        self.optimizers = []
47
        self.new_results_list = []
48
        self.all_results_list = []
49
50
    @TimesTracker.eval_time
51
    def _score(self, pos):
52
        return self.score(pos)
53
54
    @TimesTracker.iter_time
55
    def _initialization(self, init_pos, nth_iter):
56
        self.init_pos(init_pos)
57
58
        score_new = self._score(init_pos)
59
        self.evaluate(score_new)
60
61
        self.p_bar.update(score_new, init_pos, nth_iter)
62
63
    @TimesTracker.iter_time
64
    def _iteration(self, nth_iter):
65
        pos_new = self.iterate()
66
67
        score_new = self._score(pos_new)
68
        self.evaluate(score_new)
69
70
        self.p_bar.update(score_new, pos_new, nth_iter)
71
72
    def _init_search(self):
73
        if "progress_bar" in self.verbosity:
74
            self.p_bar = ProgressBarLVL1(
75
                self.nth_process, self.n_iter, self.objective_function
76
            )
77
        else:
78
            self.p_bar = ProgressBarLVL0(
79
                self.nth_process, self.n_iter, self.objective_function
80
            )
81
82
        set_random_seed(self.nth_process, self.random_state)
83
84
        # get init positions
85
        init = Initializer(self.conv)
86
        init_positions = init.set_pos(self.initialize)
87
88
        return init_positions
89
90
    def _early_stop(self):
91
        if time_exceeded(self.start_time, self.max_time):
92
            return True
93
        elif score_exceeded(self.p_bar.score_best, self.max_score):
94
            return True
95
        else:
96
            return False
97
98
    def print_info(self, *args):
99
        print_info(*args)
100
101
    def search(
102
        self,
103
        objective_function,
104
        n_iter,
105
        max_time=None,
106
        max_score=None,
107
        memory=True,
108
        memory_warm_start=None,
109
        verbosity=["progress_bar", "print_results", "print_times"],
110
        random_state=None,
111
        nth_process=None,
112
    ):
113
        self.start_time = time.time()
114
115
        if verbosity is False:
116
            verbosity = []
117
118
        self.objective_function = objective_function
119
        self.n_iter = n_iter
120
        self.max_time = max_time
121
        self.max_score = max_score
122
        self.memory = memory
123
        self.memory_warm_start = memory_warm_start
124
        self.verbosity = verbosity
125
        self.random_state = random_state
126
        self.nth_process = nth_process
127
128
        init_positions = self._init_search()
129
130
        if memory is True:
131
            mem = Memory(memory_warm_start, self.conv)
132
            self.score = self.results_mang.score(mem.memory(objective_function))
133
        else:
134
            self.score = self.results_mang.score(objective_function)
135
136
        # loop to initialize N positions
137
        for init_pos, nth_iter in zip(init_positions, range(n_iter)):
138
            if self._early_stop():
139
                break
140
            self._initialization(init_pos, nth_iter)
141
142
        # loop to do the iterations
143
        for nth_iter in range(len(init_positions), n_iter):
144
            if self._early_stop():
145
                break
146
            self._iteration(nth_iter)
147
148
        self.results = pd.DataFrame(self.results_mang.results_list)
149
150
        self.best_score = self.p_bar.score_best
151
        self.best_value = self.conv.position2value(self.p_bar.pos_best)
152
        self.best_para = self.conv.value2para(self.best_value)
153
154
        self.results["eval_time"] = self.eval_times
155
        self.results["iter_time"] = self.iter_times
156
157
        if memory is not False:
158
            self.memory_dict = mem.memory_dict
0 ignored issues
show
introduced by
The variable mem does not seem to be defined for all execution paths.
Loading history...
159
        else:
160
            self.memory_dict = {}
161
162
        self.p_bar.close()
163
164
        self.print_info(
165
            verbosity,
166
            self.objective_function,
167
            self.best_score,
168
            self.best_para,
169
            self.eval_times,
170
            self.iter_times,
171
            self.n_iter,
172
        )
173