Remove 18

spinn_machine/base_multicast_routing_entry.py

@@ -0,0 +1,115 @@
+# 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.
+from __future__ import annotations
+from typing import Iterable, Optional, Set, Union
+from spinn_machine.constants import MAX_LINKS_PER_ROUTER
+from spinn_machine.data import MachineDataView
+
+
+class BaseMulticastRoutingEntry(object):
+    """
+    Represents an entry in a SpiNNaker chip's multicast routing table.
+    There can be up to 1024 such entries per chip, but some may be reserved
+    for system purposes.
+    """
+
+    # cache of MachineDataView.get_machine_version().max_cores_per_chip
+    max_cores_per_chip: int = 0
+
+    __slots__ = ["_spinnaker_route"]
+
+    def __init__(self, processor_ids:  Union[int, Iterable[int], None],
+                 link_ids:  Union[int, Iterable[int], None],
+                 spinnaker_route: Optional[int]):
+        """
+        .. note::
+            The processor_ids and link_ids parameters are only optional if a
+            spinnaker_route is provided. If a spinnaker_route is provided
+            the processor_ids and link_ids parameters are ignored.
+
+        :param processor_ids: The destination processor IDs
+        :type processor_ids: iterable(int) or None
+        :param link_ids: The destination link IDs
+        :type link_ids: iterable(int) or None
+        :param spinnaker_route:
+            The processor_ids and link_ids expressed as a single int.
+        :type spinnaker_route: int or None
+        :raise AssertionError: if no spinnaker_route provided and either
+            processor_ids or link_ids is missing or `None`
+        """
+        if self.max_cores_per_chip == 0:
+            BaseMulticastRoutingEntry.max_cores_per_chip = \
+                MachineDataView.get_machine_version().max_cores_per_chip
+
+        # Add processor IDs, ignore duplicates
+        if spinnaker_route is None:
+            self._spinnaker_route = \
+                self._calc_spinnaker_route(processor_ids, link_ids)
+        else:
+            self._spinnaker_route = spinnaker_route
+
+    @property
+    def processor_ids(self) -> Set[int]:
+        """
+        The destination processor IDs.
+
+        :rtype: set(int)
+        """
+        return set(pi for pi in range(
+            0, BaseMulticastRoutingEntry.max_cores_per_chip)
+                   if self._spinnaker_route & 1 << (MAX_LINKS_PER_ROUTER + pi))
+
+    @property
+    def link_ids(self) -> Set[int]:
+        """
+        The destination link IDs.
+
+        :rtype: frozenset(int)
+        """
+        return set(li for li in range(0, MAX_LINKS_PER_ROUTER)
+                   if self._spinnaker_route & 1 << li)
+
+    @property
+    def spinnaker_route(self) -> int:
+        """
+        The encoded SpiNNaker route.
+
+        :rtype: int
+        """
+        return self._spinnaker_route
+
+    def _calc_spinnaker_route(
+            self, processor_ids: Union[int, Iterable[int], None],
+            link_ids: Union[int, Iterable[int], None]) -> int:
+        """
+        create a binary routing table entry usable on the machine.
+
+        :rtype: int
+        """
+        route = 0
+        if processor_ids is None:
+            pass
+        elif isinstance(processor_ids, int):
+            route |= (1 << (MAX_LINKS_PER_ROUTER + processor_ids))
+        else:
+            for processor_id in processor_ids:
+                route |= (1 << (MAX_LINKS_PER_ROUTER + processor_id))
+        if link_ids is None:
+            pass
+        elif isinstance(link_ids, int):
+            route |= (1 << link_ids)
+        else:
+            for link_id in link_ids:
+                route |= (1 << link_id)
+        return route

spinn_machine/constants.py

@@ -0,0 +1,16 @@
+# Copyright (c) 2024 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.
+
+# The maximum number of links/directions a router can handle
+MAX_LINKS_PER_ROUTER = 6

spinn_machine/fixed_route_entry.py

@@ -11,11 +11,12 @@
 # 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 FrozenSet, Iterable, Optional
+from typing import Iterable, List, Optional
+from .base_multicast_routing_entry import BaseMulticastRoutingEntry
 from .exceptions import SpinnMachineAlreadyExistsException
 
 
