8433946879

Committed 26 Mar 2024 09:48AM UTC coverage: 84.63% (-1.5%) from 86.12%

Build # 8433946879

Build Type

push

github

Committed by

Christian-B

Commit Message

Merge branch 'less_routers' into rt_entry

Run Details

2450 of 2960 branches covered (82.77%)

Branch coverage included in aggregate %.

1 of 1 new or added line in 1 file covered. (100.0%)

62 existing lines in 2 files now uncovered.

4785 of 5589 relevant lines covered (85.61%)

0.86 hits per line

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

76.7

/pacman/model/routing_table_by_partition/multicast_routing_table_by_partition_entry.py

# Copyright (c) 2015 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 logging
from typing import Iterable, Optional, Union
from spinn_utilities.log import FormatAdapter
from spinn_machine.base_multicast_routing_entry import (
    BaseMulticastRoutingEntry)
from pacman.exceptions import (
    PacmanConfigurationException, PacmanInvalidParameterException)

log = FormatAdapter(logging.getLogger(__name__))


class MulticastRoutingTableByPartitionEntry(BaseMulticastRoutingEntry):
    """
    An entry in a path of a multicast route.
    """

    __slots__ = ["_incoming_processor", "_incoming_link"]

    def __init__(self, out_going_links: Union[int, Iterable[int], None],
                 outgoing_processors: Union[int, Iterable[int], None],
                 incoming_processor: Optional[int] = None,
                 incoming_link: Optional[int] = None,
                 spinnaker_route: Optional[int] = None):
        """
        .. note::
            If a spinnaker_route is provided the out_going_links and
            outgoing_processors parameters are ignored.

        :param out_going_links:
            the edges this path entry goes down, each of which is between
            0 and 5
        :type out_going_links: int or iterable(int) or None
        :param outgoing_processors:
            the processors this path entry goes to, each of which is between
            0 and 17
        :type outgoing_processors: int or iterable(int) or None
        :param int incoming_processor:
            the direction this entry came from (between 0 and 17)
        :param int incoming_link:
            the direction this entry came from in link (between 0 and 5)
        :param spinnaker_route:
            The processor_ids and link_ids expressed as a single int.
        :type spinnaker_route: int or None

        :raises PacmanInvalidParameterException:
        """
        super().__init__(
            outgoing_processors, out_going_links,
            spinnaker_route=spinnaker_route)

        self._incoming_link: Optional[int]
        self._incoming_processor: Optional[int]
        if incoming_link is None:
            self._incoming_link = None
            self._incoming_processor = incoming_processor
        else:
            if incoming_processor is None:
                self._incoming_link = incoming_link
                self._incoming_processor = None
            else:
                raise PacmanInvalidParameterException(
                    "The incoming direction for a path can only be from "
                    "either one link or one processors, not both",
                    str(incoming_link), str(incoming_processor))

    @property
    def incoming_link(self) -> Optional[int]:
        """
        The source link for this path entry.

        :rtype: int or None
        """
        return self._incoming_link

    @incoming_link.setter
    def incoming_link(self, incoming_link: int):
        if self._incoming_processor is not None:
            raise PacmanConfigurationException(
                f"Entry already has an incoming processor "
                f"{self._incoming_processor}")
        if (self._incoming_link is not None
                and self._incoming_link != incoming_link):
            raise PacmanConfigurationException(
                f"Entry already has a different incoming value "
                f"{self._incoming_link}")
        self._incoming_link = incoming_link

    @property
    def incoming_processor(self) -> Optional[int]:
        """
        The source processor.

        :rtype: int or None
        """
        return self._incoming_processor

    @incoming_processor.setter
    def incoming_processor(self, incoming_processor: int):
        if (self._incoming_processor is not None and
                self._incoming_processor != incoming_processor):
            raise PacmanConfigurationException(
                f"Entry already has an different incoming processor "
                f"{self._incoming_processor}")
        if self._incoming_link is not None:
            raise PacmanConfigurationException(
                f"Entry already has an incoming link {self._incoming_link}")
        self._incoming_processor = self._incoming_processor

    @property
    def defaultable(self) -> bool:
        """
        The defaultable status of the entry.

        :rtype: bool
        """
        # without an incoming link is is not defaultable
        if self._incoming_link is None:
            return False
        # defaultable if the output route is exactly the inverse of the input
        invert_link = ((self._incoming_link + 3) % 6)
        # as it is faster to go from a link to a spinnaker route
        return (self._calc_spinnaker_route(None, invert_link) ==
                self.spinnaker_route)

    def merge_entry(self, other: MulticastRoutingTableByPartitionEntry) -> \
            MulticastRoutingTableByPartitionEntry:
        """
        Merges the another entry with this one and returns a new
        MulticastRoutingTableByPartitionEntry

        :param MulticastRoutingTableByPartitionEntry other:
            the entry to merge into this one
        :return: a merged MulticastRoutingTableByPartitionEntry
        :raises PacmanInvalidParameterException:
        """
        # pylint:disable=protected-access
        if not isinstance(other, MulticastRoutingTableByPartitionEntry):
            raise PacmanInvalidParameterException(
                "other", "type error",
                "The other parameter is not an instance of "
                "MulticastRoutingTableByPartitionEntry, and therefore cannot "
                "be merged.")

        if self._incoming_processor is None:
            incoming_processor = other._incoming_processor
        elif (other._incoming_processor is None or
              self._incoming_processor == other._incoming_processor):
            incoming_processor = self._incoming_processor
        else:
            raise PacmanInvalidParameterException(
                "other", other,
                "The two MulticastRoutingTableByPartitionEntry have "
                "different incoming processors, and so can't be merged")

        if self._incoming_link is None:
            incoming_link = other._incoming_link
        elif (other._incoming_link is None or
              self._incoming_link == other._incoming_link):
            incoming_link = self._incoming_link
        else:
            raise PacmanInvalidParameterException(
                "other", other,
                "The two MulticastRoutingTableByPartitionEntry have "
                "different incoming links, and so can't be merged")

        # init checks if both incoming values are not None
        return MulticastRoutingTableByPartitionEntry(
            None, None, incoming_processor, incoming_link,
            spinnaker_route=self.spinnaker_route | other.spinnaker_route)

    def __repr__(self) -> str:
        return (f"{self.incoming_link}:{self.incoming_processor}:"
                f"{self.defaultable}:"
                f"{{{', '.join(map(str, self.link_ids))}}}:"
                f"{{{', '.join(map(str, self.processor_ids))}}}")

    def has_same_route(
            self, entry: MulticastRoutingTableByPartitionEntry) -> bool:
        """
        Checks if the two Entries have the same routes after applying mask

        :param  MulticastRoutingTableByPartitionEntry entry:
        :rtype: bool
        """
        # pylint:disable=protected-access
        # False if the outgoing processor of links are different
        if self.spinnaker_route != entry.spinnaker_route:
            return False
        # True if the incoming links are the same
        return self._incoming_link == entry._incoming_link

