8143408510

Committed 04 Mar 2024 04:14PM UTC coverage: 69.531% (+1.5%) from 67.984%

Build # 8143408510

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1437 from SpiNNakerManchester/pylint_default

Pylint default

Run Details

7334 of 10026 branches covered (73.15%)

Branch coverage included in aggregate %.

352 of 398 new or added lines in 130 files covered. (88.44%)

105 existing lines in 35 files now uncovered.

12791 of 18918 relevant lines covered (67.61%)

0.68 hits per line

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

40.23

/spynnaker/pyNN/models/neuron/local_only/local_only_convolution.py

# Copyright (c) 2021 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
from math import ceil
from typing import (
    Dict, Iterable, List, cast, TYPE_CHECKING)

import numpy
from numpy import floating, uint32
from numpy.typing import NDArray

from spinn_utilities.overrides import overrides

from pacman.model.graphs.application import ApplicationVertex

from spinn_front_end_common.interface.ds import (
    DataType, DataSpecificationGenerator)
from spinn_front_end_common.utilities.constants import (
    BYTES_PER_SHORT, BYTES_PER_WORD)

from spynnaker.pyNN.exceptions import SynapticConfigurationException
from spynnaker.pyNN.models.neural_projections.connectors import (
    ConvolutionConnector, AbstractConnector)
from spynnaker.pyNN.models.neuron.synapse_dynamics import (
    AbstractSupportsSignedWeights)
from spynnaker.pyNN.models.common import PopulationApplicationVertex
from spynnaker.pyNN.types import Weight_Delay_In_Types
from spynnaker.pyNN.models.common.local_only_2d_common import (
    get_div_const, get_rinfo_for_spike_source, get_sources_for_target,
    BITS_PER_SHORT, N_COLOUR_BITS_BITS, KEY_INFO_SIZE,
    get_first_and_last_slice, Source)
from .abstract_local_only import AbstractLocalOnly

if TYPE_CHECKING:
    from spynnaker.pyNN.models.neuron.abstract_population_vertex import (
        AbstractPopulationVertex)
    from spynnaker.pyNN.models.projection import Projection
    from spynnaker.pyNN.models.neuron import (
        PopulationMachineLocalOnlyCombinedVertex)
    from spynnaker.pyNN.models.neuron.synapse_dynamics import (
        AbstractSynapseDynamics)


#: Size of convolution config main bytes
CONV_CONFIG_SIZE = (6 * BYTES_PER_SHORT) + (4 * BYTES_PER_WORD)

#: Size of source information
SOURCE_INFO_SIZE = KEY_INFO_SIZE + (6 * BYTES_PER_WORD)


