davorian
/
trafficlights
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Added some comments for main simulaation thread, and removed console print debug statements.

This commit is contained in:
Chris Davoren 2023-10-21 15:12:40 +10:00
parent d07c1eeb9c
commit aacaf7de1a
1 changed files with 75 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

94
simulation/simulation_thread.py
View File

@ -5,32 +5,59 @@ import threading
class SimulationThread(threading.Thread):
"""
Singleton class for the simulation thread.
Usage is the call the classmethod initialize first
"""
instance = None
@classmethod
def initialize(cls, periods, start_time=None, run_time=None, time_factor=1.0):
"""
Initializes the singleton Thread with the given values. The given arguments are passed directly to the instance constructor. See the `__init__` method for argument details.
"""
cls.instance = SimulationThread(periods, start_time, run_time, time_factor)
@classmethod
def get_instance(cls, periods=None, start_time=None, run_time=None, time_factor=1.0):
def get_instance(cls):
"""
Returns the simulation thread object, or None if initialize() has not been called.
"""
return cls.instance
def __init__(self, periods, start_time, run_time, time_factor):
"""
Creates a simulation with the provided data and settings.
Args:
periods (list): A list of dictionaries that represent the traffic light periods (peak/off-peak etc) as specified in the problem description. See the provided `periods.json` file for the intended format, as this variable is intended to be a json.load() version of this data.
start_time (datetime.time): The time at which the simulation should start.
run_time (int): The amount of (real) time the simulation should run for before terminating. If None, run indefinitely or until signal_stop() is called.
time_factor (float): The time "accelleration" factor at which the simulation should run. A value of 1.0 indicates indentity with real time, and higher values run faster (e.g. 4.0 = four simulated seconds for each real second).
"""
# Parent constructor call (required). Uses parameter daemon=True to ensure thread shutdown when parent program finishes executing
# See https://docs.python.org/3/library/threading.html
super().__init__(daemon=True)
self.mutex = threading.Lock()
self.running = True
self.periods = periods
self.start_time = start_time
self.run_time = run_time
self.time_factor = time_factor
# NOT a status variable, instead indicates whether signal_stop() has been called
self.running = True
self.current_period_index = 0
self.current_state_index = 0
self.next_changeover_time = None
self.simulation_start_datetime = None
# For convenience, convert all ISO format times to datetime.time objects
for period in self.periods:
period["timestart"] = datetime.time.fromisoformat(period["timestart"])
@ -38,21 +65,30 @@ class SimulationThread(threading.Thread):
self.periods.sort(key=lambda x: x["timestart"])
def signal_stop(self):
"""
Signals the thread to stop, regardless of run time
"""
self.mutex.acquire()
self.running = False
self.mutex.release()
def force_to_time(self, new_time):
"""
Resets the simulation time to the provided time. This is for testing purposes; the current state is IGNORED and the sequence immediately resets to the beginning.
"""
new_datetime = datetime.datetime.fromisoformat("1900-01-01 " + new_time.isoformat())
self.mutex.acquire()
print()
print("Forcing internal time change from {} to {}".format(self.simulation_start_datetime.time().isoformat(), new_time.isoformat()))
# print("Forcing internal time change from {} to {}".format(self.simulation_start_datetime.time().isoformat(), new_time.isoformat()))
self.current_period_index = self._get_period_index(self.periods, new_datetime.time())
self.current_state_index = 0
self.next_changeover_time = time.time() + (self.periods[self.current_period_index]["states"][self.current_state_index]["duration"] / self.time_factor)
self.mutex.release()
def get_snapshot(self):
"""
Returns a dictionary of the current simulated state
"""
self.mutex.acquire()
return_data = {
"current_period_name": self.periods[self.current_period_index]["name"],
@ -60,11 +96,15 @@ class SimulationThread(threading.Thread):
"current_state": self.periods[self.current_period_index]["states"][self.current_state_index],
"next_changeover_time": self.next_changeover_time,
"simulation_time": self.current_simulation_datetime.time(),
"time_remaining": self.next_changeover_time - time.time(),
}
self.mutex.release()
return return_data
def _get_period_index(self, states, test_time: datetime.time):
"""
Returns the index of the period corresponding to the given time
"""
for i, state in enumerate(states):
if i == len(states) - 1:
return i
@ -73,10 +113,17 @@ class SimulationThread(threading.Thread):
return i
# Should NOT be reached since last state assumed to last until midnight
# TODO: Raise exception?
return None
raise
def run(self):
"""
Main thread loop. Note that this method should NOT be called directly: as a subclass of threading.Thread, the inherited start() method should be used instead.
The loop executes until one of the following conditions is met:
- The given run_time is exceeded (in real seconds), or
- signal_stop() has been called
- The executing program finishes (daemon=True)
"""
real_start_time = time.time()
self.simulation_start_datetime = None
@ -91,16 +138,19 @@ class SimulationThread(threading.Thread):
self.next_changeover_time = real_start_time + (self.periods[self.current_period_index]["states"][self.current_state_index]["duration"] / self.time_factor)
print("Starting with simulation time {}".format(self.simulation_start_datetime.time()))
print("Initial state data: {}".format(self.periods[self.current_period_index]["states"][self.current_state_index]))
# print("Starting with simulation time {}".format(self.simulation_start_datetime.time()))
# print("Initial state data: {}".format(self.periods[self.current_period_index]["states"][self.current_state_index]))
"""
if self.run_time is None:
print("No simulation time limit.")
else:
print("Simulation duration (in real second(s)): {}".format(self.run_time))
print("Time compression factor: {}".format(self.run_time, self.time_factor))
"""
# print("Time compression factor: {}".format(self.run_time, self.time_factor))
while (self.run_time is None) or (self.run_time is not None and time.time() < real_start_time + self.run_time):
self.mutex.acquire()
if not self.running:
self.mutex.release()
break
@ -112,33 +162,39 @@ class SimulationThread(threading.Thread):
if now >= self.next_changeover_time:
timed_period_index = self._get_period_index(self.periods, self.current_simulation_datetime.time())
if timed_period_index != self.current_period_index and self.current_state_index == len(self.periods[self.current_period_index]["states"])-1:
# Safe to change states
print()
print("{} : Changing PERIOD from {} to {}".format(self.current_simulation_datetime.time(), self.periods[self.current_period_index]["name"], self.periods[timed_period_index]["name"]))
print(" Old state data: {}".format(self.periods[self.current_period_index]["states"][self.current_state_index]))
# print()
# print("{} : Changing PERIOD from {} to {}".format(self.current_simulation_datetime.time(), self.periods[self.current_period_index]["name"], self.periods[timed_period_index]["name"]))
# print(" Old state data: {}".format(self.periods[self.current_period_index]["states"][self.current_state_index]))
# Safe to change period
self.current_period_index = timed_period_index
self.current_state_index = 0
print(" New state data: {}".format(self.periods[self.current_period_index]["states"][self.current_state_index]))
print(" Next changeover in {} second(s)".format(self.periods[self.current_period_index]["states"][self.current_state_index]["duration"]))
# print(" New state data: {}".format(self.periods[self.current_period_index]["states"][self.current_state_index]))
# print(" Next changeover in {} second(s)".format(self.periods[self.current_period_index]["states"][self.current_state_index]["duration"]))
else:
# Current sequence still completing, not safe to change to next period
next_cycle_index = (self.current_state_index + 1) % len(self.periods[self.current_period_index]["states"])
print()
print("{} : {} : Changing state from {} to {}".format(self.current_simulation_datetime.time(), self.periods[self.current_period_index]["name"], self.current_state_index, next_cycle_index))
# print()
# print("{} : {} : Changing state from {} to {}".format(self.current_simulation_datetime.time(), self.periods[self.current_period_index]["name"], self.current_state_index, next_cycle_index))
"""
if self.current_period_index != timed_period_index:
print(" [period change pending]")
print(" Old state data: {}".format(self.periods[self.current_period_index]["states"][self.current_state_index]))
print(" New state data: {}".format(self.periods[self.current_period_index]["states"][next_cycle_index]))
print(" Next changeover in {} second(s)".format(self.periods[self.current_period_index]["states"][next_cycle_index]["duration"]))
"""
# print(" Old state data: {}".format(self.periods[self.current_period_index]["states"][self.current_state_index]))
# print(" New state data: {}".format(self.periods[self.current_period_index]["states"][next_cycle_index]))
# print(" Next changeover in {} second(s)".format(self.periods[self.current_period_index]["states"][next_cycle_index]["duration"]))
self.current_state_index = next_cycle_index
self.next_changeover_time = self.next_changeover_time + (self.periods[self.current_period_index]["states"][self.current_state_index]["duration"] / self.time_factor)
else:
pass
"""
print(".", end="")
sys.stdout.flush()
"""
self.mutex.release()
time.sleep(1.0 / self.time_factor)