run_search

Complexity

Total Complexity

8

Size/Duplication

Total Lines	58
Duplicated Lines	51.72 %

Importance

Changes

0

3 Functions

Rating	Name	Duplication	Size	Complexity
A	proxy()	0	2	1
A	_get_distribution()	12	12	4
A	run_search()	18	18	3

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


from .distribution import (
    single_process,
    joblib_wrapper,
    multiprocessing_wrapper,
    pathos_wrapper,
)
from .process import _process_


def proxy(args):
    return _process_(*args)


dist_dict = {
    "joblib": (joblib_wrapper, _process_),
    "multiprocessing": (multiprocessing_wrapper, proxy),
    "pathos": (pathos_wrapper, proxy),
}


def _get_distribution(distribution):

This code seems to be duplicated in your project.

    if hasattr(distribution, "__call__"):
        return (distribution, _process_), {}

    elif isinstance(distribution, dict):
        dist_key = list(distribution.keys())[0]
        dist_paras = list(distribution.values())[0]

        return dist_dict[dist_key], dist_paras

    elif isinstance(distribution, str):
        return dist_dict[distribution], {}


def run_search(opt_pros, distribution, n_processes):

This code seems to be duplicated in your project.

    opts = list(opt_pros.values())

    processes = list(opt_pros.keys())
    optimizers = list(opt_pros.values())
    process_infos = list(zip(processes, optimizers))

    if n_processes == "auto":
        n_processes = len(process_infos)

    if n_processes == 1:
        results_list = single_process(_process_, process_infos)
    else:
        (distribution, process_func), dist_paras = _get_distribution(distribution)

        results_list = distribution(process_func, process_infos, n_processes)

    return results_list