1# -*- coding: utf-8 -*-
2#
3# test_connect_arrays.py
4#
5# This file is part of NEST.
6#
7# Copyright (C) 2004 The NEST Initiative
8#
9# NEST is free software: you can redistribute it and/or modify
10# it under the terms of the GNU General Public License as published by
11# the Free Software Foundation, either version 2 of the License, or
12# (at your option) any later version.
13#
14# NEST is distributed in the hope that it will be useful,
15# but WITHOUT ANY WARRANTY; without even the implied warranty of
16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
17# GNU General Public License for more details.
18#
19# You should have received a copy of the GNU General Public License
20# along with NEST.  If not, see <http://www.gnu.org/licenses/>.
21
22import unittest
23import numpy as np
24
25import nest
26
27nest.set_verbosity('M_WARNING')
28
29HAVE_OPENMP = nest.ll_api.sli_func("is_threaded")
30
31
32class TestConnectArrays(unittest.TestCase):
33
34    non_unique = np.array([1, 1, 3, 5, 4, 5, 9, 7, 2, 8], dtype=np.uint64)
35
36    def setUp(self):
37        nest.ResetKernel()
38
39    def test_connect_arrays_unique(self):
40        """Connecting NumPy arrays of unique node IDs"""
41        n = 10
42        nest.Create('iaf_psc_alpha', n)
43        sources = np.arange(1, n+1, dtype=np.uint64)
44        targets = np.arange(1, n+1, dtype=np.uint64)
45        weights = 1.5
46        delays = 1.4
47
48        nest.Connect(sources, targets, syn_spec={'weight': weights, 'delay': delays})
49
50        conns = nest.GetConnections()
51
52        self.assertEqual(len(conns), n*n)
53
54        for c in conns:
55            np.testing.assert_approx_equal(c.weight, weights)
56            np.testing.assert_approx_equal(c.delay, delays)
57
58    def test_connect_arrays_nonunique(self):
59        """Connecting NumPy arrays with non-unique node IDs"""
60        n = 10
61        nest.Create('iaf_psc_alpha', n)
62        sources = np.arange(1, n+1, dtype=np.uint64)
63        targets = self.non_unique
64        weights = np.ones(n)
65        delays = np.ones(n)
66
67        nest.Connect(sources, targets, syn_spec={'weight': weights, 'delay': delays},
68                     conn_spec='one_to_one')
69
70        conns = nest.GetConnections()
71
72        for s, t, w, d, c in zip(sources, targets, weights, delays, conns):
73            self.assertEqual(c.source, s)
74            self.assertEqual(c.target, t)
75            self.assertEqual(c.weight, w)
76            self.assertEqual(c.delay, d)
77
78    def test_connect_arrays_nonunique_dict_conn_spec(self):
79        """Connecting NumPy arrays with non-unique node IDs and conn_spec as a dict"""
80        n = 10
81        nest.Create('iaf_psc_alpha', n)
82        sources = np.arange(1, n+1, dtype=np.uint64)
83        targets = self.non_unique
84        weights = 2 * np.ones(n)
85        delays = 1.5 * np.ones(n)
86
87        nest.Connect(sources, targets, syn_spec={'weight': weights, 'delay': delays},
88                     conn_spec={'rule': 'one_to_one'})
89
90        conns = nest.GetConnections()
91
92        for s, t, w, d, c in zip(sources, targets, weights, delays, conns):
93            self.assertEqual(c.source, s)
94            self.assertEqual(c.target, t)
95            self.assertEqual(c.weight, w)
96            self.assertEqual(c.delay, d)
97
98    def test_connect_arrays_no_conn_spec(self):
99        """Connecting NumPy arrays of node IDs without specifying conn_spec"""
100        n = 10
101        nest.Create('iaf_psc_alpha', n)
102        sources = np.arange(1, n+1, dtype=np.uint64)
103        targets = self.non_unique
104
105        with self.assertRaises(ValueError):
106            nest.Connect(sources, targets)
107
108    def test_connect_arrays_different_weights_delays(self):
109        """Connecting NumPy arrays with different weights and delays"""
110        n = 10
111        nest.Create('iaf_psc_alpha', n)
112        sources = np.arange(1, n+1, dtype=np.uint64)
113        targets = self.non_unique
114        weights = np.linspace(0.6, 1.5, n)
115        delays = np.linspace(0.4, 1.3, n)
116
117        nest.Connect(sources, targets, syn_spec={'weight': weights, 'delay': delays},
118                     conn_spec={'rule': 'one_to_one'})
119
120        conns = nest.GetConnections()
121
122        np.testing.assert_array_equal(conns.source, sources)
123        np.testing.assert_array_equal(conns.target, targets)
124        np.testing.assert_array_almost_equal(conns.weight, weights)
125        np.testing.assert_array_almost_equal(conns.delay, delays)
126
127    def test_connect_arrays_threaded(self):
128        """Connecting NumPy arrays, threaded"""
129        nest.local_num_threads = 2
130        n = 10
131        nest.Create('iaf_psc_alpha', n)
132        sources = np.arange(1, n+1, dtype=np.uint64)
133        targets = self.non_unique
134        weights = np.ones(len(sources))
135        delays = np.ones(len(sources))
136        syn_model = 'static_synapse'
137
138        nest.Connect(sources, targets, conn_spec='one_to_one',
139                     syn_spec={'weight': weights, 'delay': delays, 'synapse_model': syn_model})
140
141        conns = nest.GetConnections()
142
143        # Sorting connection information by source to make it equivalent to the reference.
144        conn_info = [(c.source, c.target, c.weight, c.delay) for c in conns]
145        conn_info.sort(key=lambda conn: conn[0])
146
147        for s, t, w, d, c in zip(sources, targets, weights, delays, conn_info):
148            conn_s, conn_t, conn_w, conn_d = c
149            self.assertEqual(conn_s, s)
150            self.assertEqual(conn_t, t)
151            self.assertEqual(conn_w, w)
152            self.assertEqual(conn_d, d)
153
154    def test_connect_arrays_no_delays(self):
155        """Connecting NumPy arrays without specifying delays"""
156        n = 10
157        nest.Create('iaf_psc_alpha', n)
158        sources = np.arange(1, n+1, dtype=np.uint64)
159        targets = self.non_unique
160        weights = np.ones(n)
161
162        nest.Connect(sources, targets, conn_spec='one_to_one', syn_spec={'weight': weights})
163
164        conns = nest.GetConnections()
165
166        for s, t, w, c in zip(sources, targets, weights, conns):
167            self.assertEqual(c.source, s)
168            self.assertEqual(c.target, t)
169            self.assertEqual(c.weight, w)
170
171    def test_connect_array_list(self):
172        """Connecting NumPy array and list"""
173        n = 10
174        nest.Create('iaf_psc_alpha', n)
175        sources = list(range(1, n + 1))
176        targets = self.non_unique
177
178        nest.Connect(sources, targets, conn_spec='one_to_one')
179
180        conns = nest.GetConnections()
181
182        for s, t, c in zip(sources, targets, conns):
183            self.assertEqual(c.source, s)
184            self.assertEqual(c.target, t)
185
186    def test_connect_arrays_no_weights(self):
187        """Connecting NumPy arrays without specifying weights"""
188        n = 10
189        neurons = nest.Create('iaf_psc_alpha', n)
190        targets = self.non_unique
191        delays = np.ones(n)
192
193        nest.Connect(neurons, targets, conn_spec='one_to_one', syn_spec={'delay': delays})
194
195        conns = nest.GetConnections()
196
197        for s, t, d, c in zip(neurons.tolist(), targets, delays, conns):
198            self.assertEqual(c.source, s)
199            self.assertEqual(c.target, t)
200            self.assertEqual(c.delay, d)
201
202    def test_connect_arrays_rtype(self):
203        """Connecting NumPy arrays with specified receptor_type"""
204        n = 10
205        nest.Create('iaf_psc_exp_multisynapse', n)
206        sources = np.arange(1, n+1, dtype=np.uint64)
207        targets = self.non_unique
208        weights = np.ones(len(sources))
209        delays = np.ones(len(sources))
210        receptor_type = np.ones(len(sources), dtype=np.uint64)
211        syn_model = 'static_synapse'
212
213        nest.Connect(sources, targets, conn_spec='one_to_one',
214                     syn_spec={'weight': weights, 'delay': delays, 'receptor_type': receptor_type})
215
216        conns = nest.GetConnections()
217
218        for s, t, w, d, r, c in zip(sources, targets, weights, delays, receptor_type, conns):
219            self.assertEqual(c.source, s)
220            self.assertEqual(c.target, t)
221            self.assertEqual(c.weight, w)
222            self.assertEqual(c.delay, d)
223            self.assertEqual(c.receptor, r)
224
225    def test_connect_arrays_additional_synspec_params(self):
226        """Connecting NumPy arrays with additional syn_spec params"""
227        n = 10
228        nest.Create('iaf_psc_exp_multisynapse', n)
229        sources = np.arange(1, n+1, dtype=np.uint64)
230        targets = self.non_unique
231        weights = np.ones(len(sources))
232        delays = np.ones(len(sources))
233        syn_model = 'vogels_sprekeler_synapse'
234        receptor_type = np.ones(len(sources), dtype=np.uint64)
235        alpha = 0.1*np.ones(len(sources))
236        tau = 20.*np.ones(len(sources))
237
238        nest.Connect(sources, targets, conn_spec='one_to_one',
239                     syn_spec={'weight': weights, 'delay': delays, 'synapse_model': syn_model,
240                               'receptor_type': receptor_type, 'alpha': alpha, 'tau': tau})
241
242        conns = nest.GetConnections()
243
244        for s, t, w, d, r, a, tau, c in zip(sources, targets, weights, delays, receptor_type, alpha, tau, conns):
245            self.assertEqual(c.source, s)
246            self.assertEqual(c.target, t)
247            self.assertEqual(c.weight, w)
248            self.assertEqual(c.delay, d)
249            self.assertEqual(c.receptor, r)
250            self.assertEqual(c.alpha, a)
251            self.assertEqual(c.tau, tau)
252
253    def test_connect_arrays_float_rtype(self):
254        """Raises exception when not using integer value for receptor_type"""
255        n = 10
256        nest.Create('iaf_psc_exp_multisynapse', n)
257        sources = np.arange(1, n+1, dtype=np.uint64)
258        targets = self.non_unique
259        weights = np.ones(n)
260        delays = np.ones(n)
261        syn_model = 'vogels_sprekeler_synapse'
262        receptor_type = 1.5*np.ones(len(sources))
263
264        with self.assertRaises(nest.kernel.NESTErrors.BadParameter):
265            nest.Connect(sources, targets, conn_spec='one_to_one',
266                         syn_spec={'weight': weights, 'delay': delays, 'synapse_model': syn_model,
267                                   'receptor_type': receptor_type})
268
269    def test_connect_arrays_wrong_dtype(self):
270        """Raises exception when connecting NumPy arrays with wrong dtype"""
271        n = 10
272        nest.Create('iaf_psc_alpha', n)
273        sources = np.arange(1, n+1, dtype=np.double)
274        targets = np.array(self.non_unique, dtype=np.double)
275        weights = np.ones(n)
276        delays = np.ones(n)
277        syn_model = 'static_synapse'
278
279        with self.assertRaises(nest.kernel.NESTErrors.ArgumentType):
280            nest.Connect(sources, targets, syn_spec={'weight': weights, 'delay': delays},
281                         conn_spec='one_to_one')
282
283    def test_connect_arrays_unknown_nodes(self):
284        """Raises exception when connecting NumPy arrays with unknown nodes"""
285        n = 10
286        nest.Create('iaf_psc_alpha', n)
287        sources = np.arange(1, n+2, dtype=np.uint64)
288        targets = np.arange(1, n+2, dtype=np.uint64)
289        weights = np.ones(len(sources))
290        delays = np.ones(len(sources))
291        syn_model = 'static_synapse'
292
293        with self.assertRaises(nest.kernel.NESTErrors.UnknownNode):
294            nest.Connect(sources, targets, syn_spec={'weight': weights, 'delay': delays,
295                                                     'synapse_model': syn_model})
296
297    @unittest.skipIf(not HAVE_OPENMP, 'NEST was compiled without multi-threading')
298    def test_connect_arrays_receptor_type(self):
299        """Connecting NumPy arrays with receptor type specified, threaded"""
300
301        nest.local_num_threads = 2
302
303        n = 10
304        nest.Create('iaf_psc_alpha', n)
305        sources = np.arange(1, n+1, dtype=np.uint64)
306        targets = self.non_unique
307
308        weights = len(sources) * [2.]
309        nest.Connect(sources, targets, conn_spec='one_to_one', syn_spec={'weight': weights, 'receptor_type': 0})
310
311        self.assertEqual(len(sources) * [0], nest.GetConnections().receptor)
312
313    @unittest.skipIf(not HAVE_OPENMP, 'NEST was compiled without multi-threading')
314    def test_connect_arrays_differnt_alpha(self):
315        """Connecting NumPy arrays with different alpha values in a threaded environment"""
316
317        nest.local_num_threads = 4
318
319        neurons = nest.Create("iaf_psc_exp", 10)
320        # syn_spec parameters are dependent on source, so we test with source id's not starting with 1
321        source = np.array([2, 5, 3, 10, 1, 9, 4, 6, 8, 7])
322        target = 1 + np.random.choice(10, 10, replace=True)
323
324        weights = len(source) * [2.]
325        alpha = np.array([0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 0.11])
326
327        # Need to make sure the correct alpha value is used with the correct source
328        src_alpha_ref = {key: val for key, val in zip(source, alpha)}
329
330        nest.Connect(source, target, conn_spec='one_to_one',
331                     syn_spec={'alpha': alpha, 'receptor_type': 0,
332                               'weight': weights, 'synapse_model': "stdp_synapse"})
333
334        conns = nest.GetConnections()
335        src = conns.source
336        alp = conns.alpha
337        src_alpha = {key: val for key, val in zip(src, alp)}
338
339        self.assertEqual(src_alpha_ref, src_alpha)
340
341
342def suite():
343    suite = unittest.TestLoader().loadTestsFromTestCase(TestConnectArrays)
344    return suite
345
346
347if __name__ == '__main__':
348    runner = unittest.TextTestRunner(verbosity=2)
349    runner.run(suite())
350