hyperactive.optimizers.search.Search._search() - Code Metrics - Inspection of "V5 api design" - SimonBlanke/Hyperactive - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#101)

by Simon

created 2025-03-16 17:31 UTC

hyperactive.optimizers.search.Search._search() B

↳ Parent: hyperactive.optimizers.search

Complexity

Conditions

Size

Total Lines	62
Code Lines	47

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	5
eloc	47
nop	2
dl	0
loc	62
rs	8.2678
c	0
b	0
f	0

How to fix Long Method

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

from .objective_function import ObjectiveFunction
from .hyper_gradient_conv import HyperGradientConv
from .optimizer_attributes import OptimizerAttributes
from .constraint import Constraint


class Search(OptimizerAttributes):
    max_time: float
    nth_process: int

    def __init__(self, optimizer_class, opt_params):
        super().__init__()
        self.optimizer_class = optimizer_class
        self.opt_params = opt_params

    def setup(
        self,
        experiment,
        s_space,
        n_iter,
        initialize,
        constraints,
        pass_through,
        max_score,
        early_stopping,
        random_state,
        memory,
        memory_warm_start,
        verbosity,
    ):
        self.experiment = experiment
        self.s_space = s_space
        self.n_iter = n_iter

        self.initialize = initialize
        self.constraints = constraints
        self.pass_through = pass_through
        self.max_score = max_score
        self.early_stopping = early_stopping
        self.random_state = random_state
        self.memory = memory
        self.memory_warm_start = memory_warm_start
        self.verbosity = verbosity

        if "progress_bar" in self.verbosity:
            self.verbosity = ["progress_bar"]
        else:
            self.verbosity = []

    def pass_args(self, max_time, nth_process):
        self.max_time = max_time
        self.nth_process = nth_process

    def convert_results2hyper(self):

        self.eval_times = sum(self.gfo_optimizer.eval_times)
        self.iter_times = sum(self.gfo_optimizer.iter_times)

        if self.gfo_optimizer.best_para is not None:
            value = self.hg_conv.para2value(self.gfo_optimizer.best_para)
            position = self.hg_conv.position2value(value)
            best_para = self.hg_conv.value2para(position)
            self.best_para = best_para
        else:
            self.best_para = None

        self.best_score = self.gfo_optimizer.best_score
        self.positions = self.gfo_optimizer.search_data
        self.search_data = self.hg_conv.positions2results(self.positions)

        results_dd = self.gfo_optimizer.search_data.drop_duplicates(
            subset=self.s_space.dim_keys, keep="first"
        )
        self.memory_values_df = results_dd[
            self.s_space.dim_keys + ["score"]
        ].reset_index(drop=True)

    def _setup_process(self):

        self.hg_conv = HyperGradientConv(self.s_space)

        initialize = self.hg_conv.conv_initialize(self.initialize)
        search_space_positions = self.s_space.positions

        # conv warm start for smbo from values into positions
        if "warm_start_smbo" in self.opt_params:
            self.opt_params["warm_start_smbo"] = (
                self.hg_conv.conv_memory_warm_start(
                    self.opt_params["warm_start_smbo"]
                )
            )

        gfo_constraints = [
            Constraint(constraint, self.s_space)
            for constraint in self.constraints
        ]

        self.gfo_optimizer = self.optimizer_class(
            search_space=search_space_positions,
            initialize=initialize,
            constraints=gfo_constraints,
            random_state=self.random_state,
            nth_process=self.nth_process,
            **self.opt_params,
        )

        self.conv = self.gfo_optimizer.conv

    def _search(self, p_bar):
        self._setup_process()

        gfo_wrapper_model = ObjectiveFunction(
            experiment=self.experiment,
        )
        gfo_wrapper_model.pass_through = self.pass_through

        memory_warm_start = self.hg_conv.conv_memory_warm_start(
            self.memory_warm_start
        )

        gfo_objective_function = gfo_wrapper_model(self.s_space())

        self.gfo_optimizer.init_search(
            gfo_objective_function,
            self.n_iter,
            self.max_time,
            self.max_score,
            self.early_stopping,
            self.memory,
            memory_warm_start,
            False,
        )
        for nth_iter in range(self.n_iter):
            if p_bar:
                p_bar.set_description(
                    "["
                    + str(self.nth_process)
                    + "] "
                    + str(self.experiment.__class__.__name__)
                    + " ("
                    + self.optimizer_class.name
                    + ")",
                )

            self.gfo_optimizer.search_step(nth_iter)
            if self.gfo_optimizer.stop.check():
                break

            if p_bar:
                p_bar.set_postfix(
                    best_score=str(self.gfo_optimizer.score_best),
                    best_pos=str(self.gfo_optimizer.pos_best),
                    best_iter=str(self.gfo_optimizer.p_bar._best_since_iter),
                )

                p_bar.update(1)
                p_bar.refresh()

        self.gfo_optimizer.finish_search()

        self.convert_results2hyper()

        self._add_result_attributes(
            self.best_para,
            self.best_score,
            self.gfo_optimizer.p_bar._best_since_iter,
            self.eval_times,
            self.iter_times,
            self.search_data,
            self.gfo_optimizer.random_seed,
        )


