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
|
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:
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:
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))
|
