248 lines
7.7 KiB
Python
248 lines
7.7 KiB
Python
"""
|
|
Flask server that enables a user to run Bot-Man-Toe games as a client from
|
|
a web browser interface.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
import uuid
|
|
from typing import Any, Optional
|
|
from urllib.parse import urlparse, urlunparse
|
|
from dataclasses import dataclass
|
|
|
|
import requests
|
|
from flask import Flask, jsonify, request
|
|
|
|
from pyclient.games.tic_tac_toe import TicTacToe
|
|
from pyclient import PyClient, ServerAgent
|
|
|
|
@dataclass(frozen=True)
|
|
class Match:
|
|
name : str
|
|
turns : list[tuple[int, Any, Any]]
|
|
winner : int | None
|
|
|
|
def append_turn(self, player : int, action : Any, state : Any) -> "Match":
|
|
"""
|
|
Create a new match value that contains a new turn.
|
|
|
|
:param player: The player taking the action
|
|
:type player: int
|
|
:param action: The action taken by the player
|
|
:type action: Any
|
|
:param state: The new game state based on the action
|
|
:type state: Any
|
|
:return: A new match type
|
|
:rtype: Match
|
|
"""
|
|
return Match(
|
|
name=self.name,
|
|
turns=self.turns + [( player, action, state )],
|
|
winner=state.winner(),
|
|
)
|
|
|
|
@classmethod
|
|
def new(cls, name : str) -> "Match":
|
|
"""
|
|
Initialize a new match.
|
|
|
|
:param name: The name of the game being played.
|
|
:type name: str
|
|
:return: Initialized match
|
|
:rtype: Match
|
|
"""
|
|
return cls(name=name, turns=[], winner=None)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
"""
|
|
Create a dictionary representation of the match.
|
|
"""
|
|
return dict(
|
|
name=self.name,
|
|
turns=[
|
|
dict(player=player, action=action, state=state.to_dict())
|
|
for player, action, state in self.turns
|
|
],
|
|
winner=self.winner,
|
|
)
|
|
|
|
class WebClient:
|
|
"""
|
|
A small Flask application aimed at running Bot-Man-Toe games.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
debug : bool = False,
|
|
import_name: str = __name__,
|
|
) -> None:
|
|
"""
|
|
Create a WebClient.
|
|
|
|
:param debug: Debug mode
|
|
:type debug: bool
|
|
:param import_name: Flask import name
|
|
:type import_name: str
|
|
"""
|
|
self.app = Flask(import_name)
|
|
self.debug = debug
|
|
self.__lock = threading.RLock()
|
|
self.__matches : dict[str, Match] = {}
|
|
self.__games : dict[str, Any] = {}
|
|
|
|
@self.app.route("/game-details", methods=["GET"])
|
|
def game_details():
|
|
payload = request.get_json(silent=True) or {}
|
|
game_id : str | None = payload.get("game_id")
|
|
|
|
if not isinstance(game_id, str):
|
|
return jsonify({
|
|
"error": "Expected field `game` as string"
|
|
}), 400
|
|
|
|
match : Match | None = self.__find_match(match_id=game_id)
|
|
|
|
if not isinstance(match, Match):
|
|
return jsonify({
|
|
"error": f"Could not find match with id `{game_id}`",
|
|
}), 400
|
|
|
|
return jsonify(match.to_dict())
|
|
|
|
@self.app.route("/profile", methods=["GET"])
|
|
def profile():
|
|
payload = request.get_json(silent=True) or {}
|
|
url = payload.get("url")
|
|
|
|
if not isinstance(url, str):
|
|
return jsonify({"error": "A string 'url' is required."}), 400
|
|
|
|
try:
|
|
agent = ServerAgent.from_url(url, debug=self.debug)
|
|
except (requests.HTTPError, requests.RequestException, ValueError) as exc:
|
|
return jsonify({"error": str(exc)}), 400
|
|
|
|
return jsonify(agent.to_dict())
|
|
|
|
@self.app.route("/start-game", methods=["GET"])
|
|
def start_game():
|
|
payload = request.get_json(silent=True) or {}
|
|
|
|
game : str | None = payload.get("game", None)
|
|
ps : list | None = payload.get("players", None)
|
|
|
|
if not isinstance(game, str):
|
|
return jsonify({
|
|
"error": "Field `game` must be a string",
|
|
}), 400
|
|
if not isinstance(ps, list):
|
|
return jsonify({
|
|
"error": "Field `players` must be a list of URLs"
|
|
}), 400
|
|
|
|
players : list[str] = [ str(player) for player in ps ]
|
|
game_cls : Any | None = self.__games.get(game, None)
|
|
|
|
if game_cls is None:
|
|
return jsonify({
|
|
"error": "Unknown game type"
|
|
}), 400
|
|
|
|
match_id = self.__register_match(game)
|
|
|
|
runner = threading.Thread(
|
|
target=self.__run_match,
|
|
args=(match_id, game, players),
|
|
daemon=True,
|
|
)
|
|
runner.start()
|
|
|
|
return match_id
|
|
|
|
def __find_match(self, match_id : str) -> Match | None:
|
|
"""
|
|
Find a match. Uses a threading lock for a safe read.
|
|
"""
|
|
with self.__lock:
|
|
return self.__matches.get(match_id, None)
|
|
|
|
def __register_match(self, name : str) -> str:
|
|
"""
|
|
Create a new match in a given game.
|
|
|
|
:param name: The name of the game to be played.
|
|
:type name: str
|
|
:param game: Game class
|
|
:type game: Any
|
|
:return: Unique match identifier
|
|
:rtype: str
|
|
"""
|
|
match_id = uuid.uuid4().hex
|
|
|
|
with self.__lock:
|
|
# Ensure uuid uniqueness
|
|
while match_id in self.__matches:
|
|
print("WARNING: Found duplicate uuid `{match_id}` in existing matches")
|
|
match_id = uuid.uuid4().hex
|
|
|
|
self.__matches[match_id] = Match.new(name)
|
|
|
|
return match_id
|
|
|
|
def __run_match(self, match_id : str, game : str, players : list[str]) -> None:
|
|
"""
|
|
Run a match. This function is usually run in a separate thread.
|
|
|
|
:param match_id: The match to process
|
|
:type match_id: str
|
|
:param game: The name of the game type to play
|
|
:type game: str
|
|
:param players: List of URLs to use for processing
|
|
:type players: list[str]
|
|
:raises KeyError: The game is not recognized or the match isn't found
|
|
:raises ValueError: None of the players were accessible
|
|
"""
|
|
game_cls = self.__games[game]
|
|
match = self.__find_match(match_id=match_id)
|
|
|
|
if match is None:
|
|
raise KeyError(
|
|
f"Could not find match with id `{match_id}`"
|
|
)
|
|
|
|
c = PyClient(hosts=players, debug=self.debug)
|
|
|
|
for player, action, state in c.gen_game(name=game, game=game_cls, urls=players, move_default_if_nonexistent=True):
|
|
match = match.append_turn(player=player, action=action, state=state)
|
|
|
|
with self.__lock:
|
|
self.__matches[match_id] = match
|
|
|
|
def register_game(self, game_name: str, game_type: type[Any]) -> None:
|
|
"""
|
|
Register a supported game.
|
|
|
|
:param game_name: The string name of the game
|
|
:type game_name: str
|
|
:param game_type: The game's object
|
|
:type game_type: Any
|
|
:raises ValueError:
|
|
"""
|
|
name = game_name.strip().strip("/")
|
|
|
|
if name == "":
|
|
raise ValueError("Game name cannot be empty.")
|
|
|
|
self.__games[name] = game_type
|
|
|
|
def start(
|
|
self,
|
|
host: str = "127.0.0.1",
|
|
port: int = 5000,
|
|
debug: bool = False,
|
|
**kwargs: Any,
|
|
) -> None:
|
|
"""Start the Flask development server."""
|
|
self.app.run(host=host, port=port, debug=debug, **kwargs)
|