6941145312

Committed 21 Nov 2023 08:27AM UTC coverage: 63.155% (+1.6%) from 61.529%

Build # 6941145312

Build Type

Pull #1342

github

Committed by

Christian-B

Commit Message

flake8

Pull Request Pull Request #1342: Type Annotations and Checking

Run Details

1960 of 4482 branches covered (0.0%)

Branch coverage included in aggregate %.

4149 of 5079 new or added lines in 233 files covered. (81.69%)

193 existing lines in 78 files now uncovered.

12726 of 18772 relevant lines covered (67.79%)

0.68 hits per line

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

84.47

/spynnaker/pyNN/extra_algorithms/delay_support_adder.py

# Copyright (c) 2020 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 math
from typing import Dict, List, Sequence, Tuple
from spinn_utilities.progress_bar import ProgressBar
from pacman.model.graphs.application import (
    ApplicationEdge, ApplicationEdgePartition)
from spynnaker.pyNN.data import SpynnakerDataView
from spynnaker.pyNN.exceptions import DelayExtensionException
from spynnaker.pyNN.extra_algorithms.splitter_components import (
    AbstractSpynnakerSplitterDelay, SplitterDelayVertexSlice)
from spynnaker.pyNN.models.neural_projections import (
    ProjectionApplicationEdge, DelayedApplicationEdge,
    DelayAfferentApplicationEdge)
from spynnaker.pyNN.models.utility_models.delays import DelayExtensionVertex
from spynnaker.pyNN.models.neuron import AbstractPopulationVertex


def delay_support_adder() -> Tuple[
        Sequence[DelayExtensionVertex], Sequence[ApplicationEdge]]:
    """
    Adds the delay extensions to the application graph, now that all the
    splitter objects have been set.

    :return: The delay vertices and delay edges that were added
    :rtype: tuple(list(DelayExtensionVertex), list(DelayedApplicationEdge or
        DelayAfferentApplicationEdge))
    """
    adder = _DelaySupportAdder()
    # pylint: disable=protected-access
    return adder.add_delays()


