Passed
Push — master ( e5596f...be5baa )
by Simon
01:48
created

memory_warning_2()   A

Complexity

Conditions 2

Size

Total Lines 6
Code Lines 6

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
cc 2
eloc 6
nop 1
dl 0
loc 6
rs 10
c 0
b 0
f 0
1
# Author: Simon Blanke
2
# Email: [email protected]
3
# License: MIT License
4
5
import warnings
6
import numpy as np
7
from itertools import compress
8
9
np.seterr(divide="ignore", invalid="ignore")
10
11
from ..base_optimizer import BaseOptimizer
12
from ...search import Search
13
14
15
def memory_warning_1(search_space_size):
16
    if search_space_size > 1000000:
17
        warning_message0 = "\n Warning:"
18
        warning_message1 = "\n search space too large for smb-optimization."
19
        warning_message3 = "\n Please reduce search space size for better performance."
20
        print(warning_message0 + warning_message1 + warning_message3)
21
22
23
def memory_warning_2(all_pos_comb):
24
    all_pos_comb_gbyte = all_pos_comb.nbytes / 1000000000
25
    if all_pos_comb_gbyte > 1:
26
        warning_message0 = "\n Warning:"
27
        warning_message2 = "\n Memory-load exceeding recommended limit."
28
        print(warning_message0 + warning_message2)
29
30
31
class SMBO(BaseOptimizer, Search):
32
    def __init__(
33
        self,
34
        search_space,
35
        initialize={"grid": 4, "random": 2, "vertices": 4},
36
        warm_start_smbo=None,
37
    ):
38
        super().__init__(search_space, initialize)
39
        self.warm_start_smbo = warm_start_smbo
40
41
        search_space_size = 1
42
        for value_ in search_space.values():
43
            search_space_size *= len(value_)
44
45
        self.X_sample = []
46
        self.Y_sample = []
47
48
        memory_warning_1(search_space_size)
49
        self.all_pos_comb = self._all_possible_pos()
50
        memory_warning_2(self.all_pos_comb)
51
52
    def init_warm_start_smbo(self):
53
        if self.warm_start_smbo is not None:
54
            X_sample_values = self.warm_start_smbo[self.conv.para_names].values
55
            Y_sample = self.warm_start_smbo["score"].values
56
57
            self.X_sample = self.conv.values2positions(X_sample_values)
58
            self.Y_sample = list(Y_sample)
59
60
            # filter out nan
61
            mask = ~np.isnan(Y_sample)
62
            self.X_sample = list(compress(self.X_sample, mask))
63
            self.Y_sample = list(compress(self.Y_sample, mask))
64
65
    def track_X_sample(func):
66
        def wrapper(self, *args, **kwargs):
67
            pos = func(self, *args, **kwargs)
68
            self.X_sample.append(pos)
69
            return pos
70
71
        return wrapper
72
73
    def _all_possible_pos(self):
74
        pos_space = []
75
        for dim_ in self.conv.max_positions:
76
            pos_space.append(np.arange(dim_))
77
78
        n_dim = len(pos_space)
79
        return np.array(np.meshgrid(*pos_space)).T.reshape(-1, n_dim)
80
81
    @track_X_sample
82
    def init_pos(self, pos):
83
        super().init_pos(pos)
84
        return pos
85