import os import platform import tempfile import pytest from numpy.testing import assert_allclose, assert_equal from brian2 import * from brian2.devices import device_module from brian2.devices.device import reinit_and_delete, reset_device, set_device from brian2.tests.utils import assert_allclose from brian2.utils.logger import catch_logs @pytest.mark.cpp_standalone @pytest.mark.standalone_only def test_cpp_standalone(): set_device("cpp_standalone", build_on_run=False) ##### Define the model tau = 1 * ms eqs = """ dV/dt = (-40*mV-V)/tau : volt (unless refractory) """ threshold = "V>-50*mV" reset = "V=-60*mV" refractory = 5 * ms N = 1000 G = NeuronGroup( N, eqs, reset=reset, threshold=threshold, refractory=refractory, name="gp" ) G.V = "-i*mV" M = SpikeMonitor(G) S = Synapses(G, G, "w : volt", on_pre="V += w") S.connect("abs(i-j)<5 and i!=j") S.w = 0.5 * mV S.delay = "0*ms" net = Network(G, M, S) net.run(100 * ms) device.build(directory=None, with_output=False) # we do an approximate equality here because depending on minor details of how it was compiled, the results # may be slightly different (if -ffast-math is on) assert len(M.i) >= 17000 and len(M.i) <= 18000 assert len(M.t) == len(M.i) assert M.t[0] == 0.0 reset_device() @pytest.mark.cpp_standalone @pytest.mark.standalone_only def test_multiple_connects(): set_device("cpp_standalone", build_on_run=False) G = NeuronGroup(10, "v:1") S = Synapses(G, G, "w:1") S.connect(i=[0], j=[0]) S.connect(i=[1], j=[1]) run(0 * ms) device.build(directory=None, with_output=False) assert len(S) == 2 and len(S.w[:]) == 2 reset_device() @pytest.mark.cpp_standalone @pytest.mark.standalone_only def test_device_cache_synapses(): # Check that we can ask for known synaptic information at runtime set_device("cpp_standalone", build_on_run=False) G = NeuronGroup(10, "v:1") S = Synapses(G, G, "w:1") S.connect(i=[0], j=[0]) assert len(S) == 1 assert_equal(S.i[:], [0]) assert_equal(S.j[:], [0]) S.connect(i=[1], j=[1]) assert len(S) == 2 assert_equal(S.i[:], [0, 1]) assert_equal(S.j[:], [0, 1]) S.connect(p=0.1) # We can't know anything about synapses anymore with pytest.raises(NotImplementedError): len(S) with pytest.raises(NotImplementedError): S.i[:] with pytest.raises(NotImplementedError): S.j[:] S.connect(i=[1], j=[1]) # Synapses are still "unknown" due to the previous p=0.1 call with pytest.raises(NotImplementedError): len(S) with pytest.raises(NotImplementedError): S.i[:] with pytest.raises(NotImplementedError): S.j[:] @pytest.mark.cpp_standalone @pytest.mark.standalone_only def test_storing_loading(): set_device("cpp_standalone", build_on_run=False) G = NeuronGroup( 10, """ v : volt x : 1 n : integer b : boolean """, ) v = np.arange(10) * volt x = np.arange(10, 20) n = np.arange(20, 30) b = np.array([True, False]).repeat(5) G.v = v G.x = x G.n = n G.b = b S = Synapses( G, G, """ v_syn : volt x_syn : 1 n_syn : integer b_syn : boolean """, ) S.connect(j="i") S.v_syn = v S.x_syn = x S.n_syn = n S.b_syn = b run(0 * ms) device.build(directory=None, with_output=False) assert_allclose(G.v[:], v) assert_allclose(S.v_syn[:], v) assert_allclose(G.x[:], x) assert_allclose(S.x_syn[:], x) assert_allclose(G.n[:], n) assert_allclose(S.n_syn[:], n) assert_allclose(G.b[:], b) assert_allclose(S.b_syn[:], b) reset_device() @pytest.mark.cpp_standalone @pytest.mark.standalone_only @pytest.mark.openmp def test_openmp_consistency(): previous_device = get_device() n_cells = 100 n_recorded = 10 numpy.random.seed(42) taum = 20 * ms taus = 5 * ms Vt = -50 * mV Vr = -60 * mV El = -49 * mV fac = 60 * 0.27 / 10 gmax = 20 * fac dApre = 0.01 taupre = 20 * ms taupost = taupre dApost = -dApre * taupre / taupost * 1.05 dApost *= 0.1 * gmax dApre *= 0.1 * gmax connectivity = numpy.random.randn(n_cells, n_cells) sources = numpy.random.randint(0, n_cells - 1, 10 * n_cells) # Only use one spike per time step (to rule out that a single source neuron # has more than one spike in a time step) times = ( numpy.random.choice(numpy.arange(10 * n_cells), 10 * n_cells, replace=False) * ms ) v_init = Vr + numpy.random.rand(n_cells) * (Vt - Vr) eqs = Equations( """ dv/dt = (g-(v-El))/taum : volt dg/dt = -g/taus : volt """ ) results = {} for n_threads, devicename in [ (0, "runtime"), (0, "cpp_standalone"), (1, "cpp_standalone"), (2, "cpp_standalone"), (3, "cpp_standalone"), (4, "cpp_standalone"), ]: set_device(devicename, build_on_run=False, with_output=False) Synapses.__instances__().clear() if devicename == "cpp_standalone": reinit_and_delete() prefs.devices.cpp_standalone.openmp_threads = n_threads P = NeuronGroup( n_cells, model=eqs, threshold="v>Vt", reset="v=Vr", refractory=5 * ms ) Q = SpikeGeneratorGroup(n_cells, sources, times) P.v = v_init P.g = 0 * mV S = Synapses( P, P, model=""" dApre/dt=-Apre/taupre : 1 (event-driven) dApost/dt=-Apost/taupost : 1 (event-driven) w : 1 """, pre=""" g += w*mV Apre += dApre w = w + Apost """, post=""" Apost += dApost w = w + Apre """, ) S.connect() S.w = fac * connectivity.flatten() T = Synapses(Q, P, model="w : 1", on_pre="g += w*mV") T.connect(j="i") T.w = 10 * fac spike_mon = SpikeMonitor(P) rate_mon = PopulationRateMonitor(P) state_mon = StateMonitor(S, "w", record=np.arange(n_recorded), dt=0.1 * second) v_mon = StateMonitor(P, "v", record=np.arange(n_recorded)) run(0.2 * second, report="text") if devicename == "cpp_standalone": device.build(directory=None, with_output=False) results[n_threads, devicename] = {} results[n_threads, devicename]["w"] = state_mon.w results[n_threads, devicename]["v"] = v_mon.v results[n_threads, devicename]["s"] = spike_mon.num_spikes results[n_threads, devicename]["r"] = rate_mon.rate[:] for key1, key2 in [ ((0, "runtime"), (0, "cpp_standalone")), ((1, "cpp_standalone"), (0, "cpp_standalone")), ((2, "cpp_standalone"), (0, "cpp_standalone")), ((3, "cpp_standalone"), (0, "cpp_standalone")), ((4, "cpp_standalone"), (0, "cpp_standalone")), ]: assert_allclose(results[key1]["w"], results[key2]["w"]) assert_allclose(results[key1]["v"], results[key2]["v"]) assert_allclose(results[key1]["r"], results[key2]["r"]) assert_allclose(results[key1]["s"], results[key2]["s"]) reset_device(previous_device) @pytest.mark.cpp_standalone @pytest.mark.standalone_only def test_duplicate_names_across_nets(): set_device("cpp_standalone", build_on_run=False) # In standalone mode, names have to be globally unique, not just unique # per network obj1 = BrianObject(name="name1") obj2 = BrianObject(name="name2") obj3 = BrianObject(name="name3") obj4 = BrianObject(name="name1") net1 = Network(obj1, obj2) net2 = Network(obj3, obj4) net1.run(0 * ms) net2.run(0 * ms) with pytest.raises(ValueError): device.build() reset_device() @pytest.mark.cpp_standalone @pytest.mark.standalone_only @pytest.mark.openmp def test_openmp_scalar_writes(): # Test that writing to a scalar variable only is done once in an OpenMP # setting (see github issue #551) set_device("cpp_standalone", build_on_run=False) prefs.devices.cpp_standalone.openmp_threads = 4 G = NeuronGroup(10, "s : 1 (shared)") G.run_regularly("s += 1") run(defaultclock.dt) device.build(directory=None, with_output=False) assert_equal(G.s[:], 1.0) reset_device() @pytest.mark.cpp_standalone @pytest.mark.standalone_only def test_time_after_run(): set_device("cpp_standalone", build_on_run=False) # Check that the clock and network time after a run is correct, even if we # have not actually run the code yet (via build) G = NeuronGroup(10, "dv/dt = -v/(10*ms) : 1") net = Network(G) assert_allclose(defaultclock.dt, 0.1 * ms) assert_allclose(defaultclock.t, 0.0 * ms) assert_allclose(G.t, 0.0 * ms) assert_allclose(net.t, 0.0 * ms) net.run(10 * ms) assert_allclose(defaultclock.t, 10.0 * ms) assert_allclose(G.t, 10.0 * ms) assert_allclose(net.t, 10.0 * ms) net.run(10 * ms) assert_allclose(defaultclock.t, 20.0 * ms) assert_allclose(G.t, 20.0 * ms) assert_allclose(net.t, 20.0 * ms) device.build(directory=None, with_output=False) # Everything should of course still be accessible assert_allclose(defaultclock.t, 20.0 * ms) assert_allclose(G.t, 20.0 * ms) assert_allclose(net.t, 20.0 * ms) reset_device() @pytest.mark.cpp_standalone @pytest.mark.standalone_only def test_array_cache(): # Check that variables are only accessible from Python when they should be set_device("cpp_standalone", build_on_run=False) G = NeuronGroup( 10, """ dv/dt = -v / (10*ms) : 1 w : 1 x : 1 y : 1 z : 1 (shared) """, threshold="v>1", ) S = Synapses(G, G, "weight: 1", on_pre="w += weight") S.connect(p=0.2) S.weight = 7 # All neurongroup values should be known assert_allclose(G.v, 0) assert_allclose(G.w, 0) assert_allclose(G.x, 0) assert_allclose(G.y, 0) assert_allclose(G.z, 0) assert_allclose(G.i, np.arange(10)) # But the synaptic variable is not -- we don't know the number of synapses with pytest.raises(NotImplementedError): S.weight[:] # Setting variables with explicit values should not change anything G.v = np.arange(10) + 1 G.w = 2 G.y = 5 G.z = 7 assert_allclose(G.v, np.arange(10) + 1) assert_allclose(G.w, 2) assert_allclose(G.y, 5) assert_allclose(G.z, 7) # But setting with code should invalidate them G.x = "i*2" with pytest.raises(NotImplementedError): G.x[:] # Make sure that the array cache does not allow to use incorrectly sized # values to pass with pytest.raises(ValueError): setattr(G, "w", [0, 2]) with pytest.raises(ValueError): G.w.__setitem__(slice(0, 4), [0, 2]) run(10 * ms) # v is now no longer known without running the network with pytest.raises(NotImplementedError): G.v[:] # Neither is w, it is updated in the synapse with pytest.raises(NotImplementedError): G.w[:] # However, no code touches y or z assert_allclose(G.y, 5) assert_allclose(G.z, 7) # i is read-only anyway assert_allclose(G.i, np.arange(10)) # After actually running the network, everything should be accessible device.build(directory=None, with_output=False) assert all(G.v > 0) assert all(G.w > 0) assert_allclose(G.x, np.arange(10) * 2) assert_allclose(G.y, 5) assert_allclose(G.z, 7) assert_allclose(G.i, np.arange(10)) assert_allclose(S.weight, 7) @pytest.mark.cpp_standalone @pytest.mark.standalone_only def test_run_with_debug(): # We just want to make sure that it works for now (i.e. not fails with a # compilation or runtime error), capturing the output is actually # a bit involved to get right. set_device("cpp_standalone", build_on_run=True, debug=True, directory=None) group = NeuronGroup(1, "v: 1", threshold="False") syn = Synapses(group, group, on_pre="v += 1") syn.connect() mon = SpikeMonitor(group) run(defaultclock.dt) @pytest.mark.cpp_standalone @pytest.mark.standalone_only def test_run_with_synapses_and_profile(): set_device("cpp_standalone", build_on_run=True, directory=None) group = NeuronGroup(1, "v: 1", threshold="False", reset="") syn = Synapses(group, group, on_pre="v += 1") syn.connect() mon = SpikeMonitor(group) run(defaultclock.dt, profile=True) @pytest.mark.cpp_standalone @pytest.mark.standalone_only def test_changing_profile_arg(): set_device("cpp_standalone", build_on_run=False) G = NeuronGroup(10000, "v : 1") op1 = G.run_regularly("v = exp(-v)", name="op1") op2 = G.run_regularly("v = exp(-v)", name="op2") op3 = G.run_regularly("v = exp(-v)", name="op3") op4 = G.run_regularly("v = exp(-v)", name="op4") # Op 1 is active only during the first profiled run # Op 2 is active during both profiled runs # Op 3 is active only during the second profiled run # Op 4 is never active (only during the unprofiled run) op1.active = True op2.active = True op3.active = False op4.active = False run(1000 * defaultclock.dt, profile=True) op1.active = True op2.active = True op3.active = True op4.active = True run(1000 * defaultclock.dt, profile=False) op1.active = False op2.active = True op3.active = True op4.active = False run(1000 * defaultclock.dt, profile=True) device.build(directory=None, with_output=False) profiling_dict = dict(magic_network.profiling_info) # Note that for now, C++ standalone creates a new CodeObject for every run, # which is most of the time unnecessary (this is partly due to the way we # handle constants: they are included as literals in the code but they can # change between runs). Therefore, the profiling info is potentially # difficult to interpret assert len(profiling_dict) == 4 # 2 during first run, 2 during last run # The two code objects that were executed during the first run assert ( "op1_codeobject" in profiling_dict and profiling_dict["op1_codeobject"] > 0 * second ) assert ( "op2_codeobject" in profiling_dict and profiling_dict["op2_codeobject"] > 0 * second ) # Four code objects were executed during the second run, but no profiling # information was saved for name in [ "op1_codeobject_1", "op2_codeobject_1", "op3_codeobject", "op4_codeobject", ]: assert name not in profiling_dict # Two code objects were exectued during the third run assert ( "op2_codeobject_2" in profiling_dict and profiling_dict["op2_codeobject_2"] > 0 * second ) assert ( "op3_codeobject_1" in profiling_dict and profiling_dict["op3_codeobject_1"] > 0 * second ) @pytest.mark.cpp_standalone @pytest.mark.standalone_only def test_profile_via_set_device_arg(): set_device("cpp_standalone", build_on_run=False, profile=True) G = NeuronGroup(10000, "v : 1") op1 = G.run_regularly("v = exp(-v)", name="op1") op2 = G.run_regularly("v = exp(-v)", name="op2") op3 = G.run_regularly("v = exp(-v)", name="op3") op4 = G.run_regularly("v = exp(-v)", name="op4") # Op 1 is active only during the first profiled run # Op 2 is active during both profiled runs # Op 3 is active only during the second profiled run # Op 4 is never active (only during the unprofiled run) op1.active = True op2.active = True op3.active = False op4.active = False run(1000 * defaultclock.dt) # Should use profile=True via set_device op1.active = True op2.active = True op3.active = True op4.active = True run(1000 * defaultclock.dt, profile=False) op1.active = False op2.active = True op3.active = True op4.active = False run(1000 * defaultclock.dt, profile=True) device.build(directory=None, with_output=False) profiling_dict = dict(magic_network.profiling_info) # Note that for now, C++ standalone creates a new CodeObject for every run, # which is most of the time unnecessary (this is partly due to the way we # handle constants: they are included as literals in the code but they can # change between runs). Therefore, the profiling info is potentially # difficult to interpret assert len(profiling_dict) == 4 # 2 during first run, 2 during last run # The two code objects that were executed during the first run assert ( "op1_codeobject" in profiling_dict and profiling_dict["op1_codeobject"] > 0 * second ) assert ( "op2_codeobject" in profiling_dict and profiling_dict["op2_codeobject"] > 0 * second ) # Four code objects were executed during the second run, but no profiling # information was saved for name in [ "op1_codeobject_1", "op2_codeobject_1", "op3_codeobject", "op4_codeobject", ]: assert name not in profiling_dict # Two code objects were exectued during the third run assert ( "op2_codeobject_2" in profiling_dict and profiling_dict["op2_codeobject_2"] > 0 * second ) assert ( "op3_codeobject_1" in profiling_dict and profiling_dict["op3_codeobject_1"] > 0 * second ) @pytest.mark.cpp_standalone @pytest.mark.standalone_only def test_delete_code_data(): set_device("cpp_standalone", build_on_run=False) group = NeuronGroup(10, "dv/dt = -v / (10*ms) : volt", method="exact") group.v = np.arange(10) * mV # uses the static array mechanism run(defaultclock.dt) # Overwrite the initial values via run_args mechanims device.build(run_args={group.v: np.arange(10)[::-1] * mV}, directory=None) results_dir = os.path.join(device.project_dir, "results") assert os.path.exists(results_dir) and os.path.isdir(results_dir) # There should be 3 files for the clock, 2 for the neurongroup (index + v), # and the "last_run_info.txt" file assert len(os.listdir(results_dir)) == 6 device.delete(data=True, run_args=False, code=False, directory=False) assert os.path.exists(results_dir) and os.path.isdir(results_dir) assert len(os.listdir(results_dir)) == 0 static_arrays_before = len( os.listdir(os.path.join(device.project_dir, "static_arrays")) ) assert static_arrays_before > 0 assert len(os.listdir(os.path.join(device.project_dir, "code_objects"))) > 0 device.delete(data=False, code=True, run_args=False, directory=False) assert ( 0 < len(os.listdir(os.path.join(device.project_dir, "static_arrays"))) < static_arrays_before ) assert len(os.listdir(os.path.join(device.project_dir, "code_objects"))) == 0 device.delete(data=False, code=False, run_args=True, directory=False) len(os.listdir(os.path.join(device.project_dir, "static_arrays"))) == 0 @pytest.mark.cpp_standalone @pytest.mark.standalone_only def test_delete_directory(): set_device("cpp_standalone", build_on_run=True, directory=None) group = NeuronGroup(10, "dv/dt = -v / (10*ms) : volt", method="exact") group.v = np.arange(10) * mV # uses the static array mechanism run(defaultclock.dt) # Add a new file dummy_file = os.path.join(device.project_dir, "results", "dummy.txt") open(dummy_file, "w").flush() assert os.path.isfile(dummy_file) with catch_logs() as logs: device.delete(directory=True) assert ( len( [ l for l in logs if l[1] == "brian2.devices.cpp_standalone.device.delete_skips_directory" ] ) == 1 ) assert os.path.isfile(dummy_file) with catch_logs() as logs: device.delete(directory=True, force=True) assert ( len( [ l for l in logs if l[1] == "brian2.devices.cpp_standalone.device.delete_skips_directory" ] ) == 0 ) # everything should be deleted assert not os.path.exists(device.project_dir) @pytest.mark.cpp_standalone @pytest.mark.standalone_only def test_multiple_standalone_runs(): # see github issue #1189 set_device("cpp_standalone", directory=None) network = Network() Pe = NeuronGroup(1, "v : 1", threshold="False") C_ee = Synapses(Pe, Pe, on_pre="v += 1") C_ee.connect() network.add(Pe, C_ee) network.run(defaultclock.dt) device.reinit() device.activate(directory=None) network2 = Network() Pe = NeuronGroup(1, "v : 1", threshold="False") C_ee = Synapses(Pe, Pe, on_pre="v += 1") C_ee.connect() network2.add(Pe, C_ee) network2.run(defaultclock.dt) @pytest.mark.cpp_standalone @pytest.mark.standalone_only def test_continued_standalone_runs(): # see github issue #1237 set_device("cpp_standalone", build_on_run=False) source = SpikeGeneratorGroup(1, [0], [0] * ms) target = NeuronGroup(1, "v : 1") C_ee = Synapses(source, target, on_pre="v += 1", delay=2 * ms) C_ee.connect() run(1 * ms) # Spike has not been delivered yet run(2 * ms) device.build(directory=None) assert target.v[0] == 1 # Make sure the spike got delivered @pytest.mark.cpp_standalone @pytest.mark.standalone_only def test_constant_replacement(): # see github issue #1276 set_device("cpp_standalone") x = 42 G = NeuronGroup(1, "y : 1") G.y = "x" run(0 * ms) assert G.y[0] == 42.0 @pytest.mark.cpp_standalone @pytest.mark.standalone_only def test_change_parameter_without_recompile(): prefs.core.default_float_dtype = np.float32 set_device("cpp_standalone", directory=None, with_output=True, debug=True) on_off = TimedArray([True, False, True], dt=defaultclock.dt, name="on_off") stim = TimedArray( np.arange(30).reshape(3, 10) * nA, dt=defaultclock.dt, name="stim" ) G = NeuronGroup( 10, """ x : 1 (constant) v : volt (constant) n : integer (constant) b : boolean (constant) s = int(on_off(t))*stim(t, i) : amp """, name="neurons", ) G.x = np.arange(10) G.n = np.arange(10) G.b = np.arange(10) % 2 == 0 G.v = np.arange(10) * volt mon = StateMonitor(G, "s", record=True) run(3 * defaultclock.dt) assert array_equal(G.x, np.arange(10)) assert array_equal(G.n, np.arange(10)) assert array_equal(G.b, np.arange(10) % 2 == 0) assert array_equal(G.v, np.arange(10) * volt) assert_allclose( mon.s.T / nA, np.array( [ [0, 1, 2, 3, 4, 5, 6, 7, 8, 9], [0, 0, 0, 0, 0, 0, 0, 0, 0, 0], # on_off(t) == False [20, 21, 22, 23, 24, 25, 26, 27, 28, 29], ] ), ) device.run( run_args=[ "neurons.x=5", "neurons.v=3", "neurons.n=17", "neurons.b=True", "on_off.values=True", ] ) assert array_equal(G.x, np.ones(10) * 5) assert array_equal(G.n, np.ones(10) * 17) assert array_equal(G.b, np.ones(10, dtype=bool)) assert array_equal(G.v, np.ones(10) * 3 * volt) assert_allclose( mon.s.T / nA, np.array( [ [0, 1, 2, 3, 4, 5, 6, 7, 8, 9], [10, 11, 12, 13, 14, 15, 16, 17, 18, 19], [20, 21, 22, 23, 24, 25, 26, 27, 28, 29], ] ), ) ar = np.arange(10) * 2.0 ar.astype(G.x.dtype).tofile(os.path.join(device.project_dir, "init_values_x1.dat")) ar.astype(G.n.dtype).tofile(os.path.join(device.project_dir, "init_values_n1.dat")) (np.arange(10) % 2 != 0).tofile( os.path.join(device.project_dir, "init_values_b1.dat") ) ar.astype(G.v.dtype).tofile(os.path.join(device.project_dir, "init_values_v1.dat")) ar2 = 2 * np.arange(30).reshape(3, 10) * nA ar2.astype(stim.values.dtype).tofile( os.path.join(device.project_dir, "init_stim_values.dat") ) device.run( run_args=[ "neurons.v=init_values_v1.dat", "neurons.x=init_values_x1.dat", "neurons.b=init_values_b1.dat", "neurons.n=init_values_n1.dat", "stim.values=init_stim_values.dat", ] ) assert array_equal(G.x, ar) assert array_equal(G.n, ar) assert array_equal(G.b, np.arange(10) % 2 != 0) assert array_equal(G.v, ar * volt) assert_allclose( mon.s.T / nA, np.array( [ [0, 2, 4, 6, 8, 10, 12, 14, 16, 18], [0, 0, 0, 0, 0, 0, 0, 0, 0, 0], # on_off(t) == False [40, 42, 44, 46, 48, 50, 52, 54, 56, 58], ] ), ) reset_device() @pytest.mark.cpp_standalone @pytest.mark.standalone_only def test_change_parameter_without_recompile_errors(): set_device("cpp_standalone", directory=None, with_output=False) G = NeuronGroup(10, "v:volt", name="neurons") G.v = np.arange(10) * volt run(0 * ms) with pytest.raises(DimensionMismatchError): device.run(run_args={G.v: 5}) with pytest.raises(DimensionMismatchError): device.run(run_args={G.v: 5 * siemens}) with pytest.raises(TypeError): device.run(run_args={G.v: np.arange(9) * volt}) reset_device() @pytest.mark.cpp_standalone @pytest.mark.standalone_only def test_change_parameter_without_recompile_dict_syntax(): set_device("cpp_standalone", directory=None, with_output=False) on_off = TimedArray([True, False, True], dt=defaultclock.dt, name="on_off") stim = TimedArray( np.arange(30).reshape(3, 10) * nA, dt=defaultclock.dt, name="stim" ) G = NeuronGroup( 10, """ x : 1 (constant) n : integer (constant) b : boolean (constant) v : volt (constant) s = int(on_off(t))*stim(t, i) : amp """, name="neurons", ) G.x = np.arange(10) G.n = np.arange(10) G.b = np.arange(10) % 2 == 0 G.v = np.arange(10) * volt mon = StateMonitor(G, "s", record=True) run(3 * defaultclock.dt) assert array_equal(G.x, np.arange(10)) assert array_equal(G.n, np.arange(10)) assert array_equal(G.b, np.arange(10) % 2 == 0) assert array_equal(G.v, np.arange(10) * volt) assert_allclose( mon.s.T / nA, np.array( [ [0, 1, 2, 3, 4, 5, 6, 7, 8, 9], [0, 0, 0, 0, 0, 0, 0, 0, 0, 0], # on_off(t) == False [20, 21, 22, 23, 24, 25, 26, 27, 28, 29], ] ), ) device.run(run_args={G.x: 5, G.v: 3 * volt, G.n: 17, G.b: True, on_off: True}) assert array_equal(G.x, np.ones(10) * 5) assert array_equal(G.n, np.ones(10) * 17) assert array_equal(G.b, np.ones(10, dtype=bool)) assert array_equal(G.v, np.ones(10) * 3 * volt) assert_allclose( mon.s.T / nA, np.array( [ [0, 1, 2, 3, 4, 5, 6, 7, 8, 9], [10, 11, 12, 13, 14, 15, 16, 17, 18, 19], [20, 21, 22, 23, 24, 25, 26, 27, 28, 29], ] ), ) ar = np.arange(10) * 2.0 ar2 = 2 * np.arange(30).reshape(3, 10) * nA device.run( run_args={ G.x: ar, G.v: ar * volt, G.n: ar, G.b: np.arange(10) % 2 != 0, stim: ar2, } ) assert array_equal(G.x, ar) assert array_equal(G.n, ar) assert array_equal(G.b, np.arange(10) % 2 != 0) assert array_equal(G.v, ar * volt) assert_allclose( mon.s.T / nA, np.array( [ [0, 2, 4, 6, 8, 10, 12, 14, 16, 18], [0, 0, 0, 0, 0, 0, 0, 0, 0, 0], # on_off(t) == False [40, 42, 44, 46, 48, 50, 52, 54, 56, 58], ] ), ) reset_device() @pytest.mark.cpp_standalone @pytest.mark.standalone_only def test_change_synapse_parameter_without_recompile_dict_syntax(): set_device("cpp_standalone", directory=None, with_output=False) G = NeuronGroup(10, "", name="neurons") S = Synapses(G, G, "w:1", name="Synapses") S.connect(j="i") S.w = np.arange(10) run(0 * ms) assert array_equal(S.w, np.arange(10)) device.run(run_args={S.w: 17}) assert array_equal(S.w, np.ones(10) * 17) ar = np.arange(10) * 2.0 device.run(run_args={S.w: ar}) assert array_equal(S.w, ar) reset_device() @pytest.mark.cpp_standalone @pytest.mark.standalone_only def test_change_parameter_without_recompile_dependencies(): set_device("cpp_standalone", directory=None, with_output=False) G = NeuronGroup( 10, """ v:volt w:1 """, name="neurons", ) G.v = np.arange(10) * volt device.apply_run_args() G.w = "v/volt*2" run(0 * ms) assert array_equal(G.v, np.arange(10)) assert array_equal(G.w, np.arange(10) * 2) device.run(run_args=["neurons.v=5"]) assert array_equal(G.v, np.ones(10) * 5 * volt) assert array_equal(G.w, np.ones(10) * 5 * 2) ar = np.arange(10) * 2.0 ar.astype(G.v.dtype).tofile(os.path.join(device.project_dir, "init_values_v2.dat")) device.run(run_args=[f"neurons.v=init_values_v2.dat"]) assert array_equal(G.v, ar * volt) assert array_equal(G.w, ar * 2) reset_device() class RunSim: def __init__(self): self.device = get_device() self.G = NeuronGroup( 10, """ v:volt w:1 x:1 """, name="neurons", ) run(0 * ms) def run_sim(self, idx): # Ugly hack needed for windows device_module.active_device = self.device device.run( results_directory=f"results_{idx}", run_args={ self.G.v: idx * volt, self.G.w: np.arange(10), # Same values for all processes self.G.x: np.arange(10) * idx, # Different values }, ) return self.G.v[:], self.G.w[:], self.G.x[:] @pytest.mark.cpp_standalone @pytest.mark.standalone_only @pytest.mark.skipif( platform.system() == "Darwin" and platform.processor() == "arm", reason="multiprocessing hangs on macOS with Apple Silicon", ) def test_change_parameters_multiprocessing(): set_device("cpp_standalone", directory=None) sim = RunSim() import multiprocessing p = multiprocessing.Pool() try: results = p.map(sim.run_sim, range(5)) finally: p.close() p.join() for idx, result in zip(range(5), results): v, w, x = result assert array_equal(v, np.ones(10) * idx * volt) assert array_equal(w, np.arange(10)) assert array_equal(x, np.arange(10) * idx) @pytest.mark.cpp_standalone @pytest.mark.standalone_only def test_header_file_inclusion(): set_device("cpp_standalone", directory=None, debug=True) with tempfile.TemporaryDirectory() as tmpdir: with open(os.path.join(tmpdir, "foo.h"), "w") as f: f.write( """ namespace brian_test_namespace { extern double test_variable; } """ ) with open(os.path.join(tmpdir, "foo.cpp"), "w") as f: f.write( """ namespace brian_test_namespace { double test_variable = 42; } """ ) @implementation( "cpp", """ double brian_function(int index) { using namespace brian_test_namespace; return test_variable * index; } """, headers=['"foo.h"'], sources=[os.path.join(tmpdir, "foo.cpp")], include_dirs=[tmpdir], ) @check_units(index=1, result=1) def brian_function(index): raise NotImplementedError() # Use the function in a somewhat convoluted way that exposes errors in the # code generation process G = PoissonGroup(5, rates="brian_function(i)*Hz") S = Synapses(G, G, "rate_copy : Hz") S.connect(j="i") S.run_regularly("rate_copy = rates_pre") run(defaultclock.dt) assert_allclose(S.rate_copy[:], np.arange(len(G)) * 42 * Hz) @pytest.mark.cpp_standalone @pytest.mark.standalone_only def test_negative_duration_in_standalone_device(): set_device("cpp_standalone") G = NeuronGroup(1, "v:1") with pytest.raises(ValueError): run(-1 * second) if __name__ == "__main__": for t in [ test_cpp_standalone, test_multiple_connects, test_storing_loading, test_openmp_consistency, test_duplicate_names_across_nets, test_openmp_scalar_writes, test_time_after_run, test_array_cache, test_run_with_debug, test_changing_profile_arg, test_profile_via_set_device_arg, test_delete_code_data, test_delete_directory, test_multiple_standalone_runs, test_change_parameter_without_recompile, test_change_parameter_without_recompile_errors, test_change_parameter_without_recompile_dict_syntax, test_change_parameter_without_recompile_dependencies, test_change_synapse_parameter_without_recompile_dict_syntax, test_change_parameters_multiprocessing, test_header_file_inclusion, ]: t() reinit_and_delete()