From eb5a869cf42ec1d64a112d49df3df2deff309a0f Mon Sep 17 00:00:00 2001 From: Bram van den Heuvel Date: Fri, 19 Jun 2026 14:34:23 +0200 Subject: [PATCH 1/4] Containerize PyServer --- .dockerignore | 179 ++++++++++++++++++++++++++++++++++++++ .gitignore | 4 + pyserver/Containerfile | 25 ++++++ requirements-pyserver.txt | 8 ++ server.py | 99 +++++++++++++++++++++ 5 files changed, 315 insertions(+) create mode 100644 .dockerignore create mode 100644 pyserver/Containerfile create mode 100644 requirements-pyserver.txt create mode 100644 server.py diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..e77097f --- /dev/null +++ b/.dockerignore @@ -0,0 +1,179 @@ +# Ignore Markdown documentation +spec/ +*.md + +# ---------------------------- +# ---> GITIGNORE CONFIGURATION +# ---------------------------- + +# Repository-specific virtual environments +.venv-pyserver +.venv-pyclient + +# ---> Elm +# elm-package generated files +elm-stuff +# elm-repl generated files +repl-temp-* + +# ---> Python +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +share/python-wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# PyInstaller +# Usually these files are written by a python script from a template +# before PyInstaller builds the exe, so as to inject date/other infos into it. +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.nox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +*.py,cover +.hypothesis/ +.pytest_cache/ +cover/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py +db.sqlite3 +db.sqlite3-journal + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +.pybuilder/ +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# IPython +profile_default/ +ipython_config.py + +# pyenv +# For a library or package, you might want to ignore these files since the code is +# intended to run in multiple environments; otherwise, check them in: +# .python-version + +# pipenv +# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control. +# However, in case of collaboration, if having platform-specific dependencies or dependencies +# having no cross-platform support, pipenv may install dependencies that don't work, or not +# install all needed dependencies. +#Pipfile.lock + +# poetry +# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control. +# This is especially recommended for binary packages to ensure reproducibility, and is more +# commonly ignored for libraries. +# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control +#poetry.lock + +# pdm +# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control. +#pdm.lock +# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it +# in version control. +# https://pdm.fming.dev/#use-with-ide +.pdm.toml + +# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm +__pypackages__/ + +# Celery stuff +celerybeat-schedule +celerybeat.pid + +# SageMath parsed files +*.sage.py + +# Environments +.env +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# Spyder project settings +.spyderproject +.spyproject + +# Rope project settings +.ropeproject + +# mkdocs documentation +/site + +# mypy +.mypy_cache/ +.dmypy.json +dmypy.json + +# Pyre type checker +.pyre/ + +# pytype static type analyzer +.pytype/ + +# Cython debug symbols +cython_debug/ + +# PyCharm +# JetBrains specific template is maintained in a separate JetBrains.gitignore that can +# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore +# and can be added to the global gitignore or merged into this file. For a more nuclear +# option (not recommended) you can uncomment the following to ignore the entire idea folder. +#.idea/ \ No newline at end of file diff --git a/.gitignore b/.gitignore index a9936b6..0b72478 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,7 @@ +# Repository-specific virtual environments +.venv-pyserver +.venv-pyclient + # ---> Elm # elm-package generated files elm-stuff diff --git a/pyserver/Containerfile b/pyserver/Containerfile new file mode 100644 index 0000000..c9d100d --- /dev/null +++ b/pyserver/Containerfile @@ -0,0 +1,25 @@ +FROM python:3.10-alpine AS builder +WORKDIR /app + +# Install build dependencies +RUN apk add --no-cache gcc musl-dev python3-dev +COPY requirements-pyserver.txt . + +# Create wheels for faster installation +RUN pip wheel --no-cache-dir --wheel-dir /wheels -r requirements-pyserver.txt + + +FROM python:3.10-alpine +WORKDIR /app + +# Install from pre-built wheels +COPY --from=builder /wheels /wheels +COPY requirements-pyserver.txt . +RUN pip install --no-cache-dir --no-index --find-links=/wheels \ + -r requirements-pyserver.txt && rm -rf /wheels + +# Install PyServer code +COPY pyserver/ pyserver/ +COPY server.py . + +CMD ["python", "server.py"] diff --git a/requirements-pyserver.txt b/requirements-pyserver.txt new file mode 100644 index 0000000..c2d75db --- /dev/null +++ b/requirements-pyserver.txt @@ -0,0 +1,8 @@ +blinker==1.9.0 +click==8.4.1 +colorama==0.4.6 +Flask==3.1.3 +itsdangerous==2.2.0 +Jinja2==3.1.6 +MarkupSafe==3.0.3 +Werkzeug==3.1.8 diff --git a/server.py b/server.py new file mode 100644 index 0000000..bf95a28 --- /dev/null +++ b/server.py @@ -0,0 +1,99 @@ +""" + This module enables a user to host a server that is able to play games. +""" + +from __future__ import annotations + +import random + +from pyserver import PyServer +from typing import Any + +def main() -> int: + """ + Start a server. + + :return: Exit code + :rtype: int + """ + player = PyServer( + # Customize this to whatever you'd like to call your player + name="My super smart robot player", + + # Custom information that you can use to tell people about this player + profile={}, + + # Unless you know what you're doing, don't touch this. + import_name=__name__, + ) + + # Register games! Comment out any you don't want your player to play. + player.add_tic_tac_toe(on_move=play_tic_tac_toe, profile={}) + + # Start listening for games + player.start( + host="0.0.0.0", # Comment out when using only locally + port=5000, + ) + + return 0 + +def play_tic_tac_toe(payload : dict[str, Any]) -> dict[str, Any]: + """ + Play a game of tic-tac-toe. + + You receive a payload that looks like this: + + { + "1": "X", "2": "", "3": "O", + "4": "X", "5": "O", "6": "", + "7": "", "8": "", "9": "", + "your_token": "X" + } + + And you're expected to return a response of which field you'd like to + place your piece in. For example, if you wish to place your token in + field 7, your response should look like this: + + { "move": 7 } + + The board is arranged as follows: + + 1 | 2 | 3 + ---+---+--- + 4 | 5 | 6 + ---+---+--- + 7 | 8 | 9 + + :param payload: The incoming JSON that contains the game state. + :type payload: dict[str, Any] + :return: The move you wish to take. + :rtype: dict[str, Any] + """ + + # Try printing the payload to see what it looks like! + print(payload) + + options = [ 1, 2, 3, 4, 5, 6, 7, 8, 9, ] + + # 1. Try filtering out the impossible moves! + # If an X or O was already placed at a field, remove it from the options + + # + + # 2. Try finding two in a row! If possible, you can try to place the third + # item on the board and get 3 in a row. + + # + + # 3. Perhaps you can block the opponent from getting 3 in a row? + + # + + # Now, pick any of the remaining options. + # This is just a simple implementation. Naturally, you're welcome to try + # your own logic. + return { "move": random.choice(options) } + +if __name__ == "__main__": + raise SystemExit(main()) From 78c461d9b2f6fe510e75e657dd08ec5fac4bdcb2 Mon Sep 17 00:00:00 2001 From: Bram van den Heuvel Date: Fri, 19 Jun 2026 15:46:03 +0200 Subject: [PATCH 2/4] Create client script with diagnostics --- client.py | 87 ++++++++++++++++++++++++++++++++++++++ main.py | 89 --------------------------------------- requirements-pyclient.txt | 5 +++ requirements.txt | 2 +- 4 files changed, 93 insertions(+), 90 deletions(-) create mode 100644 client.py delete mode 100644 main.py create mode 100644 requirements-pyclient.txt diff --git a/client.py b/client.py new file mode 100644 index 0000000..45a968d --- /dev/null +++ b/client.py @@ -0,0 +1,87 @@ +""" + This module offers a small place to start building game clients from. + + You can use this for: + + - Learning how the system works. + - Debugging a game server. + - Debugging a game that behaves weirdly. + + This module lets you host a game, let online servers participate in it, + and then analyze the outcome. + + + You can build several things with this: + + 1. You can build an AI trainer that trains on existing players. + 2. You can build an ELO evaluator that compares the performance of various + strategies. +""" + +from __future__ import annotations + +import json + +from pyclient import PyClient +from pyclient.games.tic_tac_toe import TicTacToe +from typing import Any + +def main() -> int: + """ + Start a client, then use it to analyze one or more matches. + + :return: Exit code + :rtype: int + """ + c = PyClient([ + "https://bmt001.noordstar.me", + "https://bmt001.noordstar.me", + ], debug=False) + + out = c.play_game("tic-tac-toe", TicTacToe) + + inspect_game(out) + + return 0 + +def inspect_game(game : list[tuple[int, dict[str, Any], Any]]) -> None: + """ + Print a diagnostic of a played game to the terminal. + + :param game: The results of a played game. + :type game: list[typle[int, dict[str, Any], Any]] + """ + max_width = 40 + hbar = max_width * '%' + + def title(s : str) -> None: + """ + Show a title nice and clean in the middle + + :param s: The title to display + :type s: str + """ + w = (max_width - len(s)) // 2 + + print(hbar) + print((w * ' ') + s.upper() + (w * ' ')) + print(hbar) + + # Show all moves made throughout the game + title("Turns taken") + for player, action, _ in game: + print(f"Player {player} : " + json.dumps(action)[:max_width-12]) + + # Show all remaining variables in the finishing state of the game + title("Final state") + final_state = game[-1][2] + for k, v in final_state.to_dict().items(): + print(f"{k} => {json.dumps(v)}") + + # Some final (usually the most relevant) statistics + title("Result") + print(f"Total turns taken: {len(game)}") + print("Winner: " + ("0 (tie)" if final_state.winner() == 0 else f"player {final_state.winner()}")) + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/main.py b/main.py deleted file mode 100644 index b550462..0000000 --- a/main.py +++ /dev/null @@ -1,89 +0,0 @@ -""" - This module enables a user to host a server that is able to play games. -""" - -import random - -from pyserver import PyServer -from pyclient import PyClient -from pyclient.games.tic_tac_toe import TicTacToe - -def main(): - player = PyServer( - # Customize this to whatever you'd like to call your player - name="My super smart robot player", - - # Custom information that you can use to tell people about this player - profile={}, - - # Unless you know what you're doing, don't touch this. - import_name=__name__, - ) - - player.add_tic_tac_toe(on_move=play_tic_tac_toe, profile={}) - - player.start(port=5001) - - return 0 - -def play_tic_tac_toe(payload): - """ - Play a game of tic-tac-toe. - - You receive a payload that looks like this: - - { - "1": "X", "2": "", "3": "O", - "4": "X", "5": "O", "6": "", - "7": "", "8": "", "9": "", - "your_token": "X" - } - - And you're expected to return a response of which field you'd like to - place your piece in. For example, if you wish to place your token in - field 7, your response should look like this: - - { "move": 7 } - - The board is arranged as follows: - - 1 | 2 | 3 - ---+---+--- - 4 | 5 | 6 - ---+---+--- - 7 | 8 | 9 - """ - - # Try printing the payload to see what it looks like! - print(payload) - - options = [ 1, 2, 3, 4, 5, 6, 7, 8, 9, ] - - # 1. Try filtering out the impossible moves! - # If an X or O was already placed at a field, remove it from the options - - # - - # 2. Try finding two in a row! If possible, you can try to place the third - # item on the board and get 3 in a row. - - # - - # 3. Perhaps you can block the opponent from getting 3 in a row? - - # - - # Now, pick any of the remaining options. - # This is just a simple implementation. Naturally, you're welcome to try - # your own logic. - return { "move": random.choice(options) } - -if __name__ == "__main__": - raise SystemExit(main()) - - c = PyClient([ - "http://127.0.0.1:5001", - "http://127.0.0.1:5002", - ], debug=True) - - out = c.play_game("tic-tac-toe", TicTacToe) diff --git a/requirements-pyclient.txt b/requirements-pyclient.txt new file mode 100644 index 0000000..58746c1 --- /dev/null +++ b/requirements-pyclient.txt @@ -0,0 +1,5 @@ +certifi==2026.6.17 +charset-normalizer==3.4.7 +idna==3.18 +requests==2.34.2 +urllib3==2.7.0 diff --git a/requirements.txt b/requirements.txt index ceebf86..3c1e47d 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,5 @@ blinker==1.9.0 -certifi==2026.5.20 +certifi==2026.6.17 charset-normalizer==3.4.7 click==8.4.1 colorama==0.4.6 From 04e21632ea75f4a7d6563dc797f15cb576ff4750 Mon Sep 17 00:00:00 2001 From: Bram van den Heuvel Date: Sat, 20 Jun 2026 14:53:51 +0200 Subject: [PATCH 3/4] Refactor PyClient module --- client.py | 32 ++++--- pyclient/__init__.py | 123 +++------------------------ pyclient/agent.py | 156 ++++++++++++++++++++++++++++++++++ pyclient/client.py | 71 ++++++++++++++++ pyclient/games/__init__.py | 12 +++ pyclient/games/game.py | 138 ++++++++++++++++++++++++++++++ pyclient/games/tic_tac_toe.py | 31 +++++-- pyclient/poll.py | 102 ---------------------- pyclient/replay.py | 92 ++++++++++++++++++++ pyclient/transition.py | 110 ++++++++++++++++++++++++ 10 files changed, 634 insertions(+), 233 deletions(-) create mode 100644 pyclient/agent.py create mode 100644 pyclient/client.py create mode 100644 pyclient/games/__init__.py create mode 100644 pyclient/games/game.py delete mode 100644 pyclient/poll.py create mode 100644 pyclient/replay.py create mode 100644 pyclient/transition.py diff --git a/client.py b/client.py index 45a968d..e5d131e 100644 --- a/client.py +++ b/client.py @@ -21,8 +21,9 @@ from __future__ import annotations import json +import pyclient -from pyclient import PyClient +from pyclient import Agent, PyClient from pyclient.games.tic_tac_toe import TicTacToe from typing import Any @@ -33,23 +34,28 @@ def main() -> int: :return: Exit code :rtype: int """ - c = PyClient([ - "https://bmt001.noordstar.me", - "https://bmt001.noordstar.me", - ], debug=False) + c = PyClient(debug=False) - out = c.play_game("tic-tac-toe", TicTacToe) + players : list[Agent] = [ + Agent.from_url(url="https://bmt001.noordstar.me/"), + Agent.from_url(url="https://bmt001.noordstar.me/"), + ] + + out = c.play_game( + players=players, + start=TicTacToe.empty(), + ) inspect_game(out) return 0 -def inspect_game(game : list[tuple[int, dict[str, Any], Any]]) -> None: +def inspect_game(game : pyclient.GameReplay) -> None: """ Print a diagnostic of a played game to the terminal. :param game: The results of a played game. - :type game: list[typle[int, dict[str, Any], Any]] + :type game: pyclient.GameReplay """ max_width = 40 hbar = max_width * '%' @@ -69,19 +75,19 @@ def inspect_game(game : list[tuple[int, dict[str, Any], Any]]) -> None: # Show all moves made throughout the game title("Turns taken") - for player, action, _ in game: - print(f"Player {player} : " + json.dumps(action)[:max_width-12]) + for turn in game.turns: + print(f"Player {turn.player} : " + json.dumps(turn.action)[:max_width-12]) # Show all remaining variables in the finishing state of the game title("Final state") - final_state = game[-1][2] + final_state = game.turns[-1].state for k, v in final_state.to_dict().items(): print(f"{k} => {json.dumps(v)}") # Some final (usually the most relevant) statistics title("Result") - print(f"Total turns taken: {len(game)}") - print("Winner: " + ("0 (tie)" if final_state.winner() == 0 else f"player {final_state.winner()}")) + print(f"Total turns taken: {len(game.turns)}") + print(f"Result: {final_state.winner()}") if __name__ == "__main__": raise SystemExit(main()) diff --git a/pyclient/__init__.py b/pyclient/__init__.py index 6d92dd9..d63fda3 100644 --- a/pyclient/__init__.py +++ b/pyclient/__init__.py @@ -1,111 +1,16 @@ -"""Public client entry points.""" +""" + Entry points for developers who wish to use the PyClient module. +""" -from typing import Any, Generator, List, Optional +from .agent import Agent, ServerAgent +from .client import PyClient +from .replay import GameReplay +from .transition import Transition -import requests - -from .poll import ServerAgent - - -class PyClient: - """Host games between discovered server agents.""" - - def __init__(self, hosts: List[str], debug : bool = False) -> None: - """ - Create a PyClient. - - :param hosts: URLs of servers that can participate. - :type hosts: List[str] - :raises ValueError: If no reachable servers are provided. - """ - self.agents: List[ServerAgent] = [] - self.debug = debug - self.hosts: List[str] = [] - - for host in hosts: - agent = self.__discover_host(host) - if agent is not None: - self.agents.append(agent) - self.hosts.append(agent.url) - - if len(self.hosts) <= 0: - raise ValueError( - "No valid hosts found! Check your internet connection or verify the URLs are correct." - ) - - def __discover_host(self, url: str) -> Optional[ServerAgent]: - try: - return ServerAgent.from_url(url, debug=self.debug) - except (requests.exceptions.RequestException, ValueError): - return None - - def gen_game(self, name : str, game : Any, urls : list[str], timeout : float = 1.0, move_default_if_nonexistent : bool = False) -> Generator[tuple[int, dict[str, Any], Any], None, None]: - """ - Play game. Return the results as they've been calculated. - - :param name: Unique identifier string name for the game - :type name: str - :param game: Game to be played - :type game: Any - :param urls: List of URLs where players are located - :type urls: list[str] - :param timeout: Maximum time in seconds to wait for a player to respond - :type timeout: float - :param move_default_if_nonexistent: If a required player doesn't exist, raise an error. When this is set to true, however, the program will instead assume a player that always takes the default option. - :type move_default_if_nonexistent: bool - :return: Generator of each turn consisting of a player, what action it took and what that resulted in. - :rtype: Generator[tuple[int, dict[str, Any], Any], None, None] - :raises KeyError: The number of URLs provided is too low, the game requires more players. - """ - agents = [ - ServerAgent(url=url, name="", games={}, debug=self.debug) - for url in urls - ] - - current_state = game.empty() - - while current_state.winner() is None: - player : int = current_state.player_to_move() - - if len(agents) < player: - if not move_default_if_nonexistent: - raise KeyError( - f"Game requires at least {player} players to exist, found only {len(agents)}" - ) - - current_state = current_state.move_default() - - yield player, {}, current_state - - else: - agent = agents[player-1] - payload = agent.poll( - game=(name + "/" + current_state.action_name()).strip("/"), - payload=current_state.as_seen_by(player), - timeout=timeout, - ) - - # Calculate move - current_state = current_state.move(payload) - - yield player, payload or {}, current_state - - def play_game(self, name : str, game, timeout : float = 1.0) -> list[tuple[int, dict[str, Any], Any]]: - """ - Play a given game. Ask the registered agents to participate. - """ - urls = [ agent.url for agent in self.agents if name in agent.games ] - - return list(self.gen_game( - name=name, game=game, urls=urls, timeout=timeout, - move_default_if_nonexistent=False, - )) - - def verify_host(self, url: str) -> bool: - """ - Verify whether the URL seems to contain a link to a playable server. - """ - return self.__discover_host(url) is not None - - -__all__ = ["PyClient", "ServerAgent"] +__all__ = [ + "Agent", + "GameReplay", + "PyClient", + "ServerAgent", + "Transition", +] diff --git a/pyclient/agent.py b/pyclient/agent.py new file mode 100644 index 0000000..dd8f318 --- /dev/null +++ b/pyclient/agent.py @@ -0,0 +1,156 @@ +""" + This module hosts various agents that can participate in games. + + Examples of possible agents could be: + + - An online server + - A locally running neural network + - A hacker who participates from the terminal + - A user interface that allows users to play against their own creations +""" + +from dataclasses import dataclass +from typing import Any, Dict, Optional, Tuple, Union + +import requests +import time + +@dataclass(frozen=True) +class Agent: + """ + Base class of a game participant. Mostly used to inherit from. + """ + + debug : bool + games : dict[str, dict[str, Any]] + profile : dict[str, Any] + + @classmethod + def from_url(cls, url : str, **kwargs) -> "ServerAgent": + """ + Create an agent based on a URL. + + :param url: The URL where the agent can be accessed. + :type url: str + :return: An agent that contacts a server when polled. + :rtype: ServerAgent + """ + return ServerAgent.from_server_url(url=url, **kwargs) + + @property + def name(self) -> str: + """ + The name by which the agent calls itself. + """ + return str(self.profile.get("name", "Nameless agent")) + + def poll( + self, + game : str, + payload : dict[str, Any], + **kwargs + ) -> Optional[dict[str, Any]]: + """ + Ask the agent to make a move. + + :param game: The game the ServerAgent is asked to play. + :type game: str + :param payload: The JSON payload that represents the game's state. + :type payload: dict[str, Any] + :return: The agent's response, or None if the agent doesn't respond. + :rtype: Optional[dict[str, Any]] + """ + print(f"WARNING: The `poll` method is not defined on the {self.__class__.__name__} class") + return None + +@dataclass(frozen=True) +class ServerAgent(Agent): + """ + Agent that reaches out to the internet to poll for moves. + """ + + url : str + + @classmethod + def from_server_url( + cls, + url: str, + timeout: float | tuple[float, float] | None = 10.0, + debug : bool = False + ) -> "ServerAgent": + """ + Create a server agent by polling its discovery endpoint. + + :param url: The URL that the server can be reached at. + :type url: str + :param timeout: Request timeout. + :type timeout: float | tuple[float, float] | None + :param debug: Enables debug mode + :type debug: bool + :return: The server's representation as an agent. + :rtype: ServerAgent + :raises requests.exceptions.HTTPError: If the server returns a + non-success HTTP status code. + :raises requests.exceptions.RequestException: If the request fails + before a response is received. + :raises ValueError: If the response body is not a JSON object or if the + payload contains malformed discovery fields. + """ + response = requests.get(url.rstrip("/") + "/", timeout=timeout) + response.raise_for_status() + content = response.json() + + if not isinstance(content, dict): + raise ValueError("Server discovery responses must be JSON objects.") + + raw_name = content.get("name", "") + name = "" if raw_name is None else str(raw_name) + + games: dict[str, dict[str, Any]] = {} + raw_games = content.get("games", {}) + if raw_games is not None: + if not isinstance(raw_games, dict): + raise ValueError("The 'games' field must be a JSON object when provided.") + + for game_name, profile in raw_games.items(): + if isinstance(profile, dict): + games[str(game_name)] = profile + + return cls( + debug=debug, + games=games, + profile=content, + url=url, + ) + + def poll( + self, + game : str, + payload : dict[str, Any], + **kwargs + ) -> Optional[dict[str, Any]]: + """ + Inquire a game to make a move. + + :param game: The game the ServerAgent is asked to play. + :type game: str + :param payload: The JSON payload that represents the game's state. + :type payload: dict[str, Any] + :return: The server's response, or None if the server did not respond. + :rtype: Optional[dict[str, Any]] + """ + url = f"{self.url.rstrip('/')}/{game.lstrip('/')}" + timeout = float(kwargs.get("timeout", 1.0)) + + try: + response = requests.get(url, json=payload, timeout=timeout) + response.raise_for_status() + content = response.json() + except (requests.exceptions.RequestException, ValueError): + return None + + if self.debug: + print(f"[DBG] Agent `{self.name}` returned:") + print(content) + + return content if isinstance(content, dict) else None diff --git a/pyclient/client.py b/pyclient/client.py new file mode 100644 index 0000000..316c72c --- /dev/null +++ b/pyclient/client.py @@ -0,0 +1,71 @@ +""" + This module hosts the PyClient class. You can use this class to simulate + games among mulltiple agents. +""" + +from __future__ import annotations + +from .agent import Agent +from .games import Game +from .replay import GameReplay, Turn +from dataclasses import dataclass +from typing import Any, Generator + +@dataclass(frozen=True) +class PyClient: + """ + Host games among multiple agents. + """ + + debug : bool + + def gen_game(self, players : list[Agent], start : Game) -> Generator[Turn, None, None]: + """ + Generate a game by polling the players. + + :param players: All players that wish to participate. + :type players: list[Agent] + :param start: The start state of the game. + :type start: Game + :return: A generator that yields turns. + :rtype: Generator[Turn, None, None] + """ + current_state = start + + while current_state.winner() is None: + player = current_state.player_to_move() + + if len(players) < player: + # Player not found! Make a default move. + current_state = current_state.move_default() + + yield Turn(action={}, player=player, state=current_state) + + else: + agent = players[player - 1] + payload = agent.poll( + game=current_state.game_name(), + payload=current_state.as_seen_by(player=player), + ) + + # Calculate move + current_state = current_state.move(payload=payload) + + yield Turn(action=payload, player=player, state=current_state) + + def play_game(self, players : list[Agent], start : Game) -> GameReplay: + """ + Generate a game by polling the players. Collect all moves in a + summary. + + :param players: All players that wish to participate. + :type players: list[Agent] + :param start: The start state of the game. + :type start: Game + :return: Summary describing how the game went. + :rtype: GameReplay + """ + return GameReplay( + start=start, + turns=list(self.gen_game(players=players, start=start)), + ) diff --git a/pyclient/games/__init__.py b/pyclient/games/__init__.py new file mode 100644 index 0000000..bb194f9 --- /dev/null +++ b/pyclient/games/__init__.py @@ -0,0 +1,12 @@ +""" + Entry point for collecting all known games +""" + +from .game import FinishState, Game +from .tic_tac_toe import TicTacToe + +__all__ = [ + "FinishState", + "Game", + "TicTacToe", +] diff --git a/pyclient/games/game.py b/pyclient/games/game.py new file mode 100644 index 0000000..f4eecc8 --- /dev/null +++ b/pyclient/games/game.py @@ -0,0 +1,138 @@ +""" + This module contains the base class for any game. + + If you'd like to create a new game, please copy this module and fill in + all the necessary methods in order to have a functioning game. +""" + +from __future__ import annotations + +from enum import Enum, auto +from typing import Any + +class FinishState(Enum): + draw = auto() + loss = auto() + win = auto() + +class Game: + """ + Base class for all games. + + A game is always a snapshot of a game that is paused because it waits + for a user to make a move. + """ + + def action_name(self) -> str: + """ + Return the type of action to take. Please only use alphanumeric + characters. + + This helps the player understand what they are supposed to do. + For example, in Risk, you can use this to explain whether they are + expected to "recruit" or "attack" or "move" with their troops. + + If all moves require the same action (such as with tic-tac-toe), + you can leave this to return an empty string. + """ + return "" + + def as_seen_by(self, player : int) -> dict[str, Any]: + """ + From the perspective of a given player, return the game's state. + + Note that it is very common in games for players to not have ALL + the details available. For example, in card games, you are + generally unable to see the cards in your opponents' hands. + + The game's state is formatted in a way that makes it easy to + convert it to JSON or a tensor. + + :param player: Player. Counting starts from one. + :type player: int + :return: The game's state from the perspective of the player. + :rtype: dict[str, Any] + """ + return {} + + @classmethod + def empty(cls) -> "Game": + """ + Create a new game. + + Use this method to shuffle the cards, to arrange pieces on the + board, and set up the state in a way where the first player can + start to make its move. + + :return: A game that's ready to start. + :rtype: Game + """ + return cls() + + def game_name(self) -> str: + """ + Return a non-empty string that uniquely identifies the game. + When creating a new unspecified game, please respect the Java + package naming convention. + """ + return "default-game" + + def move_default(self) -> "Game": + """ + Fallback move option for when a user isn't available or cannot + decide. Program this move to be predictable, and resemble an + attempted skip as much as possible. For example:" + + - In Go, place a tile in an empty space near the top-left corner + - In Monopoly, choose not to buy anything + - In poker, check. (Or fold after a raise.) + - In a maze, choose the left-most path + """ + # NOTE: Games are meant to be immutable values! + # As such, you should always consider creating a deep copy of oneself, + # do not simply return self if it might update. + return self + + def move(self, payload : dict[str, Any] | None = None) -> "Game": + """ + Make a move on behalf of whose turn it is. + + :param payload: Dictionary containing the player's response. + :type payload: dict[str, Any] | None + :return: A new instance of the game where a new action is required. + :rtype: Game + """ + if not isinstance(payload, dict): + return self.move_default() + + # NOTE: Games are meant to be immutable values! + # As such, you should always consider creating a deep copy of oneself, + # do not simply return self if it might update. + return self + + def player_to_move(self) -> int: + """ + Return which player is currently meant to move. + + :return: Which player to move. Counting starts from one. + :rtype: int + """ + return 1 + + def to_dict(self) -> dict[str, Any]: + """ + Return a dictionary containing the full game state. + + :return: The current game's state. + :rtype: dict[str, Any] + """ + return {} + + def winner(self) -> dict[int, FinishState] | None: + """ + Determine whether the game has ended. + + :return: A list detailing which players have won, lost or drawn - or None if the game hasn't finished. + :rtype: dict[int, FinishState] | None + """ + return {} diff --git a/pyclient/games/tic_tac_toe.py b/pyclient/games/tic_tac_toe.py index 8d268ad..4e0958e 100644 --- a/pyclient/games/tic_tac_toe.py +++ b/pyclient/games/tic_tac_toe.py @@ -1,6 +1,14 @@ +""" + This module hosts the game of tic-tac-toe, the traditional game where + you're trying to place 3 items in a row in a 3x3-grid. +""" + +from __future__ import annotations + +from .game import FinishState, Game from dataclasses import dataclass from enum import Enum, auto -from typing import Any, Dict, List, Optional +from typing import Any, Optional class Field(Enum): X = auto() @@ -20,7 +28,7 @@ class Field(Enum): return "" @dataclass(frozen=True) -class TicTacToe: +class TicTacToe(Game): field_1 : Field field_2 : Field field_3 : Field @@ -172,7 +180,7 @@ class TicTacToe: def action_name(self) -> str: return "" - def as_seen_by(self, player : int) -> Dict[str, Any]: + def as_seen_by(self, player : int) -> dict[str, Any]: """ Return the view of the game from the perspective of a given player. @@ -202,6 +210,9 @@ class TicTacToe: field_4=Field.empty, field_5=Field.empty, field_6=Field.empty, field_7=Field.empty, field_8=Field.empty, field_9=Field.empty, ) + + def game_name(self) -> str: + return "tic-tac-toe" def move_default(self) -> "TicTacToe": """ @@ -229,7 +240,7 @@ class TicTacToe: "No legal moves exist anymore on this tic-tac-toe board." ) - def move(self, payload : Optional[Dict[str, Any]] = None) -> "TicTacToe": + def move(self, payload : Optional[dict[str, Any]] = None) -> "TicTacToe": """ Have a player make a move. Based on this information, update the game. @@ -271,7 +282,7 @@ class TicTacToe: """ return 1 if self.count_x() <= self.count_o() else 2 - def to_dict(self) -> Dict[str, Any]: + def to_dict(self) -> dict[str, Any]: return { "1": str(self.field_1), "2": str(self.field_2), @@ -284,12 +295,12 @@ class TicTacToe: "9": str(self.field_9), } - def winner(self) -> int | None: + def winner(self) -> dict[int, FinishState] | None: """ Returns whether the board indicates that there's a winner. :return: The winning player, zero in case of a tie, or None if there's no winner yet. - :rtype: int | None + :rtype: dict[int, FinishState] | None """ win_lines = [ [ 1, 2, 3, ], @@ -303,15 +314,17 @@ class TicTacToe: ] d = self.to_dict() + out = { 1 : FinishState.loss, 2 : FinishState.loss, } for player, symbol in [ ( 1, str(Field.X) ), ( 2, str(Field.O) ) ]: for win_line in win_lines: if all(d[str(w)] == symbol for w in win_line): - return player + out[player] = FinishState.win + return out else: # Check for draw if all(item != "" for item in d.values()): - return 0 + return { 1 : FinishState.draw, 2 : FinishState.draw, } return None diff --git a/pyclient/poll.py b/pyclient/poll.py deleted file mode 100644 index 98bfef3..0000000 --- a/pyclient/poll.py +++ /dev/null @@ -1,102 +0,0 @@ -"""Discovery and polling helpers for Bot-Man-Toe servers.""" - -from typing import Any, Dict, Optional, Tuple, Union - -import requests -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: - """ - Create a server representation. - - :param url: The URL used to discover the server. - :type url: str - :param name: Name of the server's player. - :type name: str - :param games: Games the server is willing to play. - :type games: Dict[str, Dict[str, Any]] - """ - self.debug = debug - self.games = games - self.name = name - self.url = url.strip("/") - - @classmethod - def from_url( - cls, - url: str, - timeout: Optional[Union[float, Tuple[float, float]]] = 10.0, - debug : bool = False - ) -> "ServerAgent": - """ - Create a server agent by polling its discovery endpoint. - - The root endpoint is expected to return a JSON object with an optional - ``name`` field and an optional ``games`` mapping. - - :param url: The URL that the server can be reached at. - :type url: str - :param timeout: Request timeout passed to ``requests.get``. - :type timeout: Optional[Union[float, Tuple[float, float]]] - :return: The server's representation. - :rtype: ServerAgent - :raises requests.exceptions.HTTPError: If the server returns a - non-success HTTP status code. - :raises requests.exceptions.RequestException: If the request fails - before a response is received. - :raises ValueError: If the response body is not a JSON object or if the - payload contains malformed discovery fields. - """ - response = requests.get(url.rstrip("/") + "/", timeout=timeout) - response.raise_for_status() - content = response.json() - - if not isinstance(content, dict): - raise ValueError("Server discovery responses must be JSON objects.") - - raw_name = content.get("name", "") - name = "" if raw_name is None else str(raw_name) - - games: Dict[str, Dict[str, Any]] = {} - raw_games = content.get("games", {}) - if raw_games is not None: - if not isinstance(raw_games, dict): - raise ValueError("The 'games' field must be a JSON object when provided.") - - for game_name, profile in raw_games.items(): - if isinstance(profile, dict): - games[str(game_name)] = profile - - return cls(url=url, name=name, games=games, debug=debug) - - def poll(self, game: str, payload: Dict[str, Any], timeout: float = 1.0) -> Optional[Dict[str, Any]]: - """ - Inquire a game to make a move. - - :param game: The game the ServerAgent is asked to play. - :type game: str - :param payload: The JSON payload that represents the game's state. - :type payload: Dict[str, Any] - :param timeout: Maximum number of seconds to wait for a move. - :type timeout: float - :return: The server's response, or None if the server did not respond - in time or returned an invalid response. - :rtype: Optional[Dict[str, Any]] - """ - url = f"{self.url.rstrip('/')}/{game.lstrip('/')}" - - try: - response = requests.get(url, json=payload, timeout=timeout) - response.raise_for_status() - content = response.json() - except (requests.exceptions.RequestException, ValueError): - return None - - if self.debug: - print(f"[DBG] Agent `{self.name}` returned:") - print(content) - - return content if isinstance(content, dict) else None diff --git a/pyclient/replay.py b/pyclient/replay.py new file mode 100644 index 0000000..694c152 --- /dev/null +++ b/pyclient/replay.py @@ -0,0 +1,92 @@ +""" + This module helps describe & evaluate previously played games. +""" + +from __future__ import annotations + +from .games import Game +from .transition import Transition +from dataclasses import dataclass +from typing import Any, Generator + +@dataclass(frozen=True) +class GameReplay: + """ + Game replay detailing the events of what happened in a previously + played game. + """ + + start : Game + turns : list[Turn] + + def gen_transitions(self) -> Generator[Transition, None, None]: + """ + Generate transitions from the game's summary. + + :return: Transitions that describe consequences throughout the game. + :rtype: Generator[Transition, None, None] + """ + prev_state : Game = self.start + seen_participants : dict[int, tuple[Game, dict[str, Any]]] = {} + + for turn in self.turns: + # If a player has made a move before, show the current state as + # the result of their previous turn's action. + state, action = seen_participants.get(turn.player, ( None, {} )) + if state is not None: + yield Transition( + full_state_after=prev_state, + full_state_before=state, + move=action, + player=turn.player, + ) + + # Save the player's current action for the future. + # We'll see how this action worked out! + seen_participants[turn.player] = ( prev_state, turn.action ) + prev_state = turn.state + + else: + # Record to all players that the game has finished. + for player, ( state, action ) in seen_participants.items(): + yield Transition( + full_state_after=prev_state, + full_state_before=state, + move=action, + player=player, + ) + + def to_transitions(self, player : int | None = None) -> list[Transition]: + """ + Convert the GameSummary into a list of transitions that can be + fed to a system that can reflect on past moves. + + :param player: Filter to only show transitions of a given player. + :type player: int | None + :return: A list of transitions extracted from a game. + :rtype: list[Transition] + """ + if player is None: + return list(self.gen_transitions()) + else: + return [ t for t in self.gen_transitions() if t.player == player ] + +@dataclass(frozen=True) +class Turn: + """ + A turn is a snapshot of a moment where a user was prompted to take an + action. A turn consists of three values: + + 1. The player (number) that was prompted for an action + 2. The action that that player has decided to take + 3. The game's state after the action was prompted and before the next + prompted move. + + Note that turns might not correlate with in-game turns. A user might + take multiple actions in a Risk game (recruit, attack, move) and each + action is registered as an independent move. + """ + + player : int + action : dict[str, Any] + state : Game diff --git a/pyclient/transition.py b/pyclient/transition.py new file mode 100644 index 0000000..a326e8c --- /dev/null +++ b/pyclient/transition.py @@ -0,0 +1,110 @@ +""" + The transition module allows agents to reflect on the choices they + (and others) have made during recent games. + + Agents are generally intended to be stateless, so this allows them to + later understand how well certain moves turned out. +""" + +from __future__ import annotations + +from .games import FinishState, Game +from dataclasses import dataclass +from typing import Any, Generator + +@dataclass(frozen=True) +class Transition: + """ + A transition is a simplified view on a user's single move. + + The transition contains the game state, the move they made, + and what the game ended up looking like the next time they were + prompted for a move. If the game ends, the `state_after` field contains + the final state. + """ + + full_state_after : Game + full_state_before : Game + move : dict[str, Any] + player : int + + @property + def end_result(self) -> FinishState | None: + """ + Return the result of this move if it caused the game to end, + or None if the game still continues after this. + + :return: The player's result in the game with this move. + :rtype: FinishState | None + """ + w = self.full_state_after.winner() + + if w is None: + return None + + return w[self.player] + + @property + def game_has_ended(self) -> bool: + """ + Shorthand property for whether this was the player's last move. + + Tip: Use the methods `move_caused_draw`, `move_caused_loss` and + `move_caused_win` to determine the outcome. + """ + return self.full_state_after.winner() is not None + + @property + def move_caused_draw(self) -> bool: + """ + Shorthand property for whether this move caused the player to draw. + + :rtype: bool + """ + return self.end_result == FinishState.draw + + @property + def move_caused_loss(self) -> bool: + """ + Shorthand property for whether this move caused the player to lose. + + :rtype: bool + """ + return self.end_result == FinishState.loss + + @property + def move_caused_win(self) -> bool: + """ + Shorthand property for whether this move caused the player to win. + + :rtype: bool + """ + return self.end_result == FinishState.win + + @property + def state_after(self) -> dict[str, Any]: + """ + The initial state that the agent was shown after they were asked for + a move. + + Keep in mind that this state does not necessarily contain + ALL information, as players are often partially kept in the dark + about a game's state. + + :rtype: dict[str, Any] + """ + return self.full_state_after.as_seen_by(self.player) + + @property + def state_before(self) -> dict[str, Any]: + """ + The initial state that the agent was shown when they were asked for + a move. + + Keep in mind that this state does not necessarily contain + ALL information, as players are often partially kept in the dark + about a game's state. + + :rtype: dict[str, Any] + """ + return self.full_state_before.as_seen_by(self.player) From 8aaacfc83c03fdb1a2e285fe74bb829737cc4302 Mon Sep 17 00:00:00 2001 From: Bram van den Heuvel Date: Sat, 20 Jun 2026 14:59:58 +0200 Subject: [PATCH 4/4] Remove unnecessary imports from client.py --- client.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/client.py b/client.py index e5d131e..70e18ef 100644 --- a/client.py +++ b/client.py @@ -24,8 +24,7 @@ import json import pyclient from pyclient import Agent, PyClient -from pyclient.games.tic_tac_toe import TicTacToe -from typing import Any +from pyclient.games import TicTacToe def main() -> int: """