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