hyperactive.hyperactive.Hyperactive.add_search() - Code Metrics - Inspection of "Merge remote-tracking branch 'refs/remotes/origin/..." - SimonBlanke/Hyperactive - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 2e0454...bd7b54 )

by Simon

created 2024-09-28 09:50 UTC

hyperactive.hyperactive.Hyperactive.add_search() B

↳ Parent: hyperactive.hyperactive

Complexity

Conditions

Size

Total Lines	56
Code Lines	46

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	46
nop	17
dl	0
loc	56
rs	8.7672
c	0
b	0
f	0

How to fix Long Method Many Parameters

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


import copy
import multiprocessing as mp
import pandas as pd

from typing import Union, List, Dict, Type

from .optimizers import RandomSearchOptimizer
from .run_search import run_search

from .results import Results
from .print_results import PrintResults
from .search_space import SearchSpace


class Hyperactive:
    """
    Initialize the Hyperactive class to manage optimization processes.

    Parameters:
    - verbosity: List of verbosity levels (default: ["progress_bar", "print_results", "print_times"])
    - distribution: String indicating the distribution method (default: "multiprocessing")
    - n_processes: Number of processes to run in parallel or "auto" to determine automatically (default: "auto")

    Methods:
    - add_search: Add a new optimization search process with specified parameters
    - run: Execute the optimization searches
    - best_para: Get the best parameters for a specific search
    - best_score: Get the best score for a specific search
    - search_data: Get the search data for a specific search
    """

    def __init__(
        self,
        verbosity: list = ["progress_bar", "print_results", "print_times"],
        distribution: str = "multiprocessing",
        n_processes: Union[str, int] = "auto",
    ):
        super().__init__()
        if verbosity is False:
            verbosity = []

        self.verbosity = verbosity
        self.distribution = distribution
        self.n_processes = n_processes

        self.opt_pros = {}

    def _create_shared_memory(self):
        _bundle_opt_processes = {}

        for opt_pros in self.opt_pros.values():
            if opt_pros.memory != "share":
                continue
            name = opt_pros.objective_function.__name__

            _bundle_opt_processes.setdefault(name, []).append(opt_pros)

        for opt_pros_l in _bundle_opt_processes.values():
            # Check if the lengths of the search spaces of all optimizers in the list are the same.
            if (
                len(set(len(opt_pros.s_space()) for opt_pros in opt_pros_l))
                == 1
            ):
                manager = mp.Manager()  # get new manager.dict
                shared_memory = manager.dict()
                for opt_pros in opt_pros_l:
                    opt_pros.memory = shared_memory
            else:
                for opt_pros in opt_pros_l:
                    opt_pros.memory = opt_pros_l[
                        0
                    ].memory  # get same manager.dict

    @staticmethod
    def _default_opt(optimizer):
        if isinstance(optimizer, str):
            if optimizer == "default":
                optimizer = RandomSearchOptimizer()
        return copy.deepcopy(optimizer)

    @staticmethod
    def _default_search_id(search_id, objective_function):
        if not search_id:
            search_id = objective_function.__name__
        return search_id

    @staticmethod
    def check_list(search_space):
        for key in search_space.keys():
            search_dim = search_space[key]

            error_msg = "Value in '{}' of search space dictionary must be of type list".format(
                key
            )
            if not isinstance(search_dim, list):
                print("Warning", error_msg)
                # raise ValueError(error_msg)

    def add_search(
        self,
        objective_function: callable,
        search_space: Dict[str, list],
        n_iter: int,
        search_id=None,
        optimizer: Union[str, Type[RandomSearchOptimizer]] = "default",
        n_jobs: int = 1,
        initialize: Dict[str, int] = {"grid": 4, "random": 2, "vertices": 4},
        constraints: List[callable] = None,
        pass_through: Dict = None,
        callbacks: Dict[str, callable] = None,
        catch: Dict = None,
        max_score: float = None,
        early_stopping: Dict = None,
        random_state: int = None,
        memory: Union[str, bool] = "share",
        memory_warm_start: pd.DataFrame = None,
    ):
        """
        Set up and initialize a search process for optimizing an objective function over a given search space.
        """
        self.check_list(search_space)

        constraints = constraints or []
        pass_through = pass_through or {}
        callbacks = callbacks or {}
        catch = catch or {}
        early_stopping = early_stopping or {}

        optimizer = self._default_opt(optimizer)
        search_id = self._default_search_id(search_id, objective_function)
        s_space = SearchSpace(search_space)

        optimizer.setup_search(
            objective_function=objective_function,
            s_space=s_space,
            n_iter=n_iter,
            initialize=initialize,
            constraints=constraints,
            pass_through=pass_through,
            callbacks=callbacks,
            catch=catch,
            max_score=max_score,
            early_stopping=early_stopping,
            random_state=random_state,
            memory=memory,
            memory_warm_start=memory_warm_start,
            verbosity=self.verbosity,
        )

        n_jobs = mp.cpu_count() if n_jobs == -1 else n_jobs

        for _ in range(n_jobs):
            nth_process = len(self.opt_pros)
            self.opt_pros[nth_process] = optimizer

    def _print_info(self):
        print_res = PrintResults(self.opt_pros, self.verbosity)

        if self.verbosity:
            for _ in range(len(self.opt_pros)):
                print("")

        for results in self.results_list:
            nth_process = results["nth_process"]
            print_res.print_process(results, nth_process)

    def run(self, max_time: float = None):
        self._create_shared_memory()

        for opt in self.opt_pros.values():
            opt.max_time = max_time

        self.results_list = run_search(
            self.opt_pros, self.distribution, self.n_processes
        )

        self.results_ = Results(self.results_list, self.opt_pros)

        self._print_info()

    def best_para(self, id_):
        return self.results_.best_para(id_)

    def best_score(self, id_):
        return self.results_.best_score(id_)

    def search_data(self, id_, times=False):
        search_data_ = self.results_.search_data(id_)

        if times == False:
            search_data_.drop(
                labels=["eval_times", "iter_times"],
                axis=1,
                inplace=True,
                errors="ignore",
            )
        return search_data_


