6308825518

Committed 19 Sep 2023 06:06AM UTC coverage: 83.538% (+0.4%) from 83.103%

Build # 6308825518

Build Type

Pull #513

github

Committed by

Christian-B

Commit Message

merged in master

Pull Request Pull Request #513: Type Annotations and Checking

Run Details

1551 of 2034 branches covered (0.0%)

Branch coverage included in aggregate %.

1926 of 1926 new or added lines in 96 files covered. (100.0%)

5051 of 5869 relevant lines covered (86.06%)

0.86 hits per line

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

93.24

/pacman/model/routing_tables/multicast_routing_tables.py

# Copyright (c) 2014 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 json
import gzip
from typing import (
    Collection, Dict, Iterable, Iterator, Optional, Union, cast)
from spinn_utilities.typing.coords import XY
from spinn_utilities.typing.json import JsonObjectArray
from pacman.exceptions import PacmanAlreadyExistsException
from .abstract_multicast_routing_table import AbstractMulticastRoutingTable
from .uncompressed_multicast_routing_table import (
    UnCompressedMulticastRoutingTable)
from spinn_machine import MulticastRoutingEntry


class MulticastRoutingTables(object):
    """
    Represents the multicast routing tables for a number of chips.

    .. note::
        The tables in an instance of this class should be either all
        uncompressed tables, or all compressed tables.
    """

    __slots__ = (
        # dict of (x,y) -> routing table
        "_routing_tables_by_chip",
    )

    def __init__(self,
                 routing_tables: Iterable[AbstractMulticastRoutingTable] = ()):
        """
        :param iterable(AbstractMulticastRoutingTable) routing_tables:
            The routing tables to add
        :raise PacmanAlreadyExistsException:
            If any two routing tables are for the same chip
        """
        self._routing_tables_by_chip: Dict[
            XY, AbstractMulticastRoutingTable] = dict()

        for routing_table in routing_tables:
            self.add_routing_table(routing_table)

    def add_routing_table(self, routing_table: AbstractMulticastRoutingTable):
        """
        Add a routing table.

        :param AbstractMulticastRoutingTable routing_table:
            a routing table to add
        :raise PacmanAlreadyExistsException:
            If a routing table already exists for the chip
        """
        if (routing_table.x, routing_table.y) in self._routing_tables_by_chip:
            raise PacmanAlreadyExistsException(
                "The Routing table for chip "
                f"{routing_table.x}:{routing_table.y} already exists in this "
                "collection and therefore is deemed an error to re-add it",
                str(routing_table))
        self._routing_tables_by_chip[routing_table.x, routing_table.y] = \
            routing_table

    @property
    def routing_tables(self) -> Collection[AbstractMulticastRoutingTable]:
        """
        The routing tables stored within.

        :return: an iterable of routing tables
        :rtype: iterable(AbstractMulticastRoutingTable)
        """
        return self._routing_tables_by_chip.values()

    def get_max_number_of_entries(self) -> int:
        """
        The maximum number of multicast routing entries there are in any
        multicast routing table.

        Will return zero if there are no routing tables

        :rtype: int
        """
        if self._routing_tables_by_chip:
            return max(map((lambda x: x.number_of_entries),
                           self._routing_tables_by_chip.values()))
        else:
            return 0

    def get_total_number_of_entries(self) -> int:
        """
        The total number of multicast routing entries there are in all
        multicast routing table.

        Will return zero if there are no routing tables

        :rtype: int
        """
        if self._routing_tables_by_chip:
            return sum(map((lambda x: x.number_of_entries),
                           self._routing_tables_by_chip.values()))
        else:
            return 0

    def get_routing_table_for_chip(
            self, x: int, y: int) -> Optional[AbstractMulticastRoutingTable]:
        """
        Get a routing table for a particular chip.

        :param int x: The X-coordinate of the chip
        :param int y: The Y-coordinate of the chip
        :return: The routing table, or `None` if no such table exists
        :rtype: AbstractMulticastRoutingTable or None
        """
        return self._routing_tables_by_chip.get((x, y))

    def __iter__(self) -> Iterator[AbstractMulticastRoutingTable]:
        """
        Iterator for the multicast routing tables stored within.

        :return: iterator of multicast_routing_table
        :rtype: iterable(AbstractMulticastRoutingTable)
        """
        return iter(self._routing_tables_by_chip.values())


