from __future__ import annotations
import argparse
import json
import threading
import webbrowser
from dataclasses import dataclass
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any
import numpy as np
from sb3_contrib import MaskablePPO
from gomoku_rl.opponents import (
HeuristicOpponent,
PolicyGuidedOpponent,
RandomOpponent,
action_to_coord,
is_winning_move,
predict_masked_action,
)
HTML_PAGE = """
五子棋 AI 对战
"""
@dataclass
class GameState:
board: np.ndarray
human_turn: bool
winner: str | None
last_ai_move: tuple[int, int] | None
move_count: int
class GomokuWebApp:
def __init__(
self,
board_size: int,
win_length: int,
human_first: bool,
opponent_name: str,
model_path: Path | None = None,
seed: int = 42,
search_depth: int = 4,
candidate_radius: int = 2,
max_candidates: int = 10,
policy_weight: float = 220_000.0,
) -> None:
self.opponent_name = opponent_name
self.model = MaskablePPO.load(model_path) if opponent_name in {"model", "hybrid"} and model_path is not None else None
self.rule_opponent = self._build_rule_opponent(
opponent_name=opponent_name,
seed=seed,
search_depth=search_depth,
candidate_radius=candidate_radius,
max_candidates=max_candidates,
policy_weight=policy_weight,
)
self.board_size = board_size
self.win_length = win_length
self.human_first = human_first
self.lock = threading.Lock()
self.state = self._new_game()
def _new_game(self) -> GameState:
state = GameState(
board=np.zeros((self.board_size, self.board_size), dtype=np.int8),
human_turn=self.human_first,
winner=None,
last_ai_move=None,
move_count=0,
)
if not self.human_first:
self._ai_move(state)
return state
def _build_rule_opponent(
self,
opponent_name: str,
seed: int,
search_depth: int,
candidate_radius: int,
max_candidates: int,
policy_weight: float,
):
if opponent_name == "heuristic":
return HeuristicOpponent(seed=seed)
if opponent_name == "random":
return RandomOpponent(seed=seed)
if opponent_name == "hybrid":
if self.model is None:
raise RuntimeError("Hybrid opponent selected but no model was loaded.")
return PolicyGuidedOpponent(
model=self.model,
seed=seed,
search_depth=search_depth,
candidate_radius=candidate_radius,
max_candidates=max_candidates,
early_max_candidates=max(max_candidates + 4, max_candidates),
policy_weight=policy_weight,
)
return None
def reset(self) -> dict[str, Any]:
with self.lock:
self.state = self._new_game()
return self._serialize_state(self.state)
def get_state(self) -> dict[str, Any]:
with self.lock:
return self._serialize_state(self.state)
def make_human_move(self, row: int, col: int) -> dict[str, Any]:
with self.lock:
state = self.state
if state.winner is not None:
return self._serialize_state(state, message="对局已结束，请重新开始。")
if not state.human_turn:
return self._serialize_state(state, message="当前不是你的回合。")
if not (0 <= row < self.board_size and 0 <= col < self.board_size):
return self._serialize_state(state, message="坐标超出棋盘范围。")
if state.board[row, col] != 0:
return self._serialize_state(state, message="该位置已经有棋子。")
state.board[row, col] = -1
state.move_count += 1
if self._check_terminal(state.board, row, col, -1):
state.winner = "human"
state.human_turn = False
return self._serialize_state(state)
if not np.any(state.board == 0):
state.winner = "draw"
state.human_turn = False
return self._serialize_state(state)
state.human_turn = False
self._ai_move(state)
return self._serialize_state(state)
def _ai_move(self, state: GameState) -> None:
if state.winner is not None:
return
mask = state.board.reshape(-1) == 0
if not np.any(mask):
state.winner = "draw"
state.human_turn = False
return
action = self._select_ai_action(state.board, mask)
row, col = action_to_coord(int(action), self.board_size)
state.board[row, col] = 1
state.last_ai_move = (row, col)
state.move_count += 1
if self._check_terminal(state.board, row, col, 1):
state.winner = "ai"
state.human_turn = False
return
if not np.any(state.board == 0):
state.winner = "draw"
state.human_turn = False
return
state.human_turn = True
def _select_ai_action(self, board: np.ndarray, mask: np.ndarray) -> int:
if self.opponent_name == "model":
if self.model is None:
raise RuntimeError("Model opponent selected but no model was loaded.")
return predict_masked_action(self.model, board, player=1, deterministic=True)
if self.rule_opponent is None:
raise RuntimeError(f"Unsupported opponent type: {self.opponent_name}")
return int(self.rule_opponent.choose_action(board.copy(), player=1, win_length=self.win_length))
def _check_terminal(self, board: np.ndarray, row: int, col: int, player: int) -> bool:
return is_winning_move(board, row, col, player, self.win_length)
def _serialize_state(self, state: GameState, message: str | None = None) -> dict[str, Any]:
payload = {
"board": state.board.tolist(),
"board_size": self.board_size,
"human_turn": state.human_turn,
"winner": state.winner,
"last_ai_move": list(state.last_ai_move) if state.last_ai_move else None,
"move_count": state.move_count,
"opponent": self.opponent_name,
}
if message:
payload["message"] = message
return payload
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Run a local web UI for Gomoku against a trained model or rule-based opponent.")
parser.add_argument("--opponent", choices=["model", "hybrid", "heuristic", "random"], default="hybrid")
parser.add_argument("--model-path", type=Path)
parser.add_argument("--board-size", type=int, default=9)
parser.add_argument("--win-length", type=int, default=5)
parser.add_argument("--human-first", action="store_true")
parser.add_argument("--host", default="127.0.0.1")
parser.add_argument("--port", type=int, default=8000)
parser.add_argument("--no-browser", action="store_true")
parser.add_argument("--seed", type=int, default=42)
parser.add_argument("--search-depth", type=int, default=4)
parser.add_argument("--candidate-radius", type=int, default=2)
parser.add_argument("--max-candidates", type=int, default=10)
parser.add_argument("--policy-weight", type=float, default=220_000.0)
args = parser.parse_args()
if args.opponent in {"model", "hybrid"} and args.model_path is None:
parser.error("--model-path is required when --opponent model or hybrid is used.")
return args
def make_handler(app: GomokuWebApp):
class Handler(BaseHTTPRequestHandler):
def do_GET(self) -> None:
if self.path == "/":
self._send_html(HTML_PAGE)
return
if self.path == "/state":
self._send_json(app.get_state())
return
self.send_error(HTTPStatus.NOT_FOUND, "Not found")
def do_POST(self) -> None:
if self.path == "/reset":
self._send_json(app.reset())
return
if self.path == "/move":
payload = self._read_json()
row = int(payload.get("row", -1))
col = int(payload.get("col", -1))
self._send_json(app.make_human_move(row, col))
return
self.send_error(HTTPStatus.NOT_FOUND, "Not found")
def log_message(self, format: str, *args: Any) -> None:
return None
def _read_json(self) -> dict[str, Any]:
content_length = int(self.headers.get("Content-Length", "0"))
raw = self.rfile.read(content_length) if content_length else b"{}"
return json.loads(raw.decode("utf-8"))
def _send_html(self, html: str) -> None:
encoded = html.encode("utf-8")
self.send_response(HTTPStatus.OK)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(encoded)))
self.end_headers()
self.wfile.write(encoded)
def _send_json(self, payload: dict[str, Any]) -> None:
encoded = json.dumps(payload, ensure_ascii=False).encode("utf-8")
self.send_response(HTTPStatus.OK)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(encoded)))
self.end_headers()
self.wfile.write(encoded)
return Handler
def main() -> None:
args = parse_args()
app = GomokuWebApp(
board_size=args.board_size,
win_length=args.win_length,
human_first=args.human_first,
opponent_name=args.opponent,
model_path=args.model_path,
seed=args.seed,
search_depth=args.search_depth,
candidate_radius=args.candidate_radius,
max_candidates=args.max_candidates,
policy_weight=args.policy_weight,
)
server = ThreadingHTTPServer((args.host, args.port), make_handler(app))
url = f"http://{args.host}:{args.port}"
print(f"Web UI started at {url}")
print("Press Ctrl+C to stop the server.")
if not args.no_browser:
webbrowser.open(url)
try:
server.serve_forever()
except KeyboardInterrupt:
pass
finally:
server.server_close()
if __name__ == "__main__":
main()