target_space.py

"""Manages the optimization domain and holds points."""

from __future__ import annotations

from typing import TYPE_CHECKING, Any
from warnings import warn

import numpy as np
from colorama import Fore

from bayes_opt.exception import NotUniqueError
from bayes_opt.util import ensure_rng

if TYPE_CHECKING:
    from collections.abc import Callable, Mapping, Sequence

    from numpy.random import RandomState
    from numpy.typing import NDArray

    from bayes_opt.constraint import ConstraintModel

    Float = np.floating[Any]


def _hashable(x: NDArray[Float]) -> tuple[float, ...]:
    """Ensure that a point is hashable by a python dict."""
    return tuple(map(float, x))


class TargetSpace:
    """Holds the param-space coordinates (X) and target values (Y).

    Allows for constant-time appends.

    Parameters
    ----------
    target_func : function or None.
        Function to be maximized.

    pbounds : dict
        Dictionary with parameters names as keys and a tuple with minimum
        and maximum values.

    random_state : int, RandomState, or None
        optionally specify a seed for a random number generator

    allow_duplicate_points: bool, optional (default=False)
        If True, the optimizer will allow duplicate points to be registered.
        This behavior may be desired in high noise situations where repeatedly probing
        the same point will give different answers. In other situations, the acquisition
        may occasionally generate a duplicate point.

    Examples
    --------
    >>> def target_func(p1, p2):
    >>>     return p1 + p2
    >>> pbounds = {"p1": (0, 1), "p2": (1, 100)}
    >>> space = TargetSpace(target_func, pbounds, random_state=0)
    >>> x = np.array([4, 5])
    >>> y = target_func(x)
    >>> space.register(x, y)
    >>> assert self.max()["target"] == 9
    >>> assert self.max()["params"] == {"p1": 1.0, "p2": 2.0}
    """

    def __init__(
        self,
        target_func: Callable[..., float] | None,
        pbounds: Mapping[str, tuple[float, float]],
        constraint: ConstraintModel | None = None,
        random_state: int | RandomState | None = None,
        allow_duplicate_points: bool | None = False,
    ) -> None:
        self.random_state = ensure_rng(random_state)
        self._allow_duplicate_points = allow_duplicate_points or False
        self.n_duplicate_points = 0

        # The function to be optimized
        self.target_func = target_func

        # Get the name of the parameters
        self._keys: list[str] = sorted(pbounds)
        # Create an array with parameters bounds
        self._bounds: NDArray[Float] = np.array(
            [item[1] for item in sorted(pbounds.items(), key=lambda x: x[0])], dtype=float
        )

        # preallocated memory for X and Y points
        self._params: NDArray[Float] = np.empty(shape=(0, self.dim))
        self._target: NDArray[Float] = np.empty(shape=(0,))

        # keep track of unique points we have seen so far
        self._cache: dict[tuple[float, ...], float | tuple[float, float | NDArray[Float]]] = {}

        self._constraint: ConstraintModel | None = constraint

        if constraint is not None:
            # preallocated memory for constraint fulfillment
            self._constraint_values: NDArray[Float]
            if constraint.lb.size == 1:
                self._constraint_values = np.empty(shape=(0), dtype=float)
            else:
                self._constraint_values = np.empty(shape=(0, constraint.lb.size), dtype=float)

        self._sorting_warning_already_shown = False  # TODO: remove in future version

    def __contains__(self, x: NDArray[Float]) -> bool:
        """Check if this parameter has already been registered.

        Returns
        -------
        bool
        """
        return _hashable(x) in self._cache

    def __len__(self) -> int:
        """Return number of observations registered.

        Returns
        -------
        int
        """
        if len(self._params) != len(self._target):
            error_msg = "The number of parameters and targets do not match."
            raise ValueError(error_msg)
        return len(self._target)

    @property
    def empty(self) -> bool:
        """Check if anything has been registered.

        Returns
        -------
        bool
        """
        return len(self) == 0

    @property
    def params(self) -> NDArray[Float]:
        """Get the parameter values registered to this TargetSpace.

        Returns
        -------
        np.ndarray
        """
        return self._params

    @property
    def target(self) -> NDArray[Float]:
        """Get the target function values registered to this TargetSpace.

        Returns
        -------
        np.ndarray
        """
        return self._target

    @property
    def dim(self) -> int:
        """Get the number of parameter names.

        Returns
        -------
        int
        """
        return len(self._keys)

    @property
    def keys(self) -> list[str]:
        """Get the keys (or parameter names).

        Returns
        -------
        list of str
        """
        return self._keys

    @property
    def bounds(self) -> NDArray[Float]:
        """Get the bounds of this TargetSpace.

        Returns
        -------
        np.ndarray
        """
        return self._bounds

    @property
    def constraint(self) -> ConstraintModel | None:
        """Get the constraint model.

        Returns
        -------
        ConstraintModel
        """
        return self._constraint

    @property
    def constraint_values(self) -> NDArray[Float]:
        """Get the constraint values registered to this TargetSpace.

        Returns
        -------
        np.ndarray
        """
        if self._constraint is None:
            error_msg = "TargetSpace belongs to an unconstrained optimization"
            raise AttributeError(error_msg)

        return self._constraint_values

    @property
    def mask(self) -> NDArray[np.bool_]:
        """Return a boolean array of valid points.

        Points are valid if they satisfy both the constraint and boundary conditions.

        Returns
        -------
        np.ndarray
        """
        mask = np.ones_like(self.target, dtype=bool)

        # mask points that don't satisfy the constraint
        if self._constraint is not None:
            mask &= self._constraint.allowed(self._constraint_values)

        # mask points that are outside the bounds
        if self._bounds is not None:
            within_bounds = np.all(
                (self._bounds[:, 0] <= self._params) & (self._params <= self._bounds[:, 1]), axis=1
            )
            mask &= within_bounds

        return mask

    def params_to_array(self, params: Mapping[str, float]) -> NDArray[Float]:
        """Convert a dict representation of parameters into an array version.

        Parameters
        ----------
        params : dict
            a single point, with len(x) == self.dim.

        Returns
        -------
        np.ndarray
            Representation of the parameters as an array.
        """
        if set(params) != set(self.keys):
            error_msg = (
                f"Parameters' keys ({sorted(params)}) do "
                f"not match the expected set of keys ({self.keys})."
            )
            raise ValueError(error_msg)
        return np.asarray([params[key] for key in self.keys])

    def array_to_params(self, x: NDArray[Float]) -> dict[str, float]:
        """Convert an array representation of parameters into a dict version.

        Parameters
        ----------
        x : np.ndarray
            a single point, with len(x) == self.dim.

        Returns
        -------
        dict
            Representation of the parameters as dictionary.
        """
        if len(x) != len(self.keys):
            error_msg = (
                f"Size of array ({len(x)}) is different than the "
                f"expected number of parameters ({len(self.keys)})."
            )
            raise ValueError(error_msg)
        return dict(zip(self.keys, x))

    def _as_array(self, x: Any) -> NDArray[Float]:
        try:
            x = np.asarray(x, dtype=float)
        except TypeError:
            x = self.params_to_array(x)

        x = x.ravel()
        if x.size != self.dim:
            error_msg = (
                f"Size of array ({len(x)}) is different than the "
                f"expected number of parameters ({len(self.keys)})."
            )
            raise ValueError(error_msg)
        return x

    def register(
        self,
        params: Mapping[str, float] | Sequence[float] | NDArray[Float],
        target: float,
        constraint_value: float | NDArray[Float] | None = None,
    ) -> None:
        """Append a point and its target value to the known data.

        Parameters
        ----------
        params : np.ndarray
            a single point, with len(x) == self.dim.

        target : float
            target function value

        constraint_value : float or np.ndarray or None
            Constraint function value

        Raises
        ------
        NotUniqueError:
            if the point is not unique

        Notes
        -----
        runs in amortized constant time

        Examples
        --------
        >>> target_func = lambda p1, p2: p1 + p2
        >>> pbounds = {"p1": (0, 1), "p2": (1, 100)}
        >>> space = TargetSpace(target_func, pbounds)
        >>> len(space)
        0
        >>> x = np.array([0, 0])
        >>> y = 1
        >>> space.register(x, y)
        >>> len(space)
        1
        """
        # TODO: remove in future version
        if isinstance(params, np.ndarray) and not self._sorting_warning_already_shown:
            msg = (
                "You're attempting to register an np.ndarray. Currently, the optimizer internally sorts"
                " parameters by key and expects any registered array to respect this order. In future"
                " versions this behaviour will change and the order as given by the pbounds dictionary"
                " will be used. If you wish to retain sorted parameters, please manually sort your pbounds"
                " dictionary before constructing the optimizer."
            )
            warn(msg, stacklevel=1)
            self._sorting_warning_already_shown = True
        x = self._as_array(params)
        if x in self:
            if self._allow_duplicate_points:
                self.n_duplicate_points = self.n_duplicate_points + 1

                print(
                    Fore.RED + f"Data point {x} is not unique. {self.n_duplicate_points}"
                    " duplicates registered. Continuing ..." + Fore.RESET
                )
            else:
                error_msg = (
                    f"Data point {x} is not unique. You can set"
                    ' "allow_duplicate_points=True" to avoid this error'
                )
                raise NotUniqueError(error_msg)

        # if x is not within the bounds of the parameter space, warn the user
        if self._bounds is not None and not np.all((self._bounds[:, 0] <= x) & (x <= self._bounds[:, 1])):
            warn(f"\nData point {x} is outside the bounds of the parameter space. ", stacklevel=2)

        # Make copies of the data, so as not to modify the originals incase something fails
        # during the registration process. This prevents out-of-sync data.
        params_copy: NDArray[Float] = np.concatenate([self._params, x.reshape(1, -1)])
        target_copy: NDArray[Float] = np.concatenate([self._target, [target]])
        cache_copy = self._cache.copy()  # shallow copy suffices

        if self._constraint is None:
            # Insert data into unique dictionary
            cache_copy[_hashable(x.ravel())] = target
        else:
            if constraint_value is None:
                msg = (
                    "When registering a point to a constrained TargetSpace"
                    " a constraint value needs to be present."
                )
                raise ValueError(msg)
            # Insert data into unique dictionary
            cache_copy[_hashable(x.ravel())] = (target, constraint_value)
            constraint_values_copy: NDArray[Float] = np.concatenate(
                [self._constraint_values, [constraint_value]]
            )
            self._constraint_values = constraint_values_copy

        # Operations passed, update the variables
        self._params = params_copy
        self._target = target_copy
        self._cache = cache_copy

    def probe(
        self, params: Mapping[str, float] | Sequence[float] | NDArray[Float]
    ) -> float | tuple[float, float | NDArray[Float]]:
        """Evaluate the target function on a point and register the result.

        Notes
        -----
        If `params` has been previously seen and duplicate points are not allowed,
        returns a cached value of `result`.

        Parameters
        ----------
        params : np.ndarray
            a single point, with len(x) == self.dim

        Returns
        -------
        result : float | Tuple(float, float)
            target function value, or Tuple(target function value, constraint value)

        Example
        -------
        >>> target_func = lambda p1, p2: p1 + p2
        >>> pbounds = {"p1": (0, 1), "p2": (1, 100)}
        >>> space = TargetSpace(target_func, pbounds)
        >>> space.probe([1, 5])
        >>> assert self.max()["target"] == 6
        >>> assert self.max()["params"] == {"p1": 1.0, "p2": 5.0}
        """
        x = self._as_array(params)
        if x in self and not self._allow_duplicate_points:
            return self._cache[_hashable(x.ravel())]

        dict_params = self.array_to_params(x)
        if self.target_func is None:
            error_msg = "No target function has been provided."
            raise ValueError(error_msg)
        target = self.target_func(**dict_params)

        if self._constraint is None:
            self.register(x, target)
            return target

        constraint_value = self._constraint.eval(**dict_params)
        self.register(x, target, constraint_value)
        return target, constraint_value

    def random_sample(self) -> NDArray[Float]:
        """
        Sample a random point from within the bounds of the space.

        Returns
        -------
        data: ndarray
            [1 x dim] array with dimensions corresponding to `self._keys`

        Examples
        --------
        >>> target_func = lambda p1, p2: p1 + p2
        >>> pbounds = {"p1": (0, 1), "p2": (1, 100)}
        >>> space = TargetSpace(target_func, pbounds, random_state=0)
        >>> space.random_sample()
        array([[ 0.54488318,   55.33253689]])
        """
        data = np.empty((1, self.dim))
        for col, (lower, upper) in enumerate(self._bounds):
            data.T[col] = self.random_state.uniform(lower, upper, size=1)
        return data.ravel()

    def _target_max(self) -> float | None:
        """Get the maximum target value within the current parameter bounds.

        If there is a constraint present, the maximum value that fulfills the
        constraint within the parameter bounds is returned.

        Returns
        -------
        max: float
            The maximum target value.
        """
        if len(self.target) == 0:
            return None

        if len(self.target[self.mask]) == 0:
            return None

        return self.target[self.mask].max()

    def max(self) -> dict[str, Any] | None:
        """Get maximum target value found and corresponding parameters.

        If there is a constraint present, the maximum value that fulfills the
        constraint within the parameter bounds is returned.

        Returns
        -------
        res: dict
            A dictionary with the keys 'target' and 'params'. The value of
            'target' is the maximum target value, and the value of 'params' is
            a dictionary with the parameter names as keys and the parameter
            values as values.
        """
        target_max = self._target_max()
        if target_max is None:
            return None

        target = self.target[self.mask]
        params = self.params[self.mask]
        target_max_idx = np.argmax(target)

        res = {"target": target_max, "params": dict(zip(self.keys, params[target_max_idx]))}

        if self._constraint is not None:
            constraint_values = self.constraint_values[self.mask]
            res["constraint"] = constraint_values[target_max_idx]

        return res

    def res(self) -> list[dict[str, Any]]:
        """Get all target values and constraint fulfillment for all parameters.

        Returns
        -------
        res: list
            A list of dictionaries with the keys 'target', 'params', and
            'constraint'. The value of 'target' is the target value, the value
            of 'params' is a dictionary with the parameter names as keys and the
            parameter values as values, and the value of 'constraint' is the
            constraint fulfillment.

        Notes
        -----
        Does not report if points are within the bounds of the parameter space.
        """
        if self._constraint is None:
            params = [dict(zip(self.keys, p)) for p in self.params]

            return [{"target": target, "params": param} for target, param in zip(self.target, params)]

        params = [dict(zip(self.keys, p)) for p in self.params]

        return [
            {"target": target, "constraint": constraint_value, "params": param, "allowed": allowed}
            for target, constraint_value, param, allowed in zip(
                self.target,
                self._constraint_values,
                params,
                self._constraint.allowed(self._constraint_values),
            )
        ]

    def set_bounds(self, new_bounds: Mapping[str, NDArray[Float] | Sequence[float]]) -> None:
        """Change the lower and upper search bounds.

        Parameters
        ----------
        new_bounds : dict
            A dictionary with the parameter name and its new bounds
        """
        for row, key in enumerate(self.keys):
            if key in new_bounds:
                self._bounds[row] = new_bounds[key]