Implement mute agent

.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/

.gitignore

@@ -1,3 +1,7 @@
+# Repository-specific virtual environments
+.venv-pyserver
+.venv-pyclient
+
 # ---> Elm
 # elm-package generated files
 elm-stuff

client.py

@@ -0,0 +1,92 @@
+"""
+    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
+import pyclient
+
+from pyclient import Agent, PyClient
+from pyclient.games import TicTacToe
+
+def main() -> int:
+    """
+        Start a client, then use it to analyze one or more matches.
+
+        :return: Exit code
+        :rtype: int
+    """
+    c = PyClient(debug=False)
+
+    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 : pyclient.GameReplay) -> None:
+    """
+        Print a diagnostic of a played game to the terminal.
+
+        :param game: The results of a played game.
+        :type game: pyclient.GameReplay
+    """
+    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 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.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.turns)}")
+    print(f"Result: {final_state.winner()}")
+
+if __name__ == "__main__":
+    raise SystemExit(main())

elo.py

@@ -0,0 +1,36 @@
+"""
+    Create an ELO tracker that compares various server agents out there.
+"""
+
+from elo_tracker import EloTracker
+from pyclient.games import TicTacToe
+
+import pyclient
+import time
+
+GAME_FILE = "games.jsonl"
+PLAYER_FILE = "known_players.json"
+
+def main() -> int:
+    tracker = EloTracker(
+        game_file_name=GAME_FILE,
+        player_file_name=PLAYER_FILE,
+        debug=True,
+    )
+
+    tracker.start_periodic_matches(
+        game=TicTacToe.empty(),
+        interval_seconds=60,
+        player_count=2,
+    )
+
+    try:
+        while True:
+            pass
+    except KeyboardInterrupt:
+        tracker.stop_periodic_matches()
+
+    return 0
+
+if __name__ == "__main__":
+    raise SystemExit(main())

elo_tracker/__init__.py

@@ -0,0 +1,5 @@
+from .app import EloTracker
+
+__all__ = [
+    "EloTracker",
+]

elo_tracker/app.py

@@ -0,0 +1,734 @@
+"""
+    This app hosts the client that'll perform the ELO tracking.
+"""
+
+from __future__ import annotations
+
+import copy
+import json
+import os
+import random
+import threading
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Any, Sequence
+
+import pyclient
+
+from pyclient.games import FinishState, Game
+
+
+DEFAULT_ELO = 1000
+STD_DEV_DIFF = 400
+ELO_K_FACTOR = 32
+
+@dataclass(frozen=True)
+class PlayerIdentifier:
+    """
+        The Player Identifier uniquely identifies each player for the ELO
+        tracker.
+    """
+    name : str
+    url : str
+    version : str | None
+
+    def __key__(self) -> tuple[str, str, str | None]:
+        return (self.name, self.url, self.version)
+    
+    @classmethod
+    def from_server_agent(cls, agent : pyclient.ServerAgent) -> "PlayerIdentifier":
+        """
+            Gain a player identifier from an agent.
+        """
+        return cls(
+            name=agent.name,
+            url=agent.url,
+            version=agent.profile.get("version",
+                agent.profile.get("me.noordstar.peanuts.agent.version", None)
+            ),
+        )
+
+@dataclass()
+class EloStat:
+    """
+        The EloStat records the ELO statistics of a single player.
+        What's their score, and how much did they win?
+    """
+
+    # Identity
+    player_id : PlayerIdentifier
+
+    # Statistics
+    losses : int
+    draws : int
+    wins : int
+    elo : float
+
+    @classmethod
+    def new(cls, player_id : PlayerIdentifier) -> "EloStat":
+        """
+            Create a new ELO type based on a player.
+
+            :param player_id: Unique player identifier
+            :type player_id: PlayerIdentifier
+            :return: New empty Elo statistics for the player
+            :rtype: EloStat
+        """
+        return cls(
+            player_id=player_id, losses=0, draws=0, wins=0, elo=DEFAULT_ELO,
+        )
+
+    def to_json(self) -> dict[str, Any]:
+        """
+            Convert EloStat to a JSON-formatted dictionary
+
+            :return: The EloStat in JSON format
+            :rtype: dict[str, Any]
+        """
+        d = dict(
+            name=self.player_id.name,
+            url=self.player_id.url,
+            losses=self.losses,
+            draws=self.draws,
+            wins=self.wins,
+            elo=int(self.elo),
+        )
+
+        if self.player_id.version is not None:
+            d["version"] = self.player_id.version
+        
+        return d
+
+@dataclass(frozen=True)
+class Match:
+    """
+        A Match represents a game written to disk in JSONL format.
+    """
+    game_name : str
+    participants : list[tuple[PlayerIdentifier, FinishState]]
+    timestamp : str
+
+    @staticmethod
+    def now() -> str:
+        """
+            Get a timestamp of now.
+
+            :return: Timestamp in ISO format
+            :rtype: str
+        """
+        return datetime.now(tz=timezone.utc).isoformat()
+
+    @classmethod
+    def from_json_record(cls, record : dict[str, Any]) -> "Match":
+        """
+            Create a new match from a decoded JSON object.
+
+            :param participants: Decoded JSON object
+            :type participants: dict[str, Any]
+            :return: An initialized match
+            :rtype: Match
+            :raises KeyError: The JSON is missing required keys.
+            :raises ValueError: The JSON is formatted improperly.
+        """
+        participants : list[dict[str, Any]] = record["participants"]
+        if not isinstance(participants, list):
+            raise ValueError(
+                "Key `participants` must be list of objects"
+            )
+
+        game_name : str = str(record["name"])
+        timestamp : str = str(record["timestamp"]) # TODO: Perhaps verify ISO format?
+        new_participants : list[tuple[PlayerIdentifier, FinishState]] = []
+
+        for i, participant in enumerate(participants):
+            # Sanity assertions
+            if not isinstance(participant, dict):
+                raise ValueError(
+                    f"Participant #{i+1} must be dictionary"
+                )
+            
+            for key in ["name", "url", "result"]:
+                if not isinstance(participant[key], str):
+                    raise ValueError(
+                        f"Participant #{i+1} must have the `{key}` key as string"
+                    )
+            
+            # Initialize participant
+            name : str = participant["name"]
+            url : str = participant["url"]
+            result = FinishState.from_str(participant["result"])
+            version : str | None = participant.get("version", None)
+            if version is not None:
+                version = str(version)
+            
+            new_participants.append((
+                PlayerIdentifier(name=name, url=url, version=version),
+                result,
+            ))
+        
+        if len(new_participants) < 2:
+            raise ValueError(
+                "Expected at least 2 participants in a game for which ELO can be tracked"
+            )
+        
+        return cls(
+            game_name=game_name,
+            participants=new_participants,
+            timestamp=timestamp,
+        )
+    
+    @classmethod
+    def from_replay(
+        cls,
+        players : list[pyclient.ServerAgent],
+        replay : pyclient.GameReplay,
+        timestamp : str | None,
+    ) -> "Match":
+        """
+            Convert a GameReplay into a match.
+
+            :param players: The participants of the match.
+            :type players: list[pyclient.ServerAgent]
+            :param replay: Game summary.
+            :type replay: pyclient.GameReplay
+            :param timestamp: ISO formatted timestamp of when the game was planned.
+            :type timestamp: str
+            :return: An initialized match.
+            :rtype: Match
+            :raises ValueError: The replay shows an unfinished match.
+        """
+        results = replay.turns[-1].state.winner()
+        if results is None:
+            raise ValueError(
+                "Game hasn't finished yet."
+            )
+
+        participants : list[tuple[PlayerIdentifier, FinishState]] = []
+        for i, agent in enumerate(players):
+            finish_state = results.get(i, None)
+            if finish_state is None:
+                continue
+
+            participants.append((
+                PlayerIdentifier.from_server_agent(agent=agent),
+                finish_state,
+            ))
+
+        return cls(
+            game_name=replay.game_name,
+            participants=participants,
+            timestamp=timestamp or cls.now()
+        )
+    
+    def log(self, file_name : str) -> None:
+        """
+            Log the current match to disk.
+
+            :param file_name: File name to write the match to.
+            :type file_name: str
+        """
+        with open(file_name, "a", encoding="utf-8") as wp:
+            wp.write(json.dumps(self.to_json(), sort_keys=True) + "\n")
+     
+    def to_json(self) -> dict[str, Any]:
+        """
+            Convert the Match back to JSON.
+
+            :return: The Match in a dictionary that's can be converted to JSON.
+            :rtype: dict[str, Any]
+        """
+        participants : list[dict[str, str]] = []
+
+        for player_id, result in self.participants:
+            d : dict[str, str]= dict(
+                name=player_id.name,
+                url=player_id.url,
+                result=result.name,
+            )
+
+            if player_id.version is not None:
+                d["version"] = player_id.version
+
+            participants.append(d)
+
+        return dict(
+            name=self.game_name,
+            participants=participants,
+            timestamp=self.timestamp,
+        )
+
+class EloTracker:
+    """
+        The Elo tracker tracks matches between URLs that it is familiar with.
+    """
+    def __init__(
+        self,
+        game_file_name: str,
+        player_file_name: str,
+        debug: bool = False,
+        name: str = "Bot-Man-Toe Elo Tracker",
+    ) -> None:
+        """
+            Create an EloTracker.
+
+            :param game_file_name: The file name to write game results to.
+            :type game_file_name: str
+            :param player_file_name: The file name to read player URLs from.
+            :type player_file_name: str
+            :param debug: Whether to print scheduler errors.
+            :type debug: bool
+            :param name: Display name for the leaderboard.
+            :type name: str
+        """
+
+        # Threading variables
+        self.__lock = threading.RLock()
+        self.__scheduler_stop = threading.Event()
+        self.__scheduler_thread: threading.Thread | None = None
+
+        # Immutable values
+        self.debug: bool = debug
+        self.game_file_name: str = game_file_name
+        self.player_file_name: str = player_file_name
+        self.name: str = name
+
+        # Thread-unsafe variables
+        # Please use a lock while doing CRUD operations on them
+        self.players: list[pyclient.ServerAgent] = []
+        self.__matches: list[Match] = []
+        self.__stats: dict[PlayerIdentifier, EloStat] = {}
+
+        # Initialize tracker
+        self.__load_matches()
+        self.load_players()
+
+    def __debug(self, message: str) -> None:
+        """
+            Send a debug message to stdout. Ignored when not in debug mode.
+
+            :param message: The message to debug log
+            :type message: str
+        """
+        if self.debug:
+            with self.__lock:
+                print(f"[EloTracker] {message}")
+
+    def __get_stat(self, player_id : PlayerIdentifier) -> EloStat:
+        """
+            Get a player's statistics based on their player identifier.
+
+            If the player wasn't known, the function returns a newly
+            initialized record in the database for them.
+
+            :param player_id: Unique player identifier.
+            :type player_id: PlayerIdentifier
+            :return: Elo statistics
+        """
+        with self.__lock:
+            stat = self.__stats.get(player_id, None)
+            
+            if stat is not None:
+                return stat
+            
+            stat = EloStat.new(player_id=player_id)
+            self.__stats[player_id] = stat
+
+            return stat
+
+    def __load_matches(self) -> None:
+        """
+            Load persisted JSONL records and rebuild in-memory statistics.
+        """
+        if not os.path.exists(self.game_file_name):
+            return
+
+        with self.__lock:
+            self.__matches = []
+            self.__stats = {}
+
+            with open(self.game_file_name, encoding="utf-8") as fp:
+                for line_no, line in enumerate(fp, start=1):
+                    line = line.strip()
+                    if line == "":
+                        continue
+
+                    try:
+                        record = json.loads(line)
+                    except json.JSONDecodeError:
+                        self.__debug(
+                            f"Skipping malformed match record on line {line_no}."
+                        )
+                        continue
+
+                    if not isinstance(record, dict):
+                        self.__debug(
+                            f"Skipping non-object match record on line {line_no}."
+                        )
+                        continue
+
+                    try:
+                        m = Match.from_json_record(record=record)
+                    except ( KeyError, ValueError ):
+                        self.__debug(
+                            f"Skipping malformed JSON object on line {line_no}."
+                        )
+                    else:
+                        self.__matches.append(m)
+                        self.__register_match(m)
+    
+    def __register_match(self, m : Match) -> None:
+        """
+            Apply a newly registered match to the aggregate statistics.
+
+            :param m: Newly created match with results & outcomes.
+            :type m: Match
+        """
+
+        effective_k = ELO_K_FACTOR / (len(m.participants) - 1)
+        scores : dict[PlayerIdentifier, float] = {}
+        
+        # First, calculate the pairwise ELO results
+        # Do not apply them yet, in order to guarantee fair ELO shifts
+        for player_id1, result1 in m.participants:
+            total_k = 0.0
+            rating_1 = self.__get_stat(player_id=player_id1).elo
+
+            for player_id2, result2 in m.participants:
+                rating_2 = self.__get_stat(player_id=player_id2).elo
+
+                expected_score = 1 / (1 + 10 ** ((rating_2 - rating_1) / STD_DEV_DIFF))
+
+                actual_score = 0
+                if result1.score() > 0.0:
+                    actual_score = result1.score() / (result1.score() + result2.score())
+
+                total_k += effective_k * (actual_score - expected_score)
+            
+            scores[player_id1] = total_k
+        
+        all_scores = sum(scores.values())
+
+        if 0.001 <= abs(all_scores):
+            self.__debug(
+                f"In total, all ELO score changes added together are {all_scores} (should be 0.0)"
+            )
+
+        # Then, apply the ELO score update + count the wins, draws & losses
+        for player_id, result in m.participants:
+            player = self.__get_stat(player_id=player_id)
+            player.elo += scores[player_id]
+
+            match result:
+                case FinishState.draw:
+                    player.draws += 1
+                case FinishState.loss:
+                    player.losses += 1
+                case FinishState.win:
+                    player.wins += 1
+
+    def __scheduler_loop(
+        self,
+        game: Game,
+        interval_seconds: float,
+        player_count: int,
+    ) -> None:
+        """
+            Perform a schedule in which you play a random game.
+
+            :param game: Game to play.
+            :type game: pyclient.Game
+            :param interval_seconds: Number of seconds to sleep between games
+            :type interval_seconds: float
+            :param player_count: The number of players that are supposed to participate
+            :type player_count: int
+        """
+        while not self.__scheduler_stop.is_set():
+            try:
+                self.load_players()
+
+                with self.__lock:
+                    available = len(self.players)
+
+                if available < player_count:
+                    self.__debug(
+                        f"Skipping scheduled match: {available} players available, "
+                        f"{player_count} required."
+                    )
+                else:
+                    self.__debug(
+                        "Playing a new scheduled match"
+                    )
+                    self.play_random_match(game=game, player_count=player_count)
+            except Exception as exc:
+                self.__debug(f"Scheduled match failed: {exc}")
+                raise exc
+
+            if self.__scheduler_stop.wait(interval_seconds):
+                break
+
+    def create_flask_app(self, import_name : str) -> Any:
+        """
+            Create a Flask app that exposes tracker statistics.
+
+            :param import_name: The name of the application package.
+            :type import_name: str
+        """
+        try:
+            from flask import Flask, Response, jsonify
+        except ImportError as exc:
+            raise ImportError(
+                "Flask is required to host the EloTracker server. "
+                "Install the project requirements before calling create_app()."
+            ) from exc
+
+        app = Flask(__name__)
+
+        @app.get("/")
+        def index() -> Response:
+            return Response(
+                """
+<!doctype html>
+<html lang="en">
+  <head>
+    <meta charset="utf-8">
+    <title>Bot-Man-Toe Elo Tracker</title>
+  </head>
+  <body>
+    <main>
+      <h1>Bot-Man-Toe Elo Tracker</h1>
+      <p>The JSON API is available at /leaderboard, /matches, /players, and /health.</p>
+    </main>
+  </body>
+</html>
+""".strip(),
+                mimetype="text/html",
+            )
+
+        @app.get("/health")
+        def health() -> Response:
+            return jsonify({
+                "ok": True,
+                "periodic_matches": self.is_running_periodic_matches(),
+            })
+
+        @app.get("/leaderboard")
+        def leaderboard() -> Response:
+            return jsonify(self.get_json_leaderboard())
+
+        @app.get("/matches")
+        def matches() -> Response:
+            return jsonify(self.get_json_matches())
+
+        @app.get("/players")
+        def players() -> Response:
+            return jsonify(self.get_json_players())
+
+        return app
+
+    def is_running_periodic_matches(self) -> bool:
+        """
+            Return whether the scheduler thread is currently alive.
+
+            :return: Whether the scheduler thread is currently alive.
+            :rtype: bool
+        """
+        with self.__lock:
+            return self.__scheduler_thread is not None and self.__scheduler_thread.is_alive()
+
+    def load_players(self) -> None:
+        """
+            Load the known players from disk.
+
+            :raises ValueError: File was not properly JSON-formatted.
+        """
+        with open(self.player_file_name, encoding="utf-8") as fp:
+            obj = json.load(fp)
+            if not isinstance(obj, dict):
+                raise ValueError(
+                    "Expected list of URLs in player file."
+                )
+
+            urls = obj.get("players", [])
+            if not isinstance(urls, list):
+                raise ValueError(
+                    "Expected `players` field to be a list of strings."
+                )
+
+            players : list[pyclient.ServerAgent] = []
+            for url in urls:
+                if not isinstance(url, str):
+                    continue
+
+                try:
+                    agent = pyclient.Agent.from_url(url, debug=self.debug)
+                except ValueError:
+                    pass  # Not an available player right now
+                else:
+                    players.append(agent)
+
+            with self.__lock:
+                self.players = players
+
+                for agent in players:
+                    self.__get_stat(PlayerIdentifier.from_server_agent(agent))
+
+    def play_match(self, players: list[str], game: Game) -> pyclient.GameReplay:
+        """
+            Play a single match with appointed players.
+
+            :param players: List of URLs that participate.
+            :type players: list[str]
+            :return: A summary of the game.
+            :rtype: pyclient.GameReplay
+            :raises ValueError: One of the URLs could not be accessed.
+        """
+        agents = [
+            pyclient.Agent.from_url(url, debug=self.debug)
+            for url in players
+        ]
+
+        replay = pyclient.PyClient(debug=self.debug).play_game(
+            players=agents,
+            start=game,
+        )
+        
+        m = Match.from_replay(players=agents, replay=replay, timestamp=Match.now())
+
+        # Record match
+        m.log(self.game_file_name)
+        self.__register_match(m)
+
+        return replay
+
+    def play_random_match(
+        self,
+        game: Game,
+        player_count: int | None = None,
+    ) -> pyclient.GameReplay:
+        """
+            Play a game with any known players.
+
+            :param game: The game to start playing
+            :type game: Game
+            :param player_count: Optional number of players to select.
+            :type player_count: int | None
+            :raises ValueError: One of the randomly chosen URLs could not be accessed.
+        """
+        with self.__lock:
+            players = [agent.url for agent in self.players]
+
+        random.shuffle(players)
+
+        if player_count is not None:
+            players = players[:player_count]
+
+        return self.play_match(players=players, game=game)
+
+    def start_periodic_matches(
+        self,
+        game: Game,
+        interval_seconds: float = 300,
+        player_count: int = 2,
+    ) -> None:
+        """
+            Start running matches periodically in a daemon thread.
+
+            :param game: Game to play.
+            :type game: pyclient.Game
+            :param interval_seconds: Number of seconds to sleep between games
+            :type interval_seconds: float
+            :param player_count: The number of players that are supposed to participate
+            :type player_count: int
+        """
+        if interval_seconds <= 0:
+            raise ValueError("interval_seconds must be greater than zero.")
+        if player_count <= 1:
+            raise ValueError("player_count must be greater than one.")
+
+        with self.__lock:
+            if self.is_running_periodic_matches():
+                self.stop_periodic_matches()
+
+            self.__scheduler_stop.clear()
+            self.__scheduler_thread = threading.Thread(
+                target=self.__scheduler_loop,
+                args=(game, interval_seconds, player_count),
+                daemon=True,
+            )
+            self.__scheduler_thread.start()
+
+    def stop_periodic_matches(self) -> None:
+        """
+            Stop the periodic match scheduler and wait briefly for it to exit.
+        """
+        thread: threading.Thread | None
+
+        with self.__lock:
+            self.__scheduler_stop.set()
+            thread = self.__scheduler_thread
+
+        if thread is not None:
+            thread.join(timeout=5)
+
+        with self.__lock:
+            if self.__scheduler_thread is thread:
+                self.__scheduler_thread = None
+
+    def get_json_players(self) -> list[dict[str, Any]]:
+        """
+            Return known currently available players as notebook-friendly dicts.
+        """
+        with self.__lock:
+            players : list[EloStat] = list(self.__stats.values())
+
+        players.sort(
+            key=lambda player: (-int(player.elo), player.player_id.name)
+        )
+
+        return [ stat.to_json() for stat in players ]
+
+    def get_json_matches(self, limit: int | None = None) -> list[dict[str, Any]]:
+        """
+            Return persisted match records, newest last unless limited.
+
+            :param limit: Maximum number of most recent matches
+            :type limit: int | None
+            :return: A list of most recent matches
+            :rtype: list[dict[str, Any]]
+        """
+        with self.__lock:
+            if limit is None:
+                matches = self.__matches
+            elif limit <= 0:
+                matches = []
+            else:
+                matches = self.__matches[-limit:]
+
+            return [ m.to_json() for m in matches ]
+
+    def get_json_leaderboard(self) -> dict[str, Any]:
+        """
+            Return aggregate player statistics for local use or JSON APIs.
+        """
+        return {
+            "name": self.name,
+            "players": self.get_json_players(),
+        }
+
+    def start_server(
+        self,
+        host: str = "127.0.0.1",
+        import_name : str = __name__,
+        port: int = 5000,
+        debug: bool = False,
+        **kwargs: Any,
+    ) -> None:
+        """
+            Start a Flask development server from which the ELO scores can be
+            viewed interactively.
+        """
+        return (
+            self.create_flask_app(import_name=import_name)
+                .run(host=host, port=port, debug=debug, **kwargs)
+        )

known_players.json

@@ -0,0 +1,5 @@
+{
+    "players": [
+        "https://bmt001.noordstar.me"
+    ]
+}

main.py

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

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
+__all__ = [
-
+    "Agent",
-from .poll import ServerAgent
+    "GameReplay",
-
+    "PyClient",
-
+    "ServerAgent",
-class PyClient:
+    "Transition",
-    """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"]

