nmoo.utils.population

  1"""
  2Population related methods.
  3"""
  4__docformat__ = "google"
  5
  6from typing import Dict, List, Union
  7
  8import numpy as np
  9from pymoo.core.population import Population
 10
 11
 12def pareto_frontier_mask(arr: np.ndarray) -> np.ndarray:
 13    """
 14    Computes the Pareto frontier of a set of points. The array must have an
 15    `ndim` of 2.
 16
 17    Returns:
 18        A mask (array of booleans) on `arr` selecting the points belonging to
 19        the Pareto frontier. The actual Pareto frontier can be computed with
 20
 21            pfm = pareto_frontier_mask(arr)
 22            pf = arr[pfm]
 23
 24    Warning:
 25        The direction of the optimum is assumed to towards the "all negative"
 26        quadrant, i.e. $a$ (weakly) dominates $b$ if $a_i \\leq b_i$ for all $0
 27        \\leq i < d$.
 28
 29    Todo:
 30        * Possibility to specify a direction of optimum;
 31        * fancy tree-based optimization.
 32    """
 33    if not arr.ndim == 2:
 34        raise ValueError("The input array must be of shape (N, d).")
 35
 36    d = arr.shape[-1]
 37
 38    if d == 1:
 39        mask = np.full(len(arr), False)
 40        mask[arr.argmin()] = True
 41        return mask
 42    if d == 2:
 43        return pareto_frontier_mask_2d(arr)
 44
 45    argsort0, mask = np.argsort(arr[:, 0]), np.full(len(arr), True)
 46    for i, j in enumerate(argsort0):
 47        if not mask[j]:
 48            continue
 49        for k in filter(lambda x: mask[x], argsort0[i + 1 :]):
 50            # faster than (arr[k] <= arr[j]).any()
 51            for l in range(1, d):
 52                if arr[k][l] <= arr[j][l]:
 53                    mask[k] = True
 54                    break
 55            else:
 56                mask[k] = False
 57    return mask
 58
 59
 60def pareto_frontier_mask_2d(arr: np.ndarray) -> np.ndarray:
 61    """
 62    Computes the Pareto frontier of a set of 2D points. Faster than
 63    `pareto_frontier_mask`. Note that `pareto_frontier_mask` will automatically
 64    call this method whenever possible.
 65
 66    Returns:
 67        A mask (array of booleans) on `arr` selecting the points belonging to
 68        the Pareto frontier. The actual Pareto frontier can be computed with
 69
 70            pfm = pareto_frontier_mask(arr)
 71            pf = arr[pfm]
 72
 73    Warning:
 74        The direction of the optimum is assumed to be south-west.
 75    """
 76    if not (arr.ndim == 2 and arr.shape[-1] == 2):
 77        raise ValueError("The input array must be of shape (N, 2).")
 78    argsort0, mask = np.argsort(arr[:, 0]), np.full(len(arr), False)
 79    i = argsort0[0]  # Index of the last Pareto point
 80    mask[i] = True
 81    for j in argsort0:
 82        # Current point is in the south-east quadrant of the last Pareto point
 83        mask[j] = arr[j][1] <= arr[i][1]
 84        if mask[j]:
 85            i = j
 86    return mask
 87
 88
 89def population_list_to_dict(
 90    populations: Union[Population, List[Population]]
 91) -> Dict[str, np.ndarray]:
 92    """
 93    Transforms a list of pymoo Population (or a single Population) into a dict
 94    containing the following
 95
 96    * `X`: an `np.array` containing all `X` fields of all individuals across
 97      all populations;
 98    * `F`: an `np.array` containing all `F` fields of all individuals across
 99      all populations;
100    * `G`: an `np.array` containing all `G` fields of all individuals across
101      all populations;
102    * `dF`: an `np.array` containing all `dF` fields of all individuals across
103      all populations;
104    * `dG`: an `np.array` containing all `dG` fields of all individuals across
105      all populations;
106    * `ddF`: an `np.array` containing all `ddF` fields of all individuals
107      across all populations;
108    * `ddG`: an `np.array` containing all `ddG` fields of all individuals
109      across all populations;
110    * `CV`: an `np.array` containing all `CV` fields of all individuals across
111      all populations;
112    * `feasible`: an `np.array` containing all `feasible` fields of all
113      individuals across all populations;
114    * `_batch`: the index of the population the individual belongs to.
115
116    So all `np.arrays` have the same length, which is the total number of
117    individual across all populations. Each "row" corresponds to the data
118    associated to this individual (`X`, `F`, `G`, `dF`, `dG`, `ddF`, `ddG`,
119    `CV`, `feasible`), as well as the population index it belongs to
120    (`_batch`).
121    """
122    if isinstance(populations, Population):
123        populations = [populations]
124    fields = ["X", "F", "G", "dF", "dG", "ddF", "ddG", "CV", "feasible"]
125    data: Dict[str, List[np.ndarray]] = {f: [] for f in fields + ["_batch"]}
126    if not populations:
127        return {k: np.array([]) for k in data}
128    for i, pop in enumerate(populations):
129        for f in fields:
130            data[f].append(pop.get(f))
131        data["_batch"].append(np.full(len(pop), i + 1))
132    return {k: np.concatenate(v) for k, v in data.items()}