Merge pull request #540 from SpiNNakerManchester/placement_json

fix placement json code

pacman/utilities/json_utils.py

@@ -17,14 +17,19 @@
 
 import json
 import gzip
+from typing import cast, Iterable, List, Union
+
+from spinn_utilities.typing.json import JsonArray, JsonObject
+
 from pacman.data import PacmanDataView
+from pacman.model.graphs.machine import MachineVertex, SimpleMachineVertex
+from pacman.model.placements.placement import Placement
 from pacman.model.resources import (
     IPtagResource, ReverseIPtagResource, VariableSDRAM)
-from pacman.model.graphs.machine import SimpleMachineVertex
-from pacman.model.placements.placement import Placement
+from pacman.model.routing_info import BaseKeyAndMask
 
 
-def json_to_object(json_object):
+def json_to_object(json_object: Union[str, JsonObject]) -> JsonObject:
     """
     Makes sure this is a JSON object reading in a file if required
 
@@ -43,18 +48,30 @@ def json_to_object(json_object):
     return json_object
 
 
-def key_mask_to_json(key_mask):
+def key_mask_to_json(key_mask: BaseKeyAndMask) -> JsonObject:
+    """
+    Converts a BaseKeyAndMask into json
+
+    :param BaseKeyAndMask key_mask:
+    :rtype: dict(str, object)
+    """
     try:
-        json_object = dict()
+        json_object: JsonObject = dict()
         json_object["key"] = key_mask.key
         json_object["mask"] = key_mask.mask
     except Exception as ex:  # pylint: disable=broad-except
         json_object["exception"] = str(ex)
     return json_object
 
 
-def iptag_resource_to_json(iptag):
-    json_dict = dict()
+def iptag_resource_to_json(iptag: IPtagResource) -> JsonObject:
+    """
+    Converts an iptag to json
+
+    :param IPtagResource iptag:
+    :rtype: dict(str, object)
+    """
+    json_dict: JsonObject = dict()
     try:
         json_dict["ip_address"] = iptag.ip_address
         if iptag.port is not None:
@@ -68,30 +85,54 @@ def iptag_resource_to_json(iptag):
     return json_dict
 
 
-def iptag_resource_from_json(json_dict):
-    port = json_dict.get("port")
-    tag = json_dict.get("tag")
+def iptag_resource_from_json(json_dict: JsonObject) -> IPtagResource:
+    """
+    Creates an iptag from json
+
+    :param dict(str, object) json_dict:
+    :rtype: IPtagResource
+    """
     return IPtagResource(
-        json_dict["ip_address"], port, json_dict["strip_sdp"], tag,
-        json_dict["traffic_identifier"])
+        cast(str, json_dict["ip_address"]), cast(int, json_dict.get("port")),
+        cast(bool, json_dict["strip_sdp"]), cast(int, json_dict.get("tag")),
+        cast(str, json_dict["traffic_identifier"]))
+
 
+def iptag_resources_to_json(iptags: Iterable[IPtagResource]) -> JsonArray:
+    """
+    Converts a list of iptags to json.
 
