nmoo.wrapped_problem

  1"""
  2Base `nmoo` problem class. A `WrappedProblem` is simply a
  3[`pymoo.core.problem.Problem`](https://pymoo.org/problems/definition.html) that
  4contains another problem (`nmoo` or `pymoo`) to which calls to `_evaluate` are
  5deferred to. A `WrappedProblem` also have call history (although it is the
  6responsability of the `_evaluate` implementation to populate it).
  7
  8Since `WrappedProblem` directly inherits from
  9[`pymoo.core.problem.Problem`](https://pymoo.org/problems/definition.html),
 10wrapped problems can be used seemlessly with `pymoo`.
 11"""
 12__docformat__ = "google"
 13
 14from copy import deepcopy
 15from pathlib import Path
 16from typing import Any, Dict, List, Union
 17
 18import numpy as np
 19from loguru import logger as logging
 20from pymoo.core.problem import Problem
 21
 22
 23class WrappedProblem(Problem):
 24    """
 25    A simple Pymoo `Problem` wrapper that keeps a history of all calls
 26    made to it.
 27    """
 28
 29    _current_history_batch: int = 0
 30    """
 31    The number of batches added to history.
 32    """
 33
 34    _current_run: int = 0
 35    """
 36    Current run number. See `nmoo.utils.WrappedProblem.start_new_run`.
 37    """
 38
 39    _history: Dict[str, np.ndarray]
 40    """
 41    A history is a dictionary that maps string keys (e.g. `"X"`) to a numpy
 42    array of all values of that key (e.g. all values of `"X"`). All the numpy
 43    arrays should have the same length (0th shape component) but are not
 44    required to have the same type.
 45
 46    If you subclass this class, don't forget to carefully document the meaning
 47    of the keys of what you're story in history.
 48    """
 49
 50    _name: str
 51    """The name of this problem"""
 52
 53    _problem: Problem
 54    """Wrapped pymoo problem (or `nmoo.wrapped_problem.WrappedProblem`)"""
 55
 56    def __init__(
 57        self,
 58        problem: Problem,
 59        *,
 60        copy_problem: bool = True,
 61        name: str = "wrapped_problem",
 62    ):
 63        """
 64        Constructor.
 65
 66        Args:
 67            copy_problem (bool): Wether to deepcopy the problem. If `problem`
 68                is a `WrappedProblem`, this is recommended to avoid history
 69                clashes. However, if whatever benchmark that uses this problem
 70                uses multiprocessing (as opposed to single or multithreading),
 71                this does not seem to be necessary.
 72            problem (pymoo `Problem`): A non-noisy pymoo problem (or
 73                `nmoo.wrapped_problem.WrappedProblem`).
 74            name (str): An optional name for this problem. This will be used
 75                when creating history dump files. Defaults to
 76                `wrapped_problem`.
 77        """
 78        super().__init__(
 79            n_var=problem.n_var,
 80            n_obj=problem.n_obj,
 81            n_constr=problem.n_constr,
 82            xl=problem.xl,
 83            xu=problem.xu,
 84            check_inconsistencies=problem.check_inconsistencies,
 85            replace_nan_values_by=problem.replace_nan_values_by,
 86            exclude_from_serialization=problem.exclude_from_serialization,
 87            callback=problem.callback,
 88        )
 89        self._history = {}
 90        self._name = name
 91        self._problem = deepcopy(problem) if copy_problem else problem
 92
 93    def __str__(self):
 94        return self._name
 95
 96    def add_to_history(self, **kwargs):
 97        """
 98        Adds records to the history. The provided keys should match that of the
 99        history (this is not checked at runtime for perfomance reasons). The
100        provided values should be numpy arrays that all have the same length
101        (0th shape component).
102        """
103        self._current_history_batch += 1
104        if not kwargs:
105            # No items to add
106            return
107        lengths = {k: v.shape[0] for k, v in kwargs.items()}
108        if len(set(lengths.values())) > 1:
109            logging.warning(
110                "[add_to_history] The lengths of the arrays don't match: {}",
111                str(lengths),
112            )
113        kwargs["_batch"] = np.full(
114            (max(lengths.values()),),
115            self._current_history_batch,
116        )
117        kwargs["_run"] = np.full(
118            (max(lengths.values()),),
119            self._current_run,
120        )
121        for k, v in kwargs.items():
122            if k not in self._history:
123                self._history[k] = v.copy()
124            else:
125                self._history[k] = np.append(
126                    self._history[k], v.copy(), axis=0
127                )
128
129    def add_to_history_x_out(self, x: np.ndarray, out: dict, **kwargs):
130        """
131        Convenience function to add the `_evaluate` method's `x` and `out` to
132        history, along with potentially other items. Note that the `x` argument
133        is stored under the `X` key to remain consistent with `pymoo`'s API.
134        """
135        self.add_to_history(
136            X=x,
137            **{k: v for k, v in out.items() if isinstance(v, np.ndarray)},
138            **kwargs,
139        )
140
141    def all_layers(self) -> List["WrappedProblem"]:
142        """
143        Returns a list of all the `nmoo.wrapped_problem.WrappedProblem` wrapped
144        within (including the current one). This list is ordered from the
145        outermost one (the current problem) to the innermost one.
146        """
147        return [self] + (
148            self._problem.all_layers()
149            if isinstance(self._problem, WrappedProblem)
150            else []
151        )
152
153    def depth(self) -> int:
154        """
155        Returns the number of `WrappedProblem` that separate this
156        `WrappedProblem` to the ground problem (see `ground_problem`). For
157        example:
158
159            x = WrappedProblem(
160                WrappedProblem(
161                    WrappedProblem(
162                        ZDT1(...)
163                    )
164                )
165            )
166            x.depth()  # Returns 3
167
168        """
169        if isinstance(self._problem, WrappedProblem):
170            return 1 + self._problem.depth()
171        return 1
172
173    def dump_all_histories(
174        self,
175        dir_path: Union[Path, str],
176        name: str,
177        compressed: bool = True,
178        _idx: int = 1,
179    ):
180        """
181        Dumps this problem's history, as well as all problems (more precisely,
182        instances of `nmoo.utils.WrappedProblem`) recursively wrapped within
183        it. This will result in one `.npz` for each problem involved.
184
185        Args:
186            dir_path (Union[Path, str]): Output directory.
187            name (str): The output files will be named according to the
188                following pattern: `<name>.<_idx>-<layer_name>.npz`, where
189                `_idx` is the "depth" of the corresponding problem (1 for the
190                outermost, 2 for the one wrapped within it, etc.), and
191                `layer_name` is the name of the current `WrappedProblem`
192                instance (see `WrappedProblem.__init__`).
193            compressed (bool): Wether to compress the archive (defaults to
194                `True`).
195            _idx (int): Don't touch that.
196
197        See also:
198            `nmoo.utils.WrappedProblem.dump_history`
199        """
200        self.dump_history(
201            Path(dir_path) / f"{name}.{_idx}-{self._name}.npz", compressed
202        )
203        if isinstance(self._problem, WrappedProblem):
204            self._problem.dump_all_histories(
205                dir_path, name, compressed, _idx + 1
206            )
207
208    def dump_history(self, path: Union[Path, str], compressed: bool = True):
209        """
210        Dumps the history into an NPZ archive.
211
212        Args:
213            path (Union[Path, str]): File path of the output archive.
214            compressed (bool): Wether to compress the archive (defaults to
215                `True`).
216
217        See also:
218            `numpy.load
219            <https://numpy.org/doc/stable/reference/generated/numpy.load.html>`_
220
221        """
222        saver = np.savez_compressed if compressed else np.savez
223        saver(path, **self._history)
224
225    def ground_problem(self) -> Problem:
226        """
227        Recursively goes down the problem wrappers until an actual
228        `pymoo.Problem` is found, and returns it.
229        """
230        return self.innermost_wrapper()._problem
231
232    def innermost_wrapper(self) -> "WrappedProblem":
233        """
234        Recursively goes down the problem wrappers until a `WrappedProblem`
235        that wraps an actual `pymoo.Problem` is found, and returns it (the
236        wrapper).
237        """
238        if isinstance(self._problem, WrappedProblem):
239            return self._problem.innermost_wrapper()
240        return self
241
242    def reseed(self, seed: Any) -> None:
243        """
244        Recursively resets the internal random state of the problem. See the
245        [numpy
246        documentation](https://numpy.org/doc/stable/reference/random/generator.html?highlight=default_rng#numpy.random.default_rng)
247        for details about acceptable seeds.
248        """
249        if isinstance(self._problem, WrappedProblem):
250            self._problem.reseed(seed)
251
252    def start_new_run(self):
253        """
254        In short, it rotates the history of the current problem, and all
255        problems wrapped within.
256
257        Every entry in the history is annotated with a run number. If this
258        problem is reused (e.g. during benchmarks), you can call this method to
259        increase the run number for all subsequent history entries.
260        Additionally, the `_current_history_batch` counter is reset to `0`.
261
262        If the wrapped problem is itself a `WrappedProblem`, then this method
263        is recursively called.
264        """
265        self._current_run += 1
266        self._current_history_batch = 0
267        if isinstance(self._problem, WrappedProblem):
268            self._problem.start_new_run()
269
270    def _evaluate(self, x, out, *args, **kwargs):
271        """
272        Calls the wrapped problems's `_evaluate` method and appends its input
273        (`x`) and output (`out`) to the history.
274        """
275        self._problem._evaluate(x, out, *args, **kwargs)
276        self.add_to_history_x_out(x, out)