6903208647

Committed 17 Nov 2023 11:08AM UTC coverage: 63.17% (+1.6%) from 61.531%

Build # 6903208647

Build Type

Pull #1342

github

Committed by

Christian-B

Commit Message

spelling exception

Pull Request Pull Request #1342: Type Annotations and Checking

Run Details

1961 of 4484 branches covered (0.0%)

Branch coverage included in aggregate %.

4168 of 5102 new or added lines in 234 files covered. (81.69%)

193 existing lines in 78 files now uncovered.

12745 of 18796 relevant lines covered (67.81%)

0.68 hits per line

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

86.46

/spynnaker/pyNN/models/neural_projections/connectors/index_based_probability_connector.py

# 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.

from __future__ import annotations
import math
import numpy
from numpy import (
    arccos, arcsin, arctan, arctan2, ceil, cos, cosh, exp, fabs, floor, fmod,
    hypot, ldexp, log, log10, modf, power, sin, sinh, sqrt, tan, tanh, maximum,
    minimum, e, pi)
from numpy.typing import NDArray
from pyNN.random import NumpyRNG
from typing import Optional, TYPE_CHECKING
from spinn_utilities.overrides import overrides
from spinn_utilities.safe_eval import SafeEval
from pacman.model.graphs.common import Slice
from spynnaker.pyNN.utilities import utility_calls
from .abstract_connector import AbstractConnector
from .abstract_generate_connector_on_host import (
    AbstractGenerateConnectorOnHost)
if TYPE_CHECKING:
    from spynnaker.pyNN.models.neural_projections import SynapseInformation

# support for arbitrary expression for the indices
_index_expr_context = SafeEval(math, numpy, arccos, arcsin, arctan, arctan2,
                               ceil, cos, cosh, exp, fabs, floor, fmod, hypot,
                               ldexp, log, log10, modf, power, sin, sinh, sqrt,
                               tan, tanh, maximum, minimum, e=e, pi=pi)


