improved subprocess API with start points

in conjunction with forwarding nullary operator, it now allows versatile seeding of algorithms

Author: thomasWeise

moptipy/algorithms/so/rls.py

@@ -62,8 +62,6 @@
 from moptipy.api.algorithm import Algorithm1
 from moptipy.api.operators import Op0, Op1
 from moptipy.api.process import Process
-from moptipy.utils.logger import KeyValueLogSection
-from moptipy.utils.types import type_error
 
 
 # start book
@@ -103,61 +101,14 @@ def solve(self, process: Process) -> None:
             if new_f <= best_f:  # new_x is not worse than best_x?
                 best_f = new_f  # Store its objective value.
                 best_x, new_x = new_x, best_x  # Swap best and new.
-# end book
 
-    def __solve_seeded(self, process: Process) -> None:
-        """
-        Apply the RLS to an optimization problem starting from a seed.
+    # end book
 
-        :param process: the black-box process object
-        """
-        # Create records for old and new point in the search space.
-        best_x = process.create()  # record for best-so-far solution
-        new_x = process.create()  # record for new solution
-        # Obtain the random number generator.
-        random: Final[Generator] = process.get_random()
-
-        # Put function references in variables to save time.
-        evaluate: Final[Callable] = process.evaluate  # the objective
-        op1: Final[Callable] = self.op1.op1  # the unary operator
-        should_terminate: Final[Callable] = process.should_terminate
-
-        # Start at an existing point in the search space and get its quality.
-        process.get_copy_of_best_x(best_x)  # get the best-so-far solution
-        best_f: Union[int, float] = process.get_best_f()  # get the quality.
-
-        while not should_terminate():  # Until we need to quit...
-            op1(random, new_x, best_x)  # new_x = neighbor of best_x
-            new_f: Union[int, float] = evaluate(new_x)
-            if new_f <= best_f:  # new_x is not worse than best_x?
-                best_f = new_f  # Store its objective value.
-                best_x, new_x = new_x, best_x  # Swap best and new.
-
-    def __init__(self, op0: Op0, op1: Op1,
-                 seeded: bool = False) -> None:
+    def __init__(self, op0: Op0, op1: Op1) -> None:
         """
         Create the randomized local search (rls).
 
         :param op0: the nullary search operator
         :param op1: the unary search operator
-        :param seeded: `True` if the algorithm should be run in a seeded
-            fashion, i.e., expect an existing best solution. `False` if
-            it should run in the traditional way, starting at a random
-            solution
         """
         super().__init__("rls", op0, op1)
-        if not isinstance(seeded, bool):
-            raise type_error(seeded, "seeded", bool)
-        if seeded:
-            self.solve = self.__solve_seeded  # type: ignore
-        #: was this algorithm started in its seeded fashion?
-        self.__seeded: Final[bool] = seeded
-
-    def log_parameters_to(self, logger: KeyValueLogSection):
-        """
-        Log the parameters of the algorithm to a logger.
-
-        :param logger: the logger for the parameters
-        """
-        super().log_parameters_to(logger)
-        logger.key_value("seeded", self.__seeded)

moptipy/api/subprocesses.py

