set SpikeSourceArrayVertex

spynnaker/pyNN/models/common/parameter_holder.py

@@ -78,6 +78,9 @@ def __init__(
     def _safe_read_values(self, parameter: str) -> Union[List[float], float]:
         values = self.__get_call(parameter, self.__selector)
 
+        if not values:
+            return values
+
         # The values must be a single item, a list or a random distribution;
         # if a random distribution we must not have generated yet!
         if isinstance(values, RandomDistribution):

spynnaker/pyNN/models/common/types.py

@@ -17,6 +17,8 @@
 from numpy.typing import NDArray
 from typing_extensions import TypeAlias
 
+from spinn_utilities.ranged import RangedList
+
 from spynnaker.pyNN.random_distribution import RandomDistribution
 
 #: Type of names of parameters and state variables.
@@ -27,8 +29,5 @@
     float, Sequence[float], NDArray[numpy.floating], RandomDistribution]
 
 #: Type of spikes in spike sources.
-Spikes: TypeAlias = Union[
-    # Can be floating point values (will round)
-    Values,
-    # Can be integer values, or lists of such
-    int, Sequence[int], Sequence[Sequence[int]], NDArray[numpy.integer]]
+Spikes: TypeAlias = (Values | RangedList[float] | Sequence[Sequence[int]] |
+                     NDArray[numpy.integer])

spynnaker/pyNN/models/spike_source/spike_source_array_vertex.py

@@ -35,6 +35,7 @@
 from spinn_utilities.config_holder import get_config_int
 from spinn_utilities.log import FormatAdapter
 from spinn_utilities.overrides import overrides
+from spinn_utilities.ranged import RangedList
 from spinn_utilities.ranged.abstract_sized import Selector
 
 from pacman.model.graphs.common import Slice
@@ -51,7 +52,6 @@
 )
 from spynnaker.pyNN.models.common.types import Names, Spikes
 from spynnaker.pyNN.utilities.buffer_data_type import BufferDataType
-from spynnaker.pyNN.utilities.ranged import SpynnakerRangedList
 
 from .spike_source_array_machine_vertex import SpikeSourceArrayMachineVertex
 
@@ -115,7 +115,6 @@ class SpikeSourceArrayVertex(
         "__model_name",
         "__model",
         "__structure",
-        "_spike_times",
         "__n_colour_bits")
 
     #: ID of the recording region used for recording transmitted spikes.
