7277777398

Committed 20 Dec 2023 03:39PM UTC coverage: 62.895% (+0.1%) from 62.778%

Build # 7277777398

Build Type

Pull #1425

github

Committed by

Christian-B

Commit Message

fix incorrect param passing

Pull Request Pull Request #1425: Overrides check

Run Details

1909 of 4469 branches covered (0.0%)

Branch coverage included in aggregate %.

313 of 315 new or added lines in 65 files covered. (99.37%)

25 existing lines in 17 files now uncovered.

12787 of 18897 relevant lines covered (67.67%)

0.68 hits per line

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

32.61

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

# Copyright (c) 2016 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.typing import NDArray
from pyNN.random import NumpyRNG
from typing import Optional, Sequence, TYPE_CHECKING
from spinn_utilities.overrides import overrides
from pacman.model.graphs.common import Slice
from spinn_front_end_common.utilities.exceptions import ConfigurationException
from .abstract_connector import AbstractConnector
from .abstract_generate_connector_on_host import (
    AbstractGenerateConnectorOnHost)
if TYPE_CHECKING:
    from spynnaker.pyNN.models.neural_projections import SynapseInformation


class SmallWorldConnector(AbstractConnector, AbstractGenerateConnectorOnHost):
    """
    A connector that uses connection statistics based on the Small World
    network connectivity model.

    .. note::
        This is typically used from a population to itself.
    """
    __slots__ = (
        "__allow_self_connections",  # TODO: currently ignored
        "__degree",
        "__mask",
        "__n_connections",
        "__rewiring",
        "__rng")

    def __init__(
            self, degree: float, rewiring: float,
            allow_self_connections: bool = True,
            n_connections: Optional[int] = None,
            rng: Optional[NumpyRNG] = None,
            safe=True, callback=None, verbose=False):
        """
        :param float degree:
            the region length where nodes will be connected locally
        :param float rewiring: the probability of rewiring each edge
        :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 n_connections:
            if specified, the number of efferent synaptic connections per
            neuron
        :type n_connections: int or None
        :param rng:
            Seeded random number generator, or ``None`` to make one when
            needed.
        :type rng: ~pyNN.random.NumpyRNG or None
        :param bool safe:
            If ``True``, 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
        """
        # pylint: disable=too-many-arguments, unused-private-member
        super().__init__(safe, callback, verbose)
        self.__rewiring = rewiring
        self.__degree = degree
        self.__allow_self_connections = allow_self_connections
        self.__mask: Optional[NDArray] = None
        self.__n_connections = 0
        self.__rng = rng or NumpyRNG()

        if n_connections is not None:
            raise NotImplementedError(
                "n_connections is not implemented for"
                " SmallWorldConnector on this platform")

    @overrides(AbstractConnector.set_projection_information)
    def set_projection_information(self, synapse_info: SynapseInformation):
        super().set_projection_information(synapse_info)
        self._set_n_connections(synapse_info)

    def _set_n_connections(self, synapse_info: SynapseInformation):
        """
        :param SynapseInformation synapse_info:
        """
        if self.space is None:
            raise ConfigurationException("a metric space is required")
        # Get the probabilities up-front for now
        # TODO: Work out how this can be done statistically
        # space.distances(...) expects N,3 array in PyNN0.7, but 3,N in PyNN0.8
        pre_positions = synapse_info.pre_population.positions
        post_positions = synapse_info.post_population.positions

        distances = self.space.distances(
            pre_positions, post_positions, False)

        # PyNN 0.8 returns a flattened (C-style) array from space.distances,
        # so the easiest thing to do here is to reshape back to the "expected"
        # PyNN 0.7 shape; otherwise later code gets confusing and difficult
        if len(distances.shape) == 1:
            d = numpy.reshape(distances, (pre_positions.shape[0],
                                          post_positions.shape[0]))
        else:
            d = distances

        self.__mask = (d < self.__degree).astype(float)
        self.__n_connections = int(math.ceil(numpy.sum(self.__mask)))

    @overrides(AbstractConnector.get_delay_maximum)
    def get_delay_maximum(self, synapse_info: SynapseInformation) -> float:
        return self._get_delay_maximum(
            synapse_info.delays, self.__n_connections, synapse_info)

    @overrides(AbstractConnector.get_delay_minimum)
    def get_delay_minimum(self, synapse_info: SynapseInformation) -> float:
        return self._get_delay_minimum(
            synapse_info.delays, self.__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:
        assert self.__mask is not None
        # Break the array into n_post_atoms units
        split_positions = numpy.arange(
            0, synapse_info.n_post_neurons, n_post_atoms)
        split_array = numpy.array_split(self.__mask, split_positions)

        # Sum the 1s in each split row
        sum_rows = [numpy.sum(s, axis=1) for s in split_array]

        # Find the maximum of the rows
        n_connections = max([x for y in sum_rows for x in y])

        if min_delay is None or max_delay is None:
            return n_connections

        return self._get_n_connections_from_pre_vertex_with_delay_maximum(
            synapse_info.delays, self.__n_connections, 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:
        assert self.__mask is not None
        return numpy.amax([
            numpy.sum(self.__mask[:, i]) for i in range(
                synapse_info.n_post_neurons)])

    @overrides(AbstractConnector.get_weight_maximum)
    def get_weight_maximum(self, synapse_info: SynapseInformation) -> float:
        return self._get_weight_maximum(
            synapse_info.weights, self.__n_connections, synapse_info)

    @overrides(AbstractGenerateConnectorOnHost.create_synaptic_block)
    def create_synaptic_block(
            self, post_slices: Sequence[Slice], post_vertex_slice: Slice,
            synapse_type: int, synapse_info: SynapseInformation) -> NDArray:
        if self.__mask is None:
            return numpy.zeros(0, dtype=self.NUMPY_SYNAPSES_DTYPE)
        raster_ids = post_vertex_slice.get_raster_ids()
        ids = numpy.where(self.__mask[:, raster_ids])
        n_connections = len(ids[0])

        block = numpy.zeros(n_connections, dtype=self.NUMPY_SYNAPSES_DTYPE)
        block["source"] = synapse_info.pre_vertex.get_key_ordered_indices(
            ids[0] % synapse_info.n_pre_neurons)
        block["target"] = post_vertex_slice.get_relative_indices(
            raster_ids[ids[1]])
        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

        # Re-wire some connections
        rewired = numpy.where(
            self.__rng.next(n_connections) < self.__rewiring)[0]
        block["target"][rewired] = (
            (self.__rng.next(rewired.size) * (post_vertex_slice.n_atoms - 1)) +
            post_vertex_slice.lo_atom)

        return block

    def __repr__(self):
        return ("SmallWorldConnector"
                f"(degree={self.__degree}, rewiring={self.__rewiring})")

1	# Copyright (c) 2016 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	from __future__ import annotations	1✔
15	import math	1✔
16	import numpy	1✔
17	from numpy.typing import NDArray	1✔
18	from pyNN.random import NumpyRNG	1✔
19	from typing import Optional, Sequence, TYPE_CHECKING	1✔
20	from spinn_utilities.overrides import overrides	1✔
21	from pacman.model.graphs.common import Slice	1✔
22	from spinn_front_end_common.utilities.exceptions import ConfigurationException	1✔
23	from .abstract_connector import AbstractConnector	1✔
24	from .abstract_generate_connector_on_host import (	1✔
25	AbstractGenerateConnectorOnHost)
26	if TYPE_CHECKING:
27	from spynnaker.pyNN.models.neural_projections import SynapseInformation
28
29
30	class SmallWorldConnector(AbstractConnector, AbstractGenerateConnectorOnHost):	1✔
31	"""
32	A connector that uses connection statistics based on the Small World
33	network connectivity model.
34
35	.. note::
36	This is typically used from a population to itself.
37	"""
38	__slots__ = (	1✔
39	"__allow_self_connections", # TODO: currently ignored
40	"__degree",
41	"__mask",
42	"__n_connections",
43	"__rewiring",
44	"__rng")
45
46	def __init__(	1✔
47	self, degree: float, rewiring: float,
48	allow_self_connections: bool = True,
49	n_connections: Optional[int] = None,
50	rng: Optional[NumpyRNG] = None,
51	safe=True, callback=None, verbose=False):
52	"""
53	:param float degree:
54	the region length where nodes will be connected locally
55	:param float rewiring: the probability of rewiring each edge
56	:param bool allow_self_connections:
57	if the connector is used to connect a Population to itself, this
58	flag determines whether a neuron is allowed to connect to itself,
59	or only to other neurons in the Population.
60	:param n_connections:
61	if specified, the number of efferent synaptic connections per
62	neuron
63	:type n_connections: int or None
64	:param rng:
65	Seeded random number generator, or ``None`` to make one when
66	needed.
67	:type rng: ~pyNN.random.NumpyRNG or None
68	:param bool safe:
69	If ``True``, check that weights and delays have valid values.
70	If ``False``, this check is skipped.
71	:param callable callback:
72	if given, a callable that display a progress bar on the terminal.
73
74	.. note::
75	Not supported by sPyNNaker.
76	:param bool verbose:
77	Whether to output extra information about the connectivity to a
78	CSV file
79	"""
80	# pylint: disable=too-many-arguments, unused-private-member
81	super().__init__(safe, callback, verbose)	×
82	self.__rewiring = rewiring	×
83	self.__degree = degree	×
84	self.__allow_self_connections = allow_self_connections	×
85	self.__mask: Optional[NDArray] = None	×
86	self.__n_connections = 0	×
87	self.__rng = rng or NumpyRNG()	×
88
89	if n_connections is not None:	×
90	raise NotImplementedError(
91	"n_connections is not implemented for"
92	" SmallWorldConnector on this platform")
93
94	@overrides(AbstractConnector.set_projection_information)	1✔
95	def set_projection_information(self, synapse_info: SynapseInformation):	1✔
96	super().set_projection_information(synapse_info)	×
97	self._set_n_connections(synapse_info)	×
98
99	def _set_n_connections(self, synapse_info: SynapseInformation):	1✔
100	"""
101	:param SynapseInformation synapse_info:
102	"""
103	if self.space is None:	×
104	raise ConfigurationException("a metric space is required")	×
105	# Get the probabilities up-front for now
106	# TODO: Work out how this can be done statistically
107	# space.distances(...) expects N,3 array in PyNN0.7, but 3,N in PyNN0.8
108	pre_positions = synapse_info.pre_population.positions	×
109	post_positions = synapse_info.post_population.positions	×
110
111	distances = self.space.distances(	×
112	pre_positions, post_positions, False)
113
114	# PyNN 0.8 returns a flattened (C-style) array from space.distances,
115	# so the easiest thing to do here is to reshape back to the "expected"
116	# PyNN 0.7 shape; otherwise later code gets confusing and difficult
117	if len(distances.shape) == 1:	×
118	d = numpy.reshape(distances, (pre_positions.shape[0],	×
119	post_positions.shape[0]))
120	else:
121	d = distances	×
122
123	self.__mask = (d < self.__degree).astype(float)	×
124	self.__n_connections = int(math.ceil(numpy.sum(self.__mask)))	×
125
126	@overrides(AbstractConnector.get_delay_maximum)	1✔
127	def get_delay_maximum(self, synapse_info: SynapseInformation) -> float:	1✔
128	return self._get_delay_maximum(	×
129	synapse_info.delays, self.__n_connections, synapse_info)
130
131	@overrides(AbstractConnector.get_delay_minimum)	1✔
132	def get_delay_minimum(self, synapse_info: SynapseInformation) -> float:	1✔
133	return self._get_delay_minimum(	×
134	synapse_info.delays, self.__n_connections, synapse_info)
135
136	@overrides(AbstractConnector.get_n_connections_from_pre_vertex_maximum)	1✔
137	def get_n_connections_from_pre_vertex_maximum(	1✔
138	self, n_post_atoms: int, synapse_info: SynapseInformation,
139	min_delay: Optional[float] = None,
140	max_delay: Optional[float] = None) -> int:
UNCOV 141	assert self.__mask is not None	×
142	# Break the array into n_post_atoms units
143	split_positions = numpy.arange(	×
144	0, synapse_info.n_post_neurons, n_post_atoms)
145	split_array = numpy.array_split(self.__mask, split_positions)	×
146
147	# Sum the 1s in each split row
148	sum_rows = [numpy.sum(s, axis=1) for s in split_array]	×
149
150	# Find the maximum of the rows
151	n_connections = max([x for y in sum_rows for x in y])	×
152
153	if min_delay is None or max_delay is None:	×
154	return n_connections	×
155
156	return self._get_n_connections_from_pre_vertex_with_delay_maximum(	×
157	synapse_info.delays, self.__n_connections, n_connections,
158	min_delay, max_delay, synapse_info)
159
160	@overrides(AbstractConnector.get_n_connections_to_post_vertex_maximum)	1✔
161	def get_n_connections_to_post_vertex_maximum(	1✔
162	self, synapse_info: SynapseInformation) -> int:
163	assert self.__mask is not None	×
164	return numpy.amax([	×
165	numpy.sum(self.__mask[:, i]) for i in range(
166	synapse_info.n_post_neurons)])
167
168	@overrides(AbstractConnector.get_weight_maximum)	1✔
169	def get_weight_maximum(self, synapse_info: SynapseInformation) -> float:	1✔
170	return self._get_weight_maximum(	×
171	synapse_info.weights, self.__n_connections, synapse_info)
172
173	@overrides(AbstractGenerateConnectorOnHost.create_synaptic_block)	1✔
174	def create_synaptic_block(	1✔
175	self, post_slices: Sequence[Slice], post_vertex_slice: Slice,
176	synapse_type: int, synapse_info: SynapseInformation) -> NDArray:
177	if self.__mask is None:	×
178	return numpy.zeros(0, dtype=self.NUMPY_SYNAPSES_DTYPE)	×
179	raster_ids = post_vertex_slice.get_raster_ids()	×
180	ids = numpy.where(self.__mask[:, raster_ids])	×
181	n_connections = len(ids[0])	×
182
183	block = numpy.zeros(n_connections, dtype=self.NUMPY_SYNAPSES_DTYPE)	×
184	block["source"] = synapse_info.pre_vertex.get_key_ordered_indices(	×
185	ids[0] % synapse_info.n_pre_neurons)
186	block["target"] = post_vertex_slice.get_relative_indices(	×
187	raster_ids[ids[1]])
188	block["weight"] = self._generate_weights(	×
189	block["source"], block["target"], n_connections, post_vertex_slice,
190	synapse_info)
191	block["delay"] = self._generate_delays(	×
192	block["source"], block["target"], n_connections, post_vertex_slice,
193	synapse_info)
194	block["synapse_type"] = synapse_type	×
195
196	# Re-wire some connections
197	rewired = numpy.where(	×
198	self.__rng.next(n_connections) < self.__rewiring)[0]
199	block["target"][rewired] = (	×
200	(self.__rng.next(rewired.size) * (post_vertex_slice.n_atoms - 1)) +
201	post_vertex_slice.lo_atom)
202
203	return block	×
204
205	def __repr__(self):	1✔
206	return ("SmallWorldConnector"	×
207	f"(degree={self.__degree}, rewiring={self.__rewiring})")

SpiNNakerManchester / sPyNNaker / 7277777398

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