class LocalOnlyConvolution(AbstractLocalOnly, AbstractSupportsSignedWeights):
    """
    A convolution synapse dynamics that can process spikes with only DTCM.
    """

    __slots__ = (
        "__cached_sources",
    )

    def __init__(self, delay: Weight_Delay_In_Types = None):

        """
        :param float delay:
            The delay used in the connection; by default 1 time step
        """
        super().__init__(delay)

        # Store the sources to avoid recalculation
        self.__cached_sources: Dict[ApplicationVertex, Dict[
                PopulationApplicationVertex, List[Source]]] = dict()

    @property
    def _delay(self) -> float:
        # Guaranteed by check in init
        return cast(float, self.delay)

    @overrides(AbstractLocalOnly.merge)
    def merge(self, synapse_dynamics: AbstractSynapseDynamics
              ) -> LocalOnlyConvolution:
        if not isinstance(synapse_dynamics, LocalOnlyConvolution):
            raise SynapticConfigurationException(
                "All targets of this Population must have a synapse_type of"
                " Convolution")
        return synapse_dynamics

    @overrides(AbstractLocalOnly.get_vertex_executable_suffix)
    def get_vertex_executable_suffix(self) -> str:
        return "_conv"

    @property
    @overrides(AbstractLocalOnly.changes_during_run)
    def changes_during_run(self) -> bool:
        return False

    @overrides(AbstractLocalOnly.get_parameters_usage_in_bytes)
    def get_parameters_usage_in_bytes(
            self, n_atoms: int,
            incoming_projections: Iterable[Projection]) -> int:
        # pylint: disable=protected-access
        n_bytes = 0
        kernel_bytes = 0
        connectors_seen = set()
        edges_seen = set()
        for incoming in incoming_projections:
            s_info = incoming._synapse_information
            if not isinstance(s_info.connector, ConvolutionConnector):
                raise SynapticConfigurationException(
                    "Only ConvolutionConnector can be used with a synapse type"
                    " of Convolution")
            app_edge = incoming._projection_edge
            if app_edge not in edges_seen:
                edges_seen.add(app_edge)
                n_bytes += SOURCE_INFO_SIZE
            if s_info.connector not in connectors_seen:
                connectors_seen.add(s_info.connector)
                kernel_bytes += s_info.connector.kernel_n_bytes
            n_bytes += s_info.connector.parameters_n_bytes

        if kernel_bytes % BYTES_PER_WORD != 0:
            kernel_bytes += BYTES_PER_SHORT

        return CONV_CONFIG_SIZE + n_bytes + kernel_bytes

    @overrides(AbstractLocalOnly.write_parameters)
    def write_parameters(
            self, spec: DataSpecificationGenerator, region: int,
            machine_vertex: PopulationMachineLocalOnlyCombinedVertex,
            weight_scales: NDArray[floating]):
        # pylint: disable=unexpected-keyword-arg, protected-access

        # Get incoming sources for this vertex
        app_vertex = machine_vertex._pop_vertex
        sources = self.__get_sources_for_target(app_vertex)

        size = self.get_parameters_usage_in_bytes(
            machine_vertex.vertex_slice.n_atoms,
            app_vertex.incoming_projections)
        spec.reserve_memory_region(region, size, label="LocalOnlyConvolution")
        spec.switch_write_focus(region)

        # Get spec for each incoming source
        connector_weight_index: Dict[AbstractConnector, int] = dict()
        next_weight_index: int = 0
        source_data = list()
        connector_data: List[NDArray[uint32]] = list()
        weight_data = list()
        for pre_vertex, source_infos in sources.items():

            # Add connectors as needed
            first_conn_index = len(connector_data)
            for source in source_infos:
                # pylint: disable=protected-access
                conn = cast(
                    ConvolutionConnector,
                    source.projection._synapse_information.connector)
                app_edge = source.projection._projection_edge

                # Work out whether the connector needs a new weight index
                if conn in connector_weight_index:
                    weight_index = connector_weight_index[conn]
                else:
                    weight_index = next_weight_index
                    connector_weight_index[conn] = weight_index
                    next_weight_index += conn.kernel_n_weights
                    weight_data.append(conn.get_encoded_kernel_weights(
                        app_edge, weight_scales))

                connector_data.append(conn.get_local_only_data(
                    app_edge, source.local_delay, source.delay_stage,
                    weight_index))

            # Get the source routing information
            r_info, core_mask, mask_shift = get_rinfo_for_spike_source(
                pre_vertex)

            # Get the width / height per core / last_core
            first_slice, last_slice = get_first_and_last_slice(pre_vertex)
            width_per_core = first_slice.shape[0]
            height_per_core = first_slice.shape[1]
            width_on_last_core = last_slice.shape[0]
            height_on_last_core = last_slice.shape[1]

            # Get cores per width / height
            pre_shape = list(pre_vertex.atoms_shape)
            cores_per_width = int(ceil(pre_shape[0] / width_per_core))
            cores_per_height = int(ceil(pre_shape[1] / height_per_core))

            # Add the key and mask...
            source_data.extend([r_info.key, r_info.mask])
            # ... start connector index, n_colour_bits, count of connectors ...
            source_data.append(
                (len(source_infos) << BITS_PER_SHORT) +
                (pre_vertex.n_colour_bits <<
                 (BITS_PER_SHORT - N_COLOUR_BITS_BITS)) +
                first_conn_index)
            # ... core mask, mask shift ...
            source_data.append((mask_shift << BITS_PER_SHORT) + core_mask)
            # ... height / width per core ...
            source_data.append(
                (width_per_core << BITS_PER_SHORT) + height_per_core)
            # ... height / width last core ...
            source_data.append(
                (width_on_last_core << BITS_PER_SHORT) + height_on_last_core)
            # ... cores per height / width ...
            source_data.append(
                (cores_per_width << BITS_PER_SHORT) + cores_per_height)
            # ... 1 / width per core ...
            source_data.append(get_div_const(width_per_core))
            # ... 1 / width last core ...
            source_data.append(get_div_const(width_on_last_core))
            # ... 1 / cores_per_width
            source_data.append(get_div_const(cores_per_width))

        if next_weight_index % 2 != 0:
            weight_data.append(numpy.array([0], dtype="int16"))

        # Write the common spec
        post_slice = machine_vertex.vertex_slice
        post_start = numpy.array(post_slice.start)
        post_shape = numpy.array(post_slice.shape)
        post_end = (post_start + post_shape) - 1
        spec.write_value(post_start[1], data_type=DataType.INT16)
        spec.write_value(post_start[0], data_type=DataType.INT16)
        spec.write_value(post_end[1], data_type=DataType.INT16)
        spec.write_value(post_end[0], data_type=DataType.INT16)
        spec.write_value(post_shape[1], data_type=DataType.INT16)
        spec.write_value(post_shape[0], data_type=DataType.INT16)
        spec.write_value(len(sources))
        spec.write_value(len(connector_data))
        spec.write_value(next_weight_index)

        # Write the data
        # pylint: disable=unexpected-keyword-arg
        spec.write_array(numpy.array(source_data, dtype="uint32"))
        spec.write_array(numpy.concatenate(connector_data, dtype="uint32"))
        spec.write_array(
            numpy.concatenate(weight_data, dtype="int16").view("uint32"))

    def __get_sources_for_target(
            self, app_vertex: AbstractPopulationVertex) -> Dict[
                PopulationApplicationVertex, List[Source]]:
        """
        Get all the application vertex sources that will hit the given
        application vertex.

        :param AbstractPopulationVertex app_vertex: The vertex being targeted
        :return:
            A dict of source PopulationApplicationVertex to list of source
            information
        :rtype: dict(PopulationApplicationVertex, list(Source))
        """
        sources = self.__cached_sources.get(app_vertex)
        if sources is None:
            sources = get_sources_for_target(app_vertex)
            self.__cached_sources[app_vertex] = sources
        return sources

    @staticmethod
    def __connector(projection: Projection) -> ConvolutionConnector:
        # pylint: disable=protected-access
        return cast(ConvolutionConnector,
                    projection._synapse_information.connector)

    @staticmethod
    def __get_synapse_type(proj: Projection, target: str) -> int:
        edge = proj._projection_edge  # pylint: disable=protected-access
        synapse_type = edge.post_vertex.get_synapse_id_by_target(target)
        # Checked during connection validation, assumed constant
        assert synapse_type is not None
        return synapse_type

    @overrides(AbstractSupportsSignedWeights.get_positive_synapse_index)
    def get_positive_synapse_index(
            self, incoming_projection: Projection) -> int:
        return self.__get_synapse_type(
            incoming_projection,
            self.__connector(incoming_projection).positive_receptor_type)

    @overrides(AbstractSupportsSignedWeights.get_negative_synapse_index)
    def get_negative_synapse_index(
            self, incoming_projection: Projection) -> int:
        return self.__get_synapse_type(
            incoming_projection,
            self.__connector(incoming_projection).negative_receptor_type)

    @overrides(AbstractSupportsSignedWeights.get_maximum_positive_weight)
    def get_maximum_positive_weight(
            self, incoming_projection: Projection) -> float:
        conn = self.__connector(incoming_projection)
        # We know the connector doesn't care about the argument
        # conn.kernel_weights known to be an array of floats
        max_weight = cast(float, numpy.amax(conn.kernel_weights))
        return max_weight if max_weight > 0 else 0

    @overrides(AbstractSupportsSignedWeights.get_minimum_negative_weight)
    def get_minimum_negative_weight(
            self, incoming_projection: Projection) -> float:
        conn = self.__connector(incoming_projection)
        # This is different because the connector happens to support this
        # conn.kernel_weights known to be an array of floats
        min_weight = cast(float, numpy.amin(conn.kernel_weights))
        return min_weight if min_weight < 0 else 0

    @overrides(AbstractSupportsSignedWeights.get_mean_positive_weight)
    def get_mean_positive_weight(
            self, incoming_projection: Projection) -> float:
        conn = self.__connector(incoming_projection)
        pos_weights = conn.kernel_weights[conn.kernel_weights > 0]
        if len(pos_weights) == 0:
            return 0
        return numpy.mean(pos_weights)

    @overrides(AbstractSupportsSignedWeights.get_mean_negative_weight)
    def get_mean_negative_weight(
            self, incoming_projection: Projection) -> float:
        conn = self.__connector(incoming_projection)
        neg_weights = conn.kernel_weights[conn.kernel_weights < 0]
        if len(neg_weights) == 0:
            return 0
        return numpy.mean(neg_weights)

    @overrides(AbstractSupportsSignedWeights.get_variance_positive_weight)
    def get_variance_positive_weight(
            self, incoming_projection: Projection) -> float:
        conn = self.__connector(incoming_projection)
        pos_weights = conn.kernel_weights[conn.kernel_weights > 0]
        if len(pos_weights) == 0:
            return 0
        return numpy.var(pos_weights)

    @overrides(AbstractSupportsSignedWeights.get_variance_negative_weight)
    def get_variance_negative_weight(
            self, incoming_projection: Projection) -> float:
        conn = self.__connector(incoming_projection)
        neg_weights = conn.kernel_weights[conn.kernel_weights < 0]
        if len(neg_weights) == 0:
            return 0
        return numpy.var(neg_weights)