-def iptag_resources_to_json(iptags):
-    json_list = []
+    :param list(IPtagResource) iptags:
+    :rtype: list
+    """
+    json_list: JsonArray = []
     for iptag in iptags:
         json_list.append(iptag_resource_to_json(iptag))
     return json_list
 
 
-def iptag_resources_from_json(json_list):
+def iptag_resources_from_json(
+        json_list: List[JsonObject]) -> List[IPtagResource]:
+    """
+    Creates a list of iptags from json.
+
+    :param list(dict(str, object)) json_list:
+    :rtype: list(IPtagResource)
+    """
     iptags = []
     for json_dict in json_list:
         iptags.append(iptag_resource_from_json(json_dict))
     return iptags
 
 
-def reverse_iptag_to_json(iptag):
-    json_dict = dict()
+def reverse_iptag_to_json(iptag: ReverseIPtagResource) -> JsonObject:
+    """
+    Converts a reverse iptag to json
+
+    :param ReverseIPtagResource iptag:
+    :rtype: dict(object)
+    """
+    json_dict: JsonObject = dict()
     try:
         if iptag.port is not None:
             json_dict["port"] = iptag.port
@@ -103,29 +144,55 @@ def reverse_iptag_to_json(iptag):
     return json_dict
 
 
-def reverse_iptag_from_json(json_dict):
-    port = json_dict.get("port")
-    sdp_port = json_dict["sdp_port"]
-    tag = json_dict.get("tag")
+def reverse_iptag_from_json(json_dict: JsonObject) -> ReverseIPtagResource:
+    """
+    Creates a ReverseIPtagResource based on json
+
+    :param dict(str, object) json_dict:
+    :rtype: ReverseIPtagResource
+    """
+    port = cast(int, json_dict.get("port"))
+    sdp_port = cast(int, json_dict["sdp_port"])
+    tag = cast(int, json_dict.get("tag"))
     return ReverseIPtagResource(port, sdp_port, tag)
 
 
-def reverse_iptags_to_json(iptags):
-    json_list = []
+def reverse_iptags_to_json(
+        iptags: Iterable[ReverseIPtagResource]) -> JsonArray:
+    """
+    Converts a list of reverse iptags to json
+
+    :param list(ReverseIPtagResource) iptags:
+    :rtype: list
+    """
+    json_list: JsonArray = []
     for iptag in iptags:
         json_list.append(reverse_iptag_to_json(iptag))
     return json_list
 
 
-def reverse_iptags_from_json(json_list):
+def reverse_iptags_from_json(
+        json_list: List[JsonObject]) -> List[ReverseIPtagResource]:
+    """
+    Creates a list of ReverseIPtagResource from json
+
+    :param list json_list:
+    :type: list(ReverseIPtagResource)
+    """
     iptags = []
     for json_dict in json_list:
         iptags.append(reverse_iptag_from_json(json_dict))
     return iptags
 
 
-def vertex_to_json(vertex):
-    json_dict = dict()
+def vertex_to_json(vertex: MachineVertex) -> JsonObject:
+    """
+    Converts a Machine Vertex to json.
+
+    :param MachineVertex vertex:
+    :rtype: dict(str, object)
+    """
+    json_dict: JsonObject = dict()
     try:
         json_dict["class"] = vertex.__class__.__name__
         json_dict["label"] = vertex.label
@@ -140,24 +207,33 @@ def vertex_to_json(vertex):
     return json_dict
 
 
-def vertex_from_json(json_dict):
+def vertex_from_json(json_dict: JsonObject) -> SimpleMachineVertex:
+    """
+    Creates a simple Vertex based on the json
+
+    :param dict(str, object) json_dict:
+    :rtype:  SimpleMachineVertex
+    """
     sdram = VariableSDRAM(
-        json_dict["fixed_sdram"], json_dict["per_timestep_sdram"])
-    iptags = iptag_resources_from_json(json_dict["iptags"])
-    reverse_iptags = reverse_iptags_from_json(json_dict["reverse_iptags"])
+        cast(int, json_dict["fixed_sdram"]),
+        cast(float, json_dict["per_timestep_sdram"]))
+    iptags = iptag_resources_from_json(
+        cast(List[JsonObject], json_dict["iptags"]))
+    reverse_iptags = reverse_iptags_from_json(
+        cast(List[JsonObject], json_dict["reverse_iptags"]))
     return SimpleMachineVertex(
         sdram=sdram, label=json_dict["label"], iptags=iptags,
         reverse_iptags=reverse_iptags)
 
 
-def vertex_lookup(label, graph=None):
-    if graph:
-        return graph.vertex_by_label(label)
-    return SimpleMachineVertex(None, label)
-
+def placement_to_json(placement: Placement) -> JsonObject:
+    """
+    Converts a Placement to json
 
