nmoo.denoisers.knnavg

  1"""KNN-Averaging"""
  2__docformat__ = "google"
  3
  4from scipy.spatial.distance import cdist
  5import numpy as np
  6import pandas as pd
  7
  8from nmoo.wrapped_problem import WrappedProblem
  9
 10
 11class KNNAvg(WrappedProblem):
 12    """
 13    Implementation of the KNN-Averaging algorithm[^quatic21].
 14
 15    [^quatic21]: Klikovits, S., Arcaini, P. (2021). KNN-Averaging for Noisy
 16        Multi-objective Optimisation. In: Paiva, A.C.R., Cavalli, A.R., Ventura
 17        Martins, P., Pérez-Castillo, R. (eds) Quality of Information and
 18        Communications Technology. QUATIC 2021. Communications in Computer and
 19        Information Science, vol 1439. Springer, Cham.
 20        https://doi.org/10.1007/978-3-030-85347-1_36
 21    """
 22
 23    _distance_weight_mode: str
 24    _max_distance: float
 25    _n_neighbors: int
 26
 27    def __init__(
 28        self,
 29        problem: WrappedProblem,
 30        max_distance: float,
 31        n_neighbors: int = 5,  # KNN
 32        distance_weight_type: str = "uniform",
 33        *,
 34        name: str = "knn_avg",
 35        **kwargs,
 36    ):
 37        """
 38        Constructor.
 39
 40        Args:
 41            problem (`nmoo.wrapped_problem.WrappedProblem`): Noisy problem. For
 42                memory optimization reasons, this should be a
 43                `nmoo.wrapped_problem.WrappedProblem` as opposed to a pymoo
 44                `Problem`.
 45            distance_weight_type (str): Either `squared` or `uniform` (the
 46                default).
 47            max_distance (float): Distance cutoff.
 48            n_neighbors (int): Number of neighbors to consider (the K in KNN).
 49            name (str): An optional name for this problem. This will be used
 50                when creating history dump files. Defaults to `knn_avg`.
 51        """
 52        super().__init__(problem, name=name, **kwargs)
 53
 54        if distance_weight_type not in ["squared", "uniform"]:
 55            raise ValueError(
 56                "Parameter distance_weight_type must be either 'squared' or "
 57                "'uniform'."
 58            )
 59        self._distance_weight_mode = distance_weight_type
 60
 61        if max_distance < 0.0:
 62            raise ValueError(
 63                "Parameter max_distance must either be 'None' or >= 0."
 64            )
 65        self._max_distance = max_distance
 66
 67        if n_neighbors <= 0:
 68            raise ValueError("Parameter n_neighbors must be >= 1.")
 69        self._n_neighbors = n_neighbors
 70
 71    def _evaluate(self, x, out, *args, **kwargs):
 72        """
 73        Applies the KNN-Avg algorithm to the wrapped (noisy) problem's output.
 74        """
 75        self._problem._evaluate(x, out, *args, **kwargs)
 76        for i, sol in enumerate(x):
 77            # Store the solution history into a dataframe (note that we are
 78            # using the wrapped problem's history to make sure this dataframe
 79            # is never empty).
 80            x_hist = pd.DataFrame(self._problem._history["X"])
 81            # Compute the standardized Euclidean distances between the current
 82            # solution (sol) and all historical solutions.
 83            x_hist["_sed"] = cdist(
 84                self._problem._history["X"],
 85                sol.reshape((1, -1)),
 86                "seuclidean",
 87            )
 88            # Apply the KNN scheme: select the K closest neighbors among those
 89            # closer than the maximum allowed distance.
 90            x_hist = (
 91                x_hist[x_hist["_sed"] <= self._max_distance]
 92                .sort_values(by="_sed")
 93                .head(self._n_neighbors)
 94            )
 95            if x_hist.shape[0] <= 1:
 96                # If only the current solution remains, then skip to the next
 97                # solution.
 98                continue
 99            # Compute the weights.
100            if self._distance_weight_mode == "squared":
101                x_hist["_w"] = (self._max_distance - x_hist["_sed"]) ** 2
102            elif self._distance_weight_mode == "uniform":
103                x_hist["_w"] = 1.0
104            else:
105                raise RuntimeError(
106                    "Unknown distance weight mode: "
107                    + self._distance_weight_mode
108                )
109            # Compute the weighted averages of the (numerical) outputs.
110            for k in out:
111                if not isinstance(out[k], np.ndarray):
112                    continue
113                out_k_hist = pd.DataFrame(self._problem._history[k])
114                avg = np.average(
115                    out_k_hist.iloc[x_hist.index],
116                    axis=0,
117                    weights=x_hist["_w"],
118                )
119                out[k][i] = avg
120        self.add_to_history_x_out(x, out)