1	# Copyright (c) 2021 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	from math import ceil	1✔
16	from typing import (	1✔
17	Dict, Iterable, List, cast, TYPE_CHECKING)
18
19	import numpy	1✔
20	from numpy import floating, uint32	1✔
21	from numpy.typing import NDArray	1✔
22
23	from spinn_utilities.overrides import overrides	1✔
24
25	from pacman.model.graphs.application import ApplicationVertex	1✔
26
27	from spinn_front_end_common.interface.ds import (	1✔
28	DataType, DataSpecificationGenerator)
29	from spinn_front_end_common.utilities.constants import (	1✔
30	BYTES_PER_SHORT, BYTES_PER_WORD)
31
32	from spynnaker.pyNN.exceptions import SynapticConfigurationException	1✔
33	from spynnaker.pyNN.models.neural_projections.connectors import (	1✔
34	ConvolutionConnector, AbstractConnector)
35	from spynnaker.pyNN.models.neuron.synapse_dynamics import (	1✔
36	AbstractSupportsSignedWeights)
37	from spynnaker.pyNN.models.common import PopulationApplicationVertex	1✔
38	from spynnaker.pyNN.types import Weight_Delay_In_Types	1✔
39	from spynnaker.pyNN.models.common.local_only_2d_common import (	1✔
40	get_div_const, get_rinfo_for_spike_source, get_sources_for_target,
41	BITS_PER_SHORT, N_COLOUR_BITS_BITS, KEY_INFO_SIZE,
42	get_first_and_last_slice, Source)
43	from .abstract_local_only import AbstractLocalOnly	1✔
44
45	if TYPE_CHECKING:
46	from spynnaker.pyNN.models.neuron.abstract_population_vertex import (
47	AbstractPopulationVertex)
48	from spynnaker.pyNN.models.projection import Projection
49	from spynnaker.pyNN.models.neuron import (
50	PopulationMachineLocalOnlyCombinedVertex)
51	from spynnaker.pyNN.models.neuron.synapse_dynamics import (
52	AbstractSynapseDynamics)
53
54
55	#: Size of convolution config main bytes
56	CONV_CONFIG_SIZE = (6 * BYTES_PER_SHORT) + (4 * BYTES_PER_WORD)	1✔
57
58	#: Size of source information
59	SOURCE_INFO_SIZE = KEY_INFO_SIZE + (6 * BYTES_PER_WORD)	1✔
60
61
62	class LocalOnlyConvolution(AbstractLocalOnly, AbstractSupportsSignedWeights):	1✔
63	"""
64	A convolution synapse dynamics that can process spikes with only DTCM.
65	"""
66
67	__slots__ = (	1✔
68	"__cached_sources",
69	)
70
71	def __init__(self, delay: Weight_Delay_In_Types = None):	1✔
72
73	"""
74	:param float delay:
75	The delay used in the connection; by default 1 time step
76	"""
77	super().__init__(delay)	×
78
79	# Store the sources to avoid recalculation
80	self.__cached_sources: Dict[ApplicationVertex, Dict[	×
81	PopulationApplicationVertex, List[Source]]] = dict()
82
83	@property	1✔
84	def _delay(self) -> float:	1✔
85	# Guaranteed by check in init
86	return cast(float, self.delay)	×
87
88	@overrides(AbstractLocalOnly.merge)	1✔
89	def merge(self, synapse_dynamics: AbstractSynapseDynamics	1✔
90	) -> LocalOnlyConvolution:
91	if not isinstance(synapse_dynamics, LocalOnlyConvolution):	×
92	raise SynapticConfigurationException(	×
93	"All targets of this Population must have a synapse_type of"
94	" Convolution")
95	return synapse_dynamics	×
96
97	@overrides(AbstractLocalOnly.get_vertex_executable_suffix)	1✔
98	def get_vertex_executable_suffix(self) -> str:	1✔
99	return "_conv"	×
100
101	@property	1✔
102	@overrides(AbstractLocalOnly.changes_during_run)	1✔
103	def changes_during_run(self) -> bool:	1✔
104	return False	×
105
106	@overrides(AbstractLocalOnly.get_parameters_usage_in_bytes)	1✔
107	def get_parameters_usage_in_bytes(	1✔
108	self, n_atoms: int,
109	incoming_projections: Iterable[Projection]) -> int:
110	# pylint: disable=protected-access
111	n_bytes = 0	×
112	kernel_bytes = 0	×
113	connectors_seen = set()	×
114	edges_seen = set()	×
115	for incoming in incoming_projections:	×
116	s_info = incoming._synapse_information	×
117	if not isinstance(s_info.connector, ConvolutionConnector):	×
118	raise SynapticConfigurationException(	×
119	"Only ConvolutionConnector can be used with a synapse type"
120	" of Convolution")
121	app_edge = incoming._projection_edge	×
122	if app_edge not in edges_seen:	×
123	edges_seen.add(app_edge)	×
124	n_bytes += SOURCE_INFO_SIZE	×
125	if s_info.connector not in connectors_seen:	×
126	connectors_seen.add(s_info.connector)	×
127	kernel_bytes += s_info.connector.kernel_n_bytes	×
128	n_bytes += s_info.connector.parameters_n_bytes	×
129
130	if kernel_bytes % BYTES_PER_WORD != 0:	×
131	kernel_bytes += BYTES_PER_SHORT	×
132
133	return CONV_CONFIG_SIZE + n_bytes + kernel_bytes	×
134
135	@overrides(AbstractLocalOnly.write_parameters)	1✔
136	def write_parameters(	1✔
137	self, spec: DataSpecificationGenerator, region: int,
138	machine_vertex: PopulationMachineLocalOnlyCombinedVertex,
139	weight_scales: NDArray[floating]):
140	# pylint: disable=unexpected-keyword-arg, protected-access
141
142	# Get incoming sources for this vertex
143	app_vertex = machine_vertex._pop_vertex	×
144	sources = self.__get_sources_for_target(app_vertex)	×
145
146	size = self.get_parameters_usage_in_bytes(	×
147	machine_vertex.vertex_slice.n_atoms,
148	app_vertex.incoming_projections)
149	spec.reserve_memory_region(region, size, label="LocalOnlyConvolution")	×
150	spec.switch_write_focus(region)	×
151
152	# Get spec for each incoming source
153	connector_weight_index: Dict[AbstractConnector, int] = dict()	×
154	next_weight_index: int = 0	×
155	source_data = list()	×
156	connector_data: List[NDArray[uint32]] = list()	×
157	weight_data = list()	×
158	for pre_vertex, source_infos in sources.items():	×
159
160	# Add connectors as needed
161	first_conn_index = len(connector_data)	×
162	for source in source_infos:	×
163	# pylint: disable=protected-access
164	conn = cast(	×
165	ConvolutionConnector,
166	source.projection._synapse_information.connector)
167	app_edge = source.projection._projection_edge	×
168
169	# Work out whether the connector needs a new weight index
170	if conn in connector_weight_index:	×
171	weight_index = connector_weight_index[conn]	×
172	else:
173	weight_index = next_weight_index	×
174	connector_weight_index[conn] = weight_index	×
175	next_weight_index += conn.kernel_n_weights	×
176	weight_data.append(conn.get_encoded_kernel_weights(	×
177	app_edge, weight_scales))
178
179	connector_data.append(conn.get_local_only_data(	×
180	app_edge, source.local_delay, source.delay_stage,
181	weight_index))
182
183	# Get the source routing information
184	r_info, core_mask, mask_shift = get_rinfo_for_spike_source(	×
185	pre_vertex)
186
187	# Get the width / height per core / last_core
188	first_slice, last_slice = get_first_and_last_slice(pre_vertex)	×
189	width_per_core = first_slice.shape[0]	×
190	height_per_core = first_slice.shape[1]	×
191	width_on_last_core = last_slice.shape[0]	×
192	height_on_last_core = last_slice.shape[1]	×
193
194	# Get cores per width / height
195	pre_shape = list(pre_vertex.atoms_shape)	×
196	cores_per_width = int(ceil(pre_shape[0] / width_per_core))	×
197	cores_per_height = int(ceil(pre_shape[1] / height_per_core))	×
198
199	# Add the key and mask...
200	source_data.extend([r_info.key, r_info.mask])	×
201	# ... start connector index, n_colour_bits, count of connectors ...
202	source_data.append(	×
203	(len(source_infos) << BITS_PER_SHORT) +
204	(pre_vertex.n_colour_bits <<
205	(BITS_PER_SHORT - N_COLOUR_BITS_BITS)) +
206	first_conn_index)
207	# ... core mask, mask shift ...
208	source_data.append((mask_shift << BITS_PER_SHORT) + core_mask)	×
209	# ... height / width per core ...
210	source_data.append(	×
211	(width_per_core << BITS_PER_SHORT) + height_per_core)
212	# ... height / width last core ...
213	source_data.append(	×
214	(width_on_last_core << BITS_PER_SHORT) + height_on_last_core)
215	# ... cores per height / width ...
216	source_data.append(	×
217	(cores_per_width << BITS_PER_SHORT) + cores_per_height)
218	# ... 1 / width per core ...
219	source_data.append(get_div_const(width_per_core))	×
220	# ... 1 / width last core ...
221	source_data.append(get_div_const(width_on_last_core))	×
222	# ... 1 / cores_per_width
223	source_data.append(get_div_const(cores_per_width))	×
224
225	if next_weight_index % 2 != 0:	×
226	weight_data.append(numpy.array([0], dtype="int16"))	×
227
228	# Write the common spec
229	post_slice = machine_vertex.vertex_slice	×
230	post_start = numpy.array(post_slice.start)	×
231	post_shape = numpy.array(post_slice.shape)	×
232	post_end = (post_start + post_shape) - 1	×
233	spec.write_value(post_start[1], data_type=DataType.INT16)	×
234	spec.write_value(post_start[0], data_type=DataType.INT16)	×
235	spec.write_value(post_end[1], data_type=DataType.INT16)	×
236	spec.write_value(post_end[0], data_type=DataType.INT16)	×
237	spec.write_value(post_shape[1], data_type=DataType.INT16)	×
238	spec.write_value(post_shape[0], data_type=DataType.INT16)	×
239	spec.write_value(len(sources))	×
240	spec.write_value(len(connector_data))	×
241	spec.write_value(next_weight_index)	×
242
243	# Write the data
244	# pylint: disable=unexpected-keyword-arg
245	spec.write_array(numpy.array(source_data, dtype="uint32"))	×
246	spec.write_array(numpy.concatenate(connector_data, dtype="uint32"))	×
247	spec.write_array(	×
248	numpy.concatenate(weight_data, dtype="int16").view("uint32"))
249
250	def __get_sources_for_target(	1✔
251	self, app_vertex: AbstractPopulationVertex) -> Dict[
252	PopulationApplicationVertex, List[Source]]:
253	"""
254	Get all the application vertex sources that will hit the given
255	application vertex.
256
257	:param AbstractPopulationVertex app_vertex: The vertex being targeted
258	:return:
259	A dict of source PopulationApplicationVertex to list of source
260	information
261	:rtype: dict(PopulationApplicationVertex, list(Source))
262	"""
263	sources = self.__cached_sources.get(app_vertex)	×
264	if sources is None:	×
265	sources = get_sources_for_target(app_vertex)	×
266	self.__cached_sources[app_vertex] = sources	×
267	return sources	×
268
269	@staticmethod	1✔
270	def __connector(projection: Projection) -> ConvolutionConnector:	1✔
271	# pylint: disable=protected-access
272	return cast(ConvolutionConnector,	×
273	projection._synapse_information.connector)
274
275	@staticmethod	1✔
276	def __get_synapse_type(proj: Projection, target: str) -> int:	1✔
277	edge = proj._projection_edge # pylint: disable=protected-access	×
278	synapse_type = edge.post_vertex.get_synapse_id_by_target(target)	×
279	# Checked during connection validation, assumed constant
280	assert synapse_type is not None	×
281	return synapse_type	×
282
283	@overrides(AbstractSupportsSignedWeights.get_positive_synapse_index)	1✔
284	def get_positive_synapse_index(	1✔
285	self, incoming_projection: Projection) -> int:
286	return self.__get_synapse_type(	×
287	incoming_projection,
288	self.__connector(incoming_projection).positive_receptor_type)
289
290	@overrides(AbstractSupportsSignedWeights.get_negative_synapse_index)	1✔
291	def get_negative_synapse_index(	1✔
292	self, incoming_projection: Projection) -> int:
293	return self.__get_synapse_type(	×
294	incoming_projection,
295	self.__connector(incoming_projection).negative_receptor_type)
296
297	@overrides(AbstractSupportsSignedWeights.get_maximum_positive_weight)	1✔
298	def get_maximum_positive_weight(	1✔
299	self, incoming_projection: Projection) -> float:
300	conn = self.__connector(incoming_projection)	×
301	# We know the connector doesn't care about the argument
302	# conn.kernel_weights known to be an array of floats
NEW 303	max_weight = cast(float, numpy.amax(conn.kernel_weights))	×
UNCOV 304	return max_weight if max_weight > 0 else 0	×
305
306	@overrides(AbstractSupportsSignedWeights.get_minimum_negative_weight)	1✔
307	def get_minimum_negative_weight(	1✔
308	self, incoming_projection: Projection) -> float:
309	conn = self.__connector(incoming_projection)	×
310	# This is different because the connector happens to support this
311	# conn.kernel_weights known to be an array of floats
NEW 312	min_weight = cast(float, numpy.amin(conn.kernel_weights))	×
UNCOV 313	return min_weight if min_weight < 0 else 0	×
314
315	@overrides(AbstractSupportsSignedWeights.get_mean_positive_weight)	1✔
316	def get_mean_positive_weight(	1✔
317	self, incoming_projection: Projection) -> float:
318	conn = self.__connector(incoming_projection)	×
319	pos_weights = conn.kernel_weights[conn.kernel_weights > 0]	×
NEW 320	if len(pos_weights) == 0:	×
321	return 0	×
322	return numpy.mean(pos_weights)	×
323
324	@overrides(AbstractSupportsSignedWeights.get_mean_negative_weight)	1✔
325	def get_mean_negative_weight(	1✔
326	self, incoming_projection: Projection) -> float:
327	conn = self.__connector(incoming_projection)	×
328	neg_weights = conn.kernel_weights[conn.kernel_weights < 0]	×
NEW 329	if len(neg_weights) == 0:	×
330	return 0	×
331	return numpy.mean(neg_weights)	×
332
333	@overrides(AbstractSupportsSignedWeights.get_variance_positive_weight)	1✔
334	def get_variance_positive_weight(	1✔
335	self, incoming_projection: Projection) -> float:
336	conn = self.__connector(incoming_projection)	×
337	pos_weights = conn.kernel_weights[conn.kernel_weights > 0]	×
NEW 338	if len(pos_weights) == 0:	×
339	return 0	×
340	return numpy.var(pos_weights)	×
341
342	@overrides(AbstractSupportsSignedWeights.get_variance_negative_weight)	1✔
343	def get_variance_negative_weight(	1✔
344	self, incoming_projection: Projection) -> float:
345	conn = self.__connector(incoming_projection)	×
346	neg_weights = conn.kernel_weights[conn.kernel_weights < 0]	×
NEW 347	if len(neg_weights) == 0:	×
348	return 0	×
349	return numpy.var(neg_weights)	×

SpiNNakerManchester / sPyNNaker / 8143408510

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