move v_to_p_map from Chip to DataView

spinn_machine/chip.py

@@ -34,8 +34,7 @@ def __new__(cls, x: int, y: int, n_processors: int, router: Router,
                 ip_address: Optional[str] = None,
                 tag_ids: Optional[Iterable[int]] = None,
                 down_cores: Optional[Collection[int]] = None,
-                parent_link: Optional[int] = None,
-                v_to_p_map: Optional[bytes] = None):
+                parent_link: Optional[int] = None):
         return tuple.__new__(cls, (x, y))
 
     # pylint: disable=too-many-arguments, wrong-spelling-in-docstring
@@ -45,8 +44,7 @@ def __init__(self, x: int, y: int, n_processors: int, router: Router,
                  ip_address: Optional[str] = None,
                  tag_ids: Optional[Iterable[int]] = None,
                  down_cores: Optional[Collection[int]] = None,
-                 parent_link: Optional[int] = None,
-                 v_to_p_map: Optional[bytes] = None):
+                 parent_link: Optional[int] = None):
         """
         :param int x: the x-coordinate of the chip's position in the
             two-dimensional grid of chips
@@ -96,7 +94,6 @@ def __init__(self, x: int, y: int, n_processors: int, router: Router,
         self._nearest_ethernet_x = nearest_ethernet_x
         self._nearest_ethernet_y = nearest_ethernet_y
         self._parent_link = parent_link
-        self._v_to_p_map = v_to_p_map
 
     def __generate_processors(
             self, n_processors: int,
@@ -267,40 +264,14 @@ def parent_link(self) -> Optional[int]:
         """
         return self._parent_link
 
-    def get_physical_core_id(self, virtual_p: int) -> Optional[int]:
-        """
-        Get the physical core ID from a virtual core ID.
-
-        :param int virtual_p: The virtual core ID
-        :rtype: int or None if core not in map
-        """
-        if (self._v_to_p_map is None or virtual_p >= len(self._v_to_p_map) or
-                self._v_to_p_map[virtual_p] == 0xFF):
-            return None
-        return self._v_to_p_map[virtual_p]
-
-    def get_physical_core_string(self, virtual_p: int) -> str:
-        """
-        Get a string that can be appended to a core to show the physical
-        core, or an empty string if not possible.
-
-        :param int virtual_p: The virtual core ID
-        :rtype: str
-        """
-        physical_p = self.get_physical_core_id(virtual_p)
-        if physical_p is None:
-            return ""
-        return f" (ph: {physical_p})"
-
     def __str__(self) -> str:
         if self._ip_address:
             ip_info = f"ip_address={self.ip_address} "
         else:
             ip_info = ""
         return (
             f"[Chip: x={self[0]}, y={self[1]}, {ip_info}"
-            f"n_cores={self.n_processors}, "
-            f"mon={self.get_physical_core_id(0)}]")
+            f"n_cores={self.n_processors}]")
 
     def __repr__(self) -> str:
         return self.__str__()

spinn_machine/data/machine_data_view.py

@@ -12,11 +12,12 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 from __future__ import annotations
-from typing import Callable, Optional, TYPE_CHECKING
+from typing import Callable, Dict, Optional, TYPE_CHECKING
 from spinn_utilities.typing.coords import XY
 from spinn_utilities.data import UtilsDataView
 from spinn_machine.exceptions import SpinnMachineException
 from spinn_machine.version.version_factory import version_factory
+from spinn_machine.version.version_spin1 import VersionSpin1
 if TYPE_CHECKING:
     from spinn_machine.chip import Chip
     from spinn_machine.machine import Machine
@@ -46,7 +47,8 @@ class _MachineDataModel(object):
         "_machine",
         "_machine_generator",
         "_machine_version",
-        "_user_accessed_machine"
+        "_user_accessed_machine",
+        "_v_to_p_map"
     ]
 
     def __new__(cls) -> '_MachineDataModel':
