Bram
/
Bot-Man-Toe
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Compare commits

base: Bram:e9a25c37e15137b6ae9b2d0a84b38e260430a10c
Branches Tags
Bram:main
Bram:bram/mute
Bram:bram/agent-of-chaos
Bram:bram/deployment
Bram:web-client
Bram:agent/mute-1.0.0
Bram:agent/agent-of-chaos-1.0.0
..
compare: Bram:e2a11ae0d1663b11882fdf08de243b919bd070d4
Branches Tags
Bram:main
Bram:bram/mute
Bram:bram/agent-of-chaos
Bram:bram/deployment
Bram:web-client
Bram:agent/mute-1.0.0
Bram:agent/agent-of-chaos-1.0.0

No commits in common. "e9a25c37e15137b6ae9b2d0a84b38e260430a10c" and "e2a11ae0d1663b11882fdf08de243b919bd070d4" have entirely different histories.
e9a25c37e1 ... e2a11ae0d1

23 changed files with 341 additions and 1829 deletions
Show Stats Expand all files Collapse all files

179
.dockerignore
View File

@ -1,179 +0,0 @@
# 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/

4
.gitignore vendored
View File

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

92
client.py
View File

@ -1,92 +0,0 @@
"""
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())

36
elo.py
View File

@ -1,36 +0,0 @@
"""
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())

5
elo_tracker/__init__.py
View File

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

734
elo_tracker/app.py
View File

@ -1,734 +0,0 @@
"""
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)
)

0
games.jsonl
View File

5
known_players.json
View File

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

89
main.py Normal file
View File

@ -0,0 +1,89 @@
"""
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)

123
pyclient/__init__.py
View File

@ -1,16 +1,111 @@
""" """Public client entry points."""
Entry points for developers who wish to use the PyClient module.
"""
from .agent import Agent, ServerAgent from typing import Any, Generator, List, Optional
from .client import PyClient
from .replay import GameReplay
from .transition import Transition
__all__ = [ import requests
"Agent",
"GameReplay", from .poll import ServerAgent
"PyClient",
"ServerAgent",
"Transition", 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"]

178
pyclient/agent.py
View File

@ -1,178 +0,0 @@
"""
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,
)

72
pyclient/client.py
View File

@ -1,72 +0,0 @@
"""
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)),
)

12
pyclient/games/__init__.py
View File

@ -1,12 +0,0 @@
"""
Entry point for collecting all known games
"""
from .game import FinishState, Game
from .tic_tac_toe import TicTacToe
__all__ = [
"FinishState",
"Game",
"TicTacToe",
]

172
pyclient/games/game.py
View File

@ -1,172 +0,0 @@
"""
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 {}

31
pyclient/games/tic_tac_toe.py
View File

@ -1,14 +1,6 @@
"""
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 dataclasses import dataclass
from enum import Enum, auto from enum import Enum, auto
from typing import Any, Optional from typing import Any, Dict, List, Optional
class Field(Enum): class Field(Enum):
X = auto() X = auto()
@ -28,7 +20,7 @@ class Field(Enum):
return "" return ""
@dataclass(frozen=True) @dataclass(frozen=True)
class TicTacToe(Game): class TicTacToe:
field_1 : Field field_1 : Field
field_2 : Field field_2 : Field
field_3 : Field field_3 : Field
@ -180,7 +172,7 @@ class TicTacToe(Game):
def action_name(self) -> str: def action_name(self) -> str:
return "" 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. Return the view of the game from the perspective of a given player.
@ -210,9 +202,6 @@ class TicTacToe(Game):
field_4=Field.empty, field_5=Field.empty, field_6=Field.empty, field_4=Field.empty, field_5=Field.empty, field_6=Field.empty,
field_7=Field.empty, field_8=Field.empty, field_9=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": def move_default(self) -> "TicTacToe":
""" """
@ -240,7 +229,7 @@ class TicTacToe(Game):
"No legal moves exist anymore on this tic-tac-toe board." "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 Have a player make a move. Based on this information, update the
game. game.
@ -282,7 +271,7 @@ class TicTacToe(Game):
""" """
return 1 if self.count_x() <= self.count_o() else 2 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 { return {
"1": str(self.field_1), "1": str(self.field_1),
"2": str(self.field_2), "2": str(self.field_2),
@ -295,12 +284,12 @@ class TicTacToe(Game):
"9": str(self.field_9), "9": str(self.field_9),
} }
def winner(self) -> dict[int, FinishState] | None: def winner(self) -> int | None:
""" """
Returns whether the board indicates that there's a winner. 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. :return: The winning player, zero in case of a tie, or None if there's no winner yet.
:rtype: dict[int, FinishState] | None :rtype: int | None
""" """
win_lines = [ win_lines = [
[ 1, 2, 3, ], [ 1, 2, 3, ],
@ -314,17 +303,15 @@ class TicTacToe(Game):
] ]
d = self.to_dict() d = self.to_dict()
out = { 1 : FinishState.loss, 2 : FinishState.loss, }
for player, symbol in [ ( 1, str(Field.X) ), ( 2, str(Field.O) ) ]: for player, symbol in [ ( 1, str(Field.X) ), ( 2, str(Field.O) ) ]:
for win_line in win_lines: for win_line in win_lines:
if all(d[str(w)] == symbol for w in win_line): if all(d[str(w)] == symbol for w in win_line):
out[player] = FinishState.win return player
return out
else: else:
# Check for draw # Check for draw
if all(item != "" for item in d.values()): if all(item != "" for item in d.values()):
return { 1 : FinishState.draw, 2 : FinishState.draw, } return 0
return None return None

133
pyclient/poll.py Normal file
View File

@ -0,0 +1,133 @@
"""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,
)

93
pyclient/replay.py
View File

@ -1,93 +0,0 @@
"""
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

110
pyclient/transition.py
View File

@ -1,110 +0,0 @@
"""
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)

25
pyserver/Containerfile
View File

@ -1,25 +0,0 @@
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"]

5
requirements-pyclient.txt
View File

@ -1,5 +0,0 @@
certifi==2026.6.17
charset-normalizer==3.4.7
idna==3.18
requests==2.34.2
urllib3==2.7.0

8
requirements-pyserver.txt
View File

@ -1,8 +0,0 @@
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

2
requirements.txt
View File

@ -1,5 +1,5 @@
blinker==1.9.0 blinker==1.9.0
certifi==2026.6.17 certifi==2026.5.20
charset-normalizer==3.4.7 charset-normalizer==3.4.7
click==8.4.1 click==8.4.1
colorama==0.4.6 colorama==0.4.6

62
server.py
View File

@ -1,62 +0,0 @@
"""
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())