Spaces:
Running
Running
| import unittest | |
| import cityflow | |
| import time | |
| class TestArchive(unittest.TestCase): | |
| config_file = "./examples/config.json" | |
| period = 100 | |
| def run_steps(engine, steps): | |
| for i in range(steps): | |
| engine.next_step() | |
| def run_and_check(self, engine, record): | |
| self.run_steps(engine, self.period) | |
| new_record = self.get_record(engine) | |
| self.assertEqual(new_record, record) | |
| def get_record(engine): | |
| return engine.get_lane_vehicle_count(), engine.get_average_travel_time() | |
| def test_save_and_load(self): | |
| """Single save and single load with single threading engine""" | |
| engine = cityflow.Engine(config_file=self.config_file, thread_num=1) | |
| self.run_steps(engine, self.period) | |
| start_time = time.time() | |
| archive = engine.snapshot() | |
| save_time = time.time() - start_time | |
| self.run_steps(engine, self.period) | |
| record0 = self.get_record(engine) | |
| start_time = time.time() | |
| engine.load(archive) | |
| load_time = time.time() - start_time | |
| self.run_and_check(engine, record0) | |
| del engine | |
| print("\nsave: %.4fs load: %.4fs" % (save_time, load_time)) | |
| def test_save_and_load_multithread(self): | |
| """Single save and single load with multi-threading engine""" | |
| engine = cityflow.Engine(config_file=self.config_file, thread_num=4) | |
| self.run_steps(engine, self.period) | |
| archive = engine.snapshot() | |
| self.run_steps(engine, self.period) | |
| record0 = self.get_record(engine) | |
| engine.load(archive) | |
| self.run_and_check(engine, record0) | |
| del engine | |
| def test_save_and_multi_load(self): | |
| """Multiple saves and multiple loads with multi-threading engine""" | |
| engine = cityflow.Engine(config_file=self.config_file, thread_num=4) | |
| self.run_steps(engine, self.period) | |
| archive = engine.snapshot() | |
| self.run_steps(engine, self.period) | |
| record0 = self.get_record(engine) | |
| repeats = 2 | |
| for i in range(repeats): | |
| engine.load(archive) | |
| self.run_and_check(engine, record0) | |
| del engine | |
| def test_multi_save_and_multi_load(self): | |
| """ Multiple save and multiple loads with multi-threading engine") """ | |
| engine = cityflow.Engine(config_file=self.config_file, thread_num=4) | |
| archives, records = [], [] | |
| repeats = 5 | |
| for i in range(repeats + 1): | |
| archives.append(engine.snapshot()) | |
| records.append(self.get_record(engine)) | |
| self.run_steps(engine, self.period) | |
| for i in range(repeats): | |
| for j in range(repeats): | |
| engine.load(archives[j]) | |
| self.run_and_check(engine, records[j + 1]) | |
| del engine | |
| def test_save_to_file(self): | |
| """ Disk IO test """ | |
| engine = cityflow.Engine(config_file=self.config_file, thread_num=4) | |
| self.run_steps(engine, self.period) | |
| engine.snapshot().dump("save.json") | |
| self.run_steps(engine, self.period) | |
| record = self.get_record(engine) | |
| engine.load_from_file("save.json") | |
| self.run_and_check(engine, record) | |
| del engine | |
| def test_multi_save_to_file(self): | |
| """ Disk IO test 2""" | |
| engine = cityflow.Engine(config_file=self.config_file, thread_num=4) | |
| for i in range(2): | |
| self.run_steps(engine, self.period) | |
| engine.snapshot().dump("save.json") | |
| self.run_steps(engine, self.period) | |
| record = self.get_record(engine) | |
| for j in range(2): | |
| engine.load_from_file("save.json") | |
| self.run_and_check(engine, record) | |
| del engine | |
| if __name__ == '__main__': | |
| unittest.main(verbosity=2) | |