diff --git a/spynnaker/pyNN/spinnaker.py b/spynnaker/pyNN/spinnaker.py index 465d792e98..74c3c4e893 100644 --- a/spynnaker/pyNN/spinnaker.py +++ b/spynnaker/pyNN/spinnaker.py @@ -12,6 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. +import importlib import logging import os from typing import Any, Collection, Optional, Type, Union, cast @@ -26,7 +27,7 @@ from spinn_utilities.log import FormatAdapter -from spinn_utilities.config_holder import get_config_bool +from spinn_utilities.config_holder import get_config_bool, get_config_str_list from spinn_utilities.overrides import overrides from spinn_front_end_common.interface.abstract_spinnaker_base import ( @@ -123,6 +124,25 @@ def __init__( # Clears all previously added ceiling on the number of neurons per # core, the number of synapse cores and allowing of delay extensions AbstractPyNNNeuronModel.reset_all() + self._add_test_cfg_settings() + + def _add_test_cfg_settings(self) -> None: + """ + Ideally this function will find nothing to do. + + This is designed for switching scripts like PyNNExamples + to split mode without the need to change the scripts. + + Better to call set_model_n_synapse_cores in the script + """ + n_synapse_cores = get_config_str_list("Simulation", "n_synapse_cores") + for n_synapse_core in n_synapse_cores: + path, name, n = n_synapse_core.split(":") + model_class = getattr(importlib.import_module(path), name) + model = model_class() + model.set_model_n_synapse_cores(int(n)) + logger.warning(f"model:{name} n_synapse_core set to {n} " + f"based on cfg") @overrides(AbstractSpinnakerBase._add_cfg_defaults_and_template) def _add_cfg_defaults_and_template(self) -> None: diff --git a/spynnaker/pyNN/spynnaker.cfg b/spynnaker/pyNN/spynnaker.cfg index 569ce8bede..59a06717ab 100644 --- a/spynnaker/pyNN/spynnaker.cfg +++ b/spynnaker/pyNN/spynnaker.cfg @@ -63,6 +63,12 @@ n_colour_bits = 4 error_on_non_spynnaker_pynn = True @error_on_non_spynnaker_pynn = Whether to error or just warn on non-spynnaker-compatible PyNN +n_synapse_cores = None +@n_synapse_cores = Testing Option! + A better way is to call set_model_n_synapse_cores on the model. + Mapping of a AbstractPyNNNeuronModel to n_synapse_cores to be set + In the format model:n_synapse_cores,model:n_synapse_cores + [Recording] @ = Section for the sending of live spikes. diff --git a/spynnaker_integration_tests/test_external_devices/test_live_neuron_voltage.py b/spynnaker_integration_tests/test_external_devices/test_live_neuron_voltage.py index e72a435a9e..f6fd47b483 100644 --- a/spynnaker_integration_tests/test_external_devices/test_live_neuron_voltage.py +++ b/spynnaker_integration_tests/test_external_devices/test_live_neuron_voltage.py @@ -120,6 +120,7 @@ def live_neuron_voltage() -> None: translator_2 = Translator(devices_2) model_2 = p.external_devices.ExternalDeviceLifControl( devices_2, create_edges, translator_2) + model_2.set_model_n_synapse_cores(1) conn = p.external_devices.SpynnakerLiveSpikesConnection( receive_labels=["stim"], local_port=None) conn.add_receive_callback("stim", spike_receiver) @@ -160,6 +161,8 @@ class TestLiveNeuronVoltage(BaseTestCase): def test_live_neuron_voltage(self) -> None: self.runsafe(live_neuron_voltage) + self.check_binaries_used(["external_device_lif_control.aplx", + "external_device_lif_control_neuron.aplx"]) if __name__ == '__main__': diff --git a/spynnaker_integration_tests/test_multi_call_examples/test_stdp_random_run.py b/spynnaker_integration_tests/test_multi_call_examples/test_stdp_random_run.py index 476ff32af9..eab45e5b1a 100644 --- a/spynnaker_integration_tests/test_multi_call_examples/test_stdp_random_run.py +++ b/spynnaker_integration_tests/test_multi_call_examples/test_stdp_random_run.py @@ -12,9 +12,11 @@ # See the License for the specific language governing permissions and # limitations under the License. +import numpy import pyNN.spiNNaker as sim + from spinnaker_testbase import BaseTestCase -import numpy +from spynnaker.pyNN.models.abstract_pynn_model import AbstractPyNNModel class TestSTDPRandomRun(BaseTestCase): @@ -22,11 +24,11 @@ class TestSTDPRandomRun(BaseTestCase): # same thing being loaded twice, both with data generated off and on the # machine - def do_run(self) -> None: + def run_model(self, model: AbstractPyNNModel) -> None: sim.setup(1.0) pop = sim.Population(1, sim.IF_curr_exp(i_offset=5.0), label="pop") pop.record("spikes") - pop_2 = sim.Population(1, sim.IF_curr_exp(), label="pop_2") + pop_2 = sim.Population(1, model, label="pop_2") proj = sim.Projection( pop, pop_2, sim.AllToAllConnector(), sim.STDPMechanism( @@ -53,7 +55,28 @@ def do_run(self) -> None: assert numpy.array_equal(weights_1_1, weights_2_1) assert numpy.array_equal(weights_1_2, weights_2_2) + assert len(spikes_1[0]) > 0 assert numpy.array_equal(spikes_1, spikes_2) - def test_do_run(self) -> None: - self.runsafe(self.do_run) + def _do_if_curr_exp(self) -> None: + self.run_model(sim.IF_curr_exp()) + + def test_check_if_curr_exp(self) -> None: + self.runsafe(self._do_if_curr_exp) + self.check_binary_used("IF_curr_exp_stdp_mad_pair_additive.aplx") + + def _do_if_curr_exp_ca2_additive(self) -> None: + self.run_model(sim.extra_models.IFCurrExpCa2Adaptive()) + + def test_check_if_curr_exp_ca2_additive(self) -> None: + self.runsafe(self._do_if_curr_exp_ca2_additive) + self.check_binary_used( + "IF_curr_exp_ca2_adaptive_stdp_mad_pair_additive.aplx") + + def _do_izk_cond_exp_dual(self) -> None: + self.run_model(sim.extra_models.Izhikevich_cond_dual()) + + def test_check_izk_cond_exp_dual(self) -> None: + self.runsafe(self._do_izk_cond_exp_dual) + self.check_binary_used( + "IZK_cond_exp_dual_stdp_mad_pair_additive.aplx") diff --git a/spynnaker_integration_tests/test_stdp/test_STDP_nearest_pair_additive.py b/spynnaker_integration_tests/test_stdp/test_STDP_nearest_pair_additive.py index 46708dfe65..f10464045a 100644 --- a/spynnaker_integration_tests/test_stdp/test_STDP_nearest_pair_additive.py +++ b/spynnaker_integration_tests/test_stdp/test_STDP_nearest_pair_additive.py @@ -20,7 +20,7 @@ class TestSTDPNearestPairAdditive(BaseTestCase): - def potentiation_and_depression(self) -> None: + def potentiation_and_depression(self, n_synapse_cores: int) -> None: p.setup(1) runtime = 100 initial_run = 1000 # to negate any initial conditions @@ -54,7 +54,8 @@ def potentiation_and_depression(self) -> None: {'spike_times': extra_spikes}, label="extra") # Post-plastic-synapse population - post_pop = p.Population(1, p.IF_curr_exp(), label="post") + post_pop = p.Population(1, p.IF_curr_exp(), label="post", + n_synapse_cores=n_synapse_cores) # Create projections p.Projection( @@ -136,8 +137,20 @@ def potentiation_and_depression(self) -> None: self.assertTrue(numpy.allclose( weights[0], new_weight_exact, atol=0.001)) + def do_synapse(self) -> None: + self.potentiation_and_depression(1) + def test_potentiation_and_depression(self) -> None: - self.runsafe(self.potentiation_and_depression) + self.runsafe(self.do_synapse) + self.check_binary_used("synapses_stdp_mad_nearest_pair_additive.aplx") + + def do_combined(self) -> None: + self.potentiation_and_depression(0) + + def test_combined(self) -> None: + self.runsafe(self.do_combined) + self.check_binary_used( + "IF_curr_exp_stdp_mad_nearest_pair_additive.aplx") if __name__ == '__main__': diff --git a/spynnaker_integration_tests/test_stdp/test_STDP_nearest_pair_multiplicative.py b/spynnaker_integration_tests/test_stdp/test_STDP_nearest_pair_multiplicative.py index 5d01e8ac3d..eb83eb72e9 100644 --- a/spynnaker_integration_tests/test_stdp/test_STDP_nearest_pair_multiplicative.py +++ b/spynnaker_integration_tests/test_stdp/test_STDP_nearest_pair_multiplicative.py @@ -20,7 +20,7 @@ class TestSTDPNearestPairAdditive(BaseTestCase): - def potentiation_and_depression(self) -> None: + def potentiation_and_depression(self, n_synapse_cores: int) -> None: p.setup(1) runtime = 100 initial_run = 1000 # to negate any initial conditions @@ -54,7 +54,8 @@ def potentiation_and_depression(self) -> None: {'spike_times': extra_spikes}, label="extra") # Post-plastic-synapse population - post_pop = p.Population(1, p.IF_curr_exp(), label="post") + post_pop = p.Population(1, p.IF_curr_exp(), label="post", + n_synapse_cores=n_synapse_cores) # Create projections p.Projection( @@ -136,8 +137,21 @@ def potentiation_and_depression(self) -> None: self.assertTrue(numpy.allclose( weights[0], new_weight_exact, atol=0.001)) + def do_synapse(self) -> None: + self.potentiation_and_depression(1) + def test_potentiation_and_depression(self) -> None: - self.runsafe(self.potentiation_and_depression) + self.runsafe(self.do_synapse) + self.check_binary_used( + "synapses_stdp_mad_nearest_pair_multiplicative.aplx") + + def do_combined(self) -> None: + self.potentiation_and_depression(0) + + def test_combined(self) -> None: + self.runsafe(self.do_combined) + self.check_binary_used( + "IF_curr_exp_stdp_mad_nearest_pair_multiplicative.aplx") if __name__ == '__main__': diff --git a/spynnaker_integration_tests/test_stdp/test_STDP_neuromodulation.py b/spynnaker_integration_tests/test_stdp/test_STDP_neuromodulation.py index 53b8707726..dfb1739bc5 100644 --- a/spynnaker_integration_tests/test_stdp/test_STDP_neuromodulation.py +++ b/spynnaker_integration_tests/test_stdp/test_STDP_neuromodulation.py @@ -12,17 +12,24 @@ # See the License for the specific language governing permissions and # limitations under the License. +import math +from typing import cast, List +import unittest + +import numpy import pyNN.spiNNaker as sim + from spinnaker_testbase import BaseTestCase -import numpy -import unittest -import math +from spynnaker.pyNN.models.neuron import ConnectionHolder +from spynnaker.pyNN.models.neuron.plasticity.stdp.weight_dependence import ( + AbstractWeightDependence) -class TestSTDPNeuromodulation(BaseTestCase): - def neuromodulation(self) -> None: +class TestSTDPNeuromodulation(BaseTestCase): + def neuromodulation(self, weight_dependence: AbstractWeightDependence + ) -> ConnectionHolder: """ Simple test for neuromodulated STDP. Two pre-synaptic spikes are added, at times 1500 and 2400ms. @@ -97,8 +104,7 @@ def neuromodulation(self) -> None: timing_dependence=sim.SpikePairRule( tau_plus=10, tau_minus=12, A_plus=1, A_minus=1), - weight_dependence=sim.AdditiveWeightDependence( - w_min=0, w_max=20), + weight_dependence=weight_dependence, weight=rewarded_syn_weight) # Create a plastic connection between pre and post neurons @@ -127,6 +133,13 @@ def neuromodulation(self) -> None: print(spikes) + return weights + + def do_additive(self) -> None: + weight_dependence = sim.AdditiveWeightDependence(w_min=0, w_max=20) + weights = self.neuromodulation(weight_dependence) + + DA_concentration = 0.1 pot = 1 * math.exp(-((1504 - 1500)/10)) decay = math.exp(-((1601 - 1504)/1000)) el = pot * decay @@ -135,15 +148,30 @@ def neuromodulation(self) -> None: decay_e = math.exp(-((2400 - 1601)/1000)) weight_exact = ( ((el * DA_concentration) * const)*((decay_d * decay_e) - 1)) - print(f"Weight calculated: {weight_exact}") - print(f"Weight from SpiNNaker: {weights[0][2]}") - + weights0 = cast(List[float], weights[0]) + print(f"Weight from SpiNNaker: {weights0[2]}") self.assertTrue(numpy.allclose( - weights[0][2], weight_exact, atol=0.02)) + weights0[2], weight_exact, atol=0.02)) + + self.check_binary_used( + "synapses_stdp_izhikevich_neuromodulation_pair_additive.aplx") + + def test_additive(self) -> None: + self.runsafe(self.do_additive) + + def do_multiplicative(self) -> None: + weight_dependence = sim.MultiplicativeWeightDependence( + w_min=0, w_max=20) + self.neuromodulation(weight_dependence) + + # TODO Weights expected + self.check_binary_used( + "synapses_stdp_izhikevich_neuromodulation_" + "pair_multiplicative.aplx") - def test_neuromodulation(self) -> None: - self.runsafe(self.neuromodulation) + def test_multiplicative(self) -> None: + self.runsafe(self.do_multiplicative) if __name__ == '__main__': diff --git a/spynnaker_integration_tests/test_stdp/test_STDP_pair_multiplicative.py b/spynnaker_integration_tests/test_stdp/test_STDP_pair_multiplicative.py index 4cb779513a..995c663076 100644 --- a/spynnaker_integration_tests/test_stdp/test_STDP_pair_multiplicative.py +++ b/spynnaker_integration_tests/test_stdp/test_STDP_pair_multiplicative.py @@ -78,7 +78,7 @@ def post_spike_same_time() -> None: assert numpy.allclose(weights_1, new_weight_exact, rtol=0.001) -def potentiation_and_depression() -> None: +def potentiation_and_depression(n_synapse_cores: int) -> None: p.setup(1) runtime = 100 initial_run = 1000 # to negate any initial conditions @@ -112,7 +112,8 @@ def potentiation_and_depression() -> None: {'spike_times': extra_spikes}, label="extra") # Post-plastic-synapse population - post_pop = p.Population(1, p.IF_curr_exp(), label="post") + post_pop = p.Population(1, p.IF_curr_exp(), label="post", + n_synapse_cores=n_synapse_cores) # Create projections p.Projection( @@ -170,8 +171,21 @@ def potentiation_and_depression() -> None: class TestSTDPPairAdditive(BaseTestCase): + def do_synapse(self) -> None: + potentiation_and_depression(1) + def test_potentiation_and_depression(self) -> None: - self.runsafe(potentiation_and_depression) + self.runsafe(self.do_synapse) + self.check_binary_used( + "synapses_stdp_mad_pair_multiplicative.aplx") + + def do_combined(self) -> None: + potentiation_and_depression(0) + + def test_combined(self) -> None: + self.runsafe(self.do_combined) + self.check_binary_used( + "IF_curr_exp_stdp_mad_pair_multiplicative.aplx") def test_post_spike_same_time(self) -> None: self.runsafe(post_spike_same_time) diff --git a/spynnaker_integration_tests/test_stdp/test_STDP_triplet_additive.py b/spynnaker_integration_tests/test_stdp/test_STDP_triplet_additive.py index 11b858d29c..90d2172fd2 100644 --- a/spynnaker_integration_tests/test_stdp/test_STDP_triplet_additive.py +++ b/spynnaker_integration_tests/test_stdp/test_STDP_triplet_additive.py @@ -18,7 +18,7 @@ import numpy -def triplet_additive() -> None: +def triplet_additive(n_synapse_cores: int) -> None: # ------------------------------------------------------------------- # This test uses a single data point from the Pfister/Gerstner example # which is described and evaluated in more detail in @@ -52,7 +52,8 @@ def triplet_additive() -> None: for t in delta_t: # Neuron populations pre_pop = sim.Population(1, model(**cell_params)) - post_pop = sim.Population(1, model(**cell_params)) + post_pop = sim.Population(1, model(**cell_params), + n_synapse_cores=n_synapse_cores) # Stimulating populations pre_times = [start_time - 1 + (s * int(1000.0 / float(freq))) @@ -119,9 +120,22 @@ def triplet_additive() -> None: class TestSTDPPairAdditive(BaseTestCase): - def test_triplet_additive(self) -> None: - self.runsafe(triplet_additive) + def do_synapses(self) -> None: + triplet_additive(1) + + def test_synapses(self) -> None: + self.runsafe(self.do_synapses) + self.check_binary_used( + "synapses_stdp_mad_pfister_triplet_additive.aplx") + + def do_combined(self) -> None: + triplet_additive(0) + + def test_combined(self) -> None: + self.runsafe(self.do_combined) + self.check_binary_used( + "IF_curr_exp_stdp_mad_pfister_triplet_additive.aplx") if __name__ == '__main__': - triplet_additive() + triplet_additive(0) diff --git a/spynnaker_integration_tests/test_various/test_binaries.py b/spynnaker_integration_tests/test_various/test_binaries.py new file mode 100644 index 0000000000..04bc1157b1 --- /dev/null +++ b/spynnaker_integration_tests/test_various/test_binaries.py @@ -0,0 +1,88 @@ +# Copyright (c) 2026 The University of Manchester +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import pyNN.spiNNaker as sim +from spinnaker_testbase import BaseTestCase + +from spynnaker.pyNN.models.abstract_pynn_model import AbstractPyNNModel +from spynnaker.pyNN.models.populations import Population + +INTERVAL = 50 +N_NEURONS = 5 + + +class TestBinaries(BaseTestCase): + + def add_population(self, model: AbstractPyNNModel, weight: int, + input_pop: Population) -> Population: + population = sim.Population( + N_NEURONS, model, label=model.name) + sim.Projection(input_pop, population, sim.OneToOneConnector(), + synapse_type=sim.StaticSynapse(weight=weight, delay=1)) + population.record("spikes") + return population + + def add_neuron_population(self, model: AbstractPyNNModel, weight: int, + input_pop: Population) -> Population: + population = sim.Population( + N_NEURONS, model, label=model.name, n_synapse_cores=1) + sim.Projection(input_pop, population, sim.OneToOneConnector(), + synapse_type=sim.StaticSynapse(weight=weight, delay=1)) + population.record("spikes") + return population + + def check_population(self, population: Population) -> None: + neo = population.get_data(variables="spikes") + spikes_trains = neo.segments[0].spiketrains + print(population.label, spikes_trains) + self.assertEqual(len(spikes_trains), N_NEURONS) + for spike_trains in spikes_trains: + self.assertGreaterEqual(len(spikes_trains), 1) + + def check_binaries(self) -> None: + + sim.setup(timestep=1.0) + + spike_times = [] + for i in range(N_NEURONS): + spike_times.append([i, i + INTERVAL, i + INTERVAL * 2]) + input_pop = sim.Population( + N_NEURONS, sim.SpikeSourceArray(spike_times=spike_times), + label="input") + + # Ideally there should be better tests for these modules + # remove models tested elsewhere! + populations = [] + + # IF_curr_delta_ca2_adaptive.aplx + populations.append(self.add_population( + sim.extra_models.IFCurrDeltaCa2Adaptive(), 15, input_pop)) + + # IF_curr_delta_ca2_adaptive_neuron.aplx + populations.append(self.add_neuron_population( + sim.extra_models.IFCurrDeltaCa2Adaptive(), 15, input_pop)) + + sim.run(N_NEURONS + INTERVAL * 3) + + for population in populations: + self.check_population(population) + sim.end() + + self.check_binaries_used([ + "IF_curr_delta_ca2_adaptive.aplx", + "IF_curr_delta_ca2_adaptive_neuron.aplx", + ]) + + def test_binaries(self) -> None: + self.runsafe(self.check_binaries)