-
Notifications
You must be signed in to change notification settings - Fork 0
/
simulator.py
135 lines (98 loc) · 3.88 KB
/
simulator.py
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
123
124
125
126
127
128
129
130
131
132
133
134
135
from utils import GFrame
from sokoban import *
from utils import *
from time import sleep
from threading import Thread
class Simulator():
def __init__(self, map="sim_test", wait_time=1):
self._map = map
self._wait_time = wait_time
def _setup(self):
self._frame = GFrame("Sokoban Simulator")
self._game = Sokoban(30, self._map, True)
self._frame.display(self._game)
def swap_map(self, map):
self._map = map
def swap_wait_time(self, wait_time):
self._wait_time = wait_time
def _simulate(self, actions):
print("#------------STARTING SIMULATION------------#")
# Simulate the given actions every "speed" seconds
for action in actions:
sleep(self._wait_time)
print("Move: ", action)
self._game.move_enemies()
self._game.move_player(action)
print("#------------SIMULATION FINISHED------------#")
self._game.set_done(True)
def _simulate_agent(self, agent):
print("#------------STARTING SIMULATION------------#")
# Simulate the given actions every "speed" seconds
while self._game.get_state().get_mouse_locations():
sleep(self._wait_time)
action = agent.request_action(self._game.get_state())
print("Move: ", action)
self._game.move_player(action)
self._game.move_enemies()
self._game.set_done(True)
print("#------------SIMULATION FINISHED------------#")
def _simulate_listening_agent(self, agent):
print("#------------STARTING SIMULATION------------#")
# Simulate the given actions every "speed" seconds
while self._game.get_state().get_mouse_locations():
sleep(self._wait_time)
agent.listen(self._game.get_state())
action = agent.request_action(self._game.get_state())
print("Move: ", action)
self._game.move_player(action)
self._game.move_enemies()
self._game.set_done(True)
print("#------------SIMULATION FINISHED------------#")
def _simulate_prediction_agent(self, agent):
print("#------------STARTING SIMULATION------------#")
# Simulate the given actions every "speed" seconds
remaining_mice = self._game.get_state().get_mouse_locations()
while remaining_mice:
sleep(self._wait_time)
agent.predict(self._game.get_state())
action = agent.request_action(self._game.get_state())
print("Move: ", action)
self._game.move_player(action)
self._game.move_enemies()
print("#------------SIMULATION FINISHED------------#")
self._game.set_done(True)
def simulate(self, actions):
self._setup()
# Setup the simulation in its own thread
simulation_thread = Thread(target=self._simulate, args = (actions, ), daemon=True)
simulation_thread.start()
# Run the game and frame
self._frame.run()
def simulate_agent(self, agent):
self._setup()
# Setup the simulation in its own thread
simulation_thread = Thread(target=self._simulate_agent, args = (agent, ), daemon=True)
simulation_thread.start()
# Run the game and frame
self._frame.run()
def simulate_generic(self, agent):
self._setup()
# Setup the simulation in its own thread
simulation_thread = Thread(target=self._simulate_agent, args = (agent, ), daemon=True)
simulation_thread.start()
# Run the game and frame
self._frame.run()
def simulate_listening_agent(self, agent):
self._setup()
# Setup the simulation in its own thread
simulation_thread = Thread(target=self._simulate_listening_agent, args = (agent, ), daemon=True)
simulation_thread.start()
# Run the game and frame
self._frame.run()
def simulate_prediction_agent(self, agent):
self._setup()
# Setup the simulation in its own thread
simulation_thread = Thread(target=self._simulate_prediction_agent, args = (agent, ), daemon=True)
simulation_thread.start()
# Run the game and frame
self._frame.run()