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

81.94

/spynnaker/pyNN/extra_algorithms/splitter_components/splitter_delay_vertex_slice.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.
from typing import Dict, Sequence, Tuple
from spinn_utilities.overrides import overrides
from pacman.exceptions import (
    PacmanConfigurationException, PacmanInvalidParameterException)
from pacman.model.graphs.application import ApplicationVertex
from pacman.model.graphs.machine import MachineVertex
from pacman.model.graphs.common import Slice
from pacman.model.partitioner_splitters import AbstractSplitterCommon
from pacman.model.resources import ConstantSDRAM
from spinn_front_end_common.utilities.constants import (
    SYSTEM_BYTES_REQUIREMENT, BYTES_PER_WORD)
from spynnaker.pyNN.models.utility_models.delays import (
    DelayExtensionVertex, DelayExtensionMachineVertex)
from pacman.utilities.utility_objs.chip_counter import ChipCounter
from pacman.model.resources.abstract_sdram import AbstractSDRAM


class SplitterDelayVertexSlice(AbstractSplitterCommon[DelayExtensionVertex]):
    """
    Handles the splitting of the :py:class:`DelayExtensionVertex`
    via slice logic.
    """

    __slots__ = ("_machine_vertex_by_slice", )

    _EXPANDER_BASE_PARAMS_SIZE = 3 * BYTES_PER_WORD

    NEED_EXACT_ERROR_MESSAGE = (
        "DelayExtensionsSplitters need exact incoming slices. Please fix "
        "and try again")

    DELAY_RECORDING_ERROR = (
        "The delay extensions does not record any variables. Therefore "
        "asking for them is deemed an error.")

    def __init__(self) -> None:
        super().__init__()
        self._machine_vertex_by_slice: Dict[
            Slice, DelayExtensionMachineVertex] = dict()

    @overrides(AbstractSplitterCommon.get_out_going_vertices)
    def get_out_going_vertices(
            self, partition_id: str) -> Sequence[MachineVertex]:
        return list(self.governed_app_vertex.machine_vertices)

    @overrides(AbstractSplitterCommon.get_in_coming_vertices)
    def get_in_coming_vertices(
            self, partition_id: str) -> Sequence[MachineVertex]:
        return list(self.governed_app_vertex.machine_vertices)

    @overrides(AbstractSplitterCommon.get_source_specific_in_coming_vertices)
    def get_source_specific_in_coming_vertices(
            self, source_vertex: ApplicationVertex,
            partition_id: str) -> Sequence[Tuple[
                DelayExtensionMachineVertex, Sequence[MachineVertex]]]:
        # Only connect to the source that matches the slice
        return [
            (self._machine_vertex_by_slice[m_vertex.vertex_slice], [m_vertex])
            for m_vertex in source_vertex.splitter.get_out_going_vertices(
                partition_id)]

    @overrides(AbstractSplitterCommon.create_machine_vertices)
    def create_machine_vertices(self, chip_counter: ChipCounter):
        source_app_vertex = self.governed_app_vertex.source_vertex
        slices = source_app_vertex.splitter.get_out_going_slices()

        # create vertices correctly
        for vertex_slice in slices:
            vertex = self.create_machine_vertex(
                source_app_vertex, vertex_slice)
            self.governed_app_vertex.remember_machine_vertex(vertex)
            chip_counter.add_core(vertex.sdram_required)

    @overrides(AbstractSplitterCommon.get_in_coming_slices)
    def get_in_coming_slices(self) -> Sequence[Slice]:
        other_splitter = self.governed_app_vertex.source_vertex.splitter
        return other_splitter.get_in_coming_slices()

    @overrides(AbstractSplitterCommon.get_out_going_slices)
    def get_out_going_slices(self) -> Sequence[Slice]:
        other_splitter = self.governed_app_vertex.source_vertex.splitter
        return other_splitter.get_out_going_slices()

    @overrides(AbstractSplitterCommon.set_governed_app_vertex)
    def set_governed_app_vertex(self, app_vertex: DelayExtensionVertex):
        super().set_governed_app_vertex(app_vertex)
        if not isinstance(app_vertex, DelayExtensionVertex):
            raise PacmanConfigurationException(
                f"The vertex {app_vertex} cannot be supported by the "
                "SplitterDelayVertexSlice as the only vertex supported by "
                "this splitter is a DelayExtensionVertex. Please use the "
                "correct splitter for your vertex and try again.")

    def create_machine_vertex(
            self, source_app_vertex: ApplicationVertex, vertex_slice: Slice):
        """
        Creates a delay extension machine vertex and adds to the tracker.

        :param ~pacman.model.graphs.machine.MachineVertex source_vertex:
            The source of the delay
        :return: machine vertex
        :rtype: DelayExtensionMachineVertex
        """
        label = f"Delay extension for {source_app_vertex}"
        sdram = self.get_sdram_used_by_atoms()

        machine_vertex = DelayExtensionMachineVertex(
            sdram, label, vertex_slice, self.governed_app_vertex)

        self._machine_vertex_by_slice[vertex_slice] = machine_vertex
        return machine_vertex

    def get_sdram_used_by_atoms(self) -> AbstractSDRAM:
        """
        Gets the amount of SDRAM used by the delay extension.

        :rtype: ~pacman.model.resources.ConstantSDRAM
        """
        return ConstantSDRAM(
            SYSTEM_BYTES_REQUIREMENT +
            self.governed_app_vertex.delay_params_size() +
            DelayExtensionMachineVertex.get_provenance_data_size(
                DelayExtensionMachineVertex.N_EXTRA_PROVENANCE_DATA_ENTRIES))

    @overrides(AbstractSplitterCommon.machine_vertices_for_recording)
    def machine_vertices_for_recording(self, variable_to_record):
        raise PacmanInvalidParameterException(
            variable_to_record, variable_to_record, self.DELAY_RECORDING_ERROR)

    @overrides(AbstractSplitterCommon.reset_called)
    def reset_called(self) -> None:
        self._machine_vertex_by_slice = dict()

    def get_machine_vertex(
            self, vertex_slice: Slice) -> DelayExtensionMachineVertex:
        """
        Get a delay extension machine vertex for a given vertex slice.

        :param ~pacman.model.graphs.common.Slice vertex_slice:
            The slice to get the data for
        :rtype: DelayExtensionMachineVertex
        """
        return self._machine_vertex_by_slice[vertex_slice]

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	from typing import Dict, Sequence, Tuple	1✔
15	from spinn_utilities.overrides import overrides	1✔
16	from pacman.exceptions import (	1✔
17	PacmanConfigurationException, PacmanInvalidParameterException)
18	from pacman.model.graphs.application import ApplicationVertex	1✔
19	from pacman.model.graphs.machine import MachineVertex	1✔
20	from pacman.model.graphs.common import Slice	1✔
21	from pacman.model.partitioner_splitters import AbstractSplitterCommon	1✔
22	from pacman.model.resources import ConstantSDRAM	1✔
23	from spinn_front_end_common.utilities.constants import (	1✔
24	SYSTEM_BYTES_REQUIREMENT, BYTES_PER_WORD)
25	from spynnaker.pyNN.models.utility_models.delays import (	1✔
26	DelayExtensionVertex, DelayExtensionMachineVertex)
27	from pacman.utilities.utility_objs.chip_counter import ChipCounter	1✔
28	from pacman.model.resources.abstract_sdram import AbstractSDRAM	1✔
29
30
31	class SplitterDelayVertexSlice(AbstractSplitterCommon[DelayExtensionVertex]):	1✔
32	"""
33	Handles the splitting of the :py:class:`DelayExtensionVertex`
34	via slice logic.
35	"""
36
37	__slots__ = ("_machine_vertex_by_slice", )	1✔
38
39	_EXPANDER_BASE_PARAMS_SIZE = 3 * BYTES_PER_WORD	1✔
40
41	NEED_EXACT_ERROR_MESSAGE = (	1✔
42	"DelayExtensionsSplitters need exact incoming slices. Please fix "
43	"and try again")
44
45	DELAY_RECORDING_ERROR = (	1✔
46	"The delay extensions does not record any variables. Therefore "
47	"asking for them is deemed an error.")
48
49	def __init__(self) -> None:	1✔
50	super().__init__()	1✔
51	self._machine_vertex_by_slice: Dict[	1✔
52	Slice, DelayExtensionMachineVertex] = dict()
53
54	@overrides(AbstractSplitterCommon.get_out_going_vertices)	1✔
55	def get_out_going_vertices(	1✔
56	self, partition_id: str) -> Sequence[MachineVertex]:
57	return list(self.governed_app_vertex.machine_vertices)	1✔
58
59	@overrides(AbstractSplitterCommon.get_in_coming_vertices)	1✔
60	def get_in_coming_vertices(	1✔
61	self, partition_id: str) -> Sequence[MachineVertex]:
UNCOV 62	return list(self.governed_app_vertex.machine_vertices)	×
63
64	@overrides(AbstractSplitterCommon.get_source_specific_in_coming_vertices)	1✔
65	def get_source_specific_in_coming_vertices(	1✔
66	self, source_vertex: ApplicationVertex,
67	partition_id: str) -> Sequence[Tuple[
68	DelayExtensionMachineVertex, Sequence[MachineVertex]]]:
69	# Only connect to the source that matches the slice
70	return [	×
71	(self._machine_vertex_by_slice[m_vertex.vertex_slice], [m_vertex])
72	for m_vertex in source_vertex.splitter.get_out_going_vertices(
73	partition_id)]
74
75	@overrides(AbstractSplitterCommon.create_machine_vertices)	1✔
76	def create_machine_vertices(self, chip_counter: ChipCounter):	1✔
77	source_app_vertex = self.governed_app_vertex.source_vertex	1✔
78	slices = source_app_vertex.splitter.get_out_going_slices()	1✔
79
80	# create vertices correctly
81	for vertex_slice in slices:	1✔
82	vertex = self.create_machine_vertex(	1✔
83	source_app_vertex, vertex_slice)
84	self.governed_app_vertex.remember_machine_vertex(vertex)	1✔
85	chip_counter.add_core(vertex.sdram_required)	1✔
86
87	@overrides(AbstractSplitterCommon.get_in_coming_slices)	1✔
88	def get_in_coming_slices(self) -> Sequence[Slice]:	1✔
89	other_splitter = self.governed_app_vertex.source_vertex.splitter	×
90	return other_splitter.get_in_coming_slices()	×
91
92	@overrides(AbstractSplitterCommon.get_out_going_slices)	1✔
93	def get_out_going_slices(self) -> Sequence[Slice]:	1✔
94	other_splitter = self.governed_app_vertex.source_vertex.splitter	×
95	return other_splitter.get_out_going_slices()	×
96
97	@overrides(AbstractSplitterCommon.set_governed_app_vertex)	1✔
98	def set_governed_app_vertex(self, app_vertex: DelayExtensionVertex):	1✔
99	super().set_governed_app_vertex(app_vertex)	1✔
100	if not isinstance(app_vertex, DelayExtensionVertex):	1!
101	raise PacmanConfigurationException(	×
102	f"The vertex {app_vertex} cannot be supported by the "
103	"SplitterDelayVertexSlice as the only vertex supported by "
104	"this splitter is a DelayExtensionVertex. Please use the "
105	"correct splitter for your vertex and try again.")
106
107	def create_machine_vertex(	1✔
108	self, source_app_vertex: ApplicationVertex, vertex_slice: Slice):
109	"""
110	Creates a delay extension machine vertex and adds to the tracker.
111
112	:param ~pacman.model.graphs.machine.MachineVertex source_vertex:
113	The source of the delay
114	:return: machine vertex
115	:rtype: DelayExtensionMachineVertex
116	"""
117	label = f"Delay extension for {source_app_vertex}"	1✔
118	sdram = self.get_sdram_used_by_atoms()	1✔
119
120	machine_vertex = DelayExtensionMachineVertex(	1✔
121	sdram, label, vertex_slice, self.governed_app_vertex)
122
123	self._machine_vertex_by_slice[vertex_slice] = machine_vertex	1✔
124	return machine_vertex	1✔
125
126	def get_sdram_used_by_atoms(self) -> AbstractSDRAM:	1✔
127	"""
128	Gets the amount of SDRAM used by the delay extension.
129
130	:rtype: ~pacman.model.resources.ConstantSDRAM
131	"""
132	return ConstantSDRAM(	1✔
133	SYSTEM_BYTES_REQUIREMENT +
134	self.governed_app_vertex.delay_params_size() +
135	DelayExtensionMachineVertex.get_provenance_data_size(
136	DelayExtensionMachineVertex.N_EXTRA_PROVENANCE_DATA_ENTRIES))
137
138	@overrides(AbstractSplitterCommon.machine_vertices_for_recording)	1✔
139	def machine_vertices_for_recording(self, variable_to_record):	1✔
140	raise PacmanInvalidParameterException(	×
141	variable_to_record, variable_to_record, self.DELAY_RECORDING_ERROR)
142
143	@overrides(AbstractSplitterCommon.reset_called)	1✔
144	def reset_called(self) -> None:	1✔
145	self._machine_vertex_by_slice = dict()	×
146
147	def get_machine_vertex(	1✔
148	self, vertex_slice: Slice) -> DelayExtensionMachineVertex:
149	"""
150	Get a delay extension machine vertex for a given vertex slice.
151
152	:param ~pacman.model.graphs.common.Slice vertex_slice:
153	The slice to get the data for
154	:rtype: DelayExtensionMachineVertex
155	"""
156	return self._machine_vertex_by_slice[vertex_slice]	×

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