hyperactive.hyperactive

Complexity

Total Complexity

27

Size/Duplication

Total Lines	207
Duplicated Lines	0 %

Importance

Changes

0

11 Methods

Rating	Name	Duplication	Size	Complexity
A	HyperactiveResults.best_para()	0	2	1
A	Hyperactive.add_search()	0	37	4
A	Hyperactive._add_search_processes()	0	28	2
B	HyperactiveResults._sort_results_objFunc()	0	29	5
A	HyperactiveResults.__init__()	0	2	1
A	Hyperactive.__init__()	0	23	2
A	Hyperactive.run()	0	22	3
A	HyperactiveResults.results()	0	2	1
A	HyperactiveResults._get_one_result()	0	12	4
A	HyperactiveResults._sort_results_search_id()	0	16	3
A	HyperactiveResults.best_score()	0	2	1

# Author: Simon Blanke
# Email: simon.blanke@yahoo.com
# License: MIT License


import numpy as np
import pandas as pd
from tqdm import tqdm

from .optimizers import RandomSearchOptimizer
from .run_search import run_search
from .print_info import print_info


class HyperactiveResults:
    def __init__(*args, **kwargs):
        pass

    def _sort_results_objFunc(self, objective_function):
        best_score = -np.inf
        best_para = None
        search_data = None

        results_list = []

        for results_ in self.results_list:
            nth_process = results_["nth_process"]

            process_infos = self.process_infos[nth_process]
            objective_function_ = process_infos["objective_function"]

            if objective_function_ != objective_function:
                continue

            if results_["best_score"] > best_score:
                best_score = results_["best_score"]
                best_para = results_["best_para"]

            results_list.append(results_["results"])

        if len(results_list) > 0:
            search_data = pd.concat(results_list)

        self.objFunc2results[objective_function] = {
            "best_para": best_para,
            "best_score": best_score,
            "search_data": search_data,
        }

    def _sort_results_search_id(self, search_id):
        for results_ in self.results_list:
            nth_process = results_["nth_process"]
            search_id_ = self.process_infos[nth_process]["search_id"]

            if search_id_ != search_id:
                continue

            best_score = results_["best_score"]
            best_para = results_["best_para"]
            search_data = results_["results"]

            self.search_id2results[search_id] = {
                "best_para": best_para,
                "best_score": best_score,
                "search_data": search_data,
            }

    def _get_one_result(self, id_, result_name):
        if isinstance(id_, str):
            if id_ not in self.search_id2results:
                self._sort_results_search_id(id_)

            return self.search_id2results[id_][result_name]

        else:
            if id_ not in self.objFunc2results:
                self._sort_results_objFunc(id_)

            return self.objFunc2results[id_][result_name]

    def best_para(self, id_):
        return self._get_one_result(id_, "best_para")

    def best_score(self, id_):
        return self._get_one_result(id_, "best_score")

    def results(self, id_):
        return self._get_one_result(id_, "search_data")


class Hyperactive(HyperactiveResults):
    def __init__(
        self,
        verbosity=["progress_bar", "print_results", "print_times"],
        distribution={
            "multiprocessing": {
                "initializer": tqdm.set_lock,
                "initargs": (tqdm.get_lock(),),
            }
        },
        n_processes="auto",
    ):
        super().__init__()
        if verbosity is False:
            verbosity = []

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

        self.search_ids = []
        self.process_infos = {}
        self.objFunc2results = {}
        self.search_id2results = {}

    def _add_search_processes(
        self,
        random_state,
        objective_function,
        search_space,
        optimizer,
        n_iter,
        n_jobs,
        max_score,
        memory,
        memory_warm_start,
        search_id,
    ):
        for _ in range(n_jobs):
            nth_process = len(self.process_infos)

            self.process_infos[nth_process] = {
                "random_state": random_state,
                "verbosity": self.verbosity,
                "nth_process": nth_process,
                "objective_function": objective_function,
                "search_space": search_space,
                "optimizer": optimizer,
                "n_iter": n_iter,
                "max_score": max_score,
                "memory": memory,
                "memory_warm_start": memory_warm_start,
                "search_id": search_id,
            }

    def add_search(
        self,
        objective_function,
        search_space,
        n_iter,
        search_id=None,
        optimizer="default",
        n_jobs=1,
        initialize={"grid": 4, "random": 2, "vertices": 4},
        max_score=None,
        random_state=None,
        memory=True,
        memory_warm_start=None,
    ):
        if isinstance(optimizer, str):
            if optimizer == "default":
                optimizer = RandomSearchOptimizer()
        optimizer.init(search_space, initialize)

        if search_id is not None:
            search_id = search_id
            self.search_ids.append(search_id)
        else:
            search_id = str(len(self.search_ids))
            self.search_ids.append(search_id)

        self._add_search_processes(
            random_state,
            objective_function,
            search_space,
            optimizer,
            n_iter,
            n_jobs,
            max_score,
            memory,
            memory_warm_start,
            search_id,
        )

    def run(self, max_time=None):
        for nth_process in self.process_infos.keys():
            self.process_infos[nth_process]["max_time"] = max_time

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

        for results in self.results_list:
            nth_process = results["nth_process"]

            print_info(
                verbosity=self.process_infos[nth_process]["verbosity"],
                objective_function=self.process_infos[nth_process][
                    "objective_function"
                ],
                best_score=results["best_score"],
                best_para=results["best_para"],
                best_iter=results["best_iter"],
                eval_times=results["eval_times"],
                iter_times=results["iter_times"],
                n_iter=self.process_infos[nth_process]["n_iter"],
            )