Merge branch 'operations' into compare-operation

Author: arneboockmeyer

test/node_test.py

@@ -1,5 +1,8 @@
 from itertools import product
 
+from pytest import raises
+
+from yaramo.geo_node import Wgs84GeoNode
 from yaramo.model import Node
 
 from .helper import create_edge, create_node
@@ -134,3 +137,17 @@ def test_anschluss():
             assert switch.connected_on_right == right, (
                 "right node " f"{coords_str(switch.connected_on_right)} incorrect"
             )
+
+
+def test_implausible_anschluss():
+    """Assert that detection of "Anschluss" (head, left, right) raises
+    an exception on really implausible geographies."""
+    head = create_node(0, 0)  # layout:
+    point = create_node(2, 2)  #      l
+    left = create_node(1, 3)  #       s r
+    right = create_node(3, 2)  #     h
+
+    point.connected_nodes.extend((head, left, right))
+
+    with raises(Exception) as exception:
+        point.calc_anschluss_of_all_nodes()

yaramo/geo_node.py

@@ -19,19 +19,20 @@ def __init__(self, x, y, **kwargs):
 
     @abstractmethod
     def get_distance_to_other_geo_node(self, geo_node_b: "GeoNode"):
+        """Returns to distance to the given other GeoNode."""
         pass
 
-    def to_serializable(self):
-        return self.__dict__, {}
-
     @abstractmethod
-    def to_wgs84(self):
+    def to_wgs84(self) -> "Wgs84GeoNode":
         pass
 
     @abstractmethod
-    def to_dbref(self):
+    def to_dbref(self) -> "DbrefGeoNode":
         pass
 
+    def to_serializable(self):
+        return self.__dict__, {}
+
 
 class Wgs84GeoNode(GeoNode):
     def get_distance_to_other_geo_node(self, geo_node_b: "Wgs84GeoNode"):

yaramo/node.py

@@ -155,37 +155,42 @@ def get_anschluss_for_edge(self, edge: "Edge") -> EdgeConnectionDirection:
     def calc_anschluss_of_all_edges(self):
         """Calculates and sets the 'Anschluss' or connection side of
         the connected_nodes based on their geo location."""
-
         if len(self.connected_edges) == 1:
             self.connected_edge_on_head = self.connected_edges[0]
             return
+        almost_zero = sys.float_info.epsilon
 
-        def get_arc_between_geo_nodes(geo_node_a: "GeoNode", geo_node_b: "GeoNode") -> float:
+        def get_rad_between_nodes(geo_node_a: GeoNode, geo_node_b: GeoNode) -> float:
             """
             Returns the angle of an (maybe imaginary) line between
-            :param:`node_a` and :param:`node_b`.
+            :param:`geo_node_a` and :param:`geo_node_b`.
             """
             return atan2(geo_node_b.y - geo_node_a.y, geo_node_b.x - geo_node_a.x)
 
         # Determine which node is head, left, and right by trying the
         # permutations and checking for plausibility.
         for head, left, right in permutations(self.connected_edges):
 
-            head_angle_abs = get_arc_between_geo_nodes(head.get_next_geo_node(self), self.geo_node)
-            left_angle_abs = get_arc_between_geo_nodes(self.geo_node, left.get_next_geo_node(self))
+            head_angle_abs = get_rad_between_nodes(head.get_next_geo_node(self), self.geo_node)
+            left_angle_abs = get_rad_between_nodes(self.geo_node, left.get_next_geo_node(self))
             left_angle_rel = left_angle_abs - head_angle_abs
-            if cos(left_angle_rel) <= sys.float_info.epsilon:
+            if cos(left_angle_rel) <= almost_zero:
                 # left turn more than (or almost) 90°
                 continue
 
-            right_angle_abs = get_arc_between_geo_nodes(
-                self.geo_node, right.get_next_geo_node(self)
-            )
+            right_angle_abs = get_rad_between_nodes(self.geo_node, right.get_next_geo_node(self))
             right_angle_rel = right_angle_abs - head_angle_abs
-            if cos(right_angle_rel) <= sys.float_info.epsilon:
-                # left turn more than (or almost) 90°
+            if cos(right_angle_rel) <= almost_zero:
+                # right turn more than (or almost) 90°
                 continue
 
+            if cos(left_angle_abs - right_angle_abs) <= almost_zero:
+                # turn from left to right less than (or almost) 90°
+                raise Exception(
+                    f"Node {self} has three connected nodes but "
+                    "geometrically, it does not look like a switch."
+                )
+
             if sin(left_angle_rel) < sin(right_angle_rel):
                 # Left and right mixed up. Although the permutations do
                 # contain the correct combination, fixing this right

yaramo/topology.py

@@ -1,4 +1,6 @@
+from collections import defaultdict
 from datetime import datetime
+from enum import Enum
 from typing import List
 
 import networkx as nx
@@ -13,10 +15,28 @@
 from yaramo.vacancy_section import VacancySection
 
 
+class PlanningState(Enum):
+    erstellt = 1
+    qualitaetsgeprueft = 2
+    plangeprueft = 3
+    freigegeben = 4
+    genehmigt = 5
+    abgenommen = 6
+    uebernommen = 7
+    gleichgestellt = 8
+    sonstige = 9
+
+    def __str__(self):
+        return self.name
+
+
 class Topology(BaseElement):
     """The Topology is a collection of all track elements comprising that topology.
 
     Elements like Signals, Nodes, Edges, Routes and Vacancy Sections can be accessed by their uuid in their respective dictionary.
+
+    The status_information provides additional information for each PlanningState, with the keys of the inner dict
+    being based on the PlanPro Akteur_Zuordnung class.
     """
 
     def __init__(self, **kwargs):
@@ -26,6 +46,8 @@ def __init__(self, **kwargs):
         self.signals: dict[str, Signal] = {}
         self.routes: dict[str, Route] = {}
         self.vacancy_sections: dict[str, VacancySection] = {}
+        self.current_status: PlanningState = PlanningState.erstellt
+        self.status_information: dict[PlanningState, dict[str, str]] = defaultdict(dict)
 
         self.created_at: datetime = datetime.now()
         self.created_with: str = "unknown"
@@ -82,6 +104,15 @@ def update_edge_lengths(self):
         for edge in self.edges.values():
             edge.update_length()
 
+    def get_route_by_signal_names(self, start_signal_name, end_signal_name):
+        for route_uuid in self.routes:
+            route = self.routes[route_uuid]
+            if (
+                route.start_signal.name == start_signal_name
+                and route.end_signal.name == end_signal_name
+            ):
+                return route
+
     def to_serializable(self):
         """See the description in the BaseElement class.
 
@@ -110,12 +141,16 @@ def to_serializable(self):
             "routes": routes,
             "objects": objects,
             "vacany_sections": vacancy_sections,
+            "current_status": self.current_status.value,
+            "status_information": self.status_information,
         }, {}
 
     @classmethod
     def from_json(cls, json_str: str):
         obj = json.loads(json_str)
         topology = cls()
+        topology.current_status = PlanningState(obj.get("current_status", PlanningState.erstellt))
+        topology.status_information = obj.get("status_information", defaultdict(dict))
         for node in obj["nodes"]:
             node_obj = Node(**node)
             topology.add_node(node_obj)