@@ -74,6 +76,7 @@ def _hard_reset(self) -> None:
         """
         self._soft_reset()
         self._machine: Optional[Machine] = None
+        self._v_to_p_map: Optional[Dict[XY, bytes]] = None
         self._user_accessed_machine = False
 
     def _soft_reset(self) -> None:
@@ -263,3 +266,62 @@ def get_machine_version(cls) -> AbstractVersion:
         if cls.__data._machine_version is None:
             cls.__data._machine_version = version_factory()
         return cls.__data._machine_version
+
+    @classmethod
+    def set_v_to_p_map(cls, v_to_p_map: Dict[XY, bytes]):
+        """
+        Registers the mapping from Virtual to int physical core ids
+
+        Note: Only expected to be used in Version 1
+
+        :param dict((int, int), bytes) v_to_p_map:
+        """
+        if cls.__data._v_to_p_map is None:
+            cls.__data._v_to_p_map = v_to_p_map
+        else:
+            raise SpinnMachineException(
+                "Unexpected second call to set_v_to_p_map")
+
+    @classmethod
+    def get_physical_core_id(cls, xy: XY, virtual_p: int) -> Optional[int]:
+        """
+        Get the physical core ID from a virtual core ID.
+
+        Note: This call only works for Version 1
+
+        :param (int, int) xy: The Chip or its XY coordinates
+        :param int virtual_p: The virtual core ID
+        :rtype: int or None if core not in map
+        """
+        if cls.__data._v_to_p_map is None:
+            version = cls.get_machine_version()
+            if isinstance(version, VersionSpin1):
+                raise SpinnMachineException(
+                    "Virtual to physical mapping to yet read")
+            else:
+                raise SpinnMachineException(
+                    f"This call is not supported when using Version {version}")
+        if xy in cls.__data._v_to_p_map:
+            v_to_p_map = cls.__data._v_to_p_map[xy]
+        else:
+            return None
+        if (virtual_p >= len(v_to_p_map) or v_to_p_map[virtual_p] == 0xFF):
+            return None
+        return v_to_p_map[virtual_p]
+
+    @classmethod
+    def get_physical_core_string(cls, xy: XY, virtual_p: int) -> str:
+        """
+        Returns a String representing the physical core
+
+        :param (int, int) xy: The Chip or its XY coordinates
+        :param virtual_p: The virtual (python) id for the core
+        :rtype: str
+        """
+        if cls.__data._v_to_p_map is not None:
+            physical_p = cls.get_physical_core_id(xy, virtual_p)
+            if physical_p is None:
+                return ""
+            return f" (ph: {physical_p})"
+        else:
+            return ""

unittests/data/test_data.py

@@ -19,6 +19,7 @@
 from spinn_machine.config_setup import unittest_setup
 from spinn_machine.data import MachineDataView
 from spinn_machine.data.machine_data_writer import MachineDataWriter
+from spinn_machine.exceptions import SpinnMachineException
 
 
 class TestSimulatorData(unittest.TestCase):
@@ -94,3 +95,26 @@ def test_where_is_setup(self):
             "None",
             MachineDataView.where_is_chip(None)
         )
+
+    def test_v_to_p(self):
+        writer = MachineDataWriter.setup()
+        # Before setting
+        with self.assertRaises(SpinnMachineException):
+            writer.get_physical_core_id((1, 2), 3)
+        self.assertEqual("", writer.get_physical_core_string((1, 2), 3))
+
+        # Set a v_to_p
+        v_to_p = dict()
+        v_to_p[(1, 2)] = bytes([10, 11, 12, 13, 14])
+        writer.set_v_to_p_map(v_to_p)
+        # XY that exists
+        self.assertEqual(13, writer.get_physical_core_id((1, 2), 3))
+        self.assertEqual(" (ph: 13)",
+                         writer.get_physical_core_string((1, 2), 3))
+        # Xy that does not exist
+        self.assertEqual(None, writer.get_physical_core_id((1, 4), 3))
+        self.assertEqual("",
+                         writer.get_physical_core_string((1, 4), 3))
+        self.assertEqual(None, writer.get_physical_core_id((1, 2), 19))
+        self.assertEqual("",
+                         writer.get_physical_core_string((1, 2), 19))

unittests/test_chip.py

@@ -58,8 +58,7 @@ def test_create_chip(self):
         self.assertEqual(new_chip.n_placable_processors, self.n_processors - 1)
         print(new_chip.__repr__())
         self.assertEqual(
-            "[Chip: x=0, y=1, ip_address=192.162.240.253 "
-            "n_cores=18, mon=None]",
+            "[Chip: x=0, y=1, ip_address=192.162.240.253 n_cores=18]",
             new_chip.__repr__(),)
         self.assertEqual(new_chip.tag_ids, OrderedSet([1, 2, 3, 4, 5, 6, 7]))
         self.assertTrue(new_chip.is_processor_with_id(3))