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

Refactor project to new Agent class

sam/sneaky
Bram van den Heuvel 2026-06-24 13:31:24 +02:00
parent 5c3eb14c17
commit d8c20fd4cc
7 changed files with 56 additions and 252 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
agents/mute.py
View File

@ -8,7 +8,7 @@
from __future__ import annotations
from .agent import Agent, Payload
from .agent import ActionFunction, Agent, Payload
def respond_mute(payload : Payload) -> Payload:
"""
@ -42,4 +42,13 @@ class MuteAgent(Agent):
},
)
self.add_tic_tac_toe(on_move=respond_mute, profile={})
def get_action(self, game: str, action: str | None) -> ActionFunction:
"""
Return the mute response for any game or action.
:param game: The game that the mute ignores.
:type game: str
:param action: The action that the mute ignores.
:type action: str
"""
return respond_mute

9
client.py
View File

@ -23,7 +23,8 @@ from __future__ import annotations
import json
import pyclient
from pyclient import Agent, PyClient
from agents import Agent, RemoteAgent
from pyclient import PyClient
from pyclient.games import TicTacToe
def main() -> int:
@ -33,11 +34,11 @@ def main() -> int:
:return: Exit code
:rtype: int
"""
c = PyClient(debug=False)
c = PyClient(debug=True)
players : list[Agent] = [
Agent.from_url(url="https://bmt001.noordstar.me/"),
Agent.from_url(url="https://bmt001.noordstar.me/"),
RemoteAgent(url="https://bmt001.noordstar.me/"),
RemoteAgent(url="https://bmt002.noordstar.me/"),
]
out = c.play_game(

30
elo_tracker/app.py
View File

@ -14,6 +14,7 @@ from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any, Sequence
import agents
import pyclient
from pyclient.games import FinishState, Game
@ -37,17 +38,11 @@ class PlayerIdentifier:
return (self.name, self.url, self.version)
@classmethod
def from_server_agent(cls, agent : pyclient.ServerAgent) -> "PlayerIdentifier":
def from_server_agent(cls, agent : agents.RemoteAgent) -> "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)
),
)
return cls(name=agent.name, url=agent.url, version=agent.version)
@dataclass()
class EloStat:
@ -181,7 +176,7 @@ class Match:
@classmethod
def from_replay(
cls,
players : list[pyclient.ServerAgent],
players : list[agents.RemoteAgent],
replay : pyclient.GameReplay,
timestamp : str | None,
) -> "Match":
@ -189,7 +184,7 @@ class Match:
Convert a GameReplay into a match.
:param players: The participants of the match.
:type players: list[pyclient.ServerAgent]
:type players: list[agents.RemoteAgent]
:param replay: Game summary.
:type replay: pyclient.GameReplay
:param timestamp: ISO formatted timestamp of when the game was planned.
@ -295,7 +290,7 @@ class EloTracker:
# Thread-unsafe variables
# Please use a lock while doing CRUD operations on them
self.players: list[pyclient.ServerAgent] = []
self.players: list[agents.RemoteAgent] = []
self.__matches: list[Match] = []
self.__stats: dict[PlayerIdentifier, EloStat] = {}
@ -558,13 +553,13 @@ class EloTracker:
"Expected `players` field to be a list of strings."
)
players : list[pyclient.ServerAgent] = []
players : list[agents.RemoteAgent] = []
for url in urls:
if not isinstance(url, str):
continue
try:
agent = pyclient.Agent.from_url(url, debug=self.debug)
agent = agents.RemoteAgent(url=url)
except ValueError:
pass # Not an available player right now
else:
@ -586,17 +581,14 @@ class EloTracker:
:rtype: pyclient.GameReplay
:raises ValueError: One of the URLs could not be accessed.
"""
agents : list[Any] = [
pyclient.Agent.from_url(url, debug=self.debug)
for url in players
]
ags : list[Any] = [ agents.RemoteAgent(url=url) for url in players ]
replay = pyclient.PyClient(debug=self.debug).play_game(
players=agents,
players=ags,
start=game,
)
m = Match.from_replay(players=agents, replay=replay, timestamp=Match.now())
m = Match.from_replay(players=ags, replay=replay, timestamp=Match.now())
# Record match
m.log(self.game_file_name)

3
pyclient/__init__.py
View File

@ -2,15 +2,12 @@
Entry points for developers who wish to use the PyClient module.
"""
from .agent import Agent, ServerAgent
from .client import PyClient
from .replay import GameReplay
from .transition import Transition
__all__ = [
"Agent",
"GameReplay",
"PyClient",
"ServerAgent",
"Transition",
]

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