class _DelaySupportAdder(object):
    """
    Adds delay extension vertices into the application graph as needed.
    """

    __slots__ = (
        "_app_to_delay_map",
        "_delay_post_edge_map",
        "_new_edges",
        "_new_vertices")

    def __init__(self) -> None:
        self._app_to_delay_map: Dict[
            ApplicationEdgePartition, DelayExtensionVertex] = dict()
        self._delay_post_edge_map: Dict[
            Tuple[DelayExtensionVertex, AbstractPopulationVertex],
            DelayedApplicationEdge] = dict()
        self._new_edges: List[ApplicationEdge] = list()
        self._new_vertices: List[DelayExtensionVertex] = list()

    def add_delays(self) -> Tuple[
            List[DelayExtensionVertex], List[ApplicationEdge]]:
        """
        Adds the delay extensions to the application graph, now that all the
        splitter objects have been set.

        :rtype: tuple(list(DelayExtensionVertex), list(DelayedApplicationEdge))
        """
        progress = ProgressBar(1 + SpynnakerDataView.get_n_partitions(),
                               "Adding delay extensions as required")

        for vertex in SpynnakerDataView.get_vertices_by_type(
                DelayExtensionVertex):
            self._app_to_delay_map[vertex.partition] = vertex
            for edge in vertex.outgoing_edges:
                self._delay_post_edge_map[vertex, edge.post_vertex] = edge
        progress.update(1)

        # go through all partitions.
        for app_outgoing_edge_partition in progress.over(
                SpynnakerDataView.iterate_partitions()):
            for app_edge in app_outgoing_edge_partition.edges:
                if isinstance(app_edge, ProjectionApplicationEdge):
                    self.__examine_edge_for_delays_to_add(
                        app_edge, app_outgoing_edge_partition)
        return self._new_vertices, self._new_edges

    def __examine_edge_for_delays_to_add(
            self, edge: ProjectionApplicationEdge,
            partition: ApplicationEdgePartition):
        """
        Look at a particular edge to see if it needs a delay vertex+edge
        inserted, and add it in if it does.

        :param ProjectionApplicationEdge edge:
        :param ApplicationEdgePartition partition:
        """
        # figure the max delay and if we need a delay extension
        n_stages, steps_per_stage, need_delay_ext = self._check_delay_values(
            edge, edge.synapse_information)

        # if we need a delay, add it to the app graph.
        if need_delay_ext:
            delay_app_vertex = self._create_delay_app_vertex_and_pre_edge(
                partition, edge, steps_per_stage, n_stages)

            # add the edge from the delay extension to the dest vertex
            self._create_post_delay_edge(delay_app_vertex, edge)

    def _create_post_delay_edge(
            self, delay_app_vertex: DelayExtensionVertex,
            app_edge: ProjectionApplicationEdge):
        """
        Creates the edge between delay extension and post vertex. Stores
        for future loading to the application graph when safe to do so.

        :param DelayExtensionVertex delay_app_vertex: delay extension vertex
        :param ProjectionApplicationEdge app_edge:
            the undelayed application edge this is associated with.
        """
        # check for post edge
        delayed_edge = self._delay_post_edge_map.get(
            (delay_app_vertex, app_edge.post_vertex), None)
        if delayed_edge is None:
            delay_edge = DelayedApplicationEdge(
                delay_app_vertex, app_edge.post_vertex,
                app_edge.synapse_information,
                label=(f"{app_edge.pre_vertex.label}_delayed_"
                       f"to_{app_edge.post_vertex.label}"),
                undelayed_edge=app_edge)
            self._delay_post_edge_map[
                (delay_app_vertex, app_edge.post_vertex)] = delay_edge
            self._new_edges.append(delay_edge)
            app_edge.delay_edge = delay_edge
            delay_app_vertex.add_outgoing_edge(delay_edge)

    def _create_delay_app_vertex_and_pre_edge(
            self, app_outgoing_edge_partition: ApplicationEdgePartition,
            app_edge: ProjectionApplicationEdge, delay_per_stage: int,
            n_delay_stages: int):
        """
        Creates the delay extension application vertex and the edge from the
        source vertex to this delay extension. Adds to the graph, as safe to
        do so.

        :param ApplicationEdgePartition app_outgoing_edge_partition:
            the original outgoing edge partition.
        :param ApplicationEdge app_edge: the undelayed application edge.
        :param int delay_per_stage: delay for each delay stage
        :param int n_delay_stages: the number of delay stages needed
        :return: the DelayExtensionAppVertex
        :rtype: DelayExtensionVertex
        """
        # get delay extension vertex if it already exists.
        delay_app_vertex = self._app_to_delay_map.get(
            app_outgoing_edge_partition, None)
        if delay_app_vertex is None:
            # build delay app vertex
            delay_app_vertex = DelayExtensionVertex(
                app_outgoing_edge_partition, delay_per_stage, n_delay_stages,
                app_edge.pre_vertex.n_colour_bits,
                label=f"{app_edge.pre_vertex.label}_delayed")

            # set trackers
            delay_app_vertex.splitter = SplitterDelayVertexSlice()
            self._new_vertices.append(delay_app_vertex)
            self._app_to_delay_map[app_outgoing_edge_partition] = (
                delay_app_vertex)

            # build afferent app edge
            delay_pre_edge = DelayAfferentApplicationEdge(
                app_edge.pre_vertex, delay_app_vertex,
                label=f"{app_edge.pre_vertex.label}_to_DelayExtension")
            self._new_edges.append(delay_pre_edge)
        else:
            delay_app_vertex.set_new_n_delay_stages_and_delay_per_stage(
                n_delay_stages, delay_per_stage)
        return delay_app_vertex

    def _check_delay_values(self, app_edge, synapse_infos):
        """
        Checks the delay required from the user defined max, the max delay
        supported by the post vertex splitter and the delay Extensions.

        :param ApplicationEdge app_edge: the undelayed application edge
        :param iterable[SynapseInformation] synapse_infos:
            the synapse information objects
        :return: tuple(n_delay_stages, delay_steps_per_stage, extension_needed)
        """
        # get max delay required
        max_delay_needed_ms = max(
            synapse_info.synapse_dynamics.get_delay_maximum(
                synapse_info.connector, synapse_info)
            for synapse_info in synapse_infos)

        # get if the post vertex needs a delay extension
        post_splitter = app_edge.post_vertex.splitter
        if not isinstance(
                post_splitter, AbstractSpynnakerSplitterDelay):
            raise DelayExtensionException(
                f"The app vertex {app_edge.post_vertex} "
                f"with splitter {post_splitter} does not support delays "
                f"and yet requires a delay support for edge {app_edge}. "
                "Please use a Splitter which utilises the "
                "AbstractSpynnakerSplitterDelay interface.")
        max_delay_steps = app_edge.post_vertex.splitter.max_support_delay()
        time_step_ms = SpynnakerDataView.get_simulation_time_step_ms()
        max_delay_ms = max_delay_steps * time_step_ms

        # if does not need a delay extension, run away
        if max_delay_ms >= max_delay_needed_ms:
            return 0, max_delay_steps, False

        # Check post vertex is ok with getting a delay
        if not post_splitter.accepts_edges_from_delay_vertex():
            raise DelayExtensionException(
                f"The app vertex {app_edge.post_vertex} "
                f"with splitter {post_splitter} does not support delays "
                f"and yet requires a delay support for edge {app_edge}. "
                "Please use a Splitter which does not have "
                "accepts_edges_from_delay_vertex turned off.")

        # needs a delay extension, check can be supported with 1 delay
        # extension. coz we dont do more than 1 at the moment
        ext_provided_ms = DelayExtensionVertex.get_max_delay_ticks_supported(
                max_delay_steps) * time_step_ms
        total_delay_ms = ext_provided_ms + max_delay_ms
        if total_delay_ms < max_delay_needed_ms:
            raise DelayExtensionException(
                f"Edge:{app_edge.label} "
                f"has a max delay of {max_delay_needed_ms}. "
                f"But at a timestep of {time_step_ms} "
                f"the max delay supported is {total_delay_ms}")

        # return data for building delay extensions
        n_stages = int(math.ceil(max_delay_needed_ms / max_delay_ms)) - 1
        return n_stages, max_delay_steps, True

