wehub-resource-sync
251 lines
7.8 KiB
Python
251 lines
7.8 KiB
Python
from collections.abc import Callable
|
|
from itertools import product
|
|
from concurrent.futures import ProcessPoolExecutor
|
|
from random import random, choice
|
|
from time import perf_counter
|
|
from multiprocessing import get_context
|
|
from multiprocessing.context import BaseContext
|
|
from multiprocessing.managers import DictProxy
|
|
from _collections_abc import dict_keys, dict_values, Iterable
|
|
|
|
from tqdm import tqdm
|
|
from deap import creator, base, tools, algorithms # type: ignore
|
|
|
|
from .locale import _
|
|
|
|
OUTPUT_FUNC = Callable[[str], None]
|
|
EVALUATE_FUNC = Callable[[dict], dict]
|
|
KEY_FUNC = Callable[[tuple], float]
|
|
|
|
|
|
# Create individual class used in genetic algorithm optimization
|
|
creator.create("FitnessMax", base.Fitness, weights=(1.0,))
|
|
creator.create("Individual", list, fitness=creator.FitnessMax)
|
|
|
|
|
|
class OptimizationSetting:
|
|
"""
|
|
Setting for runnning optimization.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
""""""
|
|
self.params: dict[str, list] = {}
|
|
self.target_name: str = ""
|
|
|
|
def add_parameter(
|
|
self,
|
|
name: str,
|
|
start: float,
|
|
end: float | None = None,
|
|
step: float | None = None
|
|
) -> tuple[bool, str]:
|
|
""""""
|
|
if end is None or step is None:
|
|
self.params[name] = [start]
|
|
return True, _("固定参数添加成功")
|
|
|
|
if start >= end:
|
|
return False, _("参数优化起始点必须小于终止点")
|
|
|
|
if step <= 0:
|
|
return False, _("参数优化步进必须大于0")
|
|
|
|
value: float = start
|
|
value_list: list[float] = []
|
|
|
|
while value <= end:
|
|
value_list.append(value)
|
|
value += step
|
|
|
|
self.params[name] = value_list
|
|
|
|
return True, _("范围参数添加成功，数量{}").format(len(value_list))
|
|
|
|
def set_target(self, target_name: str) -> None:
|
|
""""""
|
|
self.target_name = target_name
|
|
|
|
def generate_settings(self) -> list[dict]:
|
|
""""""
|
|
keys: dict_keys = self.params.keys()
|
|
values: dict_values = self.params.values()
|
|
products: list = list(product(*values))
|
|
|
|
settings: list = []
|
|
for p in products:
|
|
setting: dict = dict(zip(keys, p, strict=False))
|
|
settings.append(setting)
|
|
|
|
return settings
|
|
|
|
|
|
def check_optimization_setting(
|
|
optimization_setting: OptimizationSetting,
|
|
output: OUTPUT_FUNC = print
|
|
) -> bool:
|
|
""""""
|
|
if not optimization_setting.generate_settings():
|
|
output(_("优化参数组合为空，请检查"))
|
|
return False
|
|
|
|
if not optimization_setting.target_name:
|
|
output(_("优化目标未设置，请检查"))
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
def run_bf_optimization(
|
|
evaluate_func: EVALUATE_FUNC,
|
|
optimization_setting: OptimizationSetting,
|
|
key_func: KEY_FUNC,
|
|
max_workers: int | None = None,
|
|
output: OUTPUT_FUNC = print
|
|
) -> list[tuple]:
|
|
"""Run brutal force optimization"""
|
|
settings: list[dict] = optimization_setting.generate_settings()
|
|
|
|
output(_("开始执行穷举算法优化"))
|
|
output(_("参数优化空间：{}").format(len(settings)))
|
|
|
|
start: float = perf_counter()
|
|
|
|
with ProcessPoolExecutor(
|
|
max_workers,
|
|
mp_context=get_context("spawn")
|
|
) as executor:
|
|
it: Iterable = tqdm(
|
|
executor.map(evaluate_func, settings),
|
|
total=len(settings)
|
|
)
|
|
results: list[tuple] = list(it)
|
|
results.sort(reverse=True, key=key_func)
|
|
|
|
end: float = perf_counter()
|
|
cost: int = int(end - start)
|
|
output(_("穷举算法优化完成，耗时{}秒").format(cost))
|
|
|
|
return results
|
|
|
|
|
|
def run_ga_optimization(
|
|
evaluate_func: EVALUATE_FUNC,
|
|
optimization_setting: OptimizationSetting,
|
|
key_func: KEY_FUNC,
|
|
max_workers: int | None = None,
|
|
pop_size: int = 100, # population size: number of individuals in each generation
|
|
ngen: int = 30, # number of generations: number of generations to evolve
|
|
mu: int | None = None, # mu: number of individuals to select for the next generation
|
|
lambda_: int | None = None, # lambda: number of children to produce at each generation
|
|
cxpb: float = 0.95, # crossover probability: probability that an offspring is produced by crossover
|
|
mutpb: float | None = None, # mutation probability: probability that an offspring is produced by mutation
|
|
indpb: float = 1.0, # independent probability: probability for each gene to be mutated
|
|
output: OUTPUT_FUNC = print,
|
|
) -> list[tuple]:
|
|
"""Run genetic algorithm optimization"""
|
|
# Define functions for generate parameter randomly
|
|
settings: list[dict] = optimization_setting.generate_settings()
|
|
parameter_tuples: list[list[tuple]] = [list(d.items()) for d in settings]
|
|
|
|
def generate_parameter() -> list:
|
|
""""""
|
|
return choice(parameter_tuples)
|
|
|
|
def mutate_individual(individual: list, indpb: float) -> tuple:
|
|
""""""
|
|
size: int = len(individual)
|
|
paramlist: list = generate_parameter()
|
|
for i in range(size):
|
|
if random() < indpb:
|
|
individual[i] = paramlist[i]
|
|
return individual,
|
|
|
|
# Set up multiprocessing Pool and Manager
|
|
ctx: BaseContext = get_context("spawn")
|
|
with ctx.Manager() as manager, ctx.Pool(max_workers) as pool:
|
|
# Create shared dict for result cache
|
|
cache: DictProxy[tuple, tuple] = manager.dict()
|
|
|
|
# Set up toolbox
|
|
toolbox: base.Toolbox = base.Toolbox()
|
|
toolbox.register("individual", tools.initIterate, creator.Individual, generate_parameter)
|
|
toolbox.register("population", tools.initRepeat, list, toolbox.individual)
|
|
toolbox.register("mate", tools.cxTwoPoint)
|
|
toolbox.register("mutate", mutate_individual, indpb=indpb)
|
|
toolbox.register("select", tools.selNSGA2)
|
|
toolbox.register("map", pool.map)
|
|
toolbox.register(
|
|
"evaluate",
|
|
ga_evaluate,
|
|
cache,
|
|
evaluate_func,
|
|
key_func
|
|
)
|
|
|
|
# Set default values for DEAP parameters if not specified
|
|
if mu is None:
|
|
mu = int(pop_size * 0.8)
|
|
|
|
if lambda_ is None:
|
|
lambda_ = pop_size
|
|
|
|
if mutpb is None:
|
|
mutpb = 1.0 - cxpb
|
|
|
|
total_size: int = len(parameter_tuples)
|
|
pop: list = toolbox.population(pop_size)
|
|
|
|
# Run ga optimization
|
|
output(_("开始执行遗传算法优化"))
|
|
output(_("参数优化空间：{}").format(total_size))
|
|
output(_("每代族群总数：{}").format(pop_size))
|
|
output(_("优良筛选个数：{}").format(mu))
|
|
output(_("迭代次数：{}").format(ngen))
|
|
output(_("交叉概率：{:.0%}").format(cxpb))
|
|
output(_("突变概率：{:.0%}").format(mutpb))
|
|
output(_("个体突变概率：{:.0%}").format(indpb))
|
|
|
|
start: float = perf_counter()
|
|
|
|
algorithms.eaMuPlusLambda(
|
|
pop,
|
|
toolbox,
|
|
mu,
|
|
lambda_,
|
|
cxpb,
|
|
mutpb,
|
|
ngen,
|
|
verbose=True
|
|
)
|
|
|
|
end: float = perf_counter()
|
|
cost: int = int(end - start)
|
|
|
|
output(_("遗传算法优化完成，耗时{}秒").format(cost))
|
|
|
|
results: list = list(cache.values())
|
|
results.sort(reverse=True, key=key_func)
|
|
return results
|
|
|
|
|
|
def ga_evaluate(
|
|
cache: dict,
|
|
evaluate_func: Callable,
|
|
key_func: Callable,
|
|
parameters: list
|
|
) -> tuple[float, ]:
|
|
"""
|
|
Functions to be run in genetic algorithm optimization.
|
|
"""
|
|
tp: tuple = tuple(parameters)
|
|
if tp in cache:
|
|
result: dict = cache[tp]
|
|
else:
|
|
setting: dict = dict(parameters)
|
|
result = evaluate_func(setting)
|
|
cache[tp] = result
|
|
|
|
value: float = key_func(result)
|
|
return (value, )
|