Merge branch 'yaramo2' into detect-implausible-switches

# Conflicts:
#	test/node_test.py
#	yaramo/node.py

Author: arneboockmeyer

poetry.lock

@@ -1,13 +1,13 @@
 [tool.poetry]
 name = "yaramo"
-version = "0.2"
+version = "2.0"
 description = "yaramo is a railway model focusing on interoperability between different existing planning formats."
 authors = ["OSM at HPI <arne.boockmeyer@hpi.de>"]
 readme = "README.md"
 packages = [{include = "yaramo"}]
 
 [tool.poetry.dependencies]
-python = "^3.9"
+python = "^3.11"
 pyproj = "^3.6.1"
 simplejson = "^3.19.2"
 
@@ -16,7 +16,7 @@ simplejson = "^3.19.2"
 black = "^22.12.0"
 pylint = "^2.15.10"
 pre-commit = "^3.0.0"
-pytest = "^7.2.1"
+pytest = "^8.0.0"
 
 [build-system]
 requires = ["poetry-core"]
@@ -28,11 +28,15 @@ line-length = 100
 [tool.pylint.format]
 max-line-length = "100"
 
-
 [tool.isort]
 multi_line_output = 3
 include_trailing_comma = true
 force_grid_wrap = 0
 use_parentheses = true
 ensure_newline_before_comments = true
 line_length = 100
+
+[tool.pytest.ini_options]
+addopts = [
+    "--import-mode=importlib",
+]

pyproject.toml

@@ -0,0 +1,13 @@
+from yaramo.model import Edge, Node, Wgs84GeoNode
+
+
+def create_geo_node(x, y):
+    return Wgs84GeoNode(x, y)
+
+
+def create_node(x, y):
+    return Node(geo_node=create_geo_node(x, y))
+
+
+def create_edge(node_a, node_b, inter_geo_nodes=None):
+    return Edge(node_a, node_b, intermediate_geo_nodes=inter_geo_nodes)

test/helper.py

@@ -3,15 +3,13 @@
 from pytest import raises
 
 from yaramo.geo_node import Wgs84GeoNode
-from yaramo.node import Node
+from yaramo.model import Node
 
-
-def create_node(x, y):
-    return Node(geo_node=Wgs84GeoNode(x, y))
+from .helper import create_edge, create_node
 
 
 def coords_str(node: Node):
-    return f"({node.geo_node.geo_point.x}, {node.geo_node.geo_point.y})"
+    return f"({node.geo_node.x}, {node.geo_node.y})"
 
 
 def test_anschluss():
@@ -105,10 +103,14 @@ def test_anschluss():
 
         # apply offsets to base scenario, set up nodes and actually test
         for offset_x, offset_y in product(offsets, repeat=2):
-            head = create_node(hx + offset_x, hy + offset_y)
             switch = create_node(sx + offset_x, sy + offset_y)
+
+            head = create_node(hx + offset_x, hy + offset_y)
+            create_edge(switch, head)
             left = create_node(lx + offset_x, ly + offset_y)
+            create_edge(switch, left)
             right = create_node(rx + offset_x, ry + offset_y)
+            create_edge(switch, right)
 
             print(
                 "scenario:",
@@ -122,11 +124,8 @@ def test_anschluss():
                 coords_str(right),
             )
 
-            # connect all nodes to the switch
-            switch.connected_nodes.extend((head, left, right))
-
             # finally call the procedure we want to test here
