gradient_free_optimizers.search.Search.search() - Code Metrics - Inspection of "v1.0.0" - SimonBlanke/Gradient-Free-Optimizers - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 626f23...be3b1e )

by Simon

created 2021-12-01 10:57 UTC

gradient_free_optimizers.search.Search.search() C

↳ Parent: gradient_free_optimizers.search

Complexity

Conditions

Size

Total Lines	73
Code Lines	55

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	9
eloc	55
nop	9
dl	0
loc	73
rs	6.1393
c	0
b	0
f	0

How to fix Long Method Many Parameters

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

import time
import random

import numpy as np
import pandas as pd

from multiprocessing.managers import DictProxy

from .progress_bar import ProgressBarLVL0, ProgressBarLVL1
from .times_tracker import TimesTracker
from .memory import Memory
from .print_info import print_info
from .stop_run import StopRun


def set_random_seed(nth_process, random_state):

    """
    Sets the random seed separately for each thread
    (to avoid getting the same results in each thread)
    """
    if nth_process is None:
        nth_process = 0

    if random_state is None:
        random_state = np.random.randint(0, high=2 ** 31 - 2, dtype=np.int64)

    random.seed(random_state + nth_process)
    np.random.seed(random_state + nth_process)


class Search(TimesTracker):
    def __init__(self):
        super().__init__()

        self.optimizers = []
        self.new_results_list = []
        self.all_results_list = []

        self.score_l = []
        self.pos_l = []

    @TimesTracker.eval_time
    def _score(self, pos):
        return self.score(pos)

    @TimesTracker.iter_time
    def _initialization(self, init_pos, nth_iter):
        self.nth_iter = nth_iter
        self.best_score = self.p_bar.score_best

        self.init_pos(init_pos)

        score_new = self._score(init_pos)
        self.evaluate(score_new)

        self.pos_l.append(init_pos)
        self.score_l.append(score_new)

        self.p_bar.update(score_new, init_pos, nth_iter)

    @TimesTracker.iter_time
    def _iteration(self, nth_iter):
        self.nth_iter = nth_iter
        self.best_score = self.p_bar.score_best

        pos_new = self.iterate()

        score_new = self._score(pos_new)
        self.evaluate(score_new)

        self.pos_l.append(pos_new)
        self.score_l.append(score_new)

        self.p_bar.update(score_new, pos_new, nth_iter)

    def _init_search(self):
        if "progress_bar" in self.verbosity:
            self.p_bar = ProgressBarLVL1(
                self.nth_process, self.n_iter, self.objective_function
            )
        else:
            self.p_bar = ProgressBarLVL0(
                self.nth_process, self.n_iter, self.objective_function
            )

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

    def search(
        self,
        objective_function,
        n_iter,
        max_time=None,
        max_score=None,
        early_stopping=None,
        memory=True,
        memory_warm_start=None,
        verbosity=["progress_bar", "print_results", "print_times"],
    ):
        self.start_time = time.time()

        if verbosity is False:
            verbosity = []

        self.objective_function = objective_function
        self.n_iter = n_iter
        self.max_time = max_time
        self.max_score = max_score
        self.early_stopping = early_stopping
        self.memory = memory
        self.memory_warm_start = memory_warm_start
        self.verbosity = verbosity

        self.stop = StopRun(max_time, max_score, early_stopping)

        self._init_search()

        if isinstance(memory, DictProxy):
            mem = Memory(memory_warm_start, self.conv, dict_proxy=memory)
            self.score = self.results_mang.score(mem.memory(objective_function))
        elif memory is True:
            mem = Memory(memory_warm_start, self.conv)
            self.score = self.results_mang.score(mem.memory(objective_function))
        else:
            self.score = self.results_mang.score(objective_function)

        # loop to initialize N positions
        for init_pos, nth_iter in zip(self.init_positions, range(n_iter)):
            if self.stop.check(self.start_time, self.p_bar.score_best, self.score_l):
                break
            self._initialization(init_pos, nth_iter)

        self.finish_initialization()

        # loop to do the iterations
        for nth_iter in range(len(self.init_positions), n_iter):
            if self.stop.check(self.start_time, self.p_bar.score_best, self.score_l):
                break
            self._iteration(nth_iter)

        self.search_data = pd.DataFrame(self.results_mang.results_list)

        self.best_score = self.p_bar.score_best
        self.best_value = self.conv.position2value(self.p_bar.pos_best)
        self.best_para = self.conv.value2para(self.best_value)

        if memory not in [False, None]:
            self.memory_dict = mem.memory_dict

        else:
            self.memory_dict = {}

        self.p_bar.close()

        self.print_info(
            verbosity,
            self.objective_function,
            self.best_score,
            self.best_para,
            self.eval_times,
            self.iter_times,
            self.n_iter,
        )


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 multiprocessing.managers import DictProxy
12
13		from .progress_bar import ProgressBarLVL0, ProgressBarLVL1
14		from .times_tracker import TimesTracker
15		from .memory import Memory
16		from .print_info import print_info
17		from .stop_run import StopRun
18
19
20	View Code Duplication	def set_random_seed(nth_process, random_state):
		0 ignored issues – show Duplication introduced 2021-12-01 10:59 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
