"""Provides the GaussianEmitter."""
from __future__ import annotations
import numbers
from collections.abc import Collection
import numpy as np
from numpy.typing import ArrayLike
from ribs._utils import check_batch_shape, check_shape
from ribs.archives import ArchiveBase
from ribs.emitters._emitter_base import EmitterBase
from ribs.typing import Float, Int
[docs]
class GaussianEmitter(EmitterBase):
"""Emits solutions by adding Gaussian noise to existing elites.
If the archive is empty and ``initial_solutions`` is set, a call to :meth:`ask` will
return ``initial_solutions``. If ``initial_solutions`` is not set, we draw solutions
from a Gaussian distribution centered at ``x0`` with standard deviation ``sigma``.
Otherwise, each solution is drawn from a distribution centered at a randomly chosen
elite with standard deviation ``sigma``.
Args:
archive: Archive of solutions, e.g., :class:`ribs.archives.GridArchive`.
sigma: Standard deviation of the Gaussian distribution. Note we assume the
Gaussian is diagonal, so if this argument is an array, it must have the same
dimensionality as the solutions.
x0: Center of the Gaussian distribution from which to sample solutions when the
archive is empty. This argument is ignored if ``initial_solutions`` is set.
initial_solutions: An (n, solution_dim) array of solutions to be used when the
archive is empty. If this argument is None, then solutions will be sampled
from a Gaussian distribution centered at ``x0`` with standard deviation
``sigma``.
bounds: Bounds of the solution space. Pass None to indicate there are no bounds.
Alternatively, pass an array-like to specify the bounds for each dim. Each
element in this array-like can be None to indicate no bound, or a tuple of
``(lower_bound, upper_bound)``, where ``lower_bound`` or ``upper_bound`` may
be None to indicate no bound. Unbounded upper bounds are set to +inf, and
unbounded lower bounds are set to -inf.
lower_bounds: Instead of specifying ``bounds``, ``lower_bounds`` and
``upper_bounds`` may be specified. This is useful if, for instance,
solutions are multi-dimensional. Here, pass None to indicate there are no
bounds (i.e., bounds are set to -inf), or pass an array specifying the lower
bounds of the solution space.
upper_bounds: Upper bounds of the solution space; see ``lower_bounds`` above.
Pass None to indicate there are no bounds (i.e., bounds are set to inf).
batch_size: Number of solutions to return in :meth:`ask`.
seed: Value to seed the random number generator. Set to None to avoid a fixed
seed.
Raises:
ValueError: There is an error in x0 or initial_solutions.
ValueError: There is an error in the bounds configuration.
"""
def __init__(
self,
archive: ArchiveBase,
*,
sigma: Float | ArrayLike,
x0: ArrayLike | None = None,
initial_solutions: ArrayLike | None = None,
bounds: Collection[tuple[None | Float, None | Float]] | None = None,
lower_bounds: ArrayLike | None = None,
upper_bounds: ArrayLike | None = None,
batch_size: Int = 64,
seed: Int | None = None,
) -> None:
EmitterBase.__init__(
self,
archive,
solution_dim=archive.solution_dim,
bounds=bounds,
lower_bounds=lower_bounds,
upper_bounds=upper_bounds,
)
self._rng = np.random.default_rng(seed)
self._batch_size = batch_size
self._sigma = np.asarray(sigma, dtype=archive.dtypes["solution"])
self._x0 = None
self._initial_solutions = None
self._noise_shape = (
(self.batch_size, self.solution_dim)
if isinstance(self.solution_dim, numbers.Integral)
else (self.batch_size, *self.solution_dim)
)
if x0 is None and initial_solutions is None:
raise ValueError("Either x0 or initial_solutions must be provided.")
if x0 is not None and initial_solutions is not None:
raise ValueError("x0 and initial_solutions cannot both be provided.")
if x0 is not None:
self._x0 = np.array(x0, dtype=archive.dtypes["solution"])
check_shape(self._x0, "x0", archive.solution_dim, "archive.solution_dim")
elif initial_solutions is not None:
self._initial_solutions = np.asarray(
initial_solutions, dtype=archive.dtypes["solution"]
)
check_batch_shape(
self._initial_solutions,
"initial_solutions",
archive.solution_dim,
"archive.solution_dim",
)
@property
def x0(self) -> np.ndarray | None:
"""Initial Gaussian distribution center.
Solutions are sampled from this distribution when the archive is empty (if
:attr:`initial_solutions` is not set).
"""
return self._x0
@property
def initial_solutions(self) -> np.ndarray | None:
"""Returned when the archive is empty (if :attr:`x0` is not set)."""
return self._initial_solutions
@property
def sigma(self) -> np.floating | np.ndarray:
"""Standard deviation of the (diagonal) Gaussian distribution."""
return self._sigma
@property
def batch_size(self) -> Int:
"""Number of solutions to return in :meth:`ask`."""
return self._batch_size
def _clip(self, solutions: np.ndarray) -> np.ndarray:
"""Clips solutions to the bounds of the solution space."""
return np.clip(solutions, self.lower_bounds, self.upper_bounds)
[docs]
def ask(self) -> np.ndarray:
"""Creates solutions by adding Gaussian noise to elites in the archive.
If the archive is empty and ``initial_solutions`` is set, a call to :meth:`ask`
will return ``initial_solutions``. If ``initial_solutions`` is not set, we draw
solutions from a Gaussian distribution centered at ``x0`` with standard
deviation ``sigma``. Otherwise, each solution is drawn from a distribution
centered at a randomly chosen elite with standard deviation ``sigma``.
Returns:
If the archive is not empty, ``(batch_size, solution_dim)`` array --
contains ``batch_size`` new solutions to evaluate. If the archive is empty,
we return ``initial_solutions``, which might not have ``batch_size``
solutions.
"""
if self.archive.empty and self.initial_solutions is not None:
return self._clip(self.initial_solutions)
if self.archive.empty:
parents = np.repeat(self.x0[None], repeats=self.batch_size, axis=0)
else:
parents = self.archive.sample_elites(self.batch_size)["solution"]
noise = self._rng.normal(
scale=self.sigma,
size=self._noise_shape,
).astype(self.archive.dtypes["solution"])
return self._clip(parents + noise)