-def placement_to_json(placement):
-    json_dict = dict()
+    :param Placement placement:
+    :rtype: dict(str, object)
+    """
+    json_dict: JsonObject = dict()
     try:
         json_dict["vertex_label"] = placement.vertex.label
         json_dict["x"] = placement.x
@@ -168,14 +244,25 @@ def placement_to_json(placement):
     return json_dict
 
 
-def placements_to_json():
-    json_list = []
+def placements_to_json() -> JsonArray:
+    """
+    Gets a json description of the placements (held in DataView)
+    """
+    json_list: JsonArray = []
     for placement in PacmanDataView.iterate_placemements():
         json_list.append(placement_to_json(placement))
     return json_list
 
 
-def placement_from_json(json_dict, graph=None):
-    vertex = vertex_lookup(json_dict["vertex_label"], graph)
+def placement_from_json(json_dict: JsonObject) -> Placement:
+    """
+    Gets a Placement based on the json.
+
+    :param dict(str, object) json_dict:
+    :rtype: Placement
+    """
+    vertex = SimpleMachineVertex(None, cast(str, json_dict["vertex_label"]))
+    # The cast(int tells mypy to assume the value can be converted to an int
     return Placement(
-        vertex, int(json_dict["x"]), int(json_dict["y"]), int(json_dict["p"]))
+        vertex, int(cast(int, json_dict["x"])),
+        int(cast(int, json_dict["y"])), int(cast(int, json_dict["p"])))

unittests/utilities_tests/test_json_utils.py

@@ -14,9 +14,11 @@
 import unittest
 import json
 from pacman.config_setup import unittest_setup
+from pacman.model.placements import Placement
 from pacman.model.resources import (
     ConstantSDRAM, IPtagResource, ReverseIPtagResource)
-from pacman.utilities.json_utils import (vertex_to_json, vertex_from_json)
+from pacman.utilities.json_utils import (
+    placement_from_json, placement_to_json, vertex_to_json, vertex_from_json)
 from pacman.model.graphs.machine import SimpleMachineVertex
 
 
@@ -47,6 +49,12 @@ def _compare_vertex(self, v1, v2, seen=None):
         self.assertEqual(v1.sdram_required, v2.sdram_required)
         seen.append(v1.label)
 
+    def _compare_placement(self, p1, p2, seen=None):
+        self.assertEqual(p1.x, p2.x)
+        self.assertEqual(p1.y, p2.y)
+        self.assertEqual(p1.p, p2.p)
+        self.assertEqual(p1.vertex.label, p2.vertex.label)
+
     # ------------------------------------------------------------------
     # Composite JSON round-trip testing schemes
     # ------------------------------------------------------------------
@@ -58,6 +66,13 @@ def vertex_there_and_back(self, there):
         back = vertex_from_json(j_object2)
         self._compare_vertex(there, back)
 
+    def placement_there_and_back(self, there):
+        j_object = placement_to_json(there)
+        j_str = json.dumps(j_object)
+        j_object2 = json.loads(j_str)
+        back = placement_from_json(j_object2)
+        self._compare_placement(there, back)
+
     # ------------------------------------------------------------------
     # Test cases
     # ------------------------------------------------------------------
@@ -69,3 +84,12 @@ def test_vertex(self):
             reverse_iptags=[ReverseIPtagResource(port=25, sdp_port=2, tag=5)],
             label="Vertex")
         self.vertex_there_and_back(s1)
+
+    def test_placement(self):
+        s1 = SimpleMachineVertex(
+            sdram=ConstantSDRAM(0),
+            iptags=[IPtagResource("127.0.0.1", port=None, strip_sdp=True)],
+            reverse_iptags=[ReverseIPtagResource(port=25, sdp_port=2, tag=5)],
+            label="PVertex")
+        p1 = Placement(s1, 1, 2, 3)
+        self.placement_there_and_back(p1)