Spaces:
Running
Running
File size: 3,958 Bytes
4d03ebc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 | import unittest
import cityflow
import time
class TestArchive(unittest.TestCase):
config_file = "./examples/config.json"
period = 100
@staticmethod
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)
@staticmethod
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)
|