class IndexBasedProbabilityConnector(AbstractConnector,
                                     AbstractGenerateConnectorOnHost):
    """
    Make connections using a probability distribution which varies
    dependent upon the indices of the pre- and post-populations.
    """

    __slots = [
        "__allow_self_connections",
        "__index_expression",
        "__probs",
        "__rng"]

    def __init__(
            self, index_expression: str, allow_self_connections: bool = True,
            rng: Optional[NumpyRNG] = None,
            safe=True, callback=None, verbose=False):
        """
        :param str index_expression:
            the right-hand side of a valid python expression for
            probability, involving the indices of the pre- and
            post-populations,
            that can be parsed by `eval()`, that computes a probability
            distribution;
            the indices will be given as variables ``i`` and ``j`` when the
            expression is evaluated.
        :param bool allow_self_connections:
            if the connector is used to connect a Population to itself, this
            flag determines whether a neuron is allowed to connect to itself,
            or only to other neurons in the Population.
        :param rng:
            Seeded random number generator, or ``None`` to make one when
            needed.
        :type rng: ~pyNN.random.NumpyRNG or None
        :param bool safe:
            Whether to check that weights and delays have valid values.
            If ``False``, this check is skipped.
        :param callable callback:
            if given, a callable that display a progress bar on the terminal.

            .. note::
                Not supported by sPyNNaker.
        :param bool verbose:
            Whether to output extra information about the connectivity to a
            CSV file
        """
        super().__init__(safe, callback, verbose)
        self.__rng = rng or NumpyRNG()
        self.__index_expression = index_expression
        self.__allow_self_connections = allow_self_connections
        self.__probs: Optional[NDArray] = None

    def _update_probs_from_index_expression(
            self, synapse_info: SynapseInformation) -> NDArray:
        """
        :param SynapseInformation synapse_info:
        """
        # note: this only needs to be done once
        if self.__probs is None:
            # numpy array of probabilities using the index_expression
            self.__probs = numpy.fromfunction(
                lambda i, j: _index_expr_context.eval(
                    self.__index_expression, i=i, j=j),
                (synapse_info.n_pre_neurons, synapse_info.n_post_neurons))
        return self.__probs

    @overrides(AbstractConnector.get_delay_maximum)
    def get_delay_maximum(self, synapse_info: SynapseInformation) -> float:
        probs = self._update_probs_from_index_expression(synapse_info)
        n_connections = utility_calls.get_probable_maximum_selected(
            synapse_info.n_pre_neurons * synapse_info.n_post_neurons,
            synapse_info.n_pre_neurons * synapse_info.n_post_neurons,
            numpy.amax(probs))
        return self._get_delay_maximum(
            synapse_info.delays, n_connections, synapse_info)

    @overrides(AbstractConnector.get_delay_minimum)
    def get_delay_minimum(self, synapse_info: SynapseInformation) -> float:
        probs = self._update_probs_from_index_expression(synapse_info)
        n_connections = utility_calls.get_probable_minimum_selected(
            synapse_info.n_pre_neurons * synapse_info.n_post_neurons,
            synapse_info.n_pre_neurons * synapse_info.n_post_neurons,
            numpy.amax(probs))
        return self._get_delay_minimum(
            synapse_info.delays, n_connections, synapse_info)

    @overrides(AbstractConnector.get_n_connections_from_pre_vertex_maximum)
    def get_n_connections_from_pre_vertex_maximum(
            self, n_post_atoms: int, synapse_info: SynapseInformation,
            min_delay: Optional[float] = None,
            max_delay: Optional[float] = None) -> int:
        probs = self._update_probs_from_index_expression(synapse_info)
        n_connections = utility_calls.get_probable_maximum_selected(
            synapse_info.n_pre_neurons * synapse_info.n_post_neurons,
            n_post_atoms, numpy.amax(probs))

        if min_delay is None or max_delay is None:
            return int(math.ceil(n_connections))

        return self._get_n_connections_from_pre_vertex_with_delay_maximum(
            synapse_info.delays,
            synapse_info.n_pre_neurons * synapse_info.n_post_neurons,
            n_connections, min_delay, max_delay, synapse_info)

    @overrides(AbstractConnector.get_n_connections_to_post_vertex_maximum)
    def get_n_connections_to_post_vertex_maximum(
            self, synapse_info: SynapseInformation) -> int:
        probs = self._update_probs_from_index_expression(synapse_info)
        return utility_calls.get_probable_maximum_selected(
            synapse_info.n_pre_neurons * synapse_info.n_post_neurons,
            synapse_info.n_pre_neurons, numpy.amax(probs))

    @overrides(AbstractConnector.get_weight_maximum)
    def get_weight_maximum(
            self, synapse_info: SynapseInformation) -> float:
        probs = self._update_probs_from_index_expression(synapse_info)
        n_connections = utility_calls.get_probable_maximum_selected(
            synapse_info.n_pre_neurons * synapse_info.n_post_neurons,
            synapse_info.n_pre_neurons * synapse_info.n_post_neurons,
            numpy.amax(probs))
        return self._get_weight_maximum(
            synapse_info.weights, n_connections, synapse_info)

    @overrides(AbstractGenerateConnectorOnHost.create_synaptic_block)
    def create_synaptic_block(
            self, post_slices, post_vertex_slice: Slice, synapse_type: int,
            synapse_info: SynapseInformation) -> NDArray:
        # setup probs here
        probs = self._update_probs_from_index_expression(synapse_info)

        probs = probs[:, post_vertex_slice.as_slice].reshape(-1)

        n_items = synapse_info.n_pre_neurons * post_vertex_slice.n_atoms
        items = self.__rng.next(n_items)

        # If self connections are not allowed, remove the possibility of self
        # connections by setting the probability to a value of infinity
        if not self.__allow_self_connections:
            items[0:n_items:post_vertex_slice.n_atoms + 1] = numpy.inf

        present = items < probs
        ids = numpy.where(present)[0]
        n_connections = numpy.sum(present)

        block = numpy.zeros(
            n_connections, dtype=AbstractConnector.NUMPY_SYNAPSES_DTYPE)
        block["source"] = ids / post_vertex_slice.n_atoms
        block["target"] = (
            (ids % post_vertex_slice.n_atoms) + post_vertex_slice.lo_atom)
        block["weight"] = self._generate_weights(
            block["source"], block["target"], n_connections, post_vertex_slice,
            synapse_info)
        block["delay"] = self._generate_delays(
            block["source"], block["target"], n_connections, post_vertex_slice,
            synapse_info)
        block["synapse_type"] = synapse_type
        return block

    def __repr__(self):
        return f"IndexBasedProbabilityConnector({self.__index_expression})"

    @property
    def allow_self_connections(self) -> bool:
        """
        When the connector is used to connect a Population to itself, this
        flag determines whether a neuron is allowed to connect to itself,
        or only to other neurons in the Population.

        :rtype: bool
        """
        return self.__allow_self_connections

    @allow_self_connections.setter
    def allow_self_connections(self, new_value: bool):
        self.__allow_self_connections = new_value

    @property
    def index_expression(self) -> str:
        """
        The right-hand side of a valid python expression for probability,
        involving the indices of the pre- and post-populations, that can
        be parsed by `eval()`, that computes a probability distribution.

        :rtype: str
        """
        return self.__index_expression

    @index_expression.setter
    def index_expression(self, new_value: str):
        if self.__probs is None:
            raise ValueError("connectivity matrix already fixed")
        self.__index_expression = new_value

