Source code for gemseo_pseven.lib_pseven

# Copyright 2021 IRT Saint Exupéry,
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License version 3 as published by the Free Software Foundation.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# Lesser General Public License for more details.
# You should have received a copy of the GNU Lesser General Public License
# along with this program; if not, write to the Free Software Foundation,
# Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
# Contributors:
#    INITIAL AUTHORS - initial API and implementation and/or initial
#                           documentation
#        :author: Benoit Pauwels
"""Wrapper for the Generic Tool for Optimization (GTOpt) of pSeven Core."""
from __future__ import annotations

import logging
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from typing import Mapping
from typing import MutableMapping

import numpy
from da import p7core
from gemseo.algos.opt.optimization_library import OptimizationAlgorithmDescription
from gemseo.algos.opt.optimization_library import OptimizationLibrary
from gemseo.algos.opt_result import OptimizationResult
from gemseo.algos.stop_criteria import DesvarIsNan
from gemseo.algos.stop_criteria import FtolReached
from gemseo.algos.stop_criteria import FunctionIsNan
from gemseo.algos.stop_criteria import MaxIterReachedException
from gemseo.algos.stop_criteria import MaxTimeReached
from gemseo.algos.stop_criteria import XtolReached
from gemseo.core.mdofunctions.mdo_linear_function import MDOLinearFunction
from gemseo.core.mdofunctions.mdo_quadratic_function import MDOQuadraticFunction
from numpy import ndarray

from gemseo_pseven.pseven_problem_adapter import PSevenProblem

