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
|
""" Module that defines the interface (using WebSockets) between the GUI and
the game itself. """
import contextlib
import json
import logging
from typing import Union
import websockets.exceptions
from websockets.server import WebSocketServerProtocol
from swiftstory.client import Client
from swiftstory.exception import WrongAction, JoinError
from swiftstory.game_manager import GameManager
from swiftstory.status import error
class WebsocketsInterface:
""" Interface with WebSockets for SwiftStory. """
def __init__(self, game_manager: GameManager):
self.game_manager = game_manager
async def connection_handler(self, websocket: WebSocketServerProtocol, path: str) -> None:
""" Coroutine that handles one websocket connection. """
client = Client(websocket, self.game_manager)
with contextlib.suppress(websockets.exceptions.ConnectionClosed):
async for message in client.socket:
await client.socket.send(self.message_received_handler(client, message))
client.disconnect()
def message_received_handler(self, client: Client, message: Union[bytes, str]) -> str:
""" Handle a message and return an answer. """
def join_game() -> str:
try:
game_name = json_msg['game_name']
except KeyError:
return error('field `game_name\' is required')
else:
lang = json_msg.get('lang')
return client.join_game(game_name, lang)
def designate_card() -> str:
card_id = None
try:
card_id = int(json_msg['card_id'])
except (KeyError, TypeError):
pass
finally:
return client.designate_card(card_id)
def play_white_card() -> str:
try:
card_id = int(json_msg['card_id'])
except KeyError:
return error('field `card_id\' is required')
else:
return client.play_white_card(card_id)
opcodes_map = {
"join_game": join_game,
"view_player_cards": lambda: client.view_player_cards(),
"view_black_card": lambda: client.view_black_card(),
"view_played_cards": lambda: client.view_played_cards(),
"pick_black_card": lambda: client.pick_black_card(),
"collect_cards": lambda: client.collect_cards(),
"designate_card": designate_card,
"play_white_card": play_white_card,
}
try:
json_msg = json.loads(message)
except json.JSONDecodeError:
return error('badly formatted json')
try:
return opcodes_map[json_msg["op"]]()
except (KeyError, TypeError):
return error('invalid command')
except WrongAction as e:
return error(str(e))
except JoinError as e:
logging.warning("player could not join game: %s", e.__repr__())
return error(str(e))
|
