From bf8e33ee1cfe9a83b17cc3bf09cbe769da7db3f4 Mon Sep 17 00:00:00 2001 From: Bram van den Heuvel Date: Thu, 18 Jun 2026 23:20:19 +0200 Subject: [PATCH] Add back-end service to webclient --- pyclient/poll.py | 35 +++++- web.py | 16 +++ webclient/__init__.py | 5 + webclient/app.py | 247 ++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 301 insertions(+), 2 deletions(-) create mode 100644 web.py create mode 100644 webclient/__init__.py create mode 100644 webclient/app.py diff --git a/pyclient/poll.py b/pyclient/poll.py index 98bfef3..415d7ef 100644 --- a/pyclient/poll.py +++ b/pyclient/poll.py @@ -8,7 +8,13 @@ import time class ServerAgent: """Representation of a server that can host one or more games.""" - def __init__(self, url: str, name: str, games: Dict[str, Dict[str, Any]], debug : bool = False) -> None: + def __init__(self, + url: str, + name: str, + games: Dict[str, Dict[str, Any]], + debug : bool = False, + profile : dict[str, Any] = {} + ) -> None: """ Create a server representation. @@ -18,10 +24,15 @@ class ServerAgent: :type name: str :param games: Games the server is willing to play. :type games: Dict[str, Dict[str, Any]] + :param debug: Whether to enable debug mode. + :type debug: bool + :param profile: Custom user profile containing a user's details. + :type profile: dict[str, Any] """ self.debug = debug self.games = games self.name = name + self.profile = profile self.url = url.strip("/") @classmethod @@ -69,8 +80,13 @@ class ServerAgent: for game_name, profile in raw_games.items(): if isinstance(profile, dict): games[str(game_name)] = profile + + profile: dict[str, Any] = {} + for k, v in content.items(): + if k not in [ "name", "games" ]: + profile[k] = v - return cls(url=url, name=name, games=games, debug=debug) + return cls(url=url, name=name, games=games, debug=debug, profile=profile) def poll(self, game: str, payload: Dict[str, Any], timeout: float = 1.0) -> Optional[Dict[str, Any]]: """ @@ -100,3 +116,18 @@ class ServerAgent: print(content) return content if isinstance(content, dict) else None + + def to_dict(self) -> dict[str, Any]: + """ + Represent the agent in the form of a dict. + + :return: Dictionary representation of the ServerAgent + :rtype: dict[str, Any] + """ + return dict( + name=self.name, + games=self.games, + url=self.url, + profile=self.profile, + ) + diff --git a/web.py b/web.py new file mode 100644 index 0000000..3512ac1 --- /dev/null +++ b/web.py @@ -0,0 +1,16 @@ +"""Entry point for the web client Flask server.""" + +from webclient import WebClient + + +web_client = WebClient(import_name=__name__) +app = web_client.app + + +def main() -> int: + web_client.start() + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/webclient/__init__.py b/webclient/__init__.py new file mode 100644 index 0000000..9f4fcc2 --- /dev/null +++ b/webclient/__init__.py @@ -0,0 +1,5 @@ +"""Public server entry points.""" + +from .app import WebClient + +__all__ = ["WebClient"] diff --git a/webclient/app.py b/webclient/app.py new file mode 100644 index 0000000..1193e62 --- /dev/null +++ b/webclient/app.py @@ -0,0 +1,247 @@ +""" + 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)