11
pyclient/client.py
View File

@ -5,7 +5,7 @@
from __future__ import annotations
from .agent import Agent
from agents import Agent
from .games import Game
from .replay import GameReplay, Turn
from dataclasses import dataclass
@ -43,12 +43,17 @@ class PyClient:
else:
agent = players[player - 1]
payload = agent.poll(
on_move = agent.get_action(
game=current_state.game_name(),
payload=current_state.as_seen_by(player=player),
action=current_state.action_name(),
)
# Calculate move
if on_move is None:
payload = {}
else:
payload = on_move(current_state.as_seen_by(player=player))
current_state = current_state.move(payload=payload)
yield Turn(action=payload, player=player, state=current_state)

64
pyserver/server.py
View File

@ -2,14 +2,11 @@
from __future__ import annotations
from flask import Flask, jsonify, request
from agents import Agent
from collections.abc import Callable, Mapping
from flask import Flask, Response, jsonify, request
from functools import wraps
from typing import Any, Optional
from typing import Any
PayloadType = dict[str, Any]
GameHandler = Callable[[PayloadType], PayloadType]
class PyServer:
@ -30,34 +27,15 @@ class PyServer:
self.agent = agent
self.app = Flask(import_name)
self.__add_api_endpoint("", "", lambda _ : self.__discovery())
@self.app.get("/")
def discovery():
return jsonify(self.__discovery())
# Register endpoints
for name, (_, endpoints) in self.agent.registered_games.items():
if name == "":
continue
@self.app.get("/<path:game>")
@self.app.get("/<path:game>/<path:action>")
def dispatch(game: str, action: str | None = None):
return jsonify(self.__dispatch(game=game, action=action))
for route, func in endpoints.items():
self.__add_api_endpoint(name=name, route=route, func=func)
def __add_api_endpoint(self, name : str, route : str, func : GameHandler) -> None:
"""
Create a new API endpoint.
:param name: The name of the action to undertake.
:type name: str
:param func: The player's function that determines what action to take.
:type func: Callable[[dict[str, Any]], dict[str, Any]]
:raises ValueError: The URL has already been registered.
"""
url = self.__make_url(name, route)
return self.app.add_url_rule(url,
endpoint="botman_" + name.replace("/", "_"),
view_func=self.__func_wrapper(func),
methods=["GET"],
)
def __discovery(self) -> PayloadType:
"""
Return the server's discovery information.
@ -77,21 +55,21 @@ class PyServer:
return d
def __func_wrapper(self, func : GameHandler) -> Callable[[], Response]:
def __dispatch(self, game: str, action: str | None) -> PayloadType:
"""
Wrapper that catches an incoming request, parses it, and responds
with a player's action response.
Resolve a request against the agent's action lookup.
"""
@wraps(func)
def exec():
payload = request.get_json(silent=True) or {}
result = func(payload) or {}
return jsonify(result)
return exec
payload = request.get_json(silent=True) or {}
func = self.agent.get_action(
game=game.strip("/"),
action=action.strip("/") if isinstance(action, str) else None,
)
def __make_url(self, name : str, route : str) -> str:
return "/" + "/".join([ name.strip("/"), route.strip("/") ]).strip("/")
if func is None:
return {}
result = func(payload) or {}
return result if isinstance(result, dict) else {}
def start(self, host: str = "127.0.0.1", port: int = 5000, debug: bool = False, **kwargs: Any) -> None:
"""Start the Flask development server."""