-class FixedRouteEntry(object):
+class FixedRouteEntry(BaseMulticastRoutingEntry):
     """
     Describes the sole entry in a SpiNNaker chip's fixed route routing table.
     There is only one fixed route entry per chip.
@@ -27,18 +28,14 @@ class FixedRouteEntry(object):
         "_link_ids",
         "__repr")
 
-    def __init__(self, processor_ids: Iterable[int], link_ids: Iterable[int]):
+    def __init__(self, processor_ids: List[int], link_ids: List[int]):
         """
         :param iterable(int) processor_ids:
         :param iterable(int) link_ids:
         """
+        super().__init__(processor_ids, link_ids, spinnaker_route=None)
         self.__repr: Optional[str] = None
-        # Add processor IDs, checking that there is only one of each
-        self._processor_ids = frozenset(processor_ids)
         self.__check_dupes(processor_ids, "processor ID")
-
-        # Add link IDs, checking that there is only one of each
-        self._link_ids = frozenset(link_ids)
         self.__check_dupes(link_ids, "link ID")
 
     @staticmethod
@@ -49,27 +46,9 @@ def __check_dupes(sequence: Iterable[int], name: str):
                 raise SpinnMachineAlreadyExistsException(name, str(_id))
             check.add(_id)
 
-    @property
-    def processor_ids(self) -> FrozenSet[int]:
-        """
-        The destination processor IDs.
-
-        :rtype: frozenset(int)
-        """
-        return self._processor_ids
-
-    @property
-    def link_ids(self) -> FrozenSet[int]:
-        """
-        The destination link IDs.
-
-        :rtype: frozenset(int)
-        """
-        return self._link_ids
-
     def __repr__(self) -> str:
         if not self.__repr:
             self.__repr = ("{%s}:{%s}" % (
-                ", ".join(map(str, sorted(self._link_ids))),
-                ", ".join(map(str, sorted(self._processor_ids)))))
+                ", ".join(map(str, sorted(self.link_ids))),
+                ", ".join(map(str, sorted(self.processor_ids)))))
         return self.__repr

spinn_machine/json_machine.py

@@ -17,6 +17,7 @@
 from typing import NamedTuple, Union
 from spinn_utilities.log import FormatAdapter
 from spinn_utilities.typing.json import JsonArray, JsonObject, JsonValue
+from spinn_machine.constants import MAX_LINKS_PER_ROUTER
 from spinn_machine.data import MachineDataView
 from .chip import Chip
 from .router import Router
@@ -192,7 +193,7 @@ def _describe_chip(chip: Chip, standard, ethernet) -> JsonArray:
 
     dead_links: JsonArray = [
         link_id
-        for link_id in range(Router.MAX_LINKS_PER_ROUTER)
+        for link_id in range(MAX_LINKS_PER_ROUTER)
         if not chip.router.is_link(link_id)]
     if dead_links:
         details["deadLinks"] = dead_links

spinn_machine/multicast_routing_entry.py

@@ -12,21 +12,20 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 from __future__ import annotations
-from typing import Any, FrozenSet, Iterable, Optional, Tuple, overload
-from spinn_machine.router import Router
+from typing import Any, Iterable, Optional, overload
+from .base_multicast_routing_entry import BaseMulticastRoutingEntry
 from .exceptions import SpinnMachineInvalidParameterException
 
 
-class MulticastRoutingEntry(object):
+class MulticastRoutingEntry(BaseMulticastRoutingEntry):
     """
     Represents an entry in a SpiNNaker chip's multicast routing table.
     There can be up to 1024 such entries per chip, but some may be reserved
     for system purposes.
     """
 
     __slots__ = (
-        "_routing_entry_key", "_mask", "_defaultable", "_processor_ids",
-        "_link_ids", "_spinnaker_route", "__repr")
+        "_routing_entry_key", "_mask", "_defaultable", "__repr")
 
     @overload
     def __init__(self, routing_entry_key: int, mask: int, *,
@@ -68,12 +67,10 @@ def __init__(self, routing_entry_key: int, mask: int, *,
         :param spinnaker_route:
             The processor_ids and link_ids expressed as a single int.
         :type spinnaker_route: int or None
-        :raise spinn_machine.exceptions.SpinnMachineAlreadyExistsException:
-            * If processor_ids contains the same ID more than once
-            * If link_ids contains the same ID more than once
-        :raise TypeError: if no spinnaker_route provided and either
+        :raise AssertionError: if no spinnaker_route provided and either
             processor_ids or link_ids is missing or `None`
         """
+        super().__init__(processor_ids, link_ids, spinnaker_route)
         self._routing_entry_key: int = routing_entry_key
         self._mask: int = mask
         self._defaultable: bool = defaultable
