-
Notifications
You must be signed in to change notification settings - Fork 38
/
game_state.py
147 lines (125 loc) · 5.3 KB
/
game_state.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
136
137
138
139
140
141
142
143
144
145
146
147
import asyncio
import websockets
import json
_agent_move_set = set(("up", "down", "left", "right"))
class GameState:
def __init__(self, connection_string: str):
self._connection_string = connection_string
self._state = None
self._tick_callback = None
def set_game_tick_callback(self, generate_agent_action_callback):
self._tick_callback = generate_agent_action_callback
async def connect(self):
self.connection = await websockets.client.connect(self._connection_string)
if self.connection.open:
return self.connection
async def _send(self, packet):
await self.connection.send(json.dumps(packet))
async def send_move(self, move: str):
if move in _agent_move_set:
packet = {"type": "move", "move": move}
await self._send(packet)
async def send_bomb(self):
packet = {"type": "bomb"}
await self._send(packet)
async def send_detonate(self, x, y):
packet = {"type": "detonate", "coordinates": [x, y]}
await self._send(packet)
async def _handle_messages(self, connection: str):
while True:
try:
raw_data = await connection.recv()
data = json.loads(raw_data)
await self._on_data(data)
except websockets.exceptions.ConnectionClosed:
print('Connection with server closed')
break
async def _on_data(self, data):
data_type = data.get("type")
if data_type == "info":
# no operation
pass
elif data_type == "game_state":
payload = data.get("payload")
self._on_game_state(payload)
elif data_type == "tick":
payload = data.get("payload")
await self._on_game_tick(payload)
else:
print(f"unknown packet \"{data_type}\": {data}")
def _on_game_state(self, game_state):
self._state = game_state
async def _on_game_tick(self, game_tick):
events = game_tick.get("events")
for event in events:
event_type = event.get("type")
if event_type == "entity_spawned":
self._on_entity_spawned(event)
elif event_type == "entity_expired":
self._on_entity_expired(event)
elif event_type == "agent":
agent_action = event.get("data")
agent_number = event.get("agent_number")
self._on_agent_action(agent_number, agent_action)
elif event_type == "agent_state":
payload = event.get("data")
self._on_agent_state(payload)
elif event_type == "entity_state":
x, y = event.get("coordinates")
updated_entity = event.get("updated_entity")
self._on_entity_state(x, y, updated_entity)
else:
print(f"unknown event type {event_type}: {event}")
if self._tick_callback is not None:
tick_number = game_tick.get("tick")
await self._tick_callback(tick_number, self._state)
def _on_entity_spawned(self, spawn_event):
spawn_payload = spawn_event.get("data")
self._state["entities"].append(spawn_payload)
def _on_entity_expired(self, spawn_event):
expire_payload = spawn_event.get("data")
def filter_entity_fn(entity):
[x, y] = expire_payload
entity_x = entity.get("x")
entity_y = entity.get("y")
should_remove = entity_x == x and entity_y == y
return should_remove == False
self._state["entities"] = list(filter(
filter_entity_fn, self._state["entities"]))
def _on_agent_state(self, agent_state):
agent_number = agent_state.get("number")
self._state["agent_state"][str(agent_number)] = agent_state
def _on_entity_state(self, x, y, updated_entity):
for entity in self._state.get("entities"):
if entity.get("x") == x and entity.get("y") == y:
self._state["entities"].remove(entity)
self._state["entities"].append(updated_entity)
def _on_agent_action(self, agent_number, action_packet):
agent = self._state["agent_state"][str(agent_number)]
coordinates = agent.get("coordinates")
action_type = action_packet.get("type")
if action_type == "move":
move = action_packet.get("move")
if move in _agent_move_set:
new_coordinates = self._get_new_agent_coordinates(
coordinates, move)
self._state["agent_state"][str(
agent_number)]["coordinates"] = new_coordinates
elif action_type == "bomb":
# no - op since this is redudnant info
pass
elif action_type == "detonate":
# no - op since this is redudnant info
pass
else:
print(f"Unhandled agent action recieved: {action_type}")
def _get_new_agent_coordinates(self, coordinates, move_action) -> [int, int]:
[x, y] = coordinates
if move_action == "up":
return [x, y+1]
elif move_action == "down":
return [x, y-1]
elif move_action == "right":
return [x+1, y]
elif move_action == "left":
return [x-1, y]