outsource _switcher to make it easily testable;

test vehicle props;
test traffic light _switcher;
This commit is contained in:
Marco Zeisler
2021-06-17 20:59:23 +02:00
parent 8cb45d9b23
commit 643f98f121
2 changed files with 120 additions and 15 deletions
+17 -15
View File
@@ -56,23 +56,10 @@ class TrafficLight:
def start(self):
"""
Starts the traffic light by spawning a new thread. It toggles its state every self.switching_time / SCALING
seconds. When it switches, it notifies the orchestrator with self.send_status_update().
Starts the traffic light by spawning a new thread..
"""
self.running = True
def switcher():
num_colors = len(TrafficLightColor)
# set current color to the one before starting color, because switch immediately happens in loop
# therefore it sleeps on the end of the loop and first status is immediately sent to msg broker
self.current_color = TrafficLightColor((self._starting_color.value - 1) % num_colors)
while self.running:
self.current_color = TrafficLightColor((self.current_color.value + 1) % num_colors)
self.last_switch = datetime.now()
self.send_status_update()
time.sleep(self.switching_time / SCALING)
self._t = threading.Thread(target=switcher)
self._t = threading.Thread(target=self._switcher)
self._t.start()
def stop(self):
@@ -91,6 +78,21 @@ class TrafficLight:
self._tl_mb.send(pickle.dumps(
TrafficLightState(tlid=self.tlid, color=self.current_color, last_switch=self.last_switch)))
def _switcher(self):
"""
Toggles the traffic lights state every self.switching_time / SCALING seconds.
When it switches, it notifies the orchestrator with self.send_status_update()
"""
num_colors = len(TrafficLightColor)
# set current color to the one before starting color, because switch immediately happens in loop
# therefore it sleeps on the end of the loop and first status is immediately sent to msg broker
self.current_color = TrafficLightColor((self._starting_color.value - 1) % num_colors)
while self.running:
self.current_color = TrafficLightColor((self.current_color.value + 1) % num_colors)
self.last_switch = datetime.now()
self.send_status_update()
time.sleep(self.switching_time / SCALING)
if __name__ == '__main__':
...