Free flank protection for routes

simulate-digital-rail · Jun 27, 2024 · a209a6c · a209a6c
1 parent 020f5d8
commit a209a6c
Show file tree

Hide file tree

Showing 3 changed files with 16 additions and 6 deletions.
diff --git a/interlocking/interlockingcontroller/flankprotectioncontroller.py b/interlocking/interlockingcontroller/flankprotectioncontroller.py
@@ -18,7 +18,7 @@ def reset(self):
 
     async def add_flank_protection_for_point(self, point: Point, point_orientation: str,
                                              route: Route, train_id: str) -> bool:
-        signals, points = self._get_flank_protection_elements_of_point(point, point_orientation, route)
+        signals, points = self._get_flank_protection_elements_of_point(point, point_orientation)
         results = []
         for signal in signals:
             #signal.state = OccupancyState.FLANK_PROTECTION
@@ -44,17 +44,16 @@ async def add_flank_protection_for_point(self, point: Point, point_orientation:
                 point.is_used_for_flank_protection = True
         return all(results)
 
-    def free_flank_protection_of_point(self, point: Point, point_orientation: str, route: Route, train_id: str):
-        signals, points = self._get_flank_protection_elements_of_point(point, point_orientation, route)
+    def free_flank_protection_of_point(self, point: Point, point_orientation: str):
+        signals, points = self._get_flank_protection_elements_of_point(point, point_orientation)
         for signal in signals:
             signal.is_used_for_flank_protection = False
         for point in points:
             point.is_used_for_flank_protection = False
 
     def _get_flank_protection_elements_of_point(self,
                                                 point: Point,
-                                                point_orientation: str | None,
-                                                route: Route) -> tuple[list[Signal],
+                                                point_orientation: str | None) -> tuple[list[Signal],
                                                                        dict[Point, tuple[OccupancyState, str | None]]]:
         flank_protection_tracks = []
         if point_orientation is None:
@@ -102,7 +101,7 @@ def _get_flank_protection_elements_of_point(self,
                     connection_direction = other_point.get_connection_direction_of_track(flank_protection_track)
                     if connection_direction == NodeConnectionDirection.Spitze:
                         point_results[other_point] = (OccupancyState.FLANK_PROTECTION_TRANSPORT, None)
-                        signal_results, sub_point_results = self._get_flank_protection_elements_of_point(other_point, None, route)
+                        signal_results, sub_point_results = self._get_flank_protection_elements_of_point(other_point, None)
                         point_results = point_results | sub_point_results
                     elif connection_direction == NodeConnectionDirection.Links:
                         point_results[other_point] = (OccupancyState.FLANK_PROTECTION, "right")

diff --git a/interlocking/interlockingcontroller/pointcontroller.py b/interlocking/interlockingcontroller/pointcontroller.py
@@ -89,6 +89,7 @@ def set_point_free(self, point, train_id: str):
             logging.info(f"--- Set point {point.point_id} to free")
             point.state = OccupancyState.FREE
             point.used_by.remove(train_id)
+            self.flank_protection_cqontroller.free_flank_protection_of_point(point, point.orientation)
 
     def reset_route(self, route, train_id: str):
         for point in route.get_points_of_route():

diff --git a/test/interlocking_test.py b/test/interlocking_test.py
@@ -102,6 +102,16 @@ def test_driving():
     asyncio.run(ip.tds_count_out("a8f44-0", "RB101"))
     interlockinghelper.free_route(interlocking, route_2, "RB101")
 
+    # Verify flank protection cleanup
+    signal_as1 = interlockinghelper.get_interlocking_signal_by_name(interlocking, "60AS1")
+    assert not signal_as1.is_used_for_flank_protection
+    signal_as4 = interlockinghelper.get_interlocking_signal_by_name(interlocking, "60AS1")
+    assert not signal_as4.is_used_for_flank_protection
+    point_fa9 = interlockinghelper.get_interlocking_point_by_id(interlocking, "fa9ea")
+    assert not point_fa9.is_used_for_flank_protection
+    point_21b = interlockinghelper.get_interlocking_point_by_id(interlocking, "21b88")
+    assert not point_21b.is_used_for_flank_protection
+
 
 def test_reset_route():
     topology = topologyhelper.get_topology_from_planpro_file("./complex-example.ppxml")