21		"""
22		Sets the random seed separately for each thread
23		(to avoid getting the same results in each thread)
24		"""
25		if nth_process is None:
26		nth_process = 0
27
28		if random_state is None:
29		random_state = np.random.randint(0, high=2 ** 31 - 2, dtype=np.int64)
30
31		random.seed(random_state + nth_process)
32		np.random.seed(random_state + nth_process)
33
34
35		class Search(TimesTracker):
36		def __init__(self):
37		super().__init__()
38
39		self.optimizers = []
40		self.new_results_list = []
41		self.all_results_list = []
42
43		self.score_l = []
44		self.pos_l = []
45
46		@TimesTracker.eval_time
47		def _score(self, pos):
48		return self.score(pos)
49
50		@TimesTracker.iter_time
51		def _initialization(self, init_pos, nth_iter):
52		self.nth_iter = nth_iter
53		self.best_score = self.p_bar.score_best
54
55		self.init_pos(init_pos)
56
57		score_new = self._score(init_pos)
58		self.evaluate(score_new)
59
60		self.pos_l.append(init_pos)
61		self.score_l.append(score_new)
62
63		self.p_bar.update(score_new, init_pos, nth_iter)
64
65		@TimesTracker.iter_time
66		def _iteration(self, nth_iter):
67		self.nth_iter = nth_iter
68		self.best_score = self.p_bar.score_best
69
70		pos_new = self.iterate()
71
72		score_new = self._score(pos_new)
73		self.evaluate(score_new)
74
75		self.pos_l.append(pos_new)
76		self.score_l.append(score_new)
77
78		self.p_bar.update(score_new, pos_new, nth_iter)
79
80		def _init_search(self):
81		if "progress_bar" in self.verbosity:
82		self.p_bar = ProgressBarLVL1(
83		self.nth_process, self.n_iter, self.objective_function
84		)
85		else:
86		self.p_bar = ProgressBarLVL0(
87		self.nth_process, self.n_iter, self.objective_function
88		)
89
90		def print_info(self, *args):
91		print_info(*args)
92
93		def search(
94		self,
95		objective_function,
96		n_iter,
97		max_time=None,
98		max_score=None,
99		early_stopping=None,
100		memory=True,
101		memory_warm_start=None,
102		verbosity=["progress_bar", "print_results", "print_times"],
103		):
104		self.start_time = time.time()
105
106		if verbosity is False:
107		verbosity = []
108
109		self.objective_function = objective_function
110		self.n_iter = n_iter
111		self.max_time = max_time
112		self.max_score = max_score
113		self.early_stopping = early_stopping
114		self.memory = memory
115		self.memory_warm_start = memory_warm_start
116		self.verbosity = verbosity
117
118		self.stop = StopRun(max_time, max_score, early_stopping)
119
120		self._init_search()
121
122		if isinstance(memory, DictProxy):
123		mem = Memory(memory_warm_start, self.conv, dict_proxy=memory)
124		self.score = self.results_mang.score(mem.memory(objective_function))
125		elif memory is True:
126		mem = Memory(memory_warm_start, self.conv)
127		self.score = self.results_mang.score(mem.memory(objective_function))
128		else:
129		self.score = self.results_mang.score(objective_function)
130
131		# loop to initialize N positions
132		for init_pos, nth_iter in zip(self.init_positions, range(n_iter)):
133		if self.stop.check(self.start_time, self.p_bar.score_best, self.score_l):
134		break
135		self._initialization(init_pos, nth_iter)
136
137		self.finish_initialization()
138
139		# loop to do the iterations
140		for nth_iter in range(len(self.init_positions), n_iter):
141		if self.stop.check(self.start_time, self.p_bar.score_best, self.score_l):
142		break
143		self._iteration(nth_iter)
144
145		self.search_data = pd.DataFrame(self.results_mang.results_list)
146
147		self.best_score = self.p_bar.score_best
148		self.best_value = self.conv.position2value(self.p_bar.pos_best)
149		self.best_para = self.conv.value2para(self.best_value)
150
151		if memory not in [False, None]:
152		self.memory_dict = mem.memory_dict
		0 ignored issues – show introduced 2021-01-11 15:10 UTC by Report Bug Copy Issue Report The variable `mem` does not seem to be defined for all execution paths. Loading history...
153		else:
154		self.memory_dict = {}
155
156		self.p_bar.close()
157
158		self.print_info(
159		verbosity,
160		self.objective_function,
161		self.best_score,
162		self.best_para,
163		self.eval_times,
164		self.iter_times,
165		self.n_iter,
166		)
167

SimonBlanke / Gradient-Free-Optimizers

Push — master ( 626f23...be3b1e )

gradient_free_optimizers.search.Search.search() C

Complexity

Size

Duplication

Importance

How to fix Long Method Many Parameters

Long Method

Many Parameters

Duplication Side-by-Side

Filter issues like