6853649273

Committed 13 Nov 2023 05:37PM UTC coverage: 61.313% (+0.006%) from 61.307%

Build # 6853649273

Build Type

push

github

Committed by

Christian-B

Commit Message

accept types collection of int, float as well

Run Details

1934 of 4475 branches covered (0.0%)

Branch coverage included in aggregate %.

9 of 13 new or added lines in 1 file covered. (69.23%)

10 existing lines in 1 file now uncovered.

11745 of 17835 relevant lines covered (65.85%)

0.66 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

73.79

/spynnaker/pyNN/models/neuron/synapse_dynamics/abstract_synapse_dynamics.py

# Copyright (c) 2015 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 logging
import numpy
from pyNN.random import RandomDistribution
from spinn_utilities.abstract_base import (
    AbstractBase, abstractmethod, abstractproperty)
from spinn_utilities.log import FormatAdapter
from spynnaker.pyNN.data import SpynnakerDataView
from spynnaker.pyNN.utilities.constants import POP_TABLE_MAX_ROW_LENGTH
from spynnaker.pyNN.exceptions import InvalidParameterType

logger = FormatAdapter(logging.getLogger(__name__))


class AbstractSynapseDynamics(object, metaclass=AbstractBase):
    """
    How do the dynamics of a synapse interact with the rest of the model.
    """

    __slots__ = ("__delay", "__weight")

    def __init__(self, delay, weight):
        if delay is None:
            if delay is None:
                delay = SpynnakerDataView.get_min_delay()
        self.__check_type(delay, "delay")
        self.__delay = self._round_delay(delay)
        self.__check_type(weight, "weight")
        self.__weight = weight

    def __check_type(self, value, name):
        if isinstance(value, (int, float, str, RandomDistribution)):
            return
        try:
            for x in value:
                if isinstance(x, (int, float)):
                    # assume if first in list colelction is ok all are
                    return
                else:
                    raise TypeError(
                        f"Unexpected collection of type  {type(x)} for {name}"
                        f"Expected types in collection are int and float")
        except TypeError:
            # Ok not a collection
            pass
        raise TypeError(
            f"Unexpected type for {name}: {type(value)}. "
            "Expected types are int, float, str, RandomDistribution "
            "and collections of type int or float")


    @abstractmethod
    def merge(self, synapse_dynamics):
        """
        Merge with the given synapse_dynamics and return the result, or
        error if merge is not possible.

        :param AbstractSynapseDynamics synapse_dynamics:
        :rtype: AbstractSynapseDynamics
        """

    @abstractmethod
    def get_vertex_executable_suffix(self):
        """
        Get the executable suffix for a vertex for this dynamics.

        :rtype: str
        """

    @abstractproperty
    def changes_during_run(self):
        """
        Whether the synapses change during a run.

        :rtype: bool
        """

    @property
    def weight(self):
        """
        The weight of connections.

        :rtype: float
        """
        return self.__weight

    def _round_delay(self, delay):
        """
        Round the delays to multiples of full timesteps.

        (otherwise SDRAM estimation calculations can go wrong)

        :param delay:
        :return: Rounded delay
        """
        if isinstance(delay, RandomDistribution):
            return delay
        if isinstance(delay, str):
            return delay
        new_delay = (
                numpy.rint(numpy.array(delay) *
                           SpynnakerDataView.get_simulation_time_step_per_ms())
                * SpynnakerDataView.get_simulation_time_step_ms())
        if not numpy.allclose(delay, new_delay):
            logger.warning("Rounding up delay in f{} from {} to {}",
                           self, delay, new_delay)
        return new_delay

    @property
    def delay(self):
        """
        The delay of connections.

        :rtype: float
        """
        return self.__delay

    @abstractproperty
    def is_combined_core_capable(self):
        """
        Whether the synapse dynamics can run on a core combined with
        the neuron, or if a separate core is needed.

        :rtype: bool
        """

    def get_value(self, key):
        """
        Get a property.

        :param str key: the name of the property
        :rtype: Any or float or int or list(float) or list(int)
        """
        if hasattr(self, key):
            return getattr(self, key)
        raise InvalidParameterType(
            f"Type {type(self)} does not have parameter {key}")

    def set_value(self, key, value):
        """
        Set a property.

        :param str key: the name of the parameter to change
        :param value: the new value of the parameter to assign
        :type value: Any or float or int or list(float) or list(int)
        """
        if hasattr(self, key):
            setattr(self, key, value)
            SpynnakerDataView.set_requires_mapping()
        else:
            raise InvalidParameterType(
                f"Type {type(self)} does not have parameter {key}")

    def get_delay_maximum(self, connector, synapse_info):
        """
        Get the maximum delay for the synapses.

        :param AbstractConnector connector:
        :param ~numpy.ndarray delays:
        """
        return connector.get_delay_maximum(synapse_info)

    def get_delay_minimum(self, connector, synapse_info):
        """
        Get the minimum delay for the synapses.
        This will support the filtering of the undelayed edge
        from the graph, but requires fixes in the synaptic manager to
        happen first before this can be utilised fully.

        :param AbstractConnector connector: connector
        :param ~numpy.ndarray synapse_info: synapse info
        """
        return connector.get_delay_minimum(synapse_info)

    def get_delay_variance(self, connector, delays, synapse_info):
        """
        Get the variance in delay for the synapses.

        :param AbstractConnector connector:
        :param ~numpy.ndarray delays:
        """
        return connector.get_delay_variance(delays, synapse_info)

    def get_weight_mean(self, connector, synapse_info):
        """
        Get the mean weight for the synapses.

        :param AbstractConnector connector:
        :param ~numpy.ndarray weights:
        """
        return connector.get_weight_mean(synapse_info.weights, synapse_info)

    def get_weight_maximum(self, connector, synapse_info):
        """
        Get the maximum weight for the synapses.

        :param AbstractConnector connector:
        :param ~numpy.ndarray weights:
        """
        return connector.get_weight_maximum(synapse_info)

    def get_weight_variance(self, connector, weights, synapse_info):
        """
        Get the variance in weight for the synapses.

        :param AbstractConnector connector:
        :param ~numpy.ndarray weights:
        """
        return connector.get_weight_variance(weights, synapse_info)

    def get_provenance_data(self, pre_population_label, post_population_label):
        """
        Get the provenance data from this synapse dynamics object.

        :param str pre_population_label:
        :param str post_population_label:
        :rtype:
            iterable(~spinn_front_end_common.utilities.utility_objs.ProvenanceDataItem)
        """
        # pylint: disable=unused-argument
        # TODO: is this a meaningful method any more; if so, what does it do?
        return []

    def get_synapse_id_by_target(self, target):
        """
        Get the index of the synapse type based on the name, or `None`
        if the name is not found.

        :param str target: The name of the synapse
        :rtype: int or None
        """
        # pylint: disable=unused-argument
        return None

    def get_connected_vertices(self, s_info, source_vertex, target_vertex):
        """
        Get the machine vertices that are connected to each other with
        this connector.

        :param SynapseInformation s_info:
            The synapse information of the connection
        :param source_vertex: The source of the spikes
        :type source_vertex:
            ~pacman.model.graphs.application.ApplicationVertex
        :param target_vertex: The target of the spikes
        :type target_vertex:
            ~pacman.model.graphs.application.ApplicationVertex
        :return: A list of tuples of (target machine vertex, list of sources)
        :rtype: list(tuple(~pacman.model.graphs.machine.MachineVertex,
            list(~pacman.model.graphs.AbstractVertex)))
        """
        # By default, just ask the connector
        return s_info.connector.get_connected_vertices(
            s_info, source_vertex, target_vertex)

    @property
    def absolute_max_atoms_per_core(self):
        """
        The absolute maximum number of atoms per core supported by this
        synapse dynamics object.

        :rtype: int
        """
        # By default, we can only support the maximum row length per core
        return POP_TABLE_MAX_ROW_LENGTH