@@ -143,9 +142,6 @@ def __init__(
 
         if spike_times is None:
             spike_times = []
-        self._spike_times = SpynnakerRangedList(
-            n_neurons, spike_times,
-            use_list_as_value=not _is_double_list(spike_times))
 
         time_step = SpynnakerDataView.get_simulation_time_step_us()
 
@@ -210,6 +206,9 @@ def _check_density_double_list(self, spike_times: _DoubleList) -> None:
         counter: Counter = Counter()
         for neuron_id in range(0, self.n_atoms):
             counter.update(spike_times[neuron_id])
+        if len(counter) == 0:
+            logger.warning("SpikeSourceArray all spike times lists are empty")
+            return
         top = counter.most_common(1)
         val, count = top[0]
         if count > TOO_MANY_SPIKES:
@@ -230,65 +229,32 @@ def atoms_shape(self) -> Tuple[int, ...]:
             return self.__structure.calculate_size(self.n_atoms)
         return super().atoms_shape
 
-    def _to_early_spikes_single_list(self, spike_times: _SingleList) -> None:
-        """
-        Checks if there is one or more spike_times before the current time.
-
-        Logs a warning for the first one found
-
-        :param spike_times:
-        """
-        current_time = SpynnakerDataView.get_current_run_time_ms()
-        for spike_time in spike_times:
-            if spike_time < current_time:
-                logger.warning(
-                    "SpikeSourceArray {} has spike_times that are lower than "
-                    "the current time {} For example {} - "
-                    "these will be ignored.",
-                    self, current_time, float(spike_time))
-                return
-
-    def _check_spikes_double_list(self, spike_times: _DoubleList) -> None:
-        """
-        Checks if there is one or more spike_times before the current time.
-
-        Logs a warning for the first one found
-
-        :param spike_times:
-        """
-        current_time = SpynnakerDataView.get_current_run_time_ms()
-        for neuron_id in range(0, self.n_atoms):
-            id_times = spike_times[neuron_id]
-            for id_time in id_times:
-                if id_time < current_time:
-                    logger.warning(
-                        "SpikeSourceArray {} has spike_times that are lower "
-                        "than the current time {} For example {} - "
-                        "these will be ignored.",
-                        self, current_time, float(id_time))
-                    return
-
     def __set_spike_buffer_times(self, spike_times: Spikes) -> None:
         """
         Set the spike source array's buffer spike times.
         """
         time_step = SpynnakerDataView.get_simulation_time_step_us()
-        # warn the user if they are asking for a spike time out of range
-        if _is_double_list(spike_times):
-            self._check_spikes_double_list(spike_times)
-        elif _is_single_list(spike_times):
-            self._to_early_spikes_single_list(spike_times)
-        elif _is_singleton(spike_times):
-            self._to_early_spikes_single_list([spike_times])
-        else:
-            # in case of empty list do not check
-            pass
-        self.send_buffer_times = _send_buffer_times(spike_times, time_step)
+        self.set_send_buffer_times(_send_buffer_times(spike_times, time_step))
         self._check_spike_density(spike_times)
 
     def __read_parameter(self, name: str, selector: Selector) -> Sequence:
         _ = name
-        return self._spike_times.get_values(selector)
+        time_step = SpynnakerDataView.get_simulation_time_step_us()
+        send_buffer_times = self.send_buffer_times
+        if send_buffer_times is None:
+            return []
+        numpy_times = send_buffer_times * time_step / 1000
+        double_list = _is_double_list(numpy_times)
+        if selector or double_list:
+            # Let RangeList do the heavy lifting using 2D spikes
+            spike_times = [times.tolist() for times in numpy_times]
+            range_list: RangedList[float] = (
+                RangedList(self.n_atoms, spike_times,
+                           use_list_as_value=not double_list))
+            return range_list.get_values(selector)
+
+        # A single list is fine
+        return numpy_times.tolist()
 
     @overrides(PopulationApplicationVertex.get_parameter_values)
     def get_parameter_values(
@@ -300,9 +266,26 @@ def get_parameter_values(
     def set_parameter_values(
             self, name: str, value: Spikes, selector: Selector = None) -> None:
         self._check_parameters(name, {"spike_times"})
-        self.__set_spike_buffer_times(value)
-        self._spike_times.set_value_by_selector(
-            selector, value, use_list_as_value=not _is_double_list(value))
+        SpynnakerDataView.set_requires_mapping()
+        if value is None:
+            value = []
+        if selector is None:
+            self.__set_spike_buffer_times(value)
+        else:
+            # get the existing spike times in micro seconds
+            time_step = SpynnakerDataView.get_simulation_time_step_us()
+            send_buffer_times = self.send_buffer_times
+            if send_buffer_times is None:
+                self.__set_spike_buffer_times([])
+                return
+            numpy_times = send_buffer_times * time_step / 1000
+            # Use range list to set based on selector
+            spike_times: RangedList[float] = RangedList(
+                self.n_atoms, numpy_times,
+                use_list_as_value=not _is_double_list(numpy_times))
+            spike_times.set_value_by_selector(
+                selector, value, use_list_as_value=not _is_double_list(value))
+            self.__set_spike_buffer_times(spike_times)
 
     @overrides(PopulationApplicationVertex.get_parameters)
     def get_parameters(self) -> List[str]:

spynnaker_integration_tests/test_spike_source/test_set_spikes.py

@@ -0,0 +1,78 @@
+# Copyright (c) 2017 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
+
+
+class TestSetSpikes(BaseTestCase):
+
+    def do_run(self) -> None:
+        n_neurons = 3
+
+        sim.setup(timestep=1.0)
+
+        pop_1 = sim.Population(n_neurons, sim.IF_curr_exp(), label="pop_1")
+        input_pop = sim.Population(
+            n_neurons, sim.SpikeSourceArray(), label="input")
+        sim.Projection(input_pop, pop_1, sim.OneToOneConnector(),
+                       synapse_type=sim.StaticSynapse(weight=5, delay=1))
+        pop_1.record(["spikes"])
+        input_pop.record("spikes")
+        input_pop.describe()
+
+        sim.run(100)
+        in_spikes = input_pop.spinnaker_get_data("spikes").tolist()
+        out_spikes = pop_1.spinnaker_get_data("spikes").tolist()
+        self.assertEqual(0, len(in_spikes))
+        self.assertEqual(0, len(out_spikes))
+
+        input_pop.set(spike_times=[10, 20, 30])
+        sim.reset()
+        sim.run(100)
+        in_spikes = input_pop.spinnaker_get_data("spikes").tolist()
+        out_spikes = pop_1.spinnaker_get_data("spikes").tolist()
+        self.assertSequenceEqual([
+            [0, 10], [0, 20], [0, 30],
+            [1, 10], [1, 20], [1, 30],
+            [2, 10], [2, 20], [2, 30]], in_spikes)
+        self.assertSequenceEqual([
+            [0, 17], [0, 24], [0, 33],
+            [1, 17], [1, 24], [1, 33],
+            [2, 17], [2, 24], [2, 33]], out_spikes)
+
+        # ... but set spike times here
+        input_pop.set(spike_times=[])
+        spike_times = [[12, 40], [], [23]]
+        assert len(spike_times) == n_neurons
+        for idx in range(n_neurons):
+            input_pop[idx].set(spike_times=spike_times[idx])
+        sim.reset()
+        sim.run(100)
+
+        in_spikes = input_pop.spinnaker_get_data("spikes").tolist()
+        out_spikes = pop_1.spinnaker_get_data("spikes").tolist()
+        self.assertSequenceEqual([[0, 12], [0, 40], [2, 23]], in_spikes)
+        self.assertSequenceEqual([[0, 19], [0, 45], [2, 30]], out_spikes)
+
+        input_pop.set(spike_times=[])
+        with self.assertRaises(NotImplementedError):
+            # requires mapping so hard reset required.
+            sim.run(100)
+
+        sim.end()
+
+    def test_run(self) -> None:
+        self.runsafe(self.do_run)

unittests/model_tests/neuron/test_spike_source/test_spike_source_array_vertex.py

@@ -39,11 +39,63 @@ def test_no_spikes(self) -> None:
             if "no spike" in str(record.msg):
                 found = True
         self.assertTrue(found)
+
+        # None spike_times is converted to an empty list
+        self.assertSequenceEqual(
+            [],  list(v.get_parameter_values("spike_times")))
+
+        v.set_parameter_values("spike_times", [4, 5, 6])
+        self.assertSequenceEqual(
+            [4, 5, 6],  list(v.get_parameter_values("spike_times")))
+        self.assertSequenceEqual(
+            [4, 5, 6],
+            list(v.get_parameter_values("spike_times", selector=3)))
+        self.assertSequenceEqual(
+            [[4, 5, 6], [4, 5, 6]],
+            list(v.get_parameter_values("spike_times", selector=[2, 4])))
+
+        # None is not PyNN but lets check it works anyway
+        v.set_parameter_values("spike_times", None)  # type: ignore[arg-type]
+        self.assertSequenceEqual(
+            [],  list(v.get_parameter_values("spike_times")))
+
+        v.set_parameter_values("spike_times", 7)
+        # Single value converted to a list with 1 value
+        self.assertSequenceEqual(
+            [7],  list(v.get_parameter_values("spike_times")))
+        self.assertSequenceEqual(
+            [7],
+            list(v.get_parameter_values("spike_times", selector=3)))
+        self.assertSequenceEqual(
+            [[7], [7]],
+            list(v.get_parameter_values("spike_times", selector=[2, 4])))
+
         v.set_parameter_values("spike_times", [])
+        self.assertSequenceEqual(
+            [],  list(v.get_parameter_values("spike_times")))
+
         v.set_parameter_values("spike_times", [1, 2, 3], [1, 3])
         self.assertSequenceEqual(
             [[], [1, 2, 3], [], [1, 2, 3], []],
             list(v.get_parameter_values("spike_times")))
+        self.assertSequenceEqual(
+            [[1, 2, 3], []],
+            list(v.get_parameter_values("spike_times", selector=[1, 4])))
+        self.assertSequenceEqual(
+            [1, 2, 3],
+            list(v.get_parameter_values("spike_times", selector=3)))
+
+    def test_double_no_spikes(self) -> None:
+        with LogCapture() as lc:
+            SpikeSourceArrayVertex(
+                n_neurons=5, spike_times=[[], [], [], [], []], label="test",
+                max_atoms_per_core=10, model=SpikeSourceArray(),
+                splitter=None, n_colour_bits=None)
+        found = False
+        for record in lc.records:
+            if "lists are empty" in str(record.msg):
+                found = True
+        self.assertTrue(found)
 
     def test_singleton_list(self) -> None:
         v = SpikeSourceArrayVertex(