Experiment definition

Experiment.py

from CA import CARoom, CACell
from itertools import product
import multiprocessing as mp
import numpy as np
from CustomRoom import CustomRoom

class Experiment(object):
	def __init__(self, repeats=30):
		self.repeats = repeats
		self.results = []
		self.cnt = 0


	def run_simulation(self, simu, people):
		evac_total = 0
		iters = 0
		done = False

		while not done:			
			evac = simu.step_parallel()
			evac_total+= evac
			done = people == evac_total
			iters+= 1
		return iters


	def log_results(self, result):		
		self.results.append(result)
		self.cnt+= 1


	def run(self, full, panic, exit_size):
		n_cpu = max(1, mp.cpu_count())
		pool = mp.Pool(n_cpu)

		for i in range(self.repeats):
			simu, people = CustomRoom.make_2_obstacle_room(full_factor=full, pos_seed=None, panic_prob=panic, exit_size=exit_size)
			pool.apply_async(self.run_simulation, args=(simu, people), callback=self.log_results)

		pool.close()
		pool.join()
		return self.results


	@staticmethod
	def run_grid(reps=30):			
		full = [0.3, 0.5, 0.67, 0.9]
		panic = [0.05]
		exit_size = [1, 2, 3, 4, 5, 6]
		params = [[f, p] for f in full for p in panic]
		params = product(full, panic, exit_size)

		for f, p, s in params:
			e = Experiment(repeats=reps)
			results = e.run(full=f, panic=p, exit_size=s)
			evac_mean = np.mean(np.array(results))
			evac_sd = np.std(results)
			print(f, p, s, evac_mean, evac_sd)



if __name__ == '__main__':
	Experiment.run_grid()