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
|
import asyncio
import json
import random
import typing
import weakref
import aiohttp
import chess
import chess.engine
from aiohttp import web
app = web.Application()
app['websockets'] = weakref.WeakSet()
app['chessgames'] = weakref.WeakSet()
async def index(request: web.Request) -> web.Response:
return web.Response(text="Welcome to the chess hell backend!")
async def handle_socket(request: web.Request):
ws = web.WebSocketResponse()
await ws.prepare(request)
request.app['websockets'].add(ws)
engine: typing.Optional[chess.engine.UciProtocol] = None
try:
board = chess.Board()
transport, engine = await chess.engine.popen_uci('stockfish')
await ws.send_json(dict(event="ready", player_color='white', board=board.fen()))
async for msg in ws:
msg: aiohttp.WSMessage
if msg.type == aiohttp.WSMsgType.TEXT:
req = json.loads(msg.data)
m = req["method"]
if m == "move":
user_move = chess.Move.from_uci(req['params']['move'])
if user_move not in board.legal_moves:
await ws.send_json(dict(
event="reject_move",
board=board.fen(),
))
continue
board.push(user_move)
await ws.send_json(dict(
event='accept_move',
board=board.fen(),
lastmove=user_move.uci(),
))
await asyncio.sleep(1.5)
# candidates = await engine.analyse(board, chess.engine.Limit(time=1), multipv=100)
my_move: chess.Move = random.choice(list(board.legal_moves))
board.push(my_move)
await ws.send_json(dict(
event="computer_moved", board=board.fen(),
lastmove=my_move.uci(),
))
finally:
if not ws.closed:
await ws.close()
print("Cleaning up websocket")
request.app['websockets'].discard(ws)
if engine is not None:
asyncio.create_task(engine.quit())
return ws
async def on_shutdown(app):
print("On shutdown called")
for ws in set(app['websockets']):
ws: web.WebSocketResponse
await ws.close(code=aiohttp.WSCloseCode.GOING_AWAY,
message=b'Server shutdown')
print("Closing websocket")
async def handle_status(request: web.Request):
return web.json_response(dict(
all_tasks=len(asyncio.all_tasks(asyncio.get_running_loop())),
websockets=len(request.app['websockets']),
games=len(request.app['chessgames']),
))
app.add_routes([
web.get('/', index),
web.get('/socket', handle_socket),
web.get('/status', handle_status),
])
app.on_shutdown.append(on_shutdown)
if __name__ == '__main__':
web.run_app(app, )
|
