add basic from_json() implementation to topology

yaramo/base_element.py

@@ -2,6 +2,8 @@
 from typing import Tuple
 from uuid import uuid4
 
+from yaramo.utils import CustomJSONEncoder
+
 
 class BaseElement(object):
     def __init__(self, uuid: str = None, name: str = None, **kwargs) -> None:
@@ -26,4 +28,4 @@ def to_serializable(self) -> Tuple[dict, dict]:
         return self.__dict__, {}
 
     def to_json(self) -> str:
-        return json.dumps(self.to_serializable()[0])
+        return json.dumps(self.to_serializable()[0], cls=CustomJSONEncoder)

yaramo/edge.py

@@ -15,7 +15,8 @@ class Edge(BaseElement):
     as well as any intermediate_geo_nodes. The maximum_speed of an Edge cannot be set on construction but will generally be determined based on the connected Topology and Signals.
     """
 
-    def __init__(self, node_a: Node, node_b: Node, vacancy_section: Optional[VacancySection] = None, length: float = None, **kwargs):
+    def __init__(self, node_a: Node, node_b: Node, vacancy_section: Optional[VacancySection] = None, length: float = None,
+                 intermediate_geo_nodes=None, **kwargs):
         """
         Parameters
         ----------
@@ -30,6 +31,7 @@ def __init__(self, node_a: Node, node_b: Node, vacancy_section: Optional[Vacancy
         """
 
         super().__init__(**kwargs)
+        self.intermediate_geo_nodes = intermediate_geo_nodes or []
         self.node_a = node_a
         self.node_b = node_b
         self.intermediate_geo_nodes: list[GeoNode] = []
@@ -136,7 +138,7 @@ def to_serializable(self):
             "intermediate_geo_nodes": [geo_node.uuid for geo_node in self.intermediate_geo_nodes],
             "signals": [signal.uuid for signal in self.signals],
         }
-        objects = {}
+        objects = dict()
         for geo_node in self.intermediate_geo_nodes:
             geo_node_object, serialized_geo_node = geo_node.to_serializable()
             objects = {**objects, geo_node.uuid: geo_node_object, **serialized_geo_node}

yaramo/node.py

@@ -166,7 +166,7 @@ def to_serializable(self):
             "connected_nodes": [node.uuid for node in self.connected_nodes],
             "geo_node": self.geo_node.uuid if self.geo_node else None,
         }
-        objects = {}
+        objects = dict()
         if self.geo_node:
             geo_node, serialized_geo_node = self.geo_node.to_serializable()
             objects = {**objects, self.geo_node.uuid: geo_node, **serialized_geo_node}

yaramo/signal.py

@@ -145,7 +145,7 @@ def to_serializable(self) -> Tuple[dict, dict]:
             "function": str(self.function),
             "kind": str(self.kind),
         }
-        objects = {}
+        objects = dict()
         items = [self.trip] + self.additional_signals if self.trip else self.additional_signals
         for item in items:
             item_object, serialized_item = item.to_serializable()

yaramo/topology.py

@@ -1,3 +1,5 @@
+import json
+
 from yaramo.base_element import BaseElement
 from yaramo.edge import Edge
 from yaramo.node import Node
@@ -39,10 +41,10 @@ def get_edge_by_nodes(self, node_a: Node, node_b: Node):
         for edge_uuid in self.edges:
             edge = self.edges[edge_uuid]
             if (
-                edge.node_a.uuid == node_a.uuid
-                and edge.node_b.uuid == node_b.uuid
-                or edge.node_a.uuid == node_b.uuid
-                and edge.node_b.uuid == node_a.uuid
+                    edge.node_a.uuid == node_a.uuid
+                    and edge.node_b.uuid == node_b.uuid
+                    or edge.node_a.uuid == node_b.uuid
+                    and edge.node_b.uuid == node_a.uuid
             ):
                 return edge
         return None
@@ -54,7 +56,7 @@ def to_serializable(self):
             A serializable dictionary of all the objects belonging to the Topology.
         """
         nodes, edges, signals, routes, vacancy_sections = [], [], [], [], []
-        objects = {}
+        objects = dict()
 
         for items, _list in [
             (list(self.signals.values()), signals),
@@ -76,3 +78,32 @@ def to_serializable(self):
             "objects": objects,
             "vacany_sections": vacancy_sections
         }, {}
+
+    @classmethod
+    def from_json(cls, json_str: str):
+        obj = json.loads(json_str)
+        topology = cls()
+        for node in obj["nodes"]:
+            topology.add_node(Node(**node))
+        for signal in obj["signals"]:
+            topology.add_signal(Signal(**signal))
+        for edge in obj["edges"]:
+            node_a = topology.nodes[edge["node_a"]]
+            node_b = topology.nodes[edge["node_b"]]
+            node_a.connected_nodes.append(node_b)
+            node_b.connected_nodes.append(node_a)
+            topology.add_edge(Edge(**{
+                **edge, "node_a": node_a, "node_b": node_b,
+                "signals": [topology.signals[signal_uuid] for signal_uuid in edge["signals"]],
+                "intermediate_geo_nodes": [obj["objects"][geo_node_uuid] for geo_node_uuid in edge["intermediate_geo_nodes"] if geo_node_uuid in obj["objects"]],
+            }))
+        for signal in obj["signals"]:
+            topology.signals[signal["uuid"]].edge = topology.edges[signal["edge"]]
+        return topology
+
+
+if __name__ == "__main__":
+    with open("../test.json", "r") as f:
+        json_str = f.read()
+    topology = Topology.from_json(json_str)
+    print(topology.to_json())

yaramo/utils.py

@@ -0,0 +1,8 @@
+import json
+
+
+class CustomJSONEncoder(json.JSONEncoder):
+    def default(self, obj):
+        if isinstance(obj, set):
+            return list(obj)
+        return json.JSONEncoder.default(self, obj)