1			# Author: Simon Blanke
2			# Email: [email protected]
3			# License: MIT License
4
5
6			import copy
7			import multiprocessing as mp
8			import pandas as pd
9
10			from typing import Union, List, Dict, Type
11
12			from .optimizers import RandomSearchOptimizer
13			from .run_search import run_search
14
15			from .results import Results
16			from .print_results import PrintResults
17			from .search_space import SearchSpace
18
19
20			class Hyperactive:
21			"""
22			Initialize the Hyperactive class to manage optimization processes.
23
24			Parameters:
25			- verbosity: List of verbosity levels (default: ["progress_bar", "print_results", "print_times"])
26			- distribution: String indicating the distribution method (default: "multiprocessing")
27			- n_processes: Number of processes to run in parallel or "auto" to determine automatically (default: "auto")
28
29			Methods:
30			- add_search: Add a new optimization search process with specified parameters
31			- run: Execute the optimization searches
32			- best_para: Get the best parameters for a specific search
33			- best_score: Get the best score for a specific search
34			- search_data: Get the search data for a specific search
35			"""
36
37			def __init__(
38			self,
39			verbosity: list = ["progress_bar", "print_results", "print_times"],
40			distribution: str = "multiprocessing",
41			n_processes: Union[str, int] = "auto",
42			):
43			super().__init__()
44			if verbosity is False:
45			verbosity = []
46
47			self.verbosity = verbosity
48			self.distribution = distribution
49			self.n_processes = n_processes
50
51			self.opt_pros = {}
52
53			def _create_shared_memory(self):
54			_bundle_opt_processes = {}
55
56			for opt_pros in self.opt_pros.values():
57			if opt_pros.memory != "share":
58			continue
59			name = opt_pros.objective_function.__name__
60
61			_bundle_opt_processes.setdefault(name, []).append(opt_pros)
62
63			for opt_pros_l in _bundle_opt_processes.values():
64			# Check if the lengths of the search spaces of all optimizers in the list are the same.
65			if (
66			len(set(len(opt_pros.s_space()) for opt_pros in opt_pros_l))
67			== 1
68			):
69			manager = mp.Manager() # get new manager.dict
70			shared_memory = manager.dict()
71			for opt_pros in opt_pros_l:
72			opt_pros.memory = shared_memory
73			else:
74			for opt_pros in opt_pros_l:
75			opt_pros.memory = opt_pros_l[
76			0
77			].memory # get same manager.dict
78
79			@staticmethod
80			def _default_opt(optimizer):
81			if isinstance(optimizer, str):
82			if optimizer == "default":
83			optimizer = RandomSearchOptimizer()
84			return copy.deepcopy(optimizer)
85
86			@staticmethod
87			def _default_search_id(search_id, objective_function):
88			if not search_id:
89			search_id = objective_function.__name__
90			return search_id
91
92			@staticmethod
93			def check_list(search_space):
94			for key in search_space.keys():
95			search_dim = search_space[key]
96
97			error_msg = "Value in '{}' of search space dictionary must be of type list".format(
98			key
99			)
100			if not isinstance(search_dim, list):
101			print("Warning", error_msg)
102			# raise ValueError(error_msg)
103
104			def add_search(
105			self,
106			objective_function: callable,
107			search_space: Dict[str, list],
108			n_iter: int,
109			search_id=None,
110			optimizer: Union[str, Type[RandomSearchOptimizer]] = "default",
111			n_jobs: int = 1,
112			initialize: Dict[str, int] = {"grid": 4, "random": 2, "vertices": 4},
113			constraints: List[callable] = None,
114			pass_through: Dict = None,
115			callbacks: Dict[str, callable] = None,
116			catch: Dict = None,
117			max_score: float = None,
118			early_stopping: Dict = None,
119			random_state: int = None,
120			memory: Union[str, bool] = "share",
121			memory_warm_start: pd.DataFrame = None,
122			):
123			"""
124			Set up and initialize a search process for optimizing an objective function over a given search space.
125			"""
126			self.check_list(search_space)
127
128			constraints = constraints or []
129			pass_through = pass_through or {}
130			callbacks = callbacks or {}
131			catch = catch or {}
132			early_stopping = early_stopping or {}
133
134			optimizer = self._default_opt(optimizer)
135			search_id = self._default_search_id(search_id, objective_function)
136			s_space = SearchSpace(search_space)
137
138			optimizer.setup_search(
139			objective_function=objective_function,
140			s_space=s_space,
141			n_iter=n_iter,
142			initialize=initialize,
143			constraints=constraints,
144			pass_through=pass_through,
145			callbacks=callbacks,
146			catch=catch,
147			max_score=max_score,
148			early_stopping=early_stopping,
149			random_state=random_state,
150			memory=memory,
151			memory_warm_start=memory_warm_start,
152			verbosity=self.verbosity,
153			)
154
155			n_jobs = mp.cpu_count() if n_jobs == -1 else n_jobs
156
157			for _ in range(n_jobs):
158			nth_process = len(self.opt_pros)
159			self.opt_pros[nth_process] = optimizer
160
161			def _print_info(self):
162			print_res = PrintResults(self.opt_pros, self.verbosity)
163
164			if self.verbosity:
165			for _ in range(len(self.opt_pros)):
166			print("")
167
168			for results in self.results_list:
169			nth_process = results["nth_process"]
170			print_res.print_process(results, nth_process)
171
172			def run(self, max_time: float = None):
173			self._create_shared_memory()
174
175			for opt in self.opt_pros.values():
176			opt.max_time = max_time
177
178			self.results_list = run_search(
179			self.opt_pros, self.distribution, self.n_processes
180			)
181
182			self.results_ = Results(self.results_list, self.opt_pros)
183
184			self._print_info()
185
186			def best_para(self, id_):
187			return self.results_.best_para(id_)
188
189			def best_score(self, id_):
190			return self.results_.best_score(id_)
191
192			def search_data(self, id_, times=False):
193			search_data_ = self.results_.search_data(id_)
194
195			if times == False:
196			search_data_.drop(
197			labels=["eval_times", "iter_times"],
198			axis=1,
199			inplace=True,
200			errors="ignore",
201			)
202			return search_data_
203

SimonBlanke / Hyperactive

Push — master ( 2e0454...bd7b54 )

hyperactive.hyperactive.Hyperactive.add_search() B

Complexity

Size

Duplication

Importance

How to fix Long Method Many Parameters

Long Method

Many Parameters

Duplication Side-by-Side

Filter issues like