-
Notifications
You must be signed in to change notification settings - Fork 0
/
turn_taking_client.py
91 lines (73 loc) · 3.02 KB
/
turn_taking_client.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
# coding=utf-8
from uuid import uuid4
from client import LogLevel
from ordered_turn_client import SecureOrderedClient
class TurnTakingClient(SecureOrderedClient):
"""This client handles the high level sequencing of players in
turn based games. it relies on the players having been ordered previously
and have a 'roll' field in their peer_map entry."""
END_TURN = 'end_turn'
ROOM_CODE = 'room_code'
LEAVE_ROOM = 'leave_room'
def __init__(self, cli, state=None):
super().__init__(cli, state)
self.queue_map.extend([(self.END_TURN, self.recv_end_turn),
(self.LEAVE_ROOM, self.recv_leave_room)])
self.current_turn = 0
# Pregrnerate a room code; this will be overwritten if we're not player 0
self.room_code = uuid4()
self.setup_finished = False
def recv_leave_room(self, data):
if self.is_turn_valid(data):
pass
def recv_end_turn(self, data):
if data['data'][self.ROOM_CODE] == self.room_code:
if not data[self.SENDER_ID] == self.get_current_turn():
self.cli.log(LogLevel.ERROR,
"An invalid turn was made: {}".format(data))
else:
self.current_turn += 1
self.take_turn_if_mine()
def is_first_turn(self):
return self.current_turn == 0
def alert_players_have_been_ordered(self):
if self.is_my_turn():
self.take_turn()
def is_my_turn(self):
return self.get_current_turn() == self.cli.ident
def get_current_turn(self):
for ident, peer in self.peer_map.items():
if peer.get('roll') == self.current_turn % self.max_players:
return ident
raise IndexError
def end_my_turn(self):
self.current_turn += 1
self.cli.post_message(data={self.MESSAGE_KEY: self.END_TURN,
self.ROOM_CODE: self.room_code})
def take_turn(self):
raise NotImplementedError
def take_turn_if_mine(self):
if self.is_my_turn():
self.take_turn()
def is_turn_valid(self, data):
if data['data'][self.ROOM_CODE] != self.room_code:
if self.current_turn == 0:
self.room_code = data['data'][self.ROOM_CODE]
return True
else:
raise ValueError("Invalid Message")
return True
def send_round_message(self, key, data):
data.update({self.MESSAGE_KEY: key,
self.ROOM_CODE: self.room_code})
self.cli.post_message(data=data)
def get_ident_at_position(self, position):
ident, _ = self.get_peer_at_position(position)
return ident
def get_peer_at_position(self, position, peer_map=None):
if peer_map is None:
peer_map = self.peer_map
return [peer for peer in peer_map.items() if peer[1]['roll'] ==
position % self.max_players][0]
def get_my_position(self):
return self.peer_map[self.cli.ident]['roll']