Add support for flank protection

interlocking/infrastructureprovider/infrastructureprovider.py

@@ -74,16 +74,16 @@ def is_signal_covered(self, yaramo_signal: Signal):
         return yaramo_signal.name in self.only_apply_for_signals or \
             (len(self.only_apply_for_signals) == 0 and yaramo_signal.name not in self.apply_for_all_signals_except)
 
-    async def call_set_signal_state(self, yaramo_signal: Signal, target_state: str):
+    async def call_set_signal_aspect(self, yaramo_signal: Signal, target_state: str):
         if self.is_signal_covered(yaramo_signal):
-            return await self.set_signal_state(yaramo_signal, target_state)
+            return await self.set_signal_aspect(yaramo_signal, target_state)
         # return True to skip this call and not prevent successfully turning of signal.
         return True
 
     @abstractmethod
-    async def set_signal_state(self, yaramo_signal: Signal, target_state: str):
-        """This method will be called when the interlocking controller wants to change the signal-state of a specific signal. 
-        `yaramo_signal` corresponds to the identifier of the signal in the yaramo model; `target_state` is one of `"halt"` and `"go"`.
+    async def set_signal_aspect(self, yaramo_signal: Signal, target_aspect: str):
+        """This method will be called when the interlocking controller wants to change the signal-aspect of a specific signal.
+        `yaramo_signal` corresponds to the identifier of the signal in the yaramo model; `target_aspect` is one of `"halt"` and `"go"`.
         """
         pass
 

interlocking/infrastructureprovider/logginginfrastructureprovider.py

@@ -8,8 +8,8 @@ class LoggingInfrastructureProvider(InfrastructureProvider):
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
 
-    async def set_signal_state(self, yaramo_signal, target_state):
-        logging.info(f"{time.strftime('%X')} Set signal {yaramo_signal.name} to {target_state}")
+    async def set_signal_aspect(self, yaramo_signal, target_aspect):
+        logging.info(f"{time.strftime('%X')} Set signal {yaramo_signal.name} to {target_aspect}")
         return True
 
     async def turn_point(self, yaramo_point, target_orientation: str):

interlocking/infrastructureprovider/randomwaitinfrastructureprovider.py

@@ -27,14 +27,14 @@ def __init__(self, fail_probability=0.0, signal_time_range: range = range(2, 5),
         self.always_succeed_for = always_succeed_for
         self.always_fail_for = always_fail_for
 
-    async def set_signal_state(self, yaramo_signal: Signal, target_state):
+    async def set_signal_aspect(self, yaramo_signal: Signal, target_aspect: str):
         wait = random.sample(self.signal_time_range, 1)[0]
         await asyncio.sleep(wait)
         if (random.random() >= self.fail_probability or yaramo_signal.name in self.always_succeed_for) \
                 and yaramo_signal.name not in self.always_fail_for:
-            logging.info(f"{time.strftime('%X')} Completed setting signal {yaramo_signal.name} to {target_state} (waited {wait})")
+            logging.info(f"{time.strftime('%X')} Completed setting signal {yaramo_signal.name} to {target_aspect} (waited {wait})")
             return True
-        logging.warning(f"{time.strftime('%X')} Failed setting signal {yaramo_signal.name} to {target_state} (waited {wait})")
+        logging.warning(f"{time.strftime('%X')} Failed setting signal {yaramo_signal.name} to {target_aspect} (waited {wait})")
         return False
 
     async def turn_point(self, yaramo_point: Node, target_orientation: str):

interlocking/infrastructureprovider/sumoinfrastructureprovider.py

@@ -8,10 +8,10 @@ def __init__(self, traci_instance, **kwargs):
         super().__init__(**kwargs)
         self.traci_instance = traci_instance
 
-    async def set_signal_state(self, yaramo_signal, target_state):
-        if target_state == "go":
+    async def set_signal_aspect(self, yaramo_signal, target_aspect):
+        if target_aspect == "go":
             self.traci_instance.trafficlight.setRedYellowGreenState(yaramo_signal.name, "GG")
-        elif target_state == "halt":
+        elif target_aspect == "halt":
             if yaramo_signal.direction == SignalDirection.IN:
                 self.traci_instance.trafficlight.setRedYellowGreenState(yaramo_signal.name, "rG")
             else:

interlocking/interlockingcontroller/flankprotectioncontroller.py

@@ -0,0 +1,106 @@
+from .signalcontroller import SignalController
+from interlocking.model import Route, Point, OccupancyState, Signal
+from yaramo.model import SignalDirection, NodeConnectionDirection
+import logging
+
+
+class FlankProtectionController(object):
+
+    def __init__(self, point_controller, signal_controller: SignalController):
+        self.point_controller = point_controller
+        self.signal_controller = signal_controller
+
+    def reset(self):
+        for point in self.point_controller.points.values():
+            point.is_used_for_flank_protection = False
+        for signal in self.signal_controller.signals.values():
+            signal.is_used_for_flank_protection = False
+
+    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)
+        results = []
+        for signal in signals:
+            logging.info(f"--- Use signal {signal.yaramo_signal.name} for flank protection as 'halt-zeigendes Signal'")
+            change_successful = await self.signal_controller.set_signal_halt(signal)
+            results.append(change_successful)
+            if change_successful:
+                signal.is_used_for_flank_protection = True
+        for point in points:
+            orientation = points[point]
+            if orientation is not None:
+                # In case of a Schutztansportweiche the orientation is not relevant (None).
+                logging.info(f"--- Use point {point.point_id} for flank protection as 'Schutzweiche'")
+                change_successful = await self.point_controller.turn_point(point, orientation)
+                results.append(change_successful)
+                if change_successful:
+                    point.is_used_for_flank_protection = True
+            else:
+                logging.info(f"--- Use point {point.point_id} for flank protection as 'Schutztransportweiche'")
+                point.is_used_for_flank_protection = True
+        return all(results)
+
+    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) -> tuple[list[Signal],
+                                                                       dict[Point, str | None]]:
+        flank_protection_tracks = []
+        if point_orientation is None:
+            # It's only none, iff there is a flank protection transport point (where the flank
+            # protection area comes from Spitze
+            flank_protection_tracks = [point.left, point.right]
+        elif point_orientation == "left":
+            flank_protection_tracks = [point.right]
+        elif point_orientation == "right":
+            flank_protection_tracks = [point.left]
+
+        signal_results: list[Signal] = []
+        point_results: dict[Point, str | None] = {}
+
+        for flank_protection_track in flank_protection_tracks:
+            # Search for signals
+            yaramo_edge = flank_protection_track.yaramo_edge
+            node_a = point.yaramo_node
+            node_b = yaramo_edge.get_other_node(node_a)
+            direction = yaramo_edge.get_direction_based_on_nodes(node_a, node_b)
+
+            opposite_direction = SignalDirection.IN
+            if direction == SignalDirection.IN:
+                opposite_direction = SignalDirection.GEGEN
+            yaramo_signals_in_direction = yaramo_edge.get_signals_with_direction_in_order(opposite_direction)
+            # If there is any signal, take the closest one to the point and use it as halt-showing signal.
+            found_signal = False
+            if len(yaramo_signals_in_direction) > 0:
+                yaramo_signal = yaramo_signals_in_direction[-1]  # Take the last one, which is the closest one.
+                for signal_uuid in self.signal_controller.signals:
+                    if signal_uuid == yaramo_signal.uuid:
+                        signal_results.append(self.signal_controller.signals[signal_uuid])
+                        found_signal = True
+                        break
+
+            # No Halt zeigendes Signal detected. Try to find Schutzweiche
+            if not found_signal:
+                other_point: Point | None = None
+                for point_uuid in self.point_controller.points:
+                    _point: Point = self.point_controller.points[point_uuid]
+                    if node_b.uuid == _point.yaramo_node.uuid:
+                        other_point = _point
+
+                if other_point is not None and other_point.is_point:
+                    connection_direction = other_point.get_connection_direction_of_track(flank_protection_track)
+                    if connection_direction == NodeConnectionDirection.Spitze:
+                        point_results[other_point] = None
+                        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] = "right"
+                    elif connection_direction == NodeConnectionDirection.Rechts:
+                        point_results[other_point] = "left"
+        return signal_results, point_results

interlocking/interlockingcontroller/overlapcontroller.py

@@ -15,7 +15,7 @@ async def reserve_overlap_of_route(self, route, train_id: str):
         if overlap is None:
             raise ValueError("No reservable overlap found")
         self.reserve_segments_of_overlap(overlap, train_id)
-        success = await self.reserve_points_of_overlap(overlap, train_id)
+        success = await self.reserve_points_of_overlap(overlap, route, train_id)
         route.overlap = overlap
         return success
 
@@ -50,7 +50,7 @@ def reserve_segments_of_overlap(self, overlap, train_id: str):
             segment.state = OccupancyState.RESERVED_OVERLAP
             segment.used_by.add(train_id)
 
-    async def reserve_points_of_overlap(self, overlap, train_id: str):
+    async def reserve_points_of_overlap(self, overlap, route, train_id: str):
         tasks = []
         async with asyncio.TaskGroup() as tg:
             for point in overlap.points:
@@ -69,6 +69,9 @@ async def reserve_points_of_overlap(self, overlap, train_id: str):
                         raise ValueError("Overlap contains points without 2 of their tracks")
                     necessery_orientation = point.get_necessary_orientation(found_tracks[0], found_tracks[1])
                     tasks.append(tg.create_task(self.point_controller.turn_point(point, necessery_orientation)))
+                    tasks.append(tg.create_task(self.point_controller.flank_protection_controller.
+                                                add_flank_protection_for_point(point, necessery_orientation, route,
+                                                                               train_id)))
         return all(list(map(lambda task: task.result(), tasks)))
 
     def free_overlap_of_route(self, route, train_id: str):

interlocking/interlockingcontroller/pointcontroller.py

@@ -1,20 +1,27 @@
 from interlocking.model import OccupancyState, Point
+from interlocking.model.helper import Settings
+from interlocking.infrastructureprovider import InfrastructureProvider
+from .flankprotectioncontroller import FlankProtectionController
+from .signalcontroller import SignalController
 import asyncio
 import logging
 
 
 class PointController(object):
 
-    def __init__(self, infrastructure_providers, settings):
+    def __init__(self, signal_controller: SignalController, infrastructure_providers: list[InfrastructureProvider],
+                 settings: Settings):
         self.points: dict[str, Point] = {}
         self.infrastructure_providers = infrastructure_providers
         self.settings = settings
+        self.flank_protection_controller = FlankProtectionController(self, signal_controller)
 
     def reset(self):
         for point_id in self.points:
             self.points[point_id].orientation = "undefined"
             self.points[point_id].state = OccupancyState.FREE
             self.points[point_id].used_by = set()
+        self.flank_protection_controller.reset()
 
     async def set_route(self, route, train_id: str):
         tasks = []
@@ -27,6 +34,7 @@ async def set_route(self, route, train_id: str):
                     if orientation == "left" or orientation == "right":
                         self.set_point_reserved(point, train_id)
                         tasks.append(tg.create_task(self.turn_point(point, orientation)))
+                        tasks.append(tg.create_task(self.flank_protection_controller.add_flank_protection_for_point(point, orientation, route, train_id)))
                     else:
                         raise ValueError("Turn should happen but is not possible")
 
@@ -50,6 +58,10 @@ async def turn_point(self, point, orientation):
         if point.orientation == orientation:
             # Everything is fine
             return True
+        if point.is_used_for_flank_protection is True:
+            logging.error(f"Can not turn point of point {point.point_id} to {orientation}, "
+                          f"since it is used for flank protection.")
+            return False
         logging.info(f"--- Move point {point.point_id} to {orientation}")
         # tasks = []
         results = []
@@ -77,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_controller.free_flank_protection_of_point(point, point.orientation)
 
     def reset_route(self, route, train_id: str):
         for point in route.get_points_of_route():
@@ -86,4 +99,5 @@ def print_state(self):
         logging.debug("State of Points:")
         for point_id in self.points:
             point = self.points[point_id]
-            logging.debug(f"{point.point_id}: {point.state} (Orientation: {point.orientation}) (used by: {point.used_by})")
+            logging.debug(f"{point.point_id}: {point.state} (Orientation: {point.orientation}) "
+                          f"(used by: {point.used_by}) (is used for FP: {point.is_used_for_flank_protection})")

interlocking/interlockingcontroller/signalcontroller.py

@@ -1,6 +1,6 @@
 import asyncio
 import logging
-from interlocking.model import Signal
+from interlocking.model import OccupancyState, Signal
 
 
 class SignalController(object):
@@ -11,45 +11,70 @@ def __init__(self, infrastructure_providers):
 
     async def reset(self):
         # Run non-concurrently
-        for signal_id in self.signals:
-            await self.set_signal_halt(self.signals[signal_id])
+        for signal in self.signals.values():
+            await self.set_signal_halt(signal)
+            signal.state = OccupancyState.FREE
+            signal.used_by = set()
 
-    async def set_route(self, route):
-        return await self.set_signal_go(route.start_signal)
+
+    async def set_route(self, route, train_id: str):
+        result = await self.set_signal_go(route.start_signal)
+        if result:
+            route.start_signal.state = OccupancyState.RESERVED
+            route.start_signal.used_by.add(train_id)
+        return result
 
     async def set_signal_halt(self, signal):
-        return await self.set_signal_state(signal, "halt")
+        return await self.set_signal_aspect(signal, "halt")
 
     async def set_signal_go(self, signal):
-        return await self.set_signal_state(signal, "go")
+        return await self.set_signal_aspect(signal, "go")
 
-    async def set_signal_state(self, signal, state):
-        if signal.state == state:
+    async def set_signal_aspect(self, signal, signal_aspect):
+        if signal.signal_aspect == signal_aspect:
             # Everything is fine
             return True
-        logging.info(f"--- Set signal {signal.yaramo_signal.name} to {state}")
+        if signal.is_used_for_flank_protection is True:
+            logging.error(f"Can not set signal aspect of signal {signal.yaramo_signal.name} to {signal_aspect}, "
+                          f"since it is used for flank protection.")
+            return False
+        logging.info(f"--- Set signal {signal.yaramo_signal.name} to {signal_aspect}")
 
         results = []
         for infrastructure_provider in self.infrastructure_providers:
-            results.append(await infrastructure_provider.call_set_signal_state(signal.yaramo_signal, state))
+            results.append(await infrastructure_provider.call_set_signal_aspect(signal.yaramo_signal, signal_aspect))
 
         # tasks = []
         # async with asyncio.TaskGroup() as tg:
         #    for infrastructure_provider in self.infrastructure_providers:
-        #        tasks.append(tg.create_task(infrastructure_provider.call_set_signal_state(signal.yaramo_signal, state)))
+        #        tasks.append(tg.create_task(infrastructure_provider.call_set_signal_aspect(signal.yaramo_signal, state)))
         # if all(list(map(lambda task: task.result(), tasks))):
         if all(results):
-            signal.state = state
+            signal.signal_aspect = signal_aspect
             return True
         else:
             # TODO: Incident
             return False
 
-    async def reset_route(self, route):
-        await self.set_signal_halt(route.start_signal)
+    def free_route(self, route, train_id: str):
+        if route.start_signal.signal_aspect == "go":
+            raise ValueError("Try to free route with start signal aspect is go")
+        self.free_signal(route.start_signal, train_id)
+
+    async def reset_route(self, route, train_id: str):
+        result = await self.set_signal_halt(route.start_signal)
+        if result:
+            route.start_signal.state = OccupancyState.FREE
+            if train_id in route.start_signal.used_by:
+                route.start_signal.used_by.remove(train_id)
+
+    def free_signal(self, signal: Signal, train_id: str):
+        signal.state = OccupancyState.FREE
+        signal.used_by.remove(train_id)
 
     def print_state(self):
         logging.debug("State of Signals:")
         for signal_uuid in self.signals:
             signal = self.signals[signal_uuid]
-            logging.debug(f"{signal.yaramo_signal.name}: {signal.state}")
+            logging.debug(f"{signal.yaramo_signal.name}: {signal.state} (Signal Aspect: {signal.signal_aspect}) "
+                          f"(is used for FP: {signal.is_used_for_flank_protection})")