"""A board you can click, served to your browser.
The model stays in Python -- it is a PyTorch network and belongs where torch
is -- and the page talks to it over a tiny JSON API on localhost. That keeps
the whole thing dependency-free: `http.server` is standard library, and the
page needs no framework, no CDN and no build step.
"""
import json
import threading
import webbrowser
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Optional
import chess
from engine import ChessEngine
# The filled glyphs for both sides, with the colour done in CSS. Using the
# outline set for White and the filled set for Black is the obvious idea and it
# does not work: the two render nearly identically once the page picks a single
# text colour, so on a dark board every piece looked white.
PIECES = {"k": "♚", "q": "♛", "r": "♜", "b": "♝", "n": "♞", "p": "♟"}
class Session:
"""One game, guarded by a lock so a double click cannot race an engine.
Each colour is either an engine -- anything with ``get_move(engine)``, so
the network and the minimax are interchangeable -- or None for the human.
Two engines and nobody human is the spectator mode: the game plays itself
one ply per `/api/step`, which lets the browser pace and pause it.
"""
def __init__(self, white, black, max_moves: int, labels=("white", "black")):
self.players = {chess.WHITE: white, chess.BLACK: black}
self.labels = {chess.WHITE: labels[0], chess.BLACK: labels[1]}
self.max_moves = max_moves
self.lock = threading.Lock()
self.reset()
def reset(self) -> None:
self.engine = ChessEngine()
self.history = []
self._auto_play()
@property
def watching(self) -> bool:
return all(player is not None for player in self.players.values())
@property
def human_color(self) -> Optional[bool]:
for color, player in self.players.items():
if player is None:
return color
return None
@property
def human_is_white(self) -> bool:
return self.human_color != chess.BLACK
def _finished(self) -> bool:
return self.engine.is_game_over() or self.engine.get_move_count() >= self.max_moves
def step(self) -> bool:
"""Play one engine ply. False when it is a human's turn or the game ended."""
if self._finished():
return False
player = self.players[self.engine.get_turn()]
if player is None:
return False
move = player.get_move(self.engine)
if move is None:
return False
self.history.append(self.engine.board.san(move))
self.engine.make_move(move)
return True
def _auto_play(self) -> None:
"""Let the engines move until a human is on turn.
Bounded to one ply in spectator mode so the browser sees every move
rather than a finished game.
"""
if self.watching:
return
while self.step():
pass
def play(self, uci: str) -> Optional[str]:
"""Apply the human move, then the engine's reply. Returns an error, if any."""
if self._finished() or self.watching:
return "the game is over"
try:
move = chess.Move.from_uci(uci)
except ValueError:
return "unreadable move"
# Promotion is not asked for in the UI; anything else than a queen is
# rare enough that offering it would cost more clicks than it saves.
if move not in self.engine.board.legal_moves:
promoted = chess.Move(move.from_square, move.to_square, promotion=chess.QUEEN)
if promoted in self.engine.board.legal_moves:
move = promoted
else:
return "illegal move"
self.history.append(self.engine.board.san(move))
self.engine.make_move(move)
self._auto_play()
return None
def state(self) -> dict:
board = self.engine.board
squares = []
for rank in range(7, -1, -1):
for file in range(8):
piece = board.piece_at(chess.square(file, rank))
if piece is None:
squares.append(None)
else:
squares.append(
{
"glyph": PIECES[piece.symbol().lower()],
"side": "w" if piece.color == chess.WHITE else "b",
}
)
legal = {}
if not self._finished() and board.turn == self.human_color:
for move in board.legal_moves:
legal.setdefault(chess.square_name(move.from_square), []).append(
chess.square_name(move.to_square)
)
if board.is_checkmate():
winner = self.labels[not board.turn]
status = (
f"Checkmate — {winner} wins"
if self.watching
else "Checkmate — " + ("you win" if board.turn != self.human_color else "you lose")
)
elif board.is_stalemate():
status = "Stalemate — draw"
elif board.is_insufficient_material():
status = "Insufficient material — draw"
elif board.is_repetition(3):
status = "Threefold repetition — draw"
elif board.halfmove_clock >= 100:
status = "Fifty-move rule — draw"
elif self.engine.get_move_count() >= self.max_moves:
status = "Move limit reached"
elif board.is_check():
status = "Check!"
elif self.watching:
status = f"{self.labels[board.turn]} to move"
else:
status = "Your move" if board.turn == self.human_color else "Thinking…"
# A piece with no legal move is the commonest "the board is broken"
# moment: the click does nothing and nothing says why. Ship the reason.
check_square = None
if board.is_check():
king = board.king(board.turn)
check_square = chess.square_name(king) if king is not None else None
pinned = [
chess.square_name(square)
for square in chess.SQUARES
if (piece := board.piece_at(square)) is not None
and piece.color == self.human_color
and board.is_pinned(self.human_color, square)
]
return {
"squares": squares,
"legal": legal,
"check": check_square,
"pinned": pinned,
"yourTurn": board.turn == self.human_color and not self._finished(),
"watching": self.watching,
"whiteToMove": board.turn == chess.WHITE,
"white": self.labels[chess.WHITE],
"black": self.labels[chess.BLACK],
"last": [
chess.square_name(board.peek().from_square),
chess.square_name(board.peek().to_square),
] if board.move_stack else None,
"status": status,
"finished": self._finished(),
"flipped": not self.human_is_white,
# The whole game, not a tail: truncating it while numbering from 1
# made every excerpt look like a game that opened with a king move.
"history": self.history,
"plies": self.engine.get_move_count(),
"evaluation": round(self.engine.evaluate() / 100.0, 2),
}
PAGE = """
chess-rl
"""
def serve(model, make_algo, model_label: str, human_is_white: bool = True,
max_moves: int = 300, port: int = 8000, depth: int = 2) -> int:
"""Serve the board, with mode and minimax depth switchable from the page.
`make_algo(depth)` builds a fresh minimax, so the slider can change the
search depth without restarting the server.
"""
def build(mode: str, depth: int) -> Session:
algo_label = f"minimax d{depth}"
if mode == "watch":
# The model takes White, so the game opens on its own choice.
return Session(model, make_algo(depth), max_moves, (model_label, algo_label))
rival, label = (
(make_algo(depth), algo_label) if mode == "algo" else (model, model_label)
)
if human_is_white:
return Session(None, rival, max_moves, ("you", label))
return Session(rival, None, max_moves, (label, "you"))
state = {"session": build("model", depth), "mode": "model", "depth": depth}
class Handler(BaseHTTPRequestHandler):
def log_message(self, *args):
pass # the board is the interface; request logs only get in the way
def _send(self, payload: bytes, kind: str) -> None:
self.send_response(200)
self.send_header("Content-Type", kind)
self.send_header("Content-Length", str(len(payload)))
self.end_headers()
self.wfile.write(payload)
def _payload(self) -> bytes:
session = state["session"]
with session.lock:
snapshot = session.state()
snapshot["mode"] = state["mode"]
snapshot["depth"] = state["depth"]
return json.dumps(snapshot).encode()
def do_GET(self):
if self.path.startswith("/api/state"):
self._send(self._payload(), "application/json")
else:
self._send(PAGE.encode(), "text/html; charset=utf-8")
def do_POST(self):
length = int(self.headers.get("Content-Length", 0))
body = json.loads(self.rfile.read(length) or b"{}")
if self.path.startswith("/api/mode"):
mode = body.get("mode", "model")
requested = int(body.get("depth", state["depth"]))
if mode in ("model", "algo", "watch"):
state["mode"] = mode
state["depth"] = max(1, min(5, requested))
state["session"] = build(mode, state["depth"])
else:
session = state["session"]
with session.lock:
if self.path.startswith("/api/new"):
session.reset()
elif self.path.startswith("/api/move"):
session.play(body.get("move", ""))
elif self.path.startswith("/api/step"):
session.step()
self._send(self._payload(), "application/json")
server = ThreadingHTTPServer(("127.0.0.1", port), Handler)
url = f"http://127.0.0.1:{port}/"
print(f"Board at {url} (Ctrl-C to stop)")
threading.Timer(0.5, lambda: webbrowser.open(url)).start()
try:
server.serve_forever()
except KeyboardInterrupt:
print("\nBye.")
finally:
server.server_close()
return 0