pyclient/agent.py

@@ -0,0 +1,178 @@
+"""
+    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
+            :raises ValueError: The server fails to reach out one of the URLs.
+        """
+        try:
+            return ServerAgent.from_server_url(url=url, **kwargs)
+        except (ValueError, requests.RequestException, requests.HTTPError):
+            pass
+
+        raise ValueError(
+            "URL did not lead to a willing agent"
+        )
+
+    @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
+
+    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,
+        )

pyclient/client.py

@@ -0,0 +1,72 @@
+"""
+    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(
+            game_name=start.game_name(),
+            start=start,
+            turns=list(self.gen_game(players=players, start=start)),
+        )

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",
+]

pyclient/games/game.py

@@ -0,0 +1,172 @@
+"""
+    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()
+
+    @classmethod
+    def from_str(cls, s : str) -> "FinishState":
+        """
+            Convert the finish state from a string.
+
+            :param s: String to convert.
+            :type s: str
+            :return: Finish state
+            :rtype: FinishState
+            :raises ValueError: Invalid string value.
+        """
+        for option in FinishState:
+            if s == option.name:
+                return option
+        else:
+            raise ValueError(
+                f"Unknown finish state `{s}`"
+            )
+
+    def score(self) -> float:
+        """
+            As a score between 0 and 1, convert how "good" an outcome is.
+
+            :return: A score determining how beneficial a finish state is.
+            :rtype: float
+        """
+        match self:
+            case FinishState.draw:
+                return 0.5
+            case FinishState.loss:
+                return 0.0
+            case FinishState.win:
+                return 1.0
+
+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 {}

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
 