def to_json(router_table: MulticastRoutingTables) -> JsonObjectArray:
    return [
        {
            "x": routing_table.x,
            "y": routing_table.y,
            "entries": [
                {
                    "key": entry.routing_entry_key,
                    "mask": entry.mask,
                    "defaultable": entry.defaultable,
                    "spinnaker_route": entry.spinnaker_route
                }
                for entry in routing_table.multicast_routing_entries]
        }
        for routing_table in router_table]


def from_json(j_router: Union[str, JsonObjectArray]) -> MulticastRoutingTables:
    if isinstance(j_router, str):
        if j_router.endswith(".gz"):
            with gzip.open(j_router) as j_file:
                j_router = cast(JsonObjectArray, json.load(j_file))
        else:
            with open(j_router, encoding="utf-8") as j_file:
                j_router = cast(JsonObjectArray, json.load(j_file))

    tables = MulticastRoutingTables()
    for j_table in j_router:
        x = cast(int, j_table["x"])
        y = cast(int, j_table["y"])
        table = UnCompressedMulticastRoutingTable(x, y)
        tables.add_routing_table(table)
        for j_entry in cast(JsonObjectArray, j_table["entries"]):
            table.add_multicast_routing_entry(MulticastRoutingEntry(
                cast(int, j_entry["key"]), cast(int, j_entry["mask"]),
                defaultable=cast(bool, j_entry["defaultable"]),
                spinnaker_route=cast(int, j_entry["spinnaker_route"])))
    return tables

