hyperactive.optimizers.gfo2hyper() - Code Metrics - Inspection of "add isolated tests for optimizer-wrapper-class" - SimonBlanke/Hyperactive - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 8e1ab0...b032b3 )

by Simon

created 2021-02-02 07:03 UTC

hyperactive.optimizers.gfo2hyper() A

↳ Parent: hyperactive.optimizers

Complexity

Conditions

Size

Total Lines	7
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	6
nop	2
dl	0
loc	7
rs	10
c	0
b	0
f	0

# Author: Simon Blanke
# Email: [email protected]
# License: MIT License

import numpy as np
import pandas as pd

from gradient_free_optimizers import (
    HillClimbingOptimizer as _HillClimbingOptimizer,
    StochasticHillClimbingOptimizer as _StochasticHillClimbingOptimizer,
    RepulsingHillClimbingOptimizer as _RepulsingHillClimbingOptimizer,
    RandomSearchOptimizer as _RandomSearchOptimizer,
    RandomRestartHillClimbingOptimizer as _RandomRestartHillClimbingOptimizer,
    RandomAnnealingOptimizer as _RandomAnnealingOptimizer,
    SimulatedAnnealingOptimizer as _SimulatedAnnealingOptimizer,
    ParallelTemperingOptimizer as _ParallelTemperingOptimizer,
    ParticleSwarmOptimizer as _ParticleSwarmOptimizer,
    EvolutionStrategyOptimizer as _EvolutionStrategyOptimizer,
    BayesianOptimizer as _BayesianOptimizer,
    TreeStructuredParzenEstimators as _TreeStructuredParzenEstimators,
    DecisionTreeOptimizer as _DecisionTreeOptimizer,
    EnsembleOptimizer as _EnsembleOptimizer,
)

from .hyper_gradient_trafo import HyperGradientTrafo


def gfo2hyper(search_space, para):
    values_dict = {}
    for i, key in enumerate(search_space.keys()):
        pos_ = int(para[key])
        values_dict[key] = search_space[key][pos_]

    return values_dict


class DictClass:
    def __init__(self):
        self.para_dict = {}

    def __getitem__(self, key):
        return self.para_dict[key]

    def keys(self):
        return self.para_dict.keys()

    def values(self):
        return self.para_dict.values()


class TrafoClass:
    def __init__(self, *args, **kwargs):
        pass

    def _convert_args2gfo(self, memory_warm_start):
        memory_warm_start = self.trafo.trafo_memory_warm_start(memory_warm_start)

        return memory_warm_start

    def _positions2results(self, positions):
        results_dict = {}

        for para_name in self.conv.para_names:
            values_list = self.search_space[para_name]
            pos_ = positions[para_name].values
            values_ = [values_list[idx] for idx in pos_]
            results_dict[para_name] = values_

        results = pd.DataFrame.from_dict(results_dict)

        diff_list = np.setdiff1d(positions.columns, results.columns)
        results[diff_list] = positions[diff_list]

        return results

    def _convert_results2hyper(self):
        self.eval_time = np.array(self.optimizer.eval_times).sum()
        self.iter_time = np.array(self.optimizer.iter_times).sum()

        value = self.trafo.para2value(self.optimizer.best_para)
        position = self.trafo.position2value(value)
        best_para = self.trafo.value2para(position)

        self.best_para = best_para
        self.best_score = self.optimizer.best_score
        self.positions = self.optimizer.results

        self.results = self._positions2results(self.positions)

        self.memory_positions = self.trafo._memory2dataframe(self.optimizer.memory_dict)
        self.memory_values_df = self._positions2results(self.memory_positions)


