optimize.py
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, )