1	# Copyright (c) 2017 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	from __future__ import annotations	1✔
16	import math	1✔
17	import numpy	1✔
18	from numpy import (	1✔
19	arccos, arcsin, arctan, arctan2, ceil, cos, cosh, exp, fabs, floor, fmod,
20	hypot, ldexp, log, log10, modf, power, sin, sinh, sqrt, tan, tanh, maximum,
21	minimum, e, pi)
22	from numpy.typing import NDArray	1✔
23	from pyNN.random import NumpyRNG	1✔
24	from typing import Optional, TYPE_CHECKING	1✔
25	from spinn_utilities.overrides import overrides	1✔
26	from spinn_utilities.safe_eval import SafeEval	1✔
27	from pacman.model.graphs.common import Slice	1✔
28	from spynnaker.pyNN.utilities import utility_calls	1✔
29	from .abstract_connector import AbstractConnector	1✔
30	from .abstract_generate_connector_on_host import (	1✔
31	AbstractGenerateConnectorOnHost)
32	if TYPE_CHECKING:
33	from spynnaker.pyNN.models.neural_projections import SynapseInformation
34
35	# support for arbitrary expression for the indices
36	_index_expr_context = SafeEval(math, numpy, arccos, arcsin, arctan, arctan2,	1✔
37	ceil, cos, cosh, exp, fabs, floor, fmod, hypot,
38	ldexp, log, log10, modf, power, sin, sinh, sqrt,
39	tan, tanh, maximum, minimum, e=e, pi=pi)
40
41
42	class IndexBasedProbabilityConnector(AbstractConnector,	1✔
43	AbstractGenerateConnectorOnHost):
44	"""
45	Make connections using a probability distribution which varies
46	dependent upon the indices of the pre- and post-populations.
47	"""
48
49	__slots = [	1✔
50	"__allow_self_connections",
51	"__index_expression",
52	"__probs",
53	"__rng"]
54
55	def __init__(	1✔
56	self, index_expression: str, allow_self_connections: bool = True,
57	rng: Optional[NumpyRNG] = None,
58	safe=True, callback=None, verbose=False):
59	"""
60	:param str index_expression:
61	the right-hand side of a valid python expression for
62	probability, involving the indices of the pre- and
63	post-populations,
64	that can be parsed by `eval()`, that computes a probability
65	distribution;
66	the indices will be given as variables ``i`` and ``j`` when the
67	expression is evaluated.
68	:param bool allow_self_connections:
69	if the connector is used to connect a Population to itself, this
70	flag determines whether a neuron is allowed to connect to itself,
71	or only to other neurons in the Population.
72	:param rng:
73	Seeded random number generator, or ``None`` to make one when
74	needed.
75	:type rng: ~pyNN.random.NumpyRNG or None
76	:param bool safe:
77	Whether to check that weights and delays have valid values.
78	If ``False``, this check is skipped.
79	:param callable callback:
80	if given, a callable that display a progress bar on the terminal.
81
82	.. note::
83	Not supported by sPyNNaker.
84	:param bool verbose:
85	Whether to output extra information about the connectivity to a
86	CSV file
87	"""
88	super().__init__(safe, callback, verbose)	1✔
89	self.__rng = rng or NumpyRNG()	1✔
90	self.__index_expression = index_expression	1✔
91	self.__allow_self_connections = allow_self_connections	1✔
92	self.__probs: Optional[NDArray] = None	1✔
93
94	def _update_probs_from_index_expression(	1✔
95	self, synapse_info: SynapseInformation) -> NDArray:
96	"""
97	:param SynapseInformation synapse_info:
98	"""
99	# note: this only needs to be done once
100	if self.__probs is None:	1✔
101	# numpy array of probabilities using the index_expression
102	self.__probs = numpy.fromfunction(	1✔
103	lambda i, j: _index_expr_context.eval(
104	self.__index_expression, i=i, j=j),
105	(synapse_info.n_pre_neurons, synapse_info.n_post_neurons))
106	return self.__probs	1✔
107
108	@overrides(AbstractConnector.get_delay_maximum)	1✔
109	def get_delay_maximum(self, synapse_info: SynapseInformation) -> float:	1✔
110	probs = self._update_probs_from_index_expression(synapse_info)	1✔
111	n_connections = utility_calls.get_probable_maximum_selected(	1✔
112	synapse_info.n_pre_neurons * synapse_info.n_post_neurons,
113	synapse_info.n_pre_neurons * synapse_info.n_post_neurons,
114	numpy.amax(probs))
115	return self._get_delay_maximum(	1✔
116	synapse_info.delays, n_connections, synapse_info)
117
118	@overrides(AbstractConnector.get_delay_minimum)	1✔
119	def get_delay_minimum(self, synapse_info: SynapseInformation) -> float:	1✔
NEW 120	probs = self._update_probs_from_index_expression(synapse_info)	×
UNCOV 121	n_connections = utility_calls.get_probable_minimum_selected(	×
122	synapse_info.n_pre_neurons * synapse_info.n_post_neurons,
123	synapse_info.n_pre_neurons * synapse_info.n_post_neurons,
124	numpy.amax(probs))
125	return self._get_delay_minimum(	×
126	synapse_info.delays, n_connections, synapse_info)
127
128	@overrides(AbstractConnector.get_n_connections_from_pre_vertex_maximum)	1✔
129	def get_n_connections_from_pre_vertex_maximum(	1✔
130	self, n_post_atoms: int, synapse_info: SynapseInformation,
131	min_delay: Optional[float] = None,
132	max_delay: Optional[float] = None) -> int:
133	probs = self._update_probs_from_index_expression(synapse_info)	1✔
134	n_connections = utility_calls.get_probable_maximum_selected(	1✔
135	synapse_info.n_pre_neurons * synapse_info.n_post_neurons,
136	n_post_atoms, numpy.amax(probs))
137
138	if min_delay is None or max_delay is None:	1!
UNCOV 139	return int(math.ceil(n_connections))	×
140
141	return self._get_n_connections_from_pre_vertex_with_delay_maximum(	1✔
142	synapse_info.delays,
143	synapse_info.n_pre_neurons * synapse_info.n_post_neurons,
144	n_connections, min_delay, max_delay, synapse_info)
145
146	@overrides(AbstractConnector.get_n_connections_to_post_vertex_maximum)	1✔
147	def get_n_connections_to_post_vertex_maximum(	1✔
148	self, synapse_info: SynapseInformation) -> int:
149	probs = self._update_probs_from_index_expression(synapse_info)	1✔
150	return utility_calls.get_probable_maximum_selected(	1✔
151	synapse_info.n_pre_neurons * synapse_info.n_post_neurons,
152	synapse_info.n_pre_neurons, numpy.amax(probs))
153
154	@overrides(AbstractConnector.get_weight_maximum)	1✔
155	def get_weight_maximum(	1✔
156	self, synapse_info: SynapseInformation) -> float:
157	probs = self._update_probs_from_index_expression(synapse_info)	1✔
158	n_connections = utility_calls.get_probable_maximum_selected(	1✔
159	synapse_info.n_pre_neurons * synapse_info.n_post_neurons,
160	synapse_info.n_pre_neurons * synapse_info.n_post_neurons,
161	numpy.amax(probs))
162	return self._get_weight_maximum(	1✔
163	synapse_info.weights, n_connections, synapse_info)
164
165	@overrides(AbstractGenerateConnectorOnHost.create_synaptic_block)	1✔
166	def create_synaptic_block(	1✔
167	self, post_slices, post_vertex_slice: Slice, synapse_type: int,
168	synapse_info: SynapseInformation) -> NDArray:
169	# setup probs here
170	probs = self._update_probs_from_index_expression(synapse_info)	1✔
171
172	probs = probs[:, post_vertex_slice.as_slice].reshape(-1)	1✔
173
174	n_items = synapse_info.n_pre_neurons * post_vertex_slice.n_atoms	1✔
175	items = self.__rng.next(n_items)	1✔
176
177	# If self connections are not allowed, remove the possibility of self
178	# connections by setting the probability to a value of infinity
179	if not self.__allow_self_connections:	1✔
180	items[0:n_items:post_vertex_slice.n_atoms + 1] = numpy.inf	1✔
181
182	present = items < probs	1✔
183	ids = numpy.where(present)[0]	1✔
184	n_connections = numpy.sum(present)	1✔
185
186	block = numpy.zeros(	1✔
187	n_connections, dtype=AbstractConnector.NUMPY_SYNAPSES_DTYPE)
188	block["source"] = ids / post_vertex_slice.n_atoms	1✔
189	block["target"] = (	1✔
190	(ids % post_vertex_slice.n_atoms) + post_vertex_slice.lo_atom)
191	block["weight"] = self._generate_weights(	1✔
192	block["source"], block["target"], n_connections, post_vertex_slice,
193	synapse_info)
194	block["delay"] = self._generate_delays(	1✔
195	block["source"], block["target"], n_connections, post_vertex_slice,
196	synapse_info)
197	block["synapse_type"] = synapse_type	1✔
198	return block	1✔
199
200	def __repr__(self):	1✔
201	return f"IndexBasedProbabilityConnector({self.__index_expression})"	1✔
202
203	@property	1✔
204	def allow_self_connections(self) -> bool:	1✔
205	"""
206	When the connector is used to connect a Population to itself, this
207	flag determines whether a neuron is allowed to connect to itself,
208	or only to other neurons in the Population.
209
210	:rtype: bool
211	"""
212	return self.__allow_self_connections	×
213
214	@allow_self_connections.setter	1✔
215	def allow_self_connections(self, new_value: bool):	1✔
216	self.__allow_self_connections = new_value	×
217
218	@property	1✔
219	def index_expression(self) -> str:	1✔
220	"""
221	The right-hand side of a valid python expression for probability,
222	involving the indices of the pre- and post-populations, that can
223	be parsed by `eval()`, that computes a probability distribution.
224
225	:rtype: str
226	"""
227	return self.__index_expression	×
228
229	@index_expression.setter	1✔
230	def index_expression(self, new_value: str):	1✔
NEW 231	if self.__probs is None:	×
NEW 232	raise ValueError("connectivity matrix already fixed")	×
UNCOV 233	self.__index_expression = new_value	×

SpiNNakerManchester / sPyNNaker / 6903208647

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