class _BaseOptimizer_(DictClass, TrafoClass):
    def __init__(self, **opt_params):
        super().__init__()
        self.opt_params = opt_params

    def init(self, search_space, initialize={"grid": 8, "random": 4, "vertices": 8}):
        self.search_space = search_space

        self.trafo = HyperGradientTrafo(search_space)

        initialize = self.trafo.trafo_initialize(initialize)
        search_space_positions = self.trafo.search_space_positions

        self.optimizer = self._OptimizerClass(
            search_space_positions, initialize, **self.opt_params
        )

        self.conv = self.optimizer.conv

    def print_info(self, *args):
        self.optimizer.print_info(*args)

    def search(
        self,
        objective_function,
        n_iter,
        warm_start=None,
        max_time=None,
        max_score=None,
        memory=True,
        memory_warm_start=None,
        verbosity={
            "progress_bar": True,
            "print_results": True,
            "print_times": True,
        },
        random_state=None,
        nth_process=None,
    ):
        memory_warm_start = self._convert_args2gfo(memory_warm_start)

        def gfo_wrapper_model():
            # wrapper for GFOs
            def _model(para):
                para = gfo2hyper(self.search_space, para)
                self.para_dict = para
                return objective_function(self)

            _model.__name__ = objective_function.__name__
            return _model

        self.optimizer.search(
            gfo_wrapper_model(),
            n_iter,
            max_time,
            max_score,
            memory,
            memory_warm_start,
            verbosity,
            random_state,
            nth_process,
        )

        self._convert_results2hyper()


class HillClimbingOptimizer(_BaseOptimizer_):
    def __init__(self, **opt_params):
        super().__init__(**opt_params)
        self._OptimizerClass = _HillClimbingOptimizer


class StochasticHillClimbingOptimizer(_BaseOptimizer_):
    def __init__(self, **opt_params):
        super().__init__(**opt_params)
        self._OptimizerClass = _StochasticHillClimbingOptimizer


class RepulsingHillClimbingOptimizer(_BaseOptimizer_):
    def __init__(self, **opt_params):
        super().__init__(**opt_params)
        self._OptimizerClass = _RepulsingHillClimbingOptimizer


class RandomSearchOptimizer(_BaseOptimizer_):
    def __init__(self, **opt_params):
        super().__init__(**opt_params)
        self._OptimizerClass = _RandomSearchOptimizer


class RandomRestartHillClimbingOptimizer(_BaseOptimizer_):
    def __init__(self, **opt_params):
        super().__init__(**opt_params)
        self._OptimizerClass = _RandomRestartHillClimbingOptimizer


class RandomAnnealingOptimizer(_BaseOptimizer_):
    def __init__(self, **opt_params):
        super().__init__(**opt_params)
        self._OptimizerClass = _RandomAnnealingOptimizer


class SimulatedAnnealingOptimizer(_BaseOptimizer_):
    def __init__(self, **opt_params):
        super().__init__(**opt_params)
        self._OptimizerClass = _SimulatedAnnealingOptimizer


class ParallelTemperingOptimizer(_BaseOptimizer_):
    def __init__(self, **opt_params):
        super().__init__(**opt_params)
        self._OptimizerClass = _ParallelTemperingOptimizer


class ParticleSwarmOptimizer(_BaseOptimizer_):
    def __init__(self, **opt_params):
        super().__init__(**opt_params)
        self._OptimizerClass = _ParticleSwarmOptimizer


class EvolutionStrategyOptimizer(_BaseOptimizer_):
    def __init__(self, **opt_params):
        super().__init__(**opt_params)
        self._OptimizerClass = _EvolutionStrategyOptimizer


class BayesianOptimizer(_BaseOptimizer_):
    def __init__(self, **opt_params):
        super().__init__(**opt_params)
        self._OptimizerClass = _BayesianOptimizer


class TreeStructuredParzenEstimators(_BaseOptimizer_):
    def __init__(self, **opt_params):
        super().__init__(**opt_params)
        self._OptimizerClass = _TreeStructuredParzenEstimators


class DecisionTreeOptimizer(_BaseOptimizer_):
    def __init__(self, **opt_params):
        super().__init__(**opt_params)
        self._OptimizerClass = _DecisionTreeOptimizer


class EnsembleOptimizer(_BaseOptimizer_):
    def __init__(self, **opt_params):
        super().__init__(**opt_params)
        self._OptimizerClass = _EnsembleOptimizer