@@ -266,6 +266,8 @@ def __init__(self, owner: Process, in_and_out_x: Any,
         #: the fast call to the owner's register method
         self.__register: Final[Callable[[Any, Union[int, float]], None]] \
             = owner.register
+        #: True as long as only the seed has been used
+        self.__only_seed_used: bool = True
 
     def has_best(self) -> bool:
         return True
@@ -277,6 +279,10 @@ def get_best_f(self) -> Union[int, float]:
         return self.__best_f
 
     def evaluate(self, x) -> Union[float, int]:
+        if self.__only_seed_used:
+            if self.is_equal(x, self.__best_x):
+                return self.__best_f
+            self.__only_seed_used = False
         self.__fes = fe = self.__fes + 1
         f: Final[Union[int, float]] = self.__evaluate(x)
         if f < self.__best_f:
@@ -286,8 +292,12 @@ def evaluate(self, x) -> Union[float, int]:
         return f
 
     def register(self, x, f: Union[int, float]) -> None:
-        self.__register(x, f)
+        if self.__only_seed_used:
+            if self.is_equal(x, self.__best_x):
+                return
+            self.__only_seed_used = False
         self.__fes = fe = self.__fes + 1
+        self.__register(x, f)
         if f < self.__best_f:
             self.__best_f = f
             self.copy(self.__best_x, x)
@@ -311,6 +321,18 @@ def from_starting_point(owner: Process, in_and_out_x: Any,
     """
     Create a sub-process searching from one starting point.
 
+    This process is especially useful in conjunction with class
+    :class:`~moptipy.operators.op0_forward.Op0Forward`. This class
+    allows forwarding the nullary search operator to the function
+    :meth:`~moptipy.api.process.Process.get_copy_of_best_x`. This way, the
+    first point that it sampled by a local search can be the point specified
+    as `in_and_out_x`, which effectively seeds the local search.
+
+    To dovetail with chance of seeding, no FEs are counted at the beginning of
+    the process as long as all points to be evaluated equal to the
+    `in_and_out_x`. As soon as the first point different from `in_and_out_x`
+    is evaluated, FE counting starts.
+
     :param owner: the owning process
     :param in_and_out_x: the input solution record, which will be
         overwritten with the best encountered solution

moptipy/operators/op0_forward.py

@@ -0,0 +1,44 @@
+"""A nullary operator forwarding to another function."""
+from typing import Callable, Any, Optional
+
+import numpy as np
+from numpy.random import Generator
+
+from moptipy.api.operators import Op0
+from moptipy.utils.types import type_error
+
+
+class Op0Forward(Op0):
+    """Get the forwarder function."""
+
+    def __init__(self):
+        """Initialize this operator."""
+        #: the internal blueprint for filling permutations
+        self.__call: Optional[Callable[[Any], None]] = None
+
+    def op0(self, random: Generator, dest: np.ndarray) -> None:
+        """Forward the call."""
+        self.__call(dest)
+
+    def forward_to(self, call: Callable[[Any], None]) -> None:
+        """
+        Set the `Callable` to forward all calls from :meth:`op0` to.
+
+        :param call: the `Callable` to which all calls to :meth:`op0` should
+            be delegated to.
+        """
+        if not callable(call):
+            raise type_error(call, "call", call=True)
+        self.__call = call
+
+    def stop_forwarding(self) -> None:
+        """Stop forwarding the call."""
+        self.__call = None
+
+    def __str__(self) -> str:
+        """
+        Get the name of this operator.
+
+        :return: "forward"
+        """
+        return "forward"

tests/api/test_subprocess.py

@@ -29,6 +29,7 @@
 from moptipy.operators.bitstrings.op1_flip1 import Op1Flip1
 from moptipy.operators.bitstrings.op1_m_over_n_flip import Op1MoverNflip
 from moptipy.operators.bitstrings.op2_uniform import Op2Uniform
+from moptipy.operators.op0_forward import Op0Forward
 from moptipy.spaces.bitstrings import BitStrings
 from moptipy.utils.types import type_name_of
 
@@ -39,7 +40,8 @@ class MyAlgorithm(Algorithm2):
     def __init__(self) -> None:
         super().__init__("dummy", Op0Random(), Op1Flip1(), Op2Uniform())
         self.ea = EA(self.op0, self.op1, self.op2, 10, 10, 0.3)
-        self.rls = RLS(self.op0, self.op1, seeded=True)
+        self.fwd = Op0Forward()
+        self.rls = RLS(self.fwd, self.op1)
 
     def solve(self, process: Process) -> None:
         """Apply an EA for 100 FEs, followed by RLS."""
@@ -73,21 +75,25 @@ def solve(self, process: Process) -> None:
 
         fnew2: Union[int, float]
         fes2: int
-        with for_fes(from_starting_point(process, x1, fnew), 100) as z:
-            assert str(z) == f"forFEs_100_fromStart_{process}"
-            assert z.has_best()
-            assert z.get_best_f() == fnew
-            assert z.get_consumed_fes() == 0
-            self.rls.solve(z)
-            fnew2 = z.get_best_f()
-            fes2 = z.get_consumed_fes()
-            assert fes2 > 0
-            assert (fnew2 == 0) or (fes2 == 100)
-            assert (fnew2 >= 0) and (fes2 <= 100)
-            assert fnew2 <= fnew
+        with from_starting_point(process, x1, fnew) as z1:
+            assert str(z1) == f"fromStart_{process}"
+            with for_fes(z1, 100) as z:
+                self.fwd.forward_to(z.get_copy_of_best_x)
+                assert str(z) == f"forFEs_100_fromStart_{process}"
+                assert z.has_best()
+                assert z.get_best_f() == fnew
+                assert z.get_consumed_fes() == 0
+                self.rls.solve(z)
+                self.fwd.stop_forwarding()
+                fnew2 = z.get_best_f()
+                fes2 = z.get_consumed_fes()
+                assert fes2 > 0
+                assert (fnew2 == 0) or (fes2 == 100)
+                assert (fnew2 >= 0) and (fes2 <= 100)
+                assert fnew2 <= fnew
 
         allfes = process.get_consumed_fes()
-        assert allfes == 1 + fes + fes2
+        assert allfes == fes + fes2
         assert process.get_best_f() == fnew2
         if fnew2 > 0:
             assert process.evaluate(x1) == fnew2
@@ -105,8 +111,8 @@ def test_from_start_for_fes():
     with exp.execute() as p:
         assert p.has_best()
         assert p.get_best_f() >= 0
-        assert (p.get_best_f() == 0) or (p.get_consumed_fes() == 202)
-        assert (p.get_best_f() >= 0) and (p.get_consumed_fes() <= 202)
+        assert (p.get_best_f() == 0) or (p.get_consumed_fes() == 200)
+        assert (p.get_best_f() >= 0) and (p.get_consumed_fes() <= 200)
 
 
 class MyAlgorithm2(Algorithm2):
@@ -161,6 +167,7 @@ def test_without_should_terminate():
 
 class _OneMaxRegAlgo(Algorithm0):
     """The one-max algorithm."""
+
     def __init__(self, op0: Op0Random, f: OneMax):
         """Initialize."""
         super().__init__("om", op0)
@@ -231,11 +238,11 @@ def test_for_fes_process_no_ss_no_log_reg_norm():
         algorithm2: Algorithm = _OneMaxRegAlgo(Op0Random(), objective)
         algorithm: __RegisterForFEs = __RegisterForFEs(algorithm1, algorithm2)
 
-        with Execution()\
-                .set_solution_space(space)\
-                .set_objective(objective)\
-                .set_algorithm(algorithm)\
-                .set_max_fes(100)\
+        with Execution() \
+                .set_solution_space(space) \
+                .set_objective(objective) \
+                .set_algorithm(algorithm) \
+                .set_max_fes(100) \
                 .execute() as process:
             assert type_name_of(process) \
                    == "moptipy.api._process_no_ss._ProcessNoSS"
@@ -256,6 +263,7 @@ def test_for_fes_process_no_ss_no_log_reg_norm():
 
 class _MOAlgoForFEs(MOAlgorithm, Algorithm0):
     """The algorithm for multi-objective optimization."""
+
     def __init__(self, op0: Op0Random):
         """Initialize."""
         Algorithm0.__init__(self, "om", op0)
@@ -299,14 +307,14 @@ def test_for_fes_mo_process_no_ss_no_log():
     algorithm: Algorithm = _MOAlgoForFEs(Op0Random())
     ams = int(random.integers(2, 5))
 
-    with MOExecution()\
-            .set_solution_space(space)\
-            .set_objective(problem)\
-            .set_archive_pruner(pruner)\
-            .set_archive_max_size(ams)\
-            .set_archive_pruning_limit(int(ams + random.integers(0, 3)))\
-            .set_algorithm(algorithm)\
-            .set_max_fes(100)\
+    with MOExecution() \
+            .set_solution_space(space) \
+            .set_objective(problem) \
+            .set_archive_pruner(pruner) \
+            .set_archive_max_size(ams) \
+            .set_archive_pruning_limit(int(ams + random.integers(0, 3))) \
+            .set_algorithm(algorithm) \
+            .set_max_fes(100) \
             .execute() as process:
         assert isinstance(process, MOProcess)
         assert type_name_of(process) \
@@ -332,6 +340,7 @@ def test_for_fes_mo_process_no_ss_no_log():
 
 class _MOWithoutShouldTerminate(MOAlgorithm, Algorithm0):
     """The algorithm for multi-objective optimization."""
+
     def __init__(self, op0: Op0Random):
         """Initialize."""
         Algorithm0.__init__(self, "om", op0)
@@ -370,12 +379,12 @@ def test_without_should_terminate_mo_process_no_ss_no_log():
     algorithm: Algorithm = _MOWithoutShouldTerminate(Op0Random())
     ams = int(random.integers(2, 5))
 
-    with MOExecution()\
-            .set_solution_space(space)\
-            .set_objective(problem)\
-            .set_archive_pruning_limit(int(ams + random.integers(0, 3)))\
-            .set_algorithm(algorithm)\
-            .set_max_fes(100)\
+    with MOExecution() \
+            .set_solution_space(space) \
+            .set_objective(problem) \
+            .set_archive_pruning_limit(int(ams + random.integers(0, 3))) \
+            .set_algorithm(algorithm) \
+            .set_max_fes(100) \
             .execute() as process:
         assert isinstance(process, MOProcess)
         assert type_name_of(process) \