import multiprocessing
from pymoo.algorithms.soo.nonconvex.de import DE
from pymoo.operators.sampling.lhs import LHS
from pymoo.optimize import minimize
from pymoo.parallelization.starmap import StarmapParallelization
from carbatpy.optimizations.helpers_optimization import (
CombinedTermination,
OptiProblem,
default_n_processes,
extract_boundary_with_path,
)
def _build_case(
mode: str,
boundaries: dict,
same_fluid: bool | None = None,
heat_losses: float | None = None,
COP: float | None = None,
q_dot: float | None = None,
) -> tuple[str, dict, dict]:
"""
Prepare all cycle-specific inputs for the optimization.
Depending on the optimization mode, this function returns the function
type to optimize, the boundaries for the decision variables, and additional
parameters required for the computation.
Args:
mode (str): Optimization mode. Must be one of ``'hp'``, ``'orc'``, or ``'cb'``.
boundaries (dict): Lower and upper bounds for each optimization variable.
same_fluid (bool | None): Whether the same working fluid should be used.
Required for ``'cb'`` mode.
heat_losses (float | None): Heat losses fraction.
Required for ``'cb'`` mode.
COP (float | None): Coefficient of performance.
Required for ``'orc'`` mode.
q_dot (float | None): Heat input/output rate.
Required for ``'orc'`` mode.
Raises:
ValueError: If ``mode`` is invalid or required parameters for a mode are missing.
Returns:
tuple[str, dict, dict]:
- Function type for optimization (``'hp'``, ``'orc'``, or ``'cb'``).
- Boundaries dictionary for optimization variables.
- Additional keyword arguments required by the optimization function.
"""
if mode == "hp":
return "hp", boundaries, {}
if mode == "orc":
if COP is None or q_dot is None:
raise ValueError("For mode='orc' you must provide COP and q_dot.")
return "orc", boundaries, {"COP": COP, "q_dot": q_dot}
if mode == "cb":
return "cb", boundaries, {"same_fluid": same_fluid, "heat_losses": heat_losses}
raise ValueError("mode must be one of: 'hp' | 'orc' | 'cb'")
[docs]
def optimize(
mode: str,
config: dict | str,
boundaries: dict,
same_fluid: bool = True,
heat_losses: float = 0.0,
COP: float | None = None,
q_dot: float | None = None,
verbose: bool = True,
n_processes: int | None = None,
maxtasksperchild: int = 20,
n_ieq_constr: int = 3,
pop_size: int = 50,
sampling_iterations: int = 50,
de_variant: str = "DE/rand/1/bin",
CR: float = 0.3,
dither: str = "vector",
jitter: bool = False,
n_gen: int = 100,
period: int = 20,
ftol: float = 1e-4,
return_least_infeasible: bool = False,
):
"""
Perform the global optimization for the selected cycle.
This function sets up the optimization problem, initializes the
differential evolution algorithm, handles parallelization, and
executes the optimization using Pymoo.
Args:
mode (str): Optimization mode. Must be one of ``'hp'``, ``'orc'``, or ``'cb'``.
config (dict | str): Configuration dictionary or path to configuration file.
boundaries (dict): Lower and upper bounds for all optimization variables.
same_fluid (bool): Whether the working fluid is the same for all cycles.
Only used in ``'cb'`` mode. Defaults to ``True``.
heat_losses (float): Heat losses fraction.
Only used in ``'cb'`` mode. Defaults to ``0.0``.
COP (float | None): Coefficient of performance.
Only used in ``'orc'`` mode.
q_dot (float | None): Heat input/output rate.
Only used in ``'orc'`` mode.
verbose (bool): Whether to print detailed optimization progress.
Defaults to ``True``.
n_processes (int | None): Number of processes for parallelization.
Defaults to auto-detection via :func:`default_n_processes`.
maxtasksperchild (int): Maximum tasks per worker process in the pool.
Defaults to ``20``.
n_ieq_constr (int): Number of inequality constraints. Defaults to ``3``.
pop_size (int): Population size for the DE algorithm. Defaults to ``50``.
sampling_iterations (int): Number of iterations for Latin Hypercube sampling.
Defaults to ``50``.
de_variant (str): Variant of the differential evolution algorithm.
Defaults to ``'DE/rand/1/bin'``.
CR (float): Crossover probability for DE. Defaults to ``0.3``.
dither (str): Dithering strategy for DE. Defaults to ``'vector'``.
jitter (bool): Whether to apply jitter in DE. Defaults to ``False``.
n_gen (int): Maximum number of generations. Defaults to ``100``.
period (int): Period for robust termination checking. Defaults to ``20``.
ftol (float): Function tolerance for termination. Defaults to ``1e-4``.
return_least_infeasible (bool): Whether to return the least infeasible
solution if no feasible solution is found. Defaults to ``False``.
Raises:
ValueError: If ``boundaries`` is ``None`` or required parameters for
the selected ``mode`` are missing.
Returns:
pymoo.core.result.Result: The optimization result object containing
the best solution and additional metadata.
"""
multiprocessing.freeze_support()
if boundaries is None:
raise ValueError("boundaries must be provided.")
if n_processes is None:
n_processes = default_n_processes()
fun_to_optimize, bounds, problem_kwargs = _build_case(
mode,
boundaries=boundaries,
same_fluid=same_fluid,
heat_losses=heat_losses,
COP=COP,
q_dot=q_dot,
)
paths, xl, xu = extract_boundary_with_path(bounds)
pool = multiprocessing.Pool(n_processes, maxtasksperchild=maxtasksperchild)
runner = StarmapParallelization(pool.starmap)
try:
problem = OptiProblem(
dir_config=config,
paths=paths,
opti_fun=fun_to_optimize,
n_var=len(paths),
n_obj=1,
n_ieq_constr=n_ieq_constr,
xl=xl,
xu=xu,
elementwise_runner=runner,
**problem_kwargs,
)
sampling = LHS(iterations=sampling_iterations)
algorithm = DE(
pop_size=pop_size,
sampling=sampling, # type:ignore
variant=de_variant,
CR=CR,
dither=dither,
jitter=jitter,
)
termination = CombinedTermination(max_gen=n_gen, period=period, ftol=ftol)
result = minimize(
problem,
algorithm,
termination=termination,
return_least_infeasible=return_least_infeasible,
verbose=verbose,
)
return result
finally:
pool.close()
pool.join()