pyclient/poll.py

@@ -1,133 +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,
-        profile : dict[str, Any] = {}
-    ) -> 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]]
-        :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
-    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
-        
-        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, profile=profile)
-    
-    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
-    
-    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,
-        )
-        

pyclient/replay.py

@@ -0,0 +1,93 @@
+"""
+    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.
+    """
+
+    game_name : str
+    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

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)

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

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

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

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

server.py

@@ -0,0 +1,62 @@
+"""
+    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="Mute",
+
+        # Custom information that you can use to tell people about this player
+        profile={
+            "me.noordstar.peanuts.agent.version": "1.0.0",
+            "me.noordstar.peanuts.is_ai": False,
+            "me.noordstar.peanuts.author": "Bram",
+            "me.noordstar.peanuts.containerized": True,
+            "version": "1.0.0",
+        },
+
+        # 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=respond_mute, profile={})
+
+    # Start listening for games
+    player.start(
+        host="0.0.0.0", # Comment out when using only locally
+        port=5000,
+    )
+
+    return 0
+
+def respond_mute(payload : dict[str, Any]) -> dict[str, Any]:
+    """
+        Always respond with an empty dictionary. This means the user should
+        always take the "default" move.
+
+        A well-programmed game must NOT raise an error as a result of this.
+
+        :param payload: The game state which is completely ignored.
+        :type payload: dict[str, Any]
+        :return: An empty dictionary
+        :rtype: dict[str, Any]
+    """
+    return {}
+
+if __name__ == "__main__":
+    raise SystemExit(main())