Add back-end service to webclient
parent
8f030efa9a
commit
bf8e33ee1c
|
|
@ -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,
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -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())
|
||||
|
|
@ -0,0 +1,5 @@
|
|||
"""Public server entry points."""
|
||||
|
||||
from .app import WebClient
|
||||
|
||||
__all__ = ["WebClient"]
|
||||
|
|
@ -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)
|
||||
Loading…
Reference in New Issue