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

Internal function cleanup, changed simulation test code to use 

SimulationThread instead of (obsolete) own loop.
This commit is contained in:
Chris Davoren 2023-10-21 16:03:44 +10:00
parent 5719813c46
commit 2f5d8d70bc
3 changed files with 25 additions and 84 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

92
period-test.py
View File

@ -1,86 +1,18 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
import sys
import json import json
import datetime import datetime
import time from simulation import SimulationThread
def get_period_index(periods, test_time : datetime.time):
for i, state in enumerate(periods):
if i == len(periods) - 1:
return(i)
if test_time >= state["timestart"] and test_time < periods[i+1]["timestart"]:
return(i)
# Should NOT be reached since last state assumed to last until midnight
# TODO: Raise exception?
return None
def simulate(periods, start_time, run_time=90, time_factor=1.0):
simulation_start_time = time.time()
simulation_start_datetime = datetime.datetime.fromisoformat("1900-01-01 " + start_time.isoformat())
current_period_index = get_period_index(periods, simulation_start_datetime.time())
current_state_index = 0
next_changeover_time = simulation_start_time + (periods[current_period_index]["states"][current_state_index]["duration"] / time_factor)
print("Starting with simulation time {}".format(simulation_start_datetime.time()))
print("Initial data: {}".format(periods[current_period_index]["states"][current_state_index]))
print("Starting simulation of {} second(s) with time factor {}".format(run_time, time_factor))
while (run_time is None) or (run_time is not None and time.time() < simulation_start_time + run_time):
now = time.time()
current_simulation_datetime = simulation_start_datetime + datetime.timedelta(seconds=int((now - simulation_start_time) * time_factor))
timed_state_index = get_period_index(periods, current_simulation_datetime.time())
if now >= next_changeover_time:
timed_state_index = get_period_index(periods, current_simulation_datetime.time())
if timed_state_index != current_period_index and current_state_index == len(periods[current_period_index]["states"])-1:
# Safe to change periods
print()
print("{} : Changing periods from {} to {}".format(current_simulation_datetime.time(), periods[current_period_index]["name"], periods[timed_state_index]["name"]))
print(" Old cycle data: {}".format(periods[current_period_index]["states"][current_state_index]))
current_period_index = timed_state_index
current_state_index = 0
print(" New cycle data: {}".format(periods[current_period_index]["states"][current_state_index]))
print(" Next changeover in {} second(s)".format(periods[current_period_index]["states"][current_state_index]["duration"]))
else:
next_cycle_index = (current_state_index + 1) % len(periods[current_period_index]["states"])
print()
print("{} : Changing cycle from {} to {}".format(current_simulation_datetime.time(), current_state_index, next_cycle_index))
if current_period_index != timed_state_index:
print(" [state change pending]")
print(" Old cycle data: {}".format(periods[current_period_index]["states"][current_state_index]))
print(" New cycle data: {}".format(periods[current_period_index]["states"][next_cycle_index]))
print(" Next changeover in {} second(s)".format(periods[current_period_index]["states"][next_cycle_index]["duration"]))
current_state_index = next_cycle_index
next_changeover_time = next_changeover_time + (periods[current_period_index]["states"][current_state_index]["duration"] / time_factor)
else:
print(".", end="")
sys.stdout.flush()
time.sleep(1.0 / time_factor)
def main(): def main():
periods = json.load(open("periods.json")) periods = json.load(open("periods.json"))
# Essential due to assumed ordering further below print("Period listing:")
periods.sort(key=lambda x : x["timestart"]) for period in periods:
period_start_time = datetime.time.fromisoformat(period["timestart"])
for state in periods: print("Name: {}\t\tTime start: {}".format(period["name"], period_start_time))
time = datetime.time.fromisoformat(state["timestart"]) print(" Number of states: {}".format(len(period["states"])))
print("Name: {}\t\tTime start: {}".format(state["name"], time))
print(" Number of states: {}".format(len(state["states"])))
state["timestart"] = datetime.time.fromisoformat(state["timestart"])
test_times = [ test_times = [
"00:00:00", "00:00:00",
@ -95,12 +27,20 @@ def main():
"23:59:59", "23:59:59",
] ]
SimulationThread.initialize(periods, datetime.time.fromisoformat("07:59"), 60, 4.0)
for test_time in test_times: for test_time in test_times:
test_time_obj = datetime.time.fromisoformat(test_time) test_time_obj = datetime.time.fromisoformat(test_time)
print("Period matching time {} is '{}'".format(test_time, periods[get_period_index(periods, test_time_obj)]["name"])) print("Period matching time {} is '{}'".format(test_time, periods[SimulationThread.get_instance()._get_period_index(test_time_obj)]["verbose-name"]))
print("Starting thread...")
SimulationThread.get_instance().start()
SimulationThread.get_instance().join()
print("Thread finished")
print(SimulationThread.get_instance().get_snapshot())
simulate(periods, datetime.time.fromisoformat("07:59"), 60, 4.0)
if __name__ == '__main__': if __name__ == '__main__':
main() main()

