Refactor GeneralOptimizer if-else hierarchy (#232)

This commit removes the if-else hierarchy in GeneralOptimizer
by abstracting some parts of the topologies.
Resolves #232

Signed-off-by: Lester James V. Miranda <ljvmiranda@gmail.com>

pyswarms/backend/topology/random.py

@@ -18,18 +18,23 @@
 
 
 class Random(Topology):
-    def __init__(self, static=False):
+    def __init__(self, k, static=False):
         """Initializes the class
 
         Parameters
         ----------
+        k : int
+            number of neighbors to be considered. Must be a
+            positive integer less than :code:`n_particles-1`
         static : bool (Default is :code:`False`)
             a boolean that decides whether the topology
-            is static or dynamic"""
+            is static or dynamic
+        """
         super(Random, self).__init__(static)
+        self.k = k
         self.rep = Reporter(logger=logging.getLogger(__name__))
 
-    def compute_gbest(self, swarm, k):
+    def compute_gbest(self, swarm):
         """Update the global best using a random neighborhood approach
 
         This uses random class from :code:`numpy` to give every particle k
@@ -46,9 +51,6 @@ def compute_gbest(self, swarm, k):
         ----------
         swarm : pyswarms.backend.swarms.Swarm
             a Swarm instance
-        k : int
-            number of neighbors to be considered. Must be a
-            positive integer less than :code:`n_particles-1`
 
         Returns
         -------
@@ -60,7 +62,7 @@ def compute_gbest(self, swarm, k):
         try:
             # Check if the topology is static or dynamic and assign neighbors
             if (self.static and self.neighbor_idx is None) or not self.static:
-                adj_matrix = self.__compute_neighbors(swarm, k)
+                adj_matrix = self.__compute_neighbors(swarm, self.k)
                 self.neighbor_idx = np.array(
                     [
                         adj_matrix[i].nonzero()[0]

pyswarms/backend/topology/ring.py

@@ -20,18 +20,27 @@
 
 
 class Ring(Topology):
-    def __init__(self, static=False):
+    def __init__(self, p, k, static=False):
         """Initializes the class
 
         Parameters
         ----------
         static : bool (Default is :code:`False`)
             a boolean that decides whether the topology
-            is static or dynamic"""
+            is static or dynamic
+        p: int {1,2}
+            the Minkowski p-norm to use. 1 is the
+            sum-of-absolute values (or L1 distance) while 2 is
+            the Euclidean (or L2) distance.
+        k : int
+            number of neighbors to be considered. Must be a
+            positive integer less than :code:`n_particles`
+        """
         super(Ring, self).__init__(static)
+        self.p, self.k = p, k
         self.rep = Reporter(logger=logging.getLogger(__name__))
 
-    def compute_gbest(self, swarm, p, k):
+    def compute_gbest(self, swarm):
         """Update the global best using a ring-like neighborhood approach
 
         This uses the cKDTree method from :code:`scipy` to obtain the nearest
@@ -41,13 +50,6 @@ def compute_gbest(self, swarm, p, k):
         ----------
         swarm : pyswarms.backend.swarms.Swarm
             a Swarm instance
-        k : int
-            number of neighbors to be considered. Must be a
-            positive integer less than :code:`n_particles`
-        p: int {1,2}
-            the Minkowski p-norm to use. 1 is the
-            sum-of-absolute values (or L1 distance) while 2 is
-            the Euclidean (or L2) distance.
 
         Returns
         -------
@@ -61,12 +63,14 @@ def compute_gbest(self, swarm, p, k):
             if (self.static and self.neighbor_idx is None) or not self.static:
                 # Obtain the nearest-neighbors for each particle
                 tree = cKDTree(swarm.position)
-                _, self.neighbor_idx = tree.query(swarm.position, p=p, k=k)
+                _, self.neighbor_idx = tree.query(
+                    swarm.position, p=self.p, k=self.k
+                )
 
             # Map the computed costs to the neighbour indices and take the
             # argmin. If k-neighbors is equal to 1, then the swarm acts
             # independently of each other.
-            if k == 1:
+            if self.k == 1:
                 # The minimum index is itself, no mapping needed.
                 best_neighbor = swarm.pbest_cost[self.neighbor_idx][
                     :, np.newaxis

pyswarms/backend/topology/von_neumann.py

@@ -13,11 +13,12 @@
 
 
 class VonNeumann(Ring):
-    def __init__(self):
-        super(VonNeumann, self).__init__(static=True)
+    def __init__(self, p, r):
+        super(VonNeumann, self).__init__(static=True, p=p, k=None)
+        self.r = r
         self.rep = Reporter(logger=logging.getLogger(__name__))
 
-    def compute_gbest(self, swarm, p, r):
+    def compute_gbest(self, swarm):
         """Updates the global best using a neighborhood approach
 
         The Von Neumann topology inherits from the Ring topology and uses
@@ -43,8 +44,8 @@ def compute_gbest(self, swarm, p, r):
         float
             Best cost
         """
-        neighbors = VonNeumann.delannoy(swarm.dimensions, r)
-        return super(VonNeumann, self).compute_gbest(swarm, p, neighbors)
+        self.k = VonNeumann.delannoy(swarm.dimensions, self.r)
+        return super(VonNeumann, self).compute_gbest(swarm)
 
     @staticmethod
     def delannoy(d, r):

pyswarms/discrete/binary.py

@@ -145,7 +145,7 @@ def __init__(
         # Initialize the resettable attributes
         self.reset()
         # Initialize the topology
-        self.top = Ring(static=False)
+        self.top = Ring(static=False, p=self.p, k=self.k)
         self.name = __name__
 
     def optimize(self, objective_func, iters, fast=False, **kwargs):
@@ -192,7 +192,7 @@ def optimize(self, objective_func, iters, fast=False, **kwargs):
             best_cost_yet_found = np.min(self.swarm.best_cost)
             # Update gbest from neighborhood
             self.swarm.best_pos, self.swarm.best_cost = self.top.compute_gbest(
-                self.swarm, self.p, self.k
+                self.swarm
             )
             # Print to console
             self.rep.hook(best_cost=self.swarm.best_cost)

pyswarms/single/general_optimizer.py

@@ -167,52 +167,6 @@ def __init__(
             raise TypeError("Parameter `topology` must be a Topology object")
         else:
             self.top = topology
-
-        # Case for the Ring topology
-        if isinstance(topology, (Ring, VonNeumann)):
-            # Assign p-value as attributes
-            self.p = options["p"]
-            # Exceptions for the p value
-            if "p" not in self.options:
-                raise KeyError("Missing p in options")
-            if self.p not in [1, 2]:
-                raise ValueError(
-                    "p-value should either be 1 (for L1/Minkowski) "
-                    "or 2 (for L2/Euclidean)."
-                )
-
-        # Case for Random, VonNeumann and Ring topologies
-        if isinstance(topology, (Random, Ring, VonNeumann)):
-            if not isinstance(topology, VonNeumann):
-                self.k = options["k"]
-                if not isinstance(self.k, int):
-                    raise ValueError(
-                        "No. of neighbors must be an integer between"
-                        "0 and no. of particles."
-                    )
-                if not 0 <= self.k <= self.n_particles - 1:
-                    raise ValueError(
-                        "No. of neighbors must be between 0 and no. "
-                        "of particles."
-                    )
-                if "k" not in self.options:
-                    raise KeyError("Missing k in options")
-            else:
-                # Assign range r as attribute
-                self.r = options["r"]
-                if not isinstance(self.r, int):
-                    raise ValueError("The range must be a positive integer")
-                if (
-                    self.r <= 0
-                    or not 0
-                    <= VonNeumann.delannoy(self.swarm.dimensions, self.r)
-                    <= self.n_particles - 1
-                ):
-                    raise ValueError(
-                        "The range must be set such that the computed"
-                        "Delannoy number (number of neighbours) is"
-                        "between 0 and the no. of particles."
-                    )
         self.name = __name__
 
     def optimize(self, objective_func, iters, fast=False, **kwargs):
@@ -257,33 +211,11 @@ def optimize(self, objective_func, iters, fast=False, **kwargs):
                 self.swarm
             )
             best_cost_yet_found = self.swarm.best_cost
-            # If the topology is a ring topology just use the local minimum
-            if isinstance(self.top, Ring) and not isinstance(
-                self.top, VonNeumann
-            ):
-                # Update gbest from neighborhood
-                self.swarm.best_pos, self.swarm.best_cost = self.top.compute_gbest(
-                    self.swarm, self.p, self.k
-                )
-            # If the topology is a VonNeumann topology pass the neighbour and range attribute to compute_gbest()
-            if isinstance(self.top, VonNeumann):
-                # Update gbest from neighborhood
+            # Update swarm
+            if np.min(self.swarm.pbest_cost) < self.swarm.best_cost:
                 self.swarm.best_pos, self.swarm.best_cost = self.top.compute_gbest(
-                    self.swarm, self.p, self.r
+                    self.swarm
                 )
-            # If the topology is a random topology pass the neighbor attribute to compute_gbest()
-            elif isinstance(self.top, Random):
-                # Get minima of pbest and check if it's less than gbest
-                if np.min(self.swarm.pbest_cost) < self.swarm.best_cost:
-                    self.swarm.best_pos, self.swarm.best_cost = self.top.compute_gbest(
-                        self.swarm, self.k
-                    )
-            else:
-                # Get minima of pbest and check if it's less than gbest
-                if np.min(self.swarm.pbest_cost) < self.swarm.best_cost:
-                    self.swarm.best_pos, self.swarm.best_cost = self.top.compute_gbest(
-                        self.swarm
-                    )
             # Print to console
             self.rep.hook(best_cost=self.swarm.best_cost)
             hist = self.ToHistory(

pyswarms/single/local_best.py

@@ -176,7 +176,7 @@ def __init__(
         # Initialize the resettable attributes
         self.reset()
         # Initialize the topology
-        self.top = Ring(static=static)
+        self.top = Ring(static=static, p=self.p, k=self.k)
         self.name = __name__
 
     def optimize(self, objective_func, iters, fast=False, **kwargs):
@@ -223,7 +223,7 @@ def optimize(self, objective_func, iters, fast=False, **kwargs):
             best_cost_yet_found = np.min(self.swarm.best_cost)
             # Update gbest from neighborhood
             self.swarm.best_pos, self.swarm.best_cost = self.top.compute_gbest(
-                self.swarm, self.p, self.k
+                self.swarm
             )
             self.rep.hook(best_cost=np.min(self.swarm.best_cost))
             # Save to history

pyswarms/utils/decorators/decorators.py

@@ -39,9 +39,15 @@ def cost_func(x):
     cost_dec : callable
         The vectorized output for all particles as defined by :code:`cost_func`
     """
+
     def cost_dec(particles, **kwargs):
         n_particles = particles.shape[0]
-        vector = np.array([cost_func(particles[i], **kwargs) for i in range(n_particles)])
-        assert vector.shape == (n_particles, ), "The cost function should return a single value."
+        vector = np.array(
+            [cost_func(particles[i], **kwargs) for i in range(n_particles)]
+        )
+        assert vector.shape == (
+            n_particles,
+        ), "The cost function should return a single value."
         return vector
+
     return cost_dec

tests/backend/topology/test_random.py

@@ -13,8 +13,8 @@
 @pytest.mark.parametrize("k", [1, 2])
 def test_update_gbest_neighborhood(swarm, k, static):
     """Test if update_gbest_neighborhood gives the expected return values"""
-    topology = Random(static=static)
-    pos, cost = topology.compute_gbest(swarm, k=k)
+    topology = Random(static=static, k=k)
+    pos, cost = topology.compute_gbest(swarm)
     expected_pos = np.array([9.90438476e-01, 2.50379538e-03, 1.87405987e-05])
     expected_cost = 1.0002528364353296
     assert cost == pytest.approx(expected_cost)
@@ -25,7 +25,7 @@ def test_update_gbest_neighborhood(swarm, k, static):
 @pytest.mark.parametrize("clamp", [None, (0, 1), (-1, 1)])
 def test_compute_velocity_return_values(swarm, clamp, static):
     """Test if compute_velocity() gives the expected shape and range"""
-    topology = Random(static=static)
+    topology = Random(static=static, k=1)
     v = topology.compute_velocity(swarm, clamp)
     assert v.shape == swarm.position.shape
     if clamp is not None:
@@ -39,7 +39,7 @@ def test_compute_velocity_return_values(swarm, clamp, static):
 )
 def test_compute_position_return_values(swarm, bounds, static):
     """Test if compute_position() gives the expected shape and range"""
-    topology = Random(static=static)
+    topology = Random(static=static, k=1)
     p = topology.compute_position(swarm, bounds)
     assert p.shape == swarm.velocity.shape
     if bounds is not None:
@@ -50,8 +50,8 @@ def test_compute_position_return_values(swarm, bounds, static):
 @pytest.mark.parametrize("k", [1, 2])
 def test_compute_neighbors_return_values(swarm, k, static):
     """Test if __compute_neighbors() gives the expected shape and symmetry"""
-    topology = Random(static=static)
-    adj_matrix = topology._Random__compute_neighbors(swarm, k)
+    topology = Random(static=static, k=k)
+    adj_matrix = topology._Random__compute_neighbors(swarm, k=k)
     assert adj_matrix.shape == (swarm.n_particles, swarm.n_particles)
     assert np.allclose(adj_matrix, adj_matrix.T, atol=1e-8)  # Symmetry test
 
@@ -61,8 +61,8 @@ def test_compute_neighbors_return_values(swarm, k, static):
 def test_compute_neighbors_adjacency_matrix(swarm, k, static):
     """Test if __compute_neighbors() gives the expected matrix"""
     np.random.seed(1)
-    topology = Random(static=static)
-    adj_matrix = topology._Random__compute_neighbors(swarm, k)
+    topology = Random(static=static, k=k)
+    adj_matrix = topology._Random__compute_neighbors(swarm, k=k)
     # fmt: off
     comparison_matrix = np.array([[1, 1, 1, 0, 1, 0, 0, 0, 0, 1],
                                   [1, 1, 1, 1, 1, 1, 1, 1, 1, 1],
@@ -82,6 +82,6 @@ def test_compute_neighbors_adjacency_matrix(swarm, k, static):
 @pytest.mark.parametrize("k", [1])
 def test_neighbor_idx(swarm, k, static):
     """Test if the neighbor_idx attribute is assigned"""
-    topology = Random(static=static)
-    topology.compute_gbest(swarm, k)
+    topology = Random(static=static, k=k)
+    topology.compute_gbest(swarm)
     assert topology.neighbor_idx is not None

tests/backend/topology/test_ring.py

@@ -14,8 +14,8 @@
 @pytest.mark.parametrize("p", [1, 2])
 def test_update_gbest_neighborhood(swarm, p, k, static):
     """Test if update_gbest_neighborhood gives the expected return values"""
-    topology = Ring(static=static)
-    pos, cost = topology.compute_gbest(swarm, p=p, k=k)
+    topology = Ring(static=static, p=p, k=k)
+    pos, cost = topology.compute_gbest(swarm)
     expected_pos = np.array([9.90438476e-01, 2.50379538e-03, 1.87405987e-05])
     expected_cost = 1.0002528364353296
     assert cost == pytest.approx(expected_cost)
@@ -26,7 +26,7 @@ def test_update_gbest_neighborhood(swarm, p, k, static):
 @pytest.mark.parametrize("clamp", [None, (0, 1), (-1, 1)])
 def test_compute_velocity_return_values(swarm, clamp, static):
     """Test if compute_velocity() gives the expected shape and range"""
-    topology = Ring(static=static)
+    topology = Ring(static=static, p=1, k=2)
     v = topology.compute_velocity(swarm, clamp)
     assert v.shape == swarm.position.shape
     if clamp is not None:
@@ -40,7 +40,7 @@ def test_compute_velocity_return_values(swarm, clamp, static):
 )
 def test_compute_position_return_values(swarm, bounds, static):
     """Test if compute_position() gives the expected shape and range"""
-    topology = Ring(static=static)
+    topology = Ring(static=static, p=1, k=2)
     p = topology.compute_position(swarm, bounds)
     assert p.shape == swarm.velocity.shape
     if bounds is not None:
@@ -52,6 +52,6 @@ def test_compute_position_return_values(swarm, bounds, static):
 @pytest.mark.parametrize("p", [1, 2])
 def test_neighbor_idx(swarm, static, p, k):
     """Test if the neighbor_idx attribute is assigned"""
-    topology = Ring(static=static)
-    topology.compute_gbest(swarm, p=p, k=k)
+    topology = Ring(static=static, p=p, k=k)
+    topology.compute_gbest(swarm)
     assert topology.neighbor_idx is not None