-            switch.calc_anschluss_of_all_nodes()
+            switch.calc_anschluss_of_all_edges()
 
             # assert ``calc_anschluss_of_all_nodes`` did what it should
             assert switch.connected_on_head == head, (
@@ -144,11 +143,11 @@ 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:
-    switch = create_node(2, 2)  #      l
+    point = create_node(2, 2)  #      l
     left = create_node(1, 3)  #       s r
     right = create_node(3, 2)  #     h
 
-    switch.connected_nodes.extend((head, left, right))
+    point.connected_nodes.extend((head, left, right))
 
     with raises(Exception) as exception:
-        switch.calc_anschluss_of_all_nodes()
+        point.calc_anschluss_of_all_nodes()

test/node_test.py

@@ -0,0 +1,79 @@
+import pytest
+
+from yaramo.edge import Edge
+from yaramo.node import Node
+from yaramo.topology import Topology
+
+from .helper import create_node
+
+
+@pytest.fixture
+def straight_track():
+    # Note that this kind of topology shouldn't exist
+    node1 = Node()
+    node2 = Node()
+    node3 = Node()
+
+    edge1 = Edge(node1, node2, length=100)
+    edge2 = Edge(node2, node3, length=100)
+
+    topology = Topology()
+    topology.add_node(node1)
+    topology.add_node(node2)
+    topology.add_node(node3)
+    topology.add_edge(edge1)
+    topology.add_edge(edge2)
+
+    return topology
+
+
+@pytest.fixture
+def point():
+    node1 = create_node(0, 10)
+    point = create_node(50, 10)
+    node2 = create_node(100, 10)
+    node3 = create_node(100, 20)
+
+    edge1 = Edge(node1, point, length=50)
+    edge2 = Edge(point, node2, length=50)
+    edge3 = Edge(point, node3, length=50)
+
+    topology = Topology()
+    topology.add_node(node1)
+    topology.add_node(point)
+    topology.add_node(node2)
+    topology.add_node(node3)
+    topology.add_edge(edge1)
+    topology.add_edge(edge2)
+    topology.add_edge(edge3)
+
+    return topology
+
+
+if __name__ == "__main__":
+    straight_track()
+    point()
+
+
+def test_detect_switch(point):
+    topology = point
+
+    point_count = 0
+
+    for node in topology.nodes.values():
+        if node.is_point():
+            point_count += 1
+
+    assert point_count == 1
+
+
+def test_dont_detect_straight(straight_track):
+    topology = straight_track
+
+    point_count = 0
+
+    for node in topology.nodes.values():
+        if node.is_point():
+            point_count += 1
+
+    assert point_count == 0

test/point_test.py

@@ -1,5 +1,7 @@
 from yaramo.model import Edge, Node, Topology
 
+from .helper import create_edge, create_geo_node, create_node
+
 
 def test_get_edge_by_nodes():
     topology = Topology()
@@ -42,3 +44,42 @@ def test_json_export_and_import():
     assert len(topology.edges) == len(topology_copy.edges)
     assert len(topology.signals) == len(topology_copy.signals)
     assert len(topology.routes) == len(topology_copy.routes)
+
+
+def test_station_topology():
+    """
+    This test creates this kind of topology:
+         ______
+    ____/______\\____
+    This kind wouldn't be possible in yaramo 1, since
+    two edges connect the same points.
+    """
+    end_a = create_node(0, 0)
+    point_a = create_node(10, 0)
+    inter_geo_node_a = create_geo_node(13, 1)
+    inter_geo_node_b = create_geo_node(17, 1)
+    point_b = create_node(20, 0)
+    end_b = create_node(30, 0)
+
+    edge_1 = create_edge(end_a, point_a)
+    edge_2a = create_edge(point_a, point_b)
+    edge_2b = create_edge(point_a, point_b, inter_geo_nodes=[inter_geo_node_a, inter_geo_node_b])
+    edge_3 = create_edge(point_b, end_b)
+
+    end_a.calc_anschluss_of_all_edges()
+    point_a.calc_anschluss_of_all_edges()
+    point_b.calc_anschluss_of_all_edges()
+    end_b.calc_anschluss_of_all_edges()
+
+    assert len(end_a.connected_edges) == 1
+    assert end_a.connected_edge_on_head == edge_1
+    assert len(point_a.connected_edges) == 3
+    assert point_a.connected_edge_on_head == edge_1
+    assert point_a.connected_edge_on_left == edge_2b
+    assert point_a.connected_edge_on_right == edge_2a
+    assert len(point_b.connected_edges) == 3
+    assert point_b.connected_edge_on_head == edge_3
+    assert point_b.connected_edge_on_right == edge_2b
+    assert point_b.connected_edge_on_left == edge_2a
+    assert len(end_b.connected_edges) == 1
+    assert end_b.connected_edge_on_head == edge_3

test/topology_test.py

@@ -42,7 +42,9 @@ def __init__(
         super().__init__(**kwargs)
         self.intermediate_geo_nodes = intermediate_geo_nodes or []
         self.node_a = node_a
+        self.node_a.connected_edges.append(self)
         self.node_b = node_b
+        self.node_b.connected_edges.append(self)
         self.signals = signals or []
         self.length = length
         self.maximum_speed = maximum_speed
@@ -77,6 +79,15 @@ def __get_length(self) -> float:
         )
         return total_length
 
+    def get_direction_based_on_start_node(self, start: "Node") -> "SignalDirection":
+        from yaramo.signal import SignalDirection
+
+        if self.node_a.uuid == start.uuid:
+            return SignalDirection.IN
+        elif self.node_b.uuid == start.uuid:
+            return SignalDirection.GEGEN
+        return None
+
     def get_direction_based_on_nodes(self, node_a: "Node", node_b: "Node") -> "SignalDirection":
         """Returns the direction according to whether the order of node_a and node_b is the same as in self
 
@@ -132,6 +143,45 @@ def get_signals_with_direction_in_order(self, direction: "SignalDirection") -> L
         result.sort(key=lambda x: x.distance_edge, reverse=(direction == SignalDirection.GEGEN))
         return result
 
+    def get_opposite_node(self, node: "Node") -> "Node":
+        """Returns the opposite Node of the given Node
+
+        Parameters
+        ----------
+        node : Node
+            The Node to get the opposite Node of
+
+        Returns
+        -------
+        Node
+            The opposite Node
+        """
+
+        if self.node_a.uuid == node.uuid:
+            return self.node_b
+        return self.node_a
+
+    def get_next_geo_node(self, node: "Node") -> "GeoNode":
+        """Returns the next GeoNode on Edgeof the given Top Node
+
+        Parameters
+        ----------
+        geo_node : Node
+            The Top Node to get the next GeoNode of
+
+        Returns
+        -------
+        GeoNode
+            The next GeoNode
+        """
+        if len(self.intermediate_geo_nodes) < 2:
+            return self.get_opposite_node(node).geo_node
+        if self.node_a.uuid == node.uuid:
+            return self.intermediate_geo_nodes[1]
+        if self.node_b.uuid == node.uuid:
+            return self.intermediate_geo_nodes[-2]
+        return None
+
     def to_serializable(self):
         """See the description in the BaseElement class.