16
simulation/simulation_thread.py
View File

@ -103,15 +103,15 @@ class SimulationThread(threading.Thread):
self.mutex.release() self.mutex.release()
return return_data return return_data
def _get_period_index(self, states, test_time: datetime.time): def _get_period_index(self, test_time: datetime.time):
""" """
Returns the index of the period corresponding to the given time Returns the index of the period corresponding to the given time
""" """
for i, state in enumerate(states): for i, period in enumerate(self.periods):
if i == len(states) - 1: if i == len(self.periods) - 1:
return i return i
if test_time >= state["timestart"] and \ if test_time >= period["timestart"] and \
test_time < states[i+1]["timestart"]: test_time < self.periods[i + 1]["timestart"]:
return i return i
# Should NOT be reached since last state assumed to last until midnight # Should NOT be reached since last state assumed to last until midnight
@ -135,7 +135,7 @@ class SimulationThread(threading.Thread):
self.simulation_start_datetime = datetime.datetime.fromisoformat( self.simulation_start_datetime = datetime.datetime.fromisoformat(
"1900-01-01 " + self.start_time.isoformat()) "1900-01-01 " + self.start_time.isoformat())
self.current_period_index = self._get_period_index(self.periods, self.simulation_start_datetime.time()) self.current_period_index = self._get_period_index(self.simulation_start_datetime.time())
self.current_state_index = 0 self.current_state_index = 0
self.next_changeover_time = real_start_time + (self.periods[self.current_period_index]["states"][self.current_state_index]["duration"] / self.time_factor) self.next_changeover_time = real_start_time + (self.periods[self.current_period_index]["states"][self.current_state_index]["duration"] / self.time_factor)
@ -159,10 +159,10 @@ class SimulationThread(threading.Thread):
now = time.time() now = time.time()
self.current_simulation_datetime = self.simulation_start_datetime + datetime.timedelta(seconds=int((now - real_start_time) * self.time_factor)) self.current_simulation_datetime = self.simulation_start_datetime + datetime.timedelta(seconds=int((now - real_start_time) * self.time_factor))
timed_period_index = self._get_period_index(self.periods, self.current_simulation_datetime.time()) timed_period_index = self._get_period_index(self.current_simulation_datetime.time())
if now >= self.next_changeover_time: if now >= self.next_changeover_time:
timed_period_index = self._get_period_index(self.periods, self.current_simulation_datetime.time()) timed_period_index = self._get_period_index(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: if timed_period_index != self.current_period_index and self.current_state_index == len(self.periods[self.current_period_index]["states"])-1:
# print() # 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("{} : Changing PERIOD from {} to {}".format(self.current_simulation_datetime.time(), self.periods[self.current_period_index]["name"], self.periods[timed_period_index]["name"]))

1
trafficlightfrontend/templates/trafficlightfrontend/index.html
View File

@ -88,6 +88,7 @@
let static_path_prefix = "/static/trafficlightfrontend/"; let static_path_prefix = "/static/trafficlightfrontend/";
document.getElementById("period").innerHTML = light_status.current_period_verbose_name; document.getElementById("period").innerHTML = light_status.current_period_verbose_name;
document.getElementById("time").innerHTML = light_status.simulation_time; document.getElementById("time").innerHTML = light_status.simulation_time;
// Important note: Due to marginal overshoot of scheduled timing, it is possible for time-remaining to be very slightly negative // Important note: Due to marginal overshoot of scheduled timing, it is possible for time-remaining to be very slightly negative
// It is undesirable to present this on the frontend, and therefore zero is the lowest number displayed // It is undesirable to present this on the frontend, and therefore zero is the lowest number displayed
document.getElementById("time-remaining").innerHTML = Math.max(0, Math.round(light_status.time_remaining * 1000) / 1000).toFixed(3); document.getElementById("time-remaining").innerHTML = Math.max(0, Math.round(light_status.time_remaining * 1000) / 1000).toFixed(3);