@@ -87,19 +84,6 @@ def __init__(self, routing_entry_key: int, mask: int, *,
                 " is determined to be an error in the tool chain. Please "
                 "correct this and try again.")
 
-        # Add processor IDs, ignore duplicates
-        if spinnaker_route is None:
-            assert processor_ids is not None
-            assert link_ids is not None
-            self._processor_ids: Optional[FrozenSet[int]] = \
-                frozenset(processor_ids)
-            self._link_ids: Optional[FrozenSet[int]] = frozenset(link_ids)
-            self._spinnaker_route: int = self._calc_spinnaker_route()
-        else:
-            self._spinnaker_route = spinnaker_route
-            self._processor_ids = None
-            self._link_ids = None
-
     @property
     def routing_entry_key(self) -> int:
         """
@@ -118,28 +102,6 @@ def mask(self) -> int:
         """
         return self._mask
 
-    @property
-    def processor_ids(self) -> FrozenSet[int]:
-        """
-        The destination processor IDs.
-
-        :rtype: frozenset(int)
-        """
-        if self._processor_ids is None:
-            self._processor_ids, self._link_ids = self._calc_routing_ids()
-        return self._processor_ids
-
-    @property
-    def link_ids(self) -> FrozenSet[int]:
-        """
-        The destination link IDs.
-
-        :rtype: frozenset(int)
-        """
-        if self._link_ids is None:
-            self._processor_ids, self._link_ids = self._calc_routing_ids()
-        return self._link_ids
-
     @property
     def defaultable(self) -> bool:
         """
@@ -153,15 +115,6 @@ def defaultable(self) -> bool:
         """
         return self._defaultable
 
-    @property
-    def spinnaker_route(self) -> int:
-        """
-        The encoded SpiNNaker route.
-
-        :rtype: int
-        """
-        return self._spinnaker_route
-
     def merge(self, other: MulticastRoutingEntry) -> MulticastRoutingEntry:
         """
         Merges together two multicast routing entries.  The entry to merge
@@ -189,9 +142,8 @@ def merge(self, other: MulticastRoutingEntry) -> MulticastRoutingEntry:
 
         return MulticastRoutingEntry(
             self.routing_entry_key, self.mask,
-            processor_ids=(self.processor_ids | other.processor_ids),
-            link_ids=(self.link_ids | other.link_ids),
-            defaultable=(self._defaultable and other.defaultable))
+            defaultable=(self._defaultable and other.defaultable),
+            spinnaker_route=(self._spinnaker_route | other.spinnaker_route))
 
     def __add__(self, other: MulticastRoutingEntry) -> MulticastRoutingEntry:
         """
@@ -236,44 +188,3 @@ def __repr__(self) -> str:
 
     def __str__(self) -> str:
         return self.__repr__()
-
-    def __getstate__(self):
-        return dict(
-            (slot, getattr(self, slot))
-            for slot in self.__slots__
-            if hasattr(self, slot)
-        )
-
-    def __setstate__(self, state):
-        for slot, value in state.items():
-            setattr(self, slot, value)
-
-    def _calc_spinnaker_route(self) -> int:
-        """
-        Convert a routing table entry represented in software to a
-        binary routing table entry usable on the machine.
-
-        :rtype: int
-        """
-        route_entry = 0
-        assert self._processor_ids is not None
-        for processor_id in self._processor_ids:
-            route_entry |= (1 << (Router.MAX_LINKS_PER_ROUTER + processor_id))
-        assert self._link_ids is not None
-        for link_id in self._link_ids:
-            route_entry |= (1 << link_id)
-        return route_entry
-
-    def _calc_routing_ids(self) -> Tuple[FrozenSet[int], FrozenSet[int]]:
-        """
-        Convert a binary routing table entry usable on the machine to lists of
-        route IDs usable in a routing table entry represented in software.
-
-        :rtype: tuple(frozenset(int), frozenset(int))
-        """
-        processor_ids = (pi for pi in range(0, Router.MAX_CORES_PER_ROUTER)
-                         if self._spinnaker_route & 1 <<
-                         (Router.MAX_LINKS_PER_ROUTER + pi))
-        link_ids = (li for li in range(0, Router.MAX_LINKS_PER_ROUTER)
-                    if self._spinnaker_route & 1 << li)
-        return frozenset(processor_ids), frozenset(link_ids)