1		# Author: Simon Blanke
2		# Email: [email protected]
3		# License: MIT License
4
5		from .objective_function import ObjectiveFunction
6		from .hyper_gradient_conv import HyperGradientConv
7		from .optimizer_attributes import OptimizerAttributes
8		from .constraint import Constraint
9
10
11		class Search(OptimizerAttributes):
12		max_time: float
13		nth_process: int
14
15		def __init__(self, optimizer_class, opt_params):
16		super().__init__()
17		self.optimizer_class = optimizer_class
18		self.opt_params = opt_params
19
20		def setup(
21		self,
22		experiment,
23		s_space,
24		n_iter,
25		initialize,
26		constraints,
27		pass_through,
28		max_score,
29		early_stopping,
30		random_state,
31		memory,
32		memory_warm_start,
33		verbosity,
34		):
35		self.experiment = experiment
36		self.s_space = s_space
37		self.n_iter = n_iter
38
39		self.initialize = initialize
40		self.constraints = constraints
41		self.pass_through = pass_through
42		self.max_score = max_score
43		self.early_stopping = early_stopping
44		self.random_state = random_state
45		self.memory = memory
46		self.memory_warm_start = memory_warm_start
47		self.verbosity = verbosity
48
49		if "progress_bar" in self.verbosity:
50		self.verbosity = ["progress_bar"]
51		else:
52		self.verbosity = []
53
54		def pass_args(self, max_time, nth_process):
55		self.max_time = max_time
56		self.nth_process = nth_process
57
58	View Code Duplication	def convert_results2hyper(self):
		0 ignored issues – show Duplication introduced 2025-02-23 16:47 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
59		self.eval_times = sum(self.gfo_optimizer.eval_times)
60		self.iter_times = sum(self.gfo_optimizer.iter_times)
61
62		if self.gfo_optimizer.best_para is not None:
63		value = self.hg_conv.para2value(self.gfo_optimizer.best_para)
64		position = self.hg_conv.position2value(value)
65		best_para = self.hg_conv.value2para(position)
66		self.best_para = best_para
67		else:
68		self.best_para = None
69
70		self.best_score = self.gfo_optimizer.best_score
71		self.positions = self.gfo_optimizer.search_data
72		self.search_data = self.hg_conv.positions2results(self.positions)
73
74		results_dd = self.gfo_optimizer.search_data.drop_duplicates(
75		subset=self.s_space.dim_keys, keep="first"
76		)
77		self.memory_values_df = results_dd[
78		self.s_space.dim_keys + ["score"]
79		].reset_index(drop=True)
80
81	View Code Duplication	def _setup_process(self):
		0 ignored issues – show Duplication introduced 2025-02-23 16:47 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
82		self.hg_conv = HyperGradientConv(self.s_space)
83
84		initialize = self.hg_conv.conv_initialize(self.initialize)
85		search_space_positions = self.s_space.positions
86
87		# conv warm start for smbo from values into positions
88		if "warm_start_smbo" in self.opt_params:
89		self.opt_params["warm_start_smbo"] = (
90		self.hg_conv.conv_memory_warm_start(
91		self.opt_params["warm_start_smbo"]
92		)
93		)
94
95		gfo_constraints = [
96		Constraint(constraint, self.s_space)
97		for constraint in self.constraints
98		]
99
100		self.gfo_optimizer = self.optimizer_class(
101		search_space=search_space_positions,
102		initialize=initialize,
103		constraints=gfo_constraints,
104		random_state=self.random_state,
105		nth_process=self.nth_process,
106		**self.opt_params,
107		)
108
109		self.conv = self.gfo_optimizer.conv
110
111		def _search(self, p_bar):
112		self._setup_process()
113
114		gfo_wrapper_model = ObjectiveFunction(
115		experiment=self.experiment,
116		)
117		gfo_wrapper_model.pass_through = self.pass_through
118
119		memory_warm_start = self.hg_conv.conv_memory_warm_start(
120		self.memory_warm_start
121		)
122
123		gfo_objective_function = gfo_wrapper_model(self.s_space())
124
125		self.gfo_optimizer.init_search(
126		gfo_objective_function,
127		self.n_iter,
128		self.max_time,
129		self.max_score,
130		self.early_stopping,
131		self.memory,
132		memory_warm_start,
133		False,
134		)
135		for nth_iter in range(self.n_iter):
136		if p_bar:
137		p_bar.set_description(
138		"["
139		+ str(self.nth_process)
140		+ "] "
141		+ str(self.experiment.__class__.__name__)
142		+ " ("
143		+ self.optimizer_class.name
144		+ ")",
145		)
146
147		self.gfo_optimizer.search_step(nth_iter)
148		if self.gfo_optimizer.stop.check():
149		break
150
151		if p_bar:
152		p_bar.set_postfix(
153		best_score=str(self.gfo_optimizer.score_best),
154		best_pos=str(self.gfo_optimizer.pos_best),
155		best_iter=str(self.gfo_optimizer.p_bar._best_since_iter),
156		)
157
158		p_bar.update(1)
159		p_bar.refresh()
160
161		self.gfo_optimizer.finish_search()
162
163		self.convert_results2hyper()
164
165		self._add_result_attributes(
166		self.best_para,
167		self.best_score,
168		self.gfo_optimizer.p_bar._best_since_iter,
169		self.eval_times,
170		self.iter_times,
171		self.search_data,
172		self.gfo_optimizer.random_seed,
173		)
174

SimonBlanke / Hyperactive

Pull Request — master (#101)

hyperactive.optimizers.search.Search._search() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like