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