1	# Copyright (c) 2020 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	import math	1✔
15	from typing import Dict, List, Sequence, Tuple	1✔
16	from spinn_utilities.progress_bar import ProgressBar	1✔
17	from pacman.model.graphs.application import (	1✔
18	ApplicationEdge, ApplicationEdgePartition)
19	from spynnaker.pyNN.data import SpynnakerDataView	1✔
20	from spynnaker.pyNN.exceptions import DelayExtensionException	1✔
21	from spynnaker.pyNN.extra_algorithms.splitter_components import (	1✔
22	AbstractSpynnakerSplitterDelay, SplitterDelayVertexSlice)
23	from spynnaker.pyNN.models.neural_projections import (	1✔
24	ProjectionApplicationEdge, DelayedApplicationEdge,
25	DelayAfferentApplicationEdge)
26	from spynnaker.pyNN.models.utility_models.delays import DelayExtensionVertex	1✔
27	from spynnaker.pyNN.models.neuron import AbstractPopulationVertex	1✔
28
29
30	def delay_support_adder() -> Tuple[	1✔
31	Sequence[DelayExtensionVertex], Sequence[ApplicationEdge]]:
32	"""
33	Adds the delay extensions to the application graph, now that all the
34	splitter objects have been set.
35
36	:return: The delay vertices and delay edges that were added
37	:rtype: tuple(list(DelayExtensionVertex), list(DelayedApplicationEdge or
38	DelayAfferentApplicationEdge))
39	"""
40	adder = _DelaySupportAdder()	1✔
41	# pylint: disable=protected-access
42	return adder.add_delays()	1✔
43
44
45	class _DelaySupportAdder(object):	1✔
46	"""
47	Adds delay extension vertices into the application graph as needed.
48	"""
49
50	__slots__ = (	1✔
51	"_app_to_delay_map",
52	"_delay_post_edge_map",
53	"_new_edges",
54	"_new_vertices")
55
56	def __init__(self) -> None:	1✔
57	self._app_to_delay_map: Dict[	1✔
58	ApplicationEdgePartition, DelayExtensionVertex] = dict()
59	self._delay_post_edge_map: Dict[	1✔
60	Tuple[DelayExtensionVertex, AbstractPopulationVertex],
61	DelayedApplicationEdge] = dict()
62	self._new_edges: List[ApplicationEdge] = list()	1✔
63	self._new_vertices: List[DelayExtensionVertex] = list()	1✔
64
65	def add_delays(self) -> Tuple[	1✔
66	List[DelayExtensionVertex], List[ApplicationEdge]]:
67	"""
68	Adds the delay extensions to the application graph, now that all the
69	splitter objects have been set.
70
71	:rtype: tuple(list(DelayExtensionVertex), list(DelayedApplicationEdge))
72	"""
73	progress = ProgressBar(1 + SpynnakerDataView.get_n_partitions(),	1✔
74	"Adding delay extensions as required")
75
76	for vertex in SpynnakerDataView.get_vertices_by_type(	1!
77	DelayExtensionVertex):
78	self._app_to_delay_map[vertex.partition] = vertex	×
79	for edge in vertex.outgoing_edges:	×
NEW 80	self._delay_post_edge_map[vertex, edge.post_vertex] = edge	×
81	progress.update(1)	1✔
82
83	# go through all partitions.
84	for app_outgoing_edge_partition in progress.over(	1✔
85	SpynnakerDataView.iterate_partitions()):
86	for app_edge in app_outgoing_edge_partition.edges:	1✔
87	if isinstance(app_edge, ProjectionApplicationEdge):	1!
88	self.__examine_edge_for_delays_to_add(	1✔
89	app_edge, app_outgoing_edge_partition)
90	return self._new_vertices, self._new_edges	1✔
91
92	def __examine_edge_for_delays_to_add(	1✔
93	self, edge: ProjectionApplicationEdge,
94	partition: ApplicationEdgePartition):
95	"""
96	Look at a particular edge to see if it needs a delay vertex+edge
97	inserted, and add it in if it does.
98
99	:param ProjectionApplicationEdge edge:
100	:param ApplicationEdgePartition partition:
101	"""
102	# figure the max delay and if we need a delay extension
103	n_stages, steps_per_stage, need_delay_ext = self._check_delay_values(	1✔
104	edge, edge.synapse_information)
105
106	# if we need a delay, add it to the app graph.
107	if need_delay_ext:	1✔
108	delay_app_vertex = self._create_delay_app_vertex_and_pre_edge(	1✔
109	partition, edge, steps_per_stage, n_stages)
110
111	# add the edge from the delay extension to the dest vertex
112	self._create_post_delay_edge(delay_app_vertex, edge)	1✔
113
114	def _create_post_delay_edge(	1✔
115	self, delay_app_vertex: DelayExtensionVertex,
116	app_edge: ProjectionApplicationEdge):
117	"""
118	Creates the edge between delay extension and post vertex. Stores
119	for future loading to the application graph when safe to do so.
120
121	:param DelayExtensionVertex delay_app_vertex: delay extension vertex
122	:param ProjectionApplicationEdge app_edge:
123	the undelayed application edge this is associated with.
124	"""
125	# check for post edge
126	delayed_edge = self._delay_post_edge_map.get(	1✔
127	(delay_app_vertex, app_edge.post_vertex), None)
128	if delayed_edge is None:	1!
129	delay_edge = DelayedApplicationEdge(	1✔
130	delay_app_vertex, app_edge.post_vertex,
131	app_edge.synapse_information,
132	label=(f"{app_edge.pre_vertex.label}_delayed_"
133	f"to_{app_edge.post_vertex.label}"),
134	undelayed_edge=app_edge)
135	self._delay_post_edge_map[	1✔
136	(delay_app_vertex, app_edge.post_vertex)] = delay_edge
137	self._new_edges.append(delay_edge)	1✔
138	app_edge.delay_edge = delay_edge	1✔
139	delay_app_vertex.add_outgoing_edge(delay_edge)	1✔
140
141	def _create_delay_app_vertex_and_pre_edge(	1✔
142	self, app_outgoing_edge_partition: ApplicationEdgePartition,
143	app_edge: ProjectionApplicationEdge, delay_per_stage: int,
144	n_delay_stages: int):
145	"""
146	Creates the delay extension application vertex and the edge from the
147	source vertex to this delay extension. Adds to the graph, as safe to
148	do so.
149
150	:param ApplicationEdgePartition app_outgoing_edge_partition:
151	the original outgoing edge partition.
152	:param ApplicationEdge app_edge: the undelayed application edge.
153	:param int delay_per_stage: delay for each delay stage
154	:param int n_delay_stages: the number of delay stages needed
155	:return: the DelayExtensionAppVertex
156	:rtype: DelayExtensionVertex
157	"""
158	# get delay extension vertex if it already exists.
159	delay_app_vertex = self._app_to_delay_map.get(	1✔
160	app_outgoing_edge_partition, None)
161	if delay_app_vertex is None:	1!
162	# build delay app vertex
163	delay_app_vertex = DelayExtensionVertex(	1✔
164	app_outgoing_edge_partition, delay_per_stage, n_delay_stages,
165	app_edge.pre_vertex.n_colour_bits,
166	label=f"{app_edge.pre_vertex.label}_delayed")
167
168	# set trackers
169	delay_app_vertex.splitter = SplitterDelayVertexSlice()	1✔
170	self._new_vertices.append(delay_app_vertex)	1✔
171	self._app_to_delay_map[app_outgoing_edge_partition] = (	1✔
172	delay_app_vertex)
173
174	# build afferent app edge
175	delay_pre_edge = DelayAfferentApplicationEdge(	1✔
176	app_edge.pre_vertex, delay_app_vertex,
177	label=f"{app_edge.pre_vertex.label}_to_DelayExtension")
178	self._new_edges.append(delay_pre_edge)	1✔
179	else:
180	delay_app_vertex.set_new_n_delay_stages_and_delay_per_stage(	×
181	n_delay_stages, delay_per_stage)
182	return delay_app_vertex	1✔
183
184	def _check_delay_values(self, app_edge, synapse_infos):	1✔
185	"""
186	Checks the delay required from the user defined max, the max delay
187	supported by the post vertex splitter and the delay Extensions.
188
189	:param ApplicationEdge app_edge: the undelayed application edge
190	:param iterable[SynapseInformation] synapse_infos:
191	the synapse information objects
192	:return: tuple(n_delay_stages, delay_steps_per_stage, extension_needed)
193	"""
194	# get max delay required
195	max_delay_needed_ms = max(	1✔
196	synapse_info.synapse_dynamics.get_delay_maximum(
197	synapse_info.connector, synapse_info)
198	for synapse_info in synapse_infos)
199
200	# get if the post vertex needs a delay extension
201	post_splitter = app_edge.post_vertex.splitter	1✔
202	if not isinstance(	1!
203	post_splitter, AbstractSpynnakerSplitterDelay):
204	raise DelayExtensionException(	×
205	f"The app vertex {app_edge.post_vertex} "
206	f"with splitter {post_splitter} does not support delays "
207	f"and yet requires a delay support for edge {app_edge}. "
208	"Please use a Splitter which utilises the "
209	"AbstractSpynnakerSplitterDelay interface.")
210	max_delay_steps = app_edge.post_vertex.splitter.max_support_delay()	1✔
211	time_step_ms = SpynnakerDataView.get_simulation_time_step_ms()	1✔
212	max_delay_ms = max_delay_steps * time_step_ms	1✔
213
214	# if does not need a delay extension, run away
215	if max_delay_ms >= max_delay_needed_ms:	1✔
216	return 0, max_delay_steps, False	1✔
217
218	# Check post vertex is ok with getting a delay
219	if not post_splitter.accepts_edges_from_delay_vertex():	1!
220	raise DelayExtensionException(	×
221	f"The app vertex {app_edge.post_vertex} "
222	f"with splitter {post_splitter} does not support delays "
223	f"and yet requires a delay support for edge {app_edge}. "
224	"Please use a Splitter which does not have "
225	"accepts_edges_from_delay_vertex turned off.")
226
227	# needs a delay extension, check can be supported with 1 delay
228	# extension. coz we dont do more than 1 at the moment
229	ext_provided_ms = DelayExtensionVertex.get_max_delay_ticks_supported(	1✔
230	max_delay_steps) * time_step_ms
231	total_delay_ms = ext_provided_ms + max_delay_ms	1✔
232	if total_delay_ms < max_delay_needed_ms:	1!
233	raise DelayExtensionException(	×
234	f"Edge:{app_edge.label} "
235	f"has a max delay of {max_delay_needed_ms}. "
236	f"But at a timestep of {time_step_ms} "
237	f"the max delay supported is {total_delay_ms}")
238
239	# return data for building delay extensions
240	n_stages = int(math.ceil(max_delay_needed_ms / max_delay_ms)) - 1	1✔
241	return n_stages, max_delay_steps, True	1✔

SpiNNakerManchester / sPyNNaker / 6941145312

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