1	# Copyright (c) 2015 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 logging	1✔
16	from typing import Iterable, Optional, Union	1✔
17	from spinn_utilities.log import FormatAdapter	1✔
18	from spinn_machine.base_multicast_routing_entry import (	1✔
19	BaseMulticastRoutingEntry)
20	from pacman.exceptions import (	1✔
21	PacmanConfigurationException, PacmanInvalidParameterException)
22
23	log = FormatAdapter(logging.getLogger(__name__))	1✔
24
25
26	class MulticastRoutingTableByPartitionEntry(BaseMulticastRoutingEntry):	1✔
27	"""
28	An entry in a path of a multicast route.
29	"""
30
31	__slots__ = ["_incoming_processor", "_incoming_link"]	1✔
32
33	def __init__(self, out_going_links: Union[int, Iterable[int], None],	1✔
34	outgoing_processors: Union[int, Iterable[int], None],
35	incoming_processor: Optional[int] = None,
36	incoming_link: Optional[int] = None,
37	spinnaker_route: Optional[int] = None):
38	"""
39	.. note::
40	If a spinnaker_route is provided the out_going_links and
41	outgoing_processors parameters are ignored.
42
43	:param out_going_links:
44	the edges this path entry goes down, each of which is between
45	0 and 5
46	:type out_going_links: int or iterable(int) or None
47	:param outgoing_processors:
48	the processors this path entry goes to, each of which is between
49	0 and 17
50	:type outgoing_processors: int or iterable(int) or None
51	:param int incoming_processor:
52	the direction this entry came from (between 0 and 17)
53	:param int incoming_link:
54	the direction this entry came from in link (between 0 and 5)
55	:param spinnaker_route:
56	The processor_ids and link_ids expressed as a single int.
57	:type spinnaker_route: int or None
58
59	:raises PacmanInvalidParameterException:
60	"""
61	super().__init__(	1✔
62	outgoing_processors, out_going_links,
63	spinnaker_route=spinnaker_route)
64
65	self._incoming_link: Optional[int]	1✔
66	self._incoming_processor: Optional[int]	1✔
67	if incoming_link is None:	1✔
68	self._incoming_link = None	1✔
69	self._incoming_processor = incoming_processor	1✔
70	else:
71	if incoming_processor is None:	1✔
72	self._incoming_link = incoming_link	1✔
73	self._incoming_processor = None	1✔
74	else:
75	raise PacmanInvalidParameterException(	1✔
76	"The incoming direction for a path can only be from "
77	"either one link or one processors, not both",
78	str(incoming_link), str(incoming_processor))
79
80	@property	1✔
81	def incoming_link(self) -> Optional[int]:	1✔
82	"""
83	The source link for this path entry.
84
85	:rtype: int or None
86	"""
87	return self._incoming_link	1✔
88
89	@incoming_link.setter	1✔
90	def incoming_link(self, incoming_link: int):	1✔
UNCOV 91	if self._incoming_processor is not None:	×
92	raise PacmanConfigurationException(	×
93	f"Entry already has an incoming processor "
94	f"{self._incoming_processor}")
UNCOV 95	if (self._incoming_link is not None	×
96	and self._incoming_link != incoming_link):
97	raise PacmanConfigurationException(	×
98	f"Entry already has a different incoming value "
99	f"{self._incoming_link}")
UNCOV 100	self._incoming_link = incoming_link	×
101
102	@property	1✔
103	def incoming_processor(self) -> Optional[int]:	1✔
104	"""
105	The source processor.
106
107	:rtype: int or None
108	"""
109	return self._incoming_processor	1✔
110
111	@incoming_processor.setter	1✔
112	def incoming_processor(self, incoming_processor: int):	1✔
UNCOV 113	if (self._incoming_processor is not None and	×
114	self._incoming_processor != incoming_processor):
115	raise PacmanConfigurationException(	×
116	f"Entry already has an different incoming processor "
117	f"{self._incoming_processor}")
UNCOV 118	if self._incoming_link is not None:	×
119	raise PacmanConfigurationException(	×
120	f"Entry already has an incoming link {self._incoming_link}")
UNCOV 121	self._incoming_processor = self._incoming_processor	×
122
123	@property	1✔
124	def defaultable(self) -> bool:	1✔
125	"""
126	The defaultable status of the entry.
127
128	:rtype: bool
129	"""
130	# without an incoming link is is not defaultable
131	if self._incoming_link is None:	1✔
132	return False	1✔
133	# defaultable if the output route is exactly the inverse of the input
134	invert_link = ((self._incoming_link + 3) % 6)	1✔
135	# as it is faster to go from a link to a spinnaker route
136	return (self._calc_spinnaker_route(None, invert_link) ==	1✔
137	self.spinnaker_route)
138
139	def merge_entry(self, other: MulticastRoutingTableByPartitionEntry) -> \	1✔
140	MulticastRoutingTableByPartitionEntry:
141	"""
142	Merges the another entry with this one and returns a new
143	MulticastRoutingTableByPartitionEntry
144
145	:param MulticastRoutingTableByPartitionEntry other:
146	the entry to merge into this one
147	:return: a merged MulticastRoutingTableByPartitionEntry
148	:raises PacmanInvalidParameterException:
149	"""
150	# pylint:disable=protected-access
151	if not isinstance(other, MulticastRoutingTableByPartitionEntry):	1!
152	raise PacmanInvalidParameterException(	×
153	"other", "type error",
154	"The other parameter is not an instance of "
155	"MulticastRoutingTableByPartitionEntry, and therefore cannot "
156	"be merged.")
157
158	if self._incoming_processor is None:	1✔
159	incoming_processor = other._incoming_processor	1✔
160	elif (other._incoming_processor is None or	1!
161	self._incoming_processor == other._incoming_processor):
162	incoming_processor = self._incoming_processor	1✔
163	else:
164	raise PacmanInvalidParameterException(	×
165	"other", other,
166	"The two MulticastRoutingTableByPartitionEntry have "
167	"different incoming processors, and so can't be merged")
168
169	if self._incoming_link is None:	1✔
170	incoming_link = other._incoming_link	1✔
171	elif (other._incoming_link is None or	1!
172	self._incoming_link == other._incoming_link):
173	incoming_link = self._incoming_link	1✔
174	else:
175	raise PacmanInvalidParameterException(	×
176	"other", other,
177	"The two MulticastRoutingTableByPartitionEntry have "
178	"different incoming links, and so can't be merged")
179
180	# init checks if both incoming values are not None
181	return MulticastRoutingTableByPartitionEntry(	1✔
182	None, None, incoming_processor, incoming_link,
183	spinnaker_route=self.spinnaker_route \| other.spinnaker_route)
184
185	def __repr__(self) -> str:	1✔
186	return (f"{self.incoming_link}:{self.incoming_processor}:"	1✔
187	f"{self.defaultable}:"
188	f"{{{', '.join(map(str, self.link_ids))}}}:"
189	f"{{{', '.join(map(str, self.processor_ids))}}}")
190
191	def has_same_route(	1✔
192	self, entry: MulticastRoutingTableByPartitionEntry) -> bool:
193	"""
194	Checks if the two Entries have the same routes after applying mask
195
196	:param MulticastRoutingTableByPartitionEntry entry:
197	:rtype: bool
198	"""
199	# pylint:disable=protected-access
200	# False if the outgoing processor of links are different
201	if self.spinnaker_route != entry.spinnaker_route:	1✔
202	return False	1✔
203	# True if the incoming links are the same
204	return self._incoming_link == entry._incoming_link	1✔

SpiNNakerManchester / PACMAN / 8433946879

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