Merge branch 'async' into flank-protection

.github/workflows/package-test.yml

@@ -45,4 +45,5 @@ jobs:
     - name: Test with pytest
       run: |
         cd test
-        poetry run pytest .
+        poetry run coverage run -m pytest .
+        poetry run coverage report --omit="/usr/*,*__init__*,*model.py,*reader.py" -m --skip-covered --sort=cover --fail-under=85

interlocking/infrastructureprovider/randomwaitinfrastructureprovider.py

@@ -1,4 +1,5 @@
 from .infrastructureprovider import InfrastructureProvider
+from yaramo.model import Signal, Node
 import time
 import logging
 import random
@@ -12,26 +13,36 @@ class RandomWaitInfrastructureProvider(InfrastructureProvider):
     # Time to turn point according to
     # https://de.wikipedia.org/wiki/Weiche_(Bahn)#Umlaufzeiten
 
-    def __init__(self, allow_fail, signal_time_range=range(2, 5), point_turn_time_range=range(5, 8)):
+    def __init__(self, fail_probability=0.0, signal_time_range: range = range(2, 5),
+                 point_turn_time_range: range = range(5, 8), always_succeed_for: list[str] = None,
+                 always_fail_for: list[str] = None):
         super().__init__()
-        self.allow_fail = allow_fail
+        if always_succeed_for is None:
+            always_succeed_for = []
+        if always_fail_for is None:
+            always_fail_for = []
+        self.fail_probability = fail_probability
         self.signal_time_range = signal_time_range
         self.point_turn_time_range = point_turn_time_range
+        self.always_succeed_for = always_succeed_for
+        self.always_fail_for = always_fail_for
 
-    async def set_signal_aspect(self, yaramo_signal, target_aspect):
+    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.randint(0, 3) > 0 or not self.allow_fail:
+        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_aspect} (waited {wait})")
             return True
         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, target_orientation: str):
+    async def turn_point(self, yaramo_point: Node, target_orientation: str):
         wait = random.sample(self.point_turn_time_range, 1)[0]
         point_id = yaramo_point.uuid[-5:]
         await asyncio.sleep(wait)
-        if random.randint(0, 3) > 0 or not self.allow_fail:
+        if (random.random() >= self.fail_probability or point_id in self.always_succeed_for) \
+                and point_id not in self.always_fail_for:
             logging.info(f"{time.strftime('%X')} Completed turning point {point_id} to {target_orientation} (waited {wait})")
             return True
         logging.warning(f"{time.strftime('%X')} Failed turning point {point_id} to {target_orientation} (waited {wait})")

interlocking/interlockingcontroller/signalcontroller.py

@@ -6,7 +6,7 @@
 class SignalController(object):
 
     def __init__(self, infrastructure_providers):
-        self.signals = None
+        self.signals: dict[str, Signal] = {}
         self.infrastructure_providers = infrastructure_providers
 
     async def reset(self):
@@ -57,7 +57,9 @@ def free_route(self, route, train_id: str):
     async def reset_route(self, route, train_id: str):
         result = await self.set_signal_halt(route.start_signal)
         if result:
-            self.free_signal(route.start_signal, train_id)
+            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

interlocking/interlockingcontroller/trackcontroller.py

@@ -1,5 +1,5 @@
 from .overlapcontroller import OverlapController
-from interlocking.model import OccupancyState
+from interlocking.model import OccupancyState, Route
 from yaramo.model import SignalDirection
 import logging
 
@@ -33,7 +33,7 @@ def can_route_be_set(self, route, train_id: str):
             return False
         return self.overlap_controller.can_any_overlap_be_reserved(route, train_id)
 
-    def do_two_routes_collide(self, route_1, route_2):
+    def do_two_routes_collide(self, route_1: Route, route_2: Route):
         segments_of_route_1 = set(map(lambda seg: seg.segment_id, route_1.get_segments_of_route()))
         segments_of_route_2 = set(map(lambda seg: seg.segment_id, route_2.get_segments_of_route()))
         if len(segments_of_route_1.intersection(segments_of_route_2)) > 0:

interlocking/model/route.py

@@ -81,7 +81,9 @@ def get_segments_of_route(self):
             # Start signal and end signal are on the same track
             pos_start_signal = self.start_signal.track.get_position_of_signal(self.start_signal)
             pos_end_signal = self.end_signal.track.get_position_of_signal(self.end_signal)
-            return self.start_signal.track.segments[pos_start_signal + 1:pos_end_signal + 1]
+            min_pos = min(pos_start_signal, pos_end_signal)
+            max_pos = max(pos_start_signal, pos_end_signal)
+            return self.start_signal.track.segments[min_pos + 1:max_pos + 1]
 
         result = self.start_signal.track.get_segments_from_signal(self.start_signal)
         for track in self.tracks[1:-1]:

interlocking/model/signal.py

@@ -1,11 +1,14 @@
 from .occupancystate import OccupancyState
+from yaramo.model import Signal as YaramoSignal
 
 
 class Signal(object):
 
-    def __init__(self, yaramo_signal):
+    def __init__(self, yaramo_signal: YaramoSignal):
+        from interlocking.model import Track
+
         self.yaramo_signal = yaramo_signal
         self.signal_aspect: str = "undefined"  # Either halt or go
         self.state: OccupancyState = OccupancyState.FREE
         self.used_by = set()  # If point is reserved, occupied or part of an overlap, this contains the train numbers.
-        self.track = None
+        self.track: Track or None = None