Fixed FixedTotalNumberConnector (cf #384).

It still needs to be extended to support the `parallel_safe` RNG flag, and should support `allow_self_connections` and `with_replacement` flags.

pyNN/connectors.py

@@ -142,6 +142,24 @@ class MapConnector(Connector):
     """
 
     def _standard_connect(self, projection, connection_map_generator, distance_map=None):
+        """
+
+        `connection_map_generator` should be a function or other callable, with one optional
+        argument `mask`, which returns an iterable.
+
+        The iterable should produce one element per post-synaptic neuron.
+
+        Each element should be either:
+            (i) a boolean array, indicating which of the pre-synaptic neurons
+                should be connected to,
+            (ii) an integer array indicating the same thing using indices,
+            (iii) or a single boolean, meaning connect to all/none.
+
+        The `mask` argument, a boolean array, can be used to limit processing to just
+        neurons which exist on the local MPI node.
+
+        todo: explain the argument `distance_map`.
+        """
 
         column_indices = numpy.arange(projection.post.size)
 
@@ -973,18 +991,24 @@ def __init__(self, n, allow_self_connections=True, with_replacement=True,
         self.rng = _get_rng(rng)
 
     def connect(self, projection):
+        # This implementation is not "parallel safe" for random numbers.
+        # todo: support the `parallel_safe` flag.
+
         # Determine number of processes and current rank
-        rank, num_processes = get_mpi_config()
+        rank = projection._simulator.state.mpi_rank
+        num_processes = projection._simulator.state.num_processes
 
         # Assume that targets are equally distributed over processes
         targets_per_process = int(len(projection.post)/num_processes)
 
         # Calculate the number of synapses on each process
-        bino = RandomDistribution('binomial',[self.n,targets_per_process/len(projection.post)], rng=self.rng)
-        num_conns_on_vp = numpy.zeros(num_processes)
+        bino = RandomDistribution('binomial',
+                                  [self.n, targets_per_process/len(projection.post)],
+                                  rng=self.rng)
+        num_conns_on_vp = numpy.zeros(num_processes, dtype=int)
         sum_dist = 0
         sum_partitions = 0
-        for k in xrange(num_processes) :
+        for k in range(num_processes) :
             p_local = targets_per_process / ( len(projection.post) - sum_dist)
             bino.parameters['p'] = p_local
             bino.parameters['n'] = self.n - sum_partitions
@@ -993,19 +1017,18 @@ def connect(self, projection):
             sum_partitions += num_conns_on_vp[k]
 
         # Draw random sources and targets 
-        while num_conns_on_vp[rank] > 0 :
-            s_index = self.rng.rng.randint(low=0, high=len(projection.pre.all_cells))
-            t_index = self.rng.rng.randint(low=0, high=len(projection.post.local_cells))
-            t_index = numpy.where(projection.post.all_cells == int(projection.post.local_cells[t_index]))[0][0]
-
-            # Evaluate the lazy arrays containing the synaptic parameters
-            parameter_space = self._parameters_from_synapse_type(projection)
-            connection_parameters = {}
-            for name, map in parameter_space.items():
-                if map.is_homogeneous:
-                    connection_parameters[name] = map.evaluate(simplify=True)
-                else:
-                    connection_parameters[name] = map[source_mask, col]
-
-            projection._convergent_connect(numpy.array([s_index]),t_index, **connection_parameters)
-            num_conns_on_vp[rank] -=1
+        connections = [[] for i in range(projection.post.size)]
+        possible_targets = numpy.arange(projection.post.size)[projection.post._mask_local]
+        for i in range(num_conns_on_vp[rank]):
+            source_index = self.rng.next(1, 'uniform_int',
+                                         {"low": 0, "high": projection.pre.size},
+                                         mask_local=False)[0]
+            target_index = self.rng.choice(possible_targets, size=1)
+            connections[target_index].append(source_index)
+
+        def build_source_masks(mask=None):
+            if mask is None:
+                return [numpy.array(x) for x in connections]
+            else:
+                return [numpy.array(x) for x in numpy.array(connections)[mask]]
+        self._standard_connect(projection, build_source_masks)

test/unittests/test_connectors_parallel.py

@@ -987,3 +987,21 @@ def test_check_delay(self, sim=sim):
         self.assertEqual(connectors.check_delays(2*self.MIN_DELAY, self.MIN_DELAY, 1e99), 2*self.MIN_DELAY)
         self.assertRaises(errors.ConnectionError, connectors.check_delays, 0.5*self.MIN_DELAY, self.MIN_DELAY, 1e99)
         self.assertRaises(errors.ConnectionError, connectors.check_delays, 3.0, self.MIN_DELAY, 2.0)
+
+
+@register_class()
+class TestFixedTotalNumberConnector(unittest.TestCase):
+
+    def setUp(self, sim=sim):
+        sim.setup(num_processes=2, rank=1, min_delay=0.123)
+        self.p1 = sim.Population(4, sim.IF_cond_exp(), structure=space.Line())
+        self.p2 = sim.Population(5, sim.HH_cond_exp(), structure=space.Line())
+        assert_array_equal(self.p2._mask_local, numpy.array([0, 1, 0, 1, 0], dtype=bool))
+
+    def test_1(self):
+        C = connectors.FixedTotalNumberConnector(n=12, rng=random.NumpyRNG())
+        syn = sim.StaticSynapse(weight="0.5*d")
+        prj = sim.Projection(self.p1, self.p2, C, syn)
+        connections = prj.get(["weight", "delay"], format='list', gather=False)
+        self.assertLess(len(connections), 12)    # unlikely to be 12, since we have 2 MPI nodes
+        self.assertGreater(len(connections), 0)  # unlikely to be 0

test/unittests/test_connectors_serial.py

@@ -1117,6 +1117,24 @@ def test_check_delay(self, sim=sim):
         self.assertRaises(errors.ConnectionError, connectors.check_delays, 0.5*self.MIN_DELAY, self.MIN_DELAY, 1e99)
         self.assertRaises(errors.ConnectionError, connectors.check_delays, 3.0, self.MIN_DELAY, 2.0)
 
+
+@register_class()
+class TestFixedTotalNumberConnector(unittest.TestCase):
+
+    def setUp(self, sim=sim):
+        sim.setup(num_processes=1, rank=0, min_delay=0.123)
+        self.p1 = sim.Population(4, sim.IF_cond_exp(), structure=space.Line())
+        self.p2 = sim.Population(5, sim.HH_cond_exp(), structure=space.Line())
+        assert_array_equal(self.p2._mask_local, numpy.array([1, 1, 1, 1, 1], dtype=bool))
+
+    def test_1(self):
+        C = connectors.FixedTotalNumberConnector(n=12, rng=random.NumpyRNG())
+        syn = sim.StaticSynapse(weight="0.5*d")
+        prj = sim.Projection(self.p1, self.p2, C, syn)
+        connections = prj.get(["weight", "delay"], format='list', gather=False)
+        self.assertEqual(len(connections), 12)
+
+
 if __name__ == "__main__":
     unittest.main()