1	# Copyright (c) 2014 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
15	import json	1✔
16	import gzip	1✔
17	from typing import (	1✔
18	Collection, Dict, Iterable, Iterator, Optional, Union, cast)
19	from spinn_utilities.typing.coords import XY	1✔
20	from spinn_utilities.typing.json import JsonObjectArray	1✔
21	from pacman.exceptions import PacmanAlreadyExistsException	1✔
22	from .abstract_multicast_routing_table import AbstractMulticastRoutingTable	1✔
23	from .uncompressed_multicast_routing_table import (	1✔
24	UnCompressedMulticastRoutingTable)
25	from spinn_machine import MulticastRoutingEntry	1✔
26
27
28	class MulticastRoutingTables(object):	1✔
29	"""
30	Represents the multicast routing tables for a number of chips.
31
32	.. note::
33	The tables in an instance of this class should be either all
34	uncompressed tables, or all compressed tables.
35	"""
36
37	__slots__ = (	1✔
38	# dict of (x,y) -> routing table
39	"_routing_tables_by_chip",
40	)
41
42	def __init__(self,	1✔
43	routing_tables: Iterable[AbstractMulticastRoutingTable] = ()):
44	"""
45	:param iterable(AbstractMulticastRoutingTable) routing_tables:
46	The routing tables to add
47	:raise PacmanAlreadyExistsException:
48	If any two routing tables are for the same chip
49	"""
50	self._routing_tables_by_chip: Dict[	1✔
51	XY, AbstractMulticastRoutingTable] = dict()
52
53	for routing_table in routing_tables:	1✔
54	self.add_routing_table(routing_table)	1✔
55
56	def add_routing_table(self, routing_table: AbstractMulticastRoutingTable):	1✔
57	"""
58	Add a routing table.
59
60	:param AbstractMulticastRoutingTable routing_table:
61	a routing table to add
62	:raise PacmanAlreadyExistsException:
63	If a routing table already exists for the chip
64	"""
65	if (routing_table.x, routing_table.y) in self._routing_tables_by_chip:	1✔
66	raise PacmanAlreadyExistsException(	1✔
67	"The Routing table for chip "
68	f"{routing_table.x}:{routing_table.y} already exists in this "
69	"collection and therefore is deemed an error to re-add it",
70	str(routing_table))
71	self._routing_tables_by_chip[routing_table.x, routing_table.y] = \	1✔
72	routing_table
73
74	@property	1✔
75	def routing_tables(self) -> Collection[AbstractMulticastRoutingTable]:	1✔
76	"""
77	The routing tables stored within.
78
79	:return: an iterable of routing tables
80	:rtype: iterable(AbstractMulticastRoutingTable)
81	"""
82	return self._routing_tables_by_chip.values()	1✔
83
84	def get_max_number_of_entries(self) -> int:	1✔
85	"""
86	The maximum number of multicast routing entries there are in any
87	multicast routing table.
88
89	Will return zero if there are no routing tables
90
91	:rtype: int
92	"""
93	if self._routing_tables_by_chip:	1✔
94	return max(map((lambda x: x.number_of_entries),	1✔
95	self._routing_tables_by_chip.values()))
96	else:
97	return 0	1✔
98
99	def get_total_number_of_entries(self) -> int:	1✔
100	"""
101	The total number of multicast routing entries there are in all
102	multicast routing table.
103
104	Will return zero if there are no routing tables
105
106	:rtype: int
107	"""
108	if self._routing_tables_by_chip:	1!
109	return sum(map((lambda x: x.number_of_entries),	1✔
110	self._routing_tables_by_chip.values()))
111	else:
112	return 0	×
113
114	def get_routing_table_for_chip(	1✔
115	self, x: int, y: int) -> Optional[AbstractMulticastRoutingTable]:
116	"""
117	Get a routing table for a particular chip.
118
119	:param int x: The X-coordinate of the chip
120	:param int y: The Y-coordinate of the chip
121	:return: The routing table, or `None` if no such table exists
122	:rtype: AbstractMulticastRoutingTable or None
123	"""
124	return self._routing_tables_by_chip.get((x, y))	1✔
125
126	def __iter__(self) -> Iterator[AbstractMulticastRoutingTable]:	1✔
127	"""
128	Iterator for the multicast routing tables stored within.
129
130	:return: iterator of multicast_routing_table
131	:rtype: iterable(AbstractMulticastRoutingTable)
132	"""
133	return iter(self._routing_tables_by_chip.values())	1✔
134
135
136	def to_json(router_table: MulticastRoutingTables) -> JsonObjectArray:	1✔
137	return [	1✔
138	{
139	"x": routing_table.x,
140	"y": routing_table.y,
141	"entries": [
142	{
143	"key": entry.routing_entry_key,
144	"mask": entry.mask,
145	"defaultable": entry.defaultable,
146	"spinnaker_route": entry.spinnaker_route
147	}
148	for entry in routing_table.multicast_routing_entries]
149	}
150	for routing_table in router_table]
151
152
153	def from_json(j_router: Union[str, JsonObjectArray]) -> MulticastRoutingTables:	1✔
154	if isinstance(j_router, str):	1✔
155	if j_router.endswith(".gz"):	1!
156	with gzip.open(j_router) as j_file:	1✔
157	j_router = cast(JsonObjectArray, json.load(j_file))	1✔
158	else:
159	with open(j_router, encoding="utf-8") as j_file:	×
160	j_router = cast(JsonObjectArray, json.load(j_file))	×
161
162	tables = MulticastRoutingTables()	1✔
163	for j_table in j_router:	1✔
164	x = cast(int, j_table["x"])	1✔
165	y = cast(int, j_table["y"])	1✔
166	table = UnCompressedMulticastRoutingTable(x, y)	1✔
167	tables.add_routing_table(table)	1✔
168	for j_entry in cast(JsonObjectArray, j_table["entries"]):	1✔
169	table.add_multicast_routing_entry(MulticastRoutingEntry(	1✔
170	cast(int, j_entry["key"]), cast(int, j_entry["mask"]),
171	defaultable=cast(bool, j_entry["defaultable"]),
172	spinnaker_route=cast(int, j_entry["spinnaker_route"])))
173	return tables	1✔

SpiNNakerManchester / PACMAN / 6308825518

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