Created class for pyramid topology

pyswarms/backend/topology/__init__.py

@@ -8,6 +8,7 @@
 
 from .star import Star
 from .ring import Ring
+from .pyramid import Pyramid
 
 
-__all__ = ["Star", "Ring"]
+__all__ = ["Star", "Ring", "Pyramid"]

pyswarms/backend/topology/pyramid.py

@@ -0,0 +1,131 @@
+# -*- coding: utf-8 -*-
+
+"""
+A Pyramid Network Topology
+
+This class implements a star topology where all particles are connected in a
+pyramid like fashion.
+"""
+
+# Import from stdlib
+import logging
+
+# Import modules
+import numpy as np
+from scipy.spatial import Delaunay
+
+# Import from package
+from .. import operators as ops
+from .base import Topology
+
+# Create a logger
+logger = logging.getLogger(__name__)
+
+class Pyramid(Topology):
+    def __init__(self):
+        super(Pyramid, self).__init__()
+
+    def compute_gbest(self, swarm):
+        """Updates the global best using a pyramid neighborhood approach
+
+        This uses the Delaunay method from :code:`scipy` to triangulate space
+        with simplices
+
+        Parameters
+        ----------
+        swarm : pyswarms.backend.swarms.Swarm
+            a Swarm instance
+
+        Returns
+        -------
+        numpy.ndarray
+            Best position of shape :code:`(n_dimensions, )`
+        float
+            Best cost
+        """
+        try:
+            # If there are less than 5 particles they are all connected
+            if swarm.n_particles < 5:
+                best_pos = swarm.pbest_pos[np.argmin(swarm.pbest_cost)]
+                best_cost = np.min(swarm.pbest_cost)
+            else:
+                pyramid = Delaunay(swarm.position)
+                indices, index_pointer = pyramid.vertex_neighbor_vertices
+                # Insert all the neighbors for each particle in the idx array
+                idx = np.array([index_pointer[indices[i]:indices[i+1]] for i in range(swarm.n_particles)])
+                idx_min = swarm.pbest_cost[idx].argmin(axis=1)
+                best_neighbor = idx[np.arange(len(idx)), idx_min]
+
+                # Obtain best cost and position
+                best_cost = np.min(swarm.pbest_cost[best_neighbor])
+                best_pos = swarm.pbest_pos[
+                    np.argmin(swarm.pbest_cost[best_neighbor])
+                ]
+        except AttributeError:
+            msg = "Please pass a Swarm class. You passed {}".format(
+                type(swarm)
+            )
+            logger.error(msg)
+            raise
+        else:
+            return (best_pos, best_cost)
+
+    def compute_velocity(self, swarm, clamp=None):
+        """Computes the velocity matrix
+
+        This method updates the velocity matrix using the best and current
+        positions of the swarm. The velocity matrix is computed using the
+        cognitive and social terms of the swarm.
+
+        A sample usage can be seen with the following:
+
+        .. code-block :: python
+
+            import pyswarms.backend as P
+            from pyswarms.swarms.backend import Swarm
+            from pyswarms.backend.topology import Pyramid
+
+            my_swarm = P.create_swarm(n_particles, dimensions)
+            my_topology = Pyramid()
+
+            for i in range(iters):
+                # Inside the for-loop
+                my_swarm.velocity = my_topology.update_velocity(my_swarm, clamp)
+
+        Parameters
+        ----------
+        swarm : pyswarms.backend.swarms.Swarm
+            a Swarm instance
+        clamp : tuple of floats (default is :code:`None`)
+            a tuple of size 2 where the first entry is the minimum velocity
+            and the second entry is the maximum velocity. It
+            sets the limits for velocity clamping.
+
+        Returns
+        -------
+        numpy.ndarray
+            Updated velocity matrix
+        """
+        return ops.compute_velocity(swarm, clamp)
+
+    def compute_position(self, swarm, bounds=None):
+        """Updates the position matrix
+
+        This method updates the position matrix given the current position and
+        the velocity. If bounded, it waives updating the position.
+
+        Parameters
+        ----------
+        swarm : pyswarms.backend.swarms.Swarm
+            a Swarm instance
+        bounds : tuple of :code:`np.ndarray` or list (default is :code:`None`)
+            a tuple of size 2 where the first entry is the minimum bound while
+            the second entry is the maximum bound. Each array must be of shape
+            :code:`(dimensions,)`.
+
+        Returns
+        -------
+        numpy.ndarray
+            New position-matrix
+        """
+        return ops.compute_position(swarm, bounds)

pyswarms/backend/topology/ring.py

@@ -95,7 +95,7 @@ def compute_velocity(self, swarm, clamp=None):
 
             import pyswarms.backend as P
             from pyswarms.swarms.backend import Swarm
-            from pyswarms.backend.topology import Star
+            from pyswarms.backend.topology import Ring
 
             my_swarm = P.create_swarm(n_particles, dimensions)
             my_topology = Ring()

tests/backend/topology/test_pyramid.py

@@ -0,0 +1,42 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+# Import modules
+import pytest
+import numpy as np
+
+# Import from package
+from pyswarms.backend.topology import Pyramid
+
+
+def test_compute_gbest_return_values(swarm):
+    """Test if compute_gbest() gives the expected return values"""
+    topology = Pyramid()
+    expected_cost = 1
+    expected_pos = np.array([1, 2, 3])
+    pos, cost = topology.compute_gbest(swarm)
+    assert cost == expected_cost
+    assert (pos == expected_pos).all()
+
+
+@pytest.mark.parametrize("clamp", [None, (0, 1), (-1, 1)])
+def test_compute_velocity_return_values(swarm, clamp):
+    """Test if compute_velocity() gives the expected shape and range"""
+    topology = Pyramid()
+    v = topology.compute_velocity(swarm, clamp)
+    assert v.shape == swarm.position.shape
+    if clamp is not None:
+        assert (clamp[0] <= v).all() and (clamp[1] >= v).all()
+
+
+@pytest.mark.parametrize(
+    "bounds",
+    [None, ([-5, -5, -5], [5, 5, 5]), ([-10, -10, -10], [10, 10, 10])],
+)
+def test_compute_position_return_values(swarm, bounds):
+    """Test if compute_position() gives the expected shape and range"""
+    topology = Pyramid()
+    p = topology.compute_position(swarm, bounds)
+    assert p.shape == swarm.velocity.shape
+    if bounds is not None:
+        assert (bounds[0] <= p).all() and (bounds[1] >= p).all()