1			# Author: Simon Blanke
2			# Email: [email protected]
3			# License: MIT License
4
5			import numpy as np
6			import pandas as pd
7
8			from gradient_free_optimizers import (
9			HillClimbingOptimizer as _HillClimbingOptimizer,
10			StochasticHillClimbingOptimizer as _StochasticHillClimbingOptimizer,
11			RepulsingHillClimbingOptimizer as _RepulsingHillClimbingOptimizer,
12			RandomSearchOptimizer as _RandomSearchOptimizer,
13			RandomRestartHillClimbingOptimizer as _RandomRestartHillClimbingOptimizer,
14			RandomAnnealingOptimizer as _RandomAnnealingOptimizer,
15			SimulatedAnnealingOptimizer as _SimulatedAnnealingOptimizer,
16			ParallelTemperingOptimizer as _ParallelTemperingOptimizer,
17			ParticleSwarmOptimizer as _ParticleSwarmOptimizer,
18			EvolutionStrategyOptimizer as _EvolutionStrategyOptimizer,
19			BayesianOptimizer as _BayesianOptimizer,
20			TreeStructuredParzenEstimators as _TreeStructuredParzenEstimators,
21			DecisionTreeOptimizer as _DecisionTreeOptimizer,
22			EnsembleOptimizer as _EnsembleOptimizer,
23			)
24
25			from .hyper_gradient_trafo import HyperGradientTrafo
26
27
28			def gfo2hyper(search_space, para):
29			values_dict = {}
30			for i, key in enumerate(search_space.keys()):
31			pos_ = int(para[key])
32			values_dict[key] = search_space[key][pos_]
33
34			return values_dict
35
36
37			class DictClass:
38			def __init__(self):
39			self.para_dict = {}
40
41			def __getitem__(self, key):
42			return self.para_dict[key]
43
44			def keys(self):
45			return self.para_dict.keys()
46
47			def values(self):
48			return self.para_dict.values()
49
50
51			class TrafoClass:
52			def __init__(self, args, *kwargs):
53			pass
54
55			def _convert_args2gfo(self, memory_warm_start):
56			memory_warm_start = self.trafo.trafo_memory_warm_start(memory_warm_start)
57
58			return memory_warm_start
59
60			def _positions2results(self, positions):
61			results_dict = {}
62
63			for para_name in self.conv.para_names:
64			values_list = self.search_space[para_name]
65			pos_ = positions[para_name].values
66			values_ = [values_list[idx] for idx in pos_]
67			results_dict[para_name] = values_
68
69			results = pd.DataFrame.from_dict(results_dict)
70
71			diff_list = np.setdiff1d(positions.columns, results.columns)
72			results[diff_list] = positions[diff_list]
73
74			return results
75
76			def _convert_results2hyper(self):
77			self.eval_time = np.array(self.optimizer.eval_times).sum()
78			self.iter_time = np.array(self.optimizer.iter_times).sum()
79
80			value = self.trafo.para2value(self.optimizer.best_para)
81			position = self.trafo.position2value(value)
82			best_para = self.trafo.value2para(position)
83
84			self.best_para = best_para
85			self.best_score = self.optimizer.best_score
86			self.positions = self.optimizer.results
87
88			self.results = self._positions2results(self.positions)
89
90			self.memory_positions = self.trafo._memory2dataframe(self.optimizer.memory_dict)
91			self.memory_values_df = self._positions2results(self.memory_positions)
92
93
94			class _BaseOptimizer_(DictClass, TrafoClass):
95			def __init__(self, **opt_params):
96			super().__init__()
97			self.opt_params = opt_params
98
99			def init(self, search_space, initialize={"grid": 8, "random": 4, "vertices": 8}):
100			self.search_space = search_space
101
102			self.trafo = HyperGradientTrafo(search_space)
103
104			initialize = self.trafo.trafo_initialize(initialize)
105			search_space_positions = self.trafo.search_space_positions
106
107			self.optimizer = self._OptimizerClass(
108			search_space_positions, initialize, **self.opt_params
109			)
110
111			self.conv = self.optimizer.conv
112
113			def print_info(self, *args):
114			self.optimizer.print_info(*args)
115
116			def search(
117			self,
118			objective_function,
119			n_iter,
120			warm_start=None,
121			max_time=None,
122			max_score=None,
123			memory=True,
124			memory_warm_start=None,
125			verbosity={
126			"progress_bar": True,
127			"print_results": True,
128			"print_times": True,
129			},
130			random_state=None,
131			nth_process=None,
132			):
133			memory_warm_start = self._convert_args2gfo(memory_warm_start)
134
135			def gfo_wrapper_model():
136			# wrapper for GFOs
137			def _model(para):
138			para = gfo2hyper(self.search_space, para)
139			self.para_dict = para
140			return objective_function(self)
141
142			_model.__name__ = objective_function.__name__
143			return _model
144
145			self.optimizer.search(
146			gfo_wrapper_model(),
147			n_iter,
148			max_time,
149			max_score,
150			memory,
151			memory_warm_start,
152			verbosity,
153			random_state,
154			nth_process,
155			)
156
157			self._convert_results2hyper()
158
159
160			class HillClimbingOptimizer(_BaseOptimizer_):
161			def __init__(self, **opt_params):
162			super().__init__(**opt_params)
163			self._OptimizerClass = _HillClimbingOptimizer
164
165
166			class StochasticHillClimbingOptimizer(_BaseOptimizer_):
167			def __init__(self, **opt_params):
168			super().__init__(**opt_params)
169			self._OptimizerClass = _StochasticHillClimbingOptimizer
170
171
172			class RepulsingHillClimbingOptimizer(_BaseOptimizer_):
173			def __init__(self, **opt_params):
174			super().__init__(**opt_params)
175			self._OptimizerClass = _RepulsingHillClimbingOptimizer
176
177
178			class RandomSearchOptimizer(_BaseOptimizer_):
179			def __init__(self, **opt_params):
180			super().__init__(**opt_params)
181			self._OptimizerClass = _RandomSearchOptimizer
182
183
184			class RandomRestartHillClimbingOptimizer(_BaseOptimizer_):
185			def __init__(self, **opt_params):
186			super().__init__(**opt_params)
187			self._OptimizerClass = _RandomRestartHillClimbingOptimizer
188
189
190			class RandomAnnealingOptimizer(_BaseOptimizer_):
191			def __init__(self, **opt_params):
192			super().__init__(**opt_params)
193			self._OptimizerClass = _RandomAnnealingOptimizer
194
195
196			class SimulatedAnnealingOptimizer(_BaseOptimizer_):
197			def __init__(self, **opt_params):
198			super().__init__(**opt_params)
199			self._OptimizerClass = _SimulatedAnnealingOptimizer
200
201
202			class ParallelTemperingOptimizer(_BaseOptimizer_):
203			def __init__(self, **opt_params):
204			super().__init__(**opt_params)
205			self._OptimizerClass = _ParallelTemperingOptimizer
206
207
208			class ParticleSwarmOptimizer(_BaseOptimizer_):
209			def __init__(self, **opt_params):
210			super().__init__(**opt_params)
211			self._OptimizerClass = _ParticleSwarmOptimizer
212
213
214			class EvolutionStrategyOptimizer(_BaseOptimizer_):
215			def __init__(self, **opt_params):
216			super().__init__(**opt_params)
217			self._OptimizerClass = _EvolutionStrategyOptimizer
218
219
220			class BayesianOptimizer(_BaseOptimizer_):
221			def __init__(self, **opt_params):
222			super().__init__(**opt_params)
223			self._OptimizerClass = _BayesianOptimizer
224
225
226			class TreeStructuredParzenEstimators(_BaseOptimizer_):
227			def __init__(self, **opt_params):
228			super().__init__(**opt_params)
229			self._OptimizerClass = _TreeStructuredParzenEstimators
230
231
232			class DecisionTreeOptimizer(_BaseOptimizer_):
233			def __init__(self, **opt_params):
234			super().__init__(**opt_params)
235			self._OptimizerClass = _DecisionTreeOptimizer
236
237
238			class EnsembleOptimizer(_BaseOptimizer_):
239			def __init__(self, **opt_params):
240			super().__init__(**opt_params)
241			self._OptimizerClass = _EnsembleOptimizer
242

SimonBlanke / Hyperactive

Push — master ( 8e1ab0...b032b3 )

hyperactive.optimizers.gfo2hyper() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like