Source code for gemseo.core.mdofunctions.norm_db_function
# Copyright 2021 IRT Saint Exupéry, https://www.irt-saintexupery.com
#
# 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
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# 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 - API and implementation and/or documentation
# :author: Francois Gallard
# OTHER AUTHORS - MACROSCOPIC CHANGES
# :author: Benoit Pauwels - Stacked data management
# (e.g. iteration index)
# :author: Gilberto Ruiz Jimenez
"""An MDOFunction subclass to support formulations."""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING
from numpy import any as np_any
from numpy import isnan as np_isnan
from numpy import ndarray
from gemseo.algos.database import Database
from gemseo.algos.stop_criteria import DesvarIsNan
from gemseo.algos.stop_criteria import FunctionIsNan
from gemseo.algos.stop_criteria import MaxIterReachedException
from gemseo.core.mdofunctions.mdo_function import MDOFunction
if TYPE_CHECKING:
from gemseo.algos.opt_problem import OptimizationProblem
LOGGER = logging.getLogger(__name__)
[docs]class NormDBFunction(MDOFunction):
"""An :class:`.MDOFunction` object to be evaluated from a database."""
def __init__(
self,
orig_func: MDOFunction,
normalize: bool,
is_observable: bool,
optimization_problem: OptimizationProblem,
) -> None:
"""
Args:
orig_func: The original function to be wrapped.
normalize: If True, then normalize the function's input vector.
is_observable: If True, new_iter_listeners are not called
when function is called (avoid recursive call).
optimization_problem: The optimization problem object that contains
the function.
"""
self.__normalize = normalize
self.__orig_func = orig_func
self.__is_observable = is_observable
self.__optimization_problem = optimization_problem
# For performance
design_space = self.__optimization_problem.design_space
self.__unnormalize_vect = design_space.unnormalize_vect
# self.__round_vect = design_space.round_vect
self.__unnormalize_grad = design_space.unnormalize_grad
self.__evaluate_orig_func = self.__orig_func.evaluate
self.__jac_orig_func = orig_func.jac
self.__is_max_iter_reached = self.__optimization_problem.is_max_iter_reached
super().__init__(
self._func,
orig_func.name,
jac=self._jac,
f_type=orig_func.f_type,
expr=orig_func.expr,
args=orig_func.args,
dim=orig_func.dim,
outvars=orig_func.outvars,
)
def _func(
self,
x_vect: ndarray,
) -> ndarray:
"""Compute the function to be passed to the optimizer.
Args:
x_vect: The value of the design variables.
Returns:
The evaluation of the function for this value of the design variables.
Raises:
DesvarIsNan: If the design variables contain a NaN value.
FunctionIsNan: If a function returns a NaN value when evaluated.
MaxIterReachedException: If the maximum number of iterations has been
reached.
"""
if np_any(np_isnan(x_vect)):
raise DesvarIsNan(f"Design Variables contain a NaN value: {x_vect}")
normalize = self.__normalize
if normalize:
xn_vect = x_vect
xu_vect = self.__unnormalize_vect(xn_vect)
else:
xu_vect = x_vect
xn_vect = None
# For performance, hash once, and reuse in get/store methods
database = self.__optimization_problem.database
hashed_xu = database.get_hashed_key(xu_vect, False)
# try to retrieve the evaluation
value = database.get_f_of_x(self.name, hashed_xu)
if value is None:
new_eval = database.is_new_eval(hashed_xu)
if new_eval and self.__is_max_iter_reached():
raise MaxIterReachedException()
# if not evaluated yet, evaluate
if normalize:
value = self.__evaluate_orig_func(xn_vect)
else:
value = self.__evaluate_orig_func(xu_vect)
if self.__optimization_problem.stop_if_nan and np_any(np_isnan(value)):
raise FunctionIsNan(f"The function {self.name} is NaN for x={xu_vect}")
# store (x, f(x)) in database
database.store(hashed_xu, {self.name: value})
return value
def _jac(self, x_vect: ndarray) -> ndarray:
"""Compute the gradient of the function to be passed to the optimizer.
Args:
x_vect: The value of the design variables.
Returns:
The evaluation of the gradient for this value of the design variables.
Raises:
FunctionIsNan: If the design variables contain a NaN value.
If the evaluation of the jacobian results in a NaN value.
"""
if np_any(np_isnan(x_vect)):
raise FunctionIsNan(f"Design Variables contain a NaN value: {x_vect}")
normalize = self.__normalize
if normalize:
xn_vect = x_vect
xu_vect = self.__unnormalize_vect(xn_vect)
else:
xu_vect = x_vect
xn_vect = None
database = self.__optimization_problem.database
design_space = self.__optimization_problem.design_space
# try to retrieve the evaluation
jac_u = database.get_f_of_x(Database.get_gradient_name(self.name), xu_vect)
if jac_u is None:
new_eval = database.is_new_eval(xu_vect)
if new_eval and self.__is_max_iter_reached():
raise MaxIterReachedException()
# if not evaluated yet, evaluate
if self.__normalize:
jac_n = self.__jac_orig_func(xn_vect)
jac_u = self.__unnormalize_grad(jac_n)
else:
jac_u = self.__jac_orig_func(xu_vect)
jac_n = None
if np_any(np_isnan(jac_u)) and self.__optimization_problem.stop_if_nan:
raise FunctionIsNan(
"Function {}'s Jacobian is NaN "
"for x={}".format(self.name, xu_vect)
)
func_name_to_value = {Database.get_gradient_name(self.name): jac_u}
# store (x, j(x)) in database
database.store(xu_vect, func_name_to_value)
else:
jac_n = design_space.normalize_grad(jac_u)
if self.__normalize:
return jac_n.real
else:
return jac_u.real
@property
def expects_normalized_inputs(self) -> bool:
return self.__normalize