[docs]@dataclass class PSevenAlgorithmDescription(OptimizationAlgorithmDescription): """The description of an optimization algorithm from the NLopt library.""" library_name: str = "pSeven" website: str = ""
LOGGER = logging.getLogger(__name__)
[docs]class PSevenOpt(OptimizationLibrary): """Interface for the Generic Tool for Optimization (GTOpt) of pSeven Core.""" LIB_COMPUTE_GRAD = True OPTIONS_MAP = { "constraints_smoothness": "GTOpt/ConstraintsSmoothness", "global_phase_intensity": "GTOpt/GlobalPhaseIntensity", "max_expensive_func_iter": "GTOpt/MaximumExpensiveIterations", "max_func_iter": "GTOpt/MaximumIterations", "objectives_smoothness": "GTOpt/ObjectivesSmoothness", "deterministic": "GTOpt/Deterministic", "log_level": "GTOpt/LogLevel", "verbose_log": "GTOpt/VerboseOutput", "max_threads": "GTOpt/MaxParallel", "seed": "GTOpt/Seed", "time_limit": "GTOpt/TimeLimit", "max_batch_size": "GTOpt/BatchSize", "detect_nan_clusters": "GTOpt/DetectNaNClusters", "diff_scheme": "GTOpt/DiffScheme", "diff_type": "GTOpt/DiffType", "diff_step": "GTOpt/NumDiffStepSize", "ensure_feasibility": "GTOpt/EnsureFeasibility", "local_search": "GTOpt/LocalSearch", "restore_analytic_func": "GTOpt/RestoreAnalyticResponses", "responses_scalability": "GTOpt/ResponsesScalability", } # Initial sample __SAMPLE_X = "sample_x" __SAMPLE_F = "sample_f" __SAMPLE_C = "sample_c" LIBRARY_NAME = "pSeven" # Governing methods # Not yet used. # SO = "SO" # MO = "MO" __SBO = "SBO" # Local methods __FD = "FD" __MOM = "MoM" __NCG = "NCG" __NLS = "NLS" __POWELL = "Powell" __QP = "QP" __SQP = "SQP" __SQ2P = "S2P" __LOCAL_METHODS = (__FD, __MOM, __NCG, __NLS, __POWELL, __QP, __SQP, __SQ2P) def __init__(self) -> None: # noqa: D107 super().__init__() self.descriptions = { "PSEVEN": PSevenAlgorithmDescription( algorithm_name="PSEVEN", description="pSeven's Generic Tool for Optimization (GTOpt).", handle_equality_constraints=True, handle_inequality_constraints=True, handle_integer_variables=True, internal_algorithm_name="pSeven", ), "PSEVEN_FD": PSevenAlgorithmDescription( algorithm_name="Feasible direction", description="pSeven's feasible direction method.", handle_equality_constraints=True, handle_inequality_constraints=True, internal_algorithm_name=self.__FD, ), "PSEVEN_MOM": PSevenAlgorithmDescription( algorithm_name="MOM", description="pSeven's method of multipliers.", handle_equality_constraints=True, handle_inequality_constraints=True, internal_algorithm_name=self.__MOM, ), "PSEVEN_NCG": PSevenAlgorithmDescription( algorithm_name="NCG", description="pSeven's nonlinear conjugate gradient method.", internal_algorithm_name=self.__NCG, ), "PSEVEN_NLS": PSevenAlgorithmDescription( algorithm_name="NLS", description="pSeven's nonlinear simplex method.", internal_algorithm_name=self.__NLS, ), "PSEVEN_POWELL": PSevenAlgorithmDescription( algorithm_name="POWELL", description="pSeven's Powell conjugate direction method.", internal_algorithm_name=self.__POWELL, ), "PSEVEN_QP": PSevenAlgorithmDescription( algorithm_name="QP", description="pSeven's quadratic programming method.", handle_equality_constraints=True, handle_inequality_constraints=True, internal_algorithm_name=self.__QP, ), "PSEVEN_SQP": PSevenAlgorithmDescription( algorithm_name="SQP", description="pSeven's sequential quadratic programming method.", handle_equality_constraints=True, handle_inequality_constraints=True, internal_algorithm_name=self.__SQP, ), "PSEVEN_SQ2P": PSevenAlgorithmDescription( algorithm_name="SQ2P", description=( "pSeven's sequential quadratic constrained quadratic " "programming method." ), handle_equality_constraints=True, handle_inequality_constraints=True, internal_algorithm_name=self.__SQ2P, ), } def _get_options( self, max_iter: int = 99, evaluation_cost_type: str | Mapping[str, str] | None = None, expensive_evaluations: Mapping[str, int] | None = None, sample_x: list[float] | list[ndarray] | None = None, sample_f: list[float] | list[ndarray] | None = None, sample_c: list[float] | list[ndarray] | None = None, constraints_smoothness: str = "Auto", global_phase_intensity: str | float = "Auto", max_expensive_func_iter: int = 0, objectives_smoothness: str = "Auto", deterministic: str | bool = "Auto", log_level: str = "Error", verbose_log: bool = False, max_threads: int = 0, seed: int = 100, time_limit: int = 0, max_batch_size: int = 0, detect_nan_clusters: bool = True, diff_scheme: str = "Auto", diff_type: str = "Auto", diff_step: float = 1.1920929e-06, ensure_feasibility: bool = False, local_search: str = "Disabled", restore_analytic_func: str | bool = "Auto", responses_scalability: int = 1, globalization_method: str | None = None, surrogate_based: bool | None = None, use_gradient: bool = True, ftol_abs: float = 1e-14, xtol_abs: float = 1e-14, ftol_rel: float = 1e-8, xtol_rel: float = 1e-8, stop_crit_n_x: int = 3, normalize_design_space: bool = True, eq_tolerance: float = 1e-2, ineq_tolerance: float = 1e-4, log_path: str | None = None, use_threading: bool = False, **kwargs: Any, ) -> dict: """Set the default options values. Args: max_iter: The maximum number of evaluations. evaluation_cost_type: The evaluation cost type of each function of the problem: "Cheap" or "Expensive". If a string, then the same cost type is set for all the functions. If None, the evaluation cost types are set by pSeven. expensive_evaluations: The maximal number of expensive evaluations for each function of the problem. By default, set automatically by pSeven. sample_x: A sample of design points (in addition to the problem initial design). sample_f: The objectives values at the design points of the sample. sample_c: The constraints values at the design points of the sample. constraints_smoothness: The assumed smoothness of the constraints functions: "Smooth", "Noisy" or "Auto". global_phase_intensity: The configuration of global searching algorithms. This option has different meanings for expensive and non-expensive optimization problems. Refer to the pSeven Core API documentation. Defaults to "Auto". max_expensive_func_iter: The maximum number of evaluations for each expensive response, excluding the evaluations of initial guesses. objectives_smoothness: The assumed smoothness of the objective functions: "Smooth", "Noisy" or "Auto". deterministic: Whether to require optimization process to be reproducible using the passed seed value. Defaults to "Auto". log_level: The minimum log level: "Debug", "Info", "Warn", "Error" or "Fatal". verbose_log: Whether to enable verbose logging. max_threads: The maximum number of parallel threads to use when solving. seed: The random seed for deterministic mode. time_limit: The maximum allowed time to solve a problem in seconds. Defaults to 0, unlimited. max_batch_size: The maximum number of points in an evaluation batch. Batches of size more than one are evaluated in parallel. The (default) value 0 allows the optimizer to use any batch size. The value 1 implements sequential evaluation. detect_nan_clusters: Whether to detect and avoid design space areas that yield NaN values (for at least one function). This option has no effect in the absence of "expensive" functions. diff_scheme: The order of the differentiation scheme (when the analytic derivatives are unavailable): "FirstOrder", "SecondOrder", "Adaptive" or "Auto". diff_type: The strategy for differentiation (when the analytic derivatives are unavailable): "Numerical", "Framed" or "Auto". diff_step: The numerical differentiation step size. ensure_feasibility: Whether to restrict the evaluations of the objectives to feasible designs only. local_search: Whether to force the surrogate models to explore the design space locally near the current optimum, or to disable the local search and let the surrogate models explore the whole design space. restore_analytic_func: Whether to restore the analytic forms of the linear and quadratic functions. Once the analytic forms are restored the original functions will not be evaluated anymore. responses_scalability: The maximum number of concurrent response evaluations supported by the problem. globalization_method: The globalization method: "RL" (random linkages), "PM" (plain multistart), or "MS" (surrogate model-based multistart) If None, set automatically by pSeven depending on the problem. surrogate_based: Whether to use surrogate models. If None, set automatically depending on the problem. use_gradient: Whether to use the functions derivatives. ftol_abs: The absolute tolerance on the objective function. xtol_abs: The absolute tolerance on the design parameters. ftol_rel: The relative tolerance on the objective function. xtol_rel: The relative tolerance on the design parameters. normalize_design_space: If True, normalize the design variables between 0 and 1. stop_crit_n_x: The number of design vectors to take into account in the stopping criteria. eq_tolerance: The tolerance on the equality constraints. ineq_tolerance: The tolerance on the inequality constraints. log_path: The path where to save the pSeven log. If None, the pSeven log will not be saved. use_threading: Whether to use threads instead of processes to parallelize the evaluation of the functions. **kwargs: Other driver options. Returns: The processed options. """ processed_options = self._process_options( max_iter=max_iter, evaluation_cost_type=evaluation_cost_type, expensive_evaluations=expensive_evaluations, sample_x=sample_x, sample_f=sample_f, sample_c=sample_c, constraints_smoothness=constraints_smoothness, global_phase_intensity=global_phase_intensity, max_expensive_func_iter=max_expensive_func_iter, max_func_iter=max_iter, objectives_smoothness=objectives_smoothness, deterministic=deterministic, log_level=log_level, verbose_log=verbose_log, max_threads=max_threads, seed=seed, time_limit=time_limit, max_batch_size=max_batch_size, detect_nan_clusters=detect_nan_clusters, diff_scheme=diff_scheme, diff_type=diff_type, diff_step=diff_step, ensure_feasibility=ensure_feasibility, local_search=local_search, restore_analytic_func=restore_analytic_func, responses_scalability=responses_scalability, globalization_method=globalization_method, surrogate_based=surrogate_based, use_gradient=use_gradient, ftol_rel=ftol_rel, ftol_abs=ftol_abs, xtol_rel=xtol_rel, xtol_abs=xtol_abs, stop_crit_n_x=stop_crit_n_x, normalize_design_space=normalize_design_space, eq_tolerance=eq_tolerance, ineq_tolerance=ineq_tolerance, log_path=log_path, use_threading=use_threading, **kwargs, ) # Set the pSeven's techniques self.__set_pseven_techniques(processed_options) return processed_options def __set_pseven_techniques( self, options: MutableMapping[str, Any], ) -> None: """Get the pSeven techniques from the options.""" technique_names = list() internal_algo_name = self.descriptions[self.algo_name].internal_algorithm_name if internal_algo_name in self.__LOCAL_METHODS: technique_names.append(internal_algo_name) globalization_method = options.pop("globalization_method", None) if globalization_method is not None: technique_names.append(globalization_method) surrogate_based = options.pop("surrogate_based", None) if surrogate_based is not None and surrogate_based: technique_names.append(self.__SBO) if technique_names: options["GTOpt/Techniques"] = "[" + ", ".join(technique_names) + "]" def _run(self, **options: Any) -> OptimizationResult: """Run the algorithm. Args: **options: The options of the algorithm. Returns: The result of the optimization. """ self.__check_functions() # Pop GEMSEO options normalize_ds = options.pop(self.NORMALIZE_DESIGN_SPACE_OPTION) max_iter = options.pop(self.MAX_ITER) del options[self.F_TOL_ABS] del options[self.F_TOL_REL] del options[self.X_TOL_ABS] del options[self.X_TOL_REL] del options[self.STOP_CRIT_NX] # Create the pSeven problem initial_x, lower_bnd, upper_bnd = self.get_x0_and_bounds_vects(normalize_ds) pseven_problem = PSevenProblem( self.problem, options.pop("evaluation_cost_type"), options.pop("expensive_evaluations"), lower_bnd, upper_bnd, initial_x, use_gradient=options.pop("use_gradient"), use_threading=options.pop("use_threading"), normalize_design_space=normalize_ds, ) # Set up the solver and its logger solver = p7core.gtopt.Solver() log_path = options.pop("log_path") if log_path is None: # Direct the logging to the standard output stream = sys.stdout else: # Direct the logging to a file stream = Path(log_path).open("w") solver.set_logger( p7core.loggers.StreamLogger(stream, options["GTOpt/LogLevel"]) ) # Disable pSeven stopping criteria options["GTOpt/CoordinateTolerance"] = 0.0 options["GTOpt/ObjectiveTolerance"] = 0.0 options["GTOpt/GradientTolerance"] = 0.0 # Set the tolerance on the constraints to the minimum: the tolerances on the # constraints are effectively enforced in the definition of the constraints of # the PSevenProblem. options["GTOpt/ConstraintsTolerance"] = 0.01 * numpy.finfo(numpy.float32).eps # Grab the initial samples from the options samples = self.__get_samples(options, normalize_ds) # Check whether the evaluations budget is sufficient for the expensive functions self.__check_expensive_evaluations_budget(options, samples, max_iter) # Run the algorithm and return the result try: result = solver.solve(pseven_problem, options=options, **samples) except p7core.exceptions.UserEvaluateException as exception: # Check whether a GEMSEO stopping criterion was raised during an # evaluation called by pSeven for criterion in [ MaxIterReachedException, FunctionIsNan, DesvarIsNan, XtolReached, FtolReached, MaxTimeReached, ]: if str(exception).startswith(criterion.__name__): raise criterion raise exception else: status = result.status # Close the log file if log_path is not None: stream.close() return self.get_optimum_from_database(str(status), def __get_samples( self, options: MutableMapping[str, Any], normalize_design_space: bool, ) -> dict[str, ndarray]: """Get the pSeven initial samples. Args: options: The processed options. normalize_design_space: Whether the design space is normalized. Returns: The pSeven initial samples. """ samples = dict() sample_x = options.pop(self.__SAMPLE_X) if sample_x is not None: if normalize_design_space: sample_x = [ self.problem.design_space.normalize_vect(point) for point in sample_x ] samples[self.__SAMPLE_X] = sample_x for option in [self.__SAMPLE_F, self.__SAMPLE_C]: sample_functions = options.pop(option) if sample_functions is not None: samples[option] = sample_functions else: del options[self.__SAMPLE_F] del options[self.__SAMPLE_C] return samples def __check_functions(self) -> None: """Check that the algorithm is consistent with the problem functions. Raises: RuntimeError: If a solver for constrained problems is called on a problem without constraint. TypeError: If the QP solver is called on a problem whose objective is neither quadratic nor linear or one of whose constraints are not linear. """ if self.internal_algo_name in [ self.__MOM, self.__QP, self.__SQP, self.__SQ2P, ]: if not self.problem.constraints: raise RuntimeError( f"{self.algo_name} requires at least one constraint." ) if self.internal_algo_name == self.__QP: if not isinstance( self.problem.nonproc_objective, (MDOQuadraticFunction, MDOLinearFunction), ): # TODO: support several objectives raise TypeError( "{} requires the objective to be quadratic or linear".format( self.algo_name ) ) for constraint in self.problem.nonproc_constraints: if not isinstance(constraint, MDOLinearFunction): raise TypeError( "{} requires the constraints to be linear," " the following is not: {}".format( self.algo_name, ) ) def __check_expensive_evaluations_budget( self, options: dict[str, Any], samples: dict[str, ndarray], max_iter: int, ) -> None: """Check whether the expensive evaluations budget is sufficient. Args: options: The pSeven options. samples: The additional initial guesses. max_iter: The maximum number of evaluated designs. """ max_expensive_iter = options.get("GTOpt/MaximumExpensiveIterations") if max_expensive_iter is None: return number_of_initial_guesses = 0 if self.problem.design_space.has_current_value(): number_of_initial_guesses += 1 if self.__SAMPLE_X in samples: number_of_initial_guesses += len(samples[self.__SAMPLE_X]) if max_iter < max_expensive_iter + number_of_initial_guesses: LOGGER.warning( "The evaluations budget (%s=%d) is to small to compute the " "expensive functions at both the initial guesses (%d) and the " "iterates (%d).", self.MAX_ITER, max_iter, number_of_initial_guesses, max_expensive_iter, )