1	# Copyright (c) 2015 The University of Manchester
2	#
3	# Licensed under the Apache License, Version 2.0 (the "License");
4	# you may not use this file except in compliance with the License.
5	# You may obtain a copy of the License at
6	#
7	# https://www.apache.org/licenses/LICENSE-2.0
8	#
9	# Unless required by applicable law or agreed to in writing, software
10	# distributed under the License is distributed on an "AS IS" BASIS,
11	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12	# See the License for the specific language governing permissions and
13	# limitations under the License.
14
15
16	import logging	1✔
17	import numpy	1✔
18	from pyNN.random import RandomDistribution	1✔
19	from spinn_utilities.abstract_base import (	1✔
20	AbstractBase, abstractmethod, abstractproperty)
21	from spinn_utilities.log import FormatAdapter	1✔
22	from spynnaker.pyNN.data import SpynnakerDataView	1✔
23	from spynnaker.pyNN.utilities.constants import POP_TABLE_MAX_ROW_LENGTH	1✔
24	from spynnaker.pyNN.exceptions import InvalidParameterType	1✔
25
26	logger = FormatAdapter(logging.getLogger(__name__))	1✔
27
28
29	class AbstractSynapseDynamics(object, metaclass=AbstractBase):	1✔
30	"""
31	How do the dynamics of a synapse interact with the rest of the model.
32	"""
33
34	__slots__ = ("__delay", "__weight")	1✔
35
36	def __init__(self, delay, weight):	1✔
37	if delay is None:	1✔
38	if delay is None:	1!
39	delay = SpynnakerDataView.get_min_delay()	1✔
40	self.__check_type(delay, "delay")	1✔
41	self.__delay = self._round_delay(delay)	1✔
42	self.__check_type(weight, "weight")	1✔
43	self.__weight = weight	1✔
44
45	def __check_type(self, value, name):	1✔
46	if isinstance(value, (int, float, str, RandomDistribution)):	1✔
47	return	1✔
48	try:	1✔
49	for x in value:	1!
50	if isinstance(x, (int, float)):	1!
51	# assume if first in list colelction is ok all are
52	return	1✔
53	else:
NEW 54	raise TypeError(	×
55	f"Unexpected collection of type {type(x)} for {name}"
56	f"Expected types in collection are int and float")
NEW 57	except TypeError:	×
58	# Ok not a collection
NEW 59	pass	×
NEW 60	raise TypeError(	×
61	f"Unexpected type for {name}: {type(value)}. "
62	"Expected types are int, float, str, RandomDistribution "
63	"and collections of type int or float")
64
65
66	@abstractmethod	1✔
67	def merge(self, synapse_dynamics):	1✔
68	"""
69	Merge with the given synapse_dynamics and return the result, or
70	error if merge is not possible.
71
72	:param AbstractSynapseDynamics synapse_dynamics:
73	:rtype: AbstractSynapseDynamics
74	"""
75
76	@abstractmethod	1✔
77	def get_vertex_executable_suffix(self):	1✔
78	"""
79	Get the executable suffix for a vertex for this dynamics.
80
81	:rtype: str
82	"""
83
84	@abstractproperty	1✔
85	def changes_during_run(self):	1✔
86	"""
87	Whether the synapses change during a run.
88
89	:rtype: bool
90	"""
91
92	@property	1✔
93	def weight(self):	1✔
94	"""
95	The weight of connections.
96
97	:rtype: float
98	"""
99	return self.__weight	1✔
100
101	def _round_delay(self, delay):	1✔
102	"""
103	Round the delays to multiples of full timesteps.
104
105	(otherwise SDRAM estimation calculations can go wrong)
106
107	:param delay:
108	:return: Rounded delay
109	"""
110	if isinstance(delay, RandomDistribution):	1!
UNCOV 111	return delay	×
112	if isinstance(delay, str):	1!
UNCOV 113	return delay	×
114	new_delay = (	1✔
115	numpy.rint(numpy.array(delay) *
116	SpynnakerDataView.get_simulation_time_step_per_ms())
117	* SpynnakerDataView.get_simulation_time_step_ms())
118	if not numpy.allclose(delay, new_delay):	1!
UNCOV 119	logger.warning("Rounding up delay in f{} from {} to {}",	×
120	self, delay, new_delay)
121	return new_delay	1✔
122
123	@property	1✔
124	def delay(self):	1✔
125	"""
126	The delay of connections.
127
128	:rtype: float
129	"""
130	return self.__delay	1✔
131
132	@abstractproperty	1✔
133	def is_combined_core_capable(self):	1✔
134	"""
135	Whether the synapse dynamics can run on a core combined with
136	the neuron, or if a separate core is needed.
137
138	:rtype: bool
139	"""
140
141	def get_value(self, key):	1✔
142	"""
143	Get a property.
144
145	:param str key: the name of the property
146	:rtype: Any or float or int or list(float) or list(int)
147	"""
148	if hasattr(self, key):	×
149	return getattr(self, key)	×
150	raise InvalidParameterType(	×
151	f"Type {type(self)} does not have parameter {key}")
152
153	def set_value(self, key, value):	1✔
154	"""
155	Set a property.
156
157	:param str key: the name of the parameter to change
158	:param value: the new value of the parameter to assign
159	:type value: Any or float or int or list(float) or list(int)
160	"""
UNCOV 161	if hasattr(self, key):	×
UNCOV 162	setattr(self, key, value)	×
UNCOV 163	SpynnakerDataView.set_requires_mapping()	×
164	else:
UNCOV 165	raise InvalidParameterType(	×
166	f"Type {type(self)} does not have parameter {key}")
167
168	def get_delay_maximum(self, connector, synapse_info):	1✔
169	"""
170	Get the maximum delay for the synapses.
171
172	:param AbstractConnector connector:
173	:param ~numpy.ndarray delays:
174	"""
175	return connector.get_delay_maximum(synapse_info)	1✔
176
177	def get_delay_minimum(self, connector, synapse_info):	1✔
178	"""
179	Get the minimum delay for the synapses.
180	This will support the filtering of the undelayed edge
181	from the graph, but requires fixes in the synaptic manager to
182	happen first before this can be utilised fully.
183
184	:param AbstractConnector connector: connector
185	:param ~numpy.ndarray synapse_info: synapse info
186	"""
UNCOV 187	return connector.get_delay_minimum(synapse_info)	×
188
189	def get_delay_variance(self, connector, delays, synapse_info):	1✔
190	"""
191	Get the variance in delay for the synapses.
192
193	:param AbstractConnector connector:
194	:param ~numpy.ndarray delays:
195	"""
196	return connector.get_delay_variance(delays, synapse_info)	1✔
197
198	def get_weight_mean(self, connector, synapse_info):	1✔
199	"""
200	Get the mean weight for the synapses.
201
202	:param AbstractConnector connector:
203	:param ~numpy.ndarray weights:
204	"""
205	return connector.get_weight_mean(synapse_info.weights, synapse_info)	1✔
206
207	def get_weight_maximum(self, connector, synapse_info):	1✔
208	"""
209	Get the maximum weight for the synapses.
210
211	:param AbstractConnector connector:
212	:param ~numpy.ndarray weights:
213	"""
214	return connector.get_weight_maximum(synapse_info)	1✔
215
216	def get_weight_variance(self, connector, weights, synapse_info):	1✔
217	"""
218	Get the variance in weight for the synapses.
219
220	:param AbstractConnector connector:
221	:param ~numpy.ndarray weights:
222	"""
223	return connector.get_weight_variance(weights, synapse_info)	1✔
224
225	def get_provenance_data(self, pre_population_label, post_population_label):	1✔
226	"""
227	Get the provenance data from this synapse dynamics object.
228
229	:param str pre_population_label:
230	:param str post_population_label:
231	:rtype:
232	iterable(~spinn_front_end_common.utilities.utility_objs.ProvenanceDataItem)
233	"""
234	# pylint: disable=unused-argument
235	# TODO: is this a meaningful method any more; if so, what does it do?
UNCOV 236	return []	×
237
238	def get_synapse_id_by_target(self, target):	1✔
239	"""
240	Get the index of the synapse type based on the name, or `None`
241	if the name is not found.
242
243	:param str target: The name of the synapse
244	:rtype: int or None
245	"""
246	# pylint: disable=unused-argument
UNCOV 247	return None	×
248
249	def get_connected_vertices(self, s_info, source_vertex, target_vertex):	1✔
250	"""
251	Get the machine vertices that are connected to each other with
252	this connector.
253
254	:param SynapseInformation s_info:
255	The synapse information of the connection
256	:param source_vertex: The source of the spikes
257	:type source_vertex:
258	~pacman.model.graphs.application.ApplicationVertex
259	:param target_vertex: The target of the spikes
260	:type target_vertex:
261	~pacman.model.graphs.application.ApplicationVertex
262	:return: A list of tuples of (target machine vertex, list of sources)
263	:rtype: list(tuple(~pacman.model.graphs.machine.MachineVertex,
264	list(~pacman.model.graphs.AbstractVertex)))
265	"""
266	# By default, just ask the connector
267	return s_info.connector.get_connected_vertices(	1✔
268	s_info, source_vertex, target_vertex)
269
270	@property	1✔
271	def absolute_max_atoms_per_core(self):	1✔
272	"""
273	The absolute maximum number of atoms per core supported by this
274	synapse dynamics object.
275
276	:rtype: int
277	"""
278	# By default, we can only support the maximum row length per core
279	return POP_TABLE_MAX_ROW_LENGTH	1✔

SpiNNakerManchester / sPyNNaker / 6853649273

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous