171 lines
5.3 KiB
Python
171 lines
5.3 KiB
Python
"""
|
|
This module hosts various agents that can participate in games.
|
|
|
|
Examples of possible agents could be:
|
|
|
|
- An online server
|
|
- A locally running neural network
|
|
- A hacker who participates from the terminal
|
|
- A user interface that allows users to play against their own creations
|
|
"""
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any, Dict, Optional, Tuple, Union
|
|
|
|
import requests
|
|
import time
|
|
|
|
@dataclass(frozen=True)
|
|
class Agent:
|
|
"""
|
|
Base class of a game participant. Mostly used to inherit from.
|
|
"""
|
|
|
|
debug : bool
|
|
games : dict[str, dict[str, Any]]
|
|
profile : dict[str, Any]
|
|
|
|
@classmethod
|
|
def from_url(cls, url : str, **kwargs) -> "ServerAgent":
|
|
"""
|
|
Create an agent based on a URL.
|
|
|
|
:param url: The URL where the agent can be accessed.
|
|
:type url: str
|
|
:return: An agent that contacts a server when polled.
|
|
:rtype: ServerAgent
|
|
"""
|
|
return ServerAgent.from_server_url(url=url, **kwargs)
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
"""
|
|
The name by which the agent calls itself.
|
|
"""
|
|
return str(self.profile.get("name", "Nameless agent"))
|
|
|
|
def poll(
|
|
self,
|
|
game : str,
|
|
payload : dict[str, Any],
|
|
**kwargs
|
|
) -> Optional[dict[str, Any]]:
|
|
"""
|
|
Ask the agent to make a move.
|
|
|
|
:param game: The game the ServerAgent is asked to play.
|
|
:type game: str
|
|
:param payload: The JSON payload that represents the game's state.
|
|
:type payload: dict[str, Any]
|
|
:return: The agent's response, or None if the agent doesn't respond.
|
|
:rtype: Optional[dict[str, Any]]
|
|
"""
|
|
print(f"WARNING: The `poll` method is not defined on the {self.__class__.__name__} class")
|
|
return None
|
|
|
|
@dataclass(frozen=True)
|
|
class ServerAgent(Agent):
|
|
"""
|
|
Agent that reaches out to the internet to poll for moves.
|
|
"""
|
|
|
|
url : str
|
|
|
|
@classmethod
|
|
def from_server_url(
|
|
cls,
|
|
url: str,
|
|
timeout: float | tuple[float, float] | None = 10.0,
|
|
debug : bool = False
|
|
) -> "ServerAgent":
|
|
"""
|
|
Create a server agent by polling its discovery endpoint.
|
|
|
|
:param url: The URL that the server can be reached at.
|
|
:type url: str
|
|
:param timeout: Request timeout.
|
|
:type timeout: float | tuple[float, float] | None
|
|
:param debug: Enables debug mode
|
|
:type debug: bool
|
|
:return: The server's representation as an agent.
|
|
:rtype: ServerAgent
|
|
:raises requests.exceptions.HTTPError: If the server returns a
|
|
non-success HTTP status code.
|
|
:raises requests.exceptions.RequestException: If the request fails
|
|
before a response is received.
|
|
:raises ValueError: If the response body is not a JSON object or if the
|
|
payload contains malformed discovery fields.
|
|
"""
|
|
response = requests.get(url.rstrip("/") + "/", timeout=timeout)
|
|
response.raise_for_status()
|
|
content = response.json()
|
|
|
|
if not isinstance(content, dict):
|
|
raise ValueError("Server discovery responses must be JSON objects.")
|
|
|
|
raw_name = content.get("name", "")
|
|
name = "" if raw_name is None else str(raw_name)
|
|
|
|
games: dict[str, dict[str, Any]] = {}
|
|
raw_games = content.get("games", {})
|
|
if raw_games is not None:
|
|
if not isinstance(raw_games, dict):
|
|
raise ValueError("The 'games' field must be a JSON object when provided.")
|
|
|
|
for game_name, profile in raw_games.items():
|
|
if isinstance(profile, dict):
|
|
games[str(game_name)] = profile
|
|
|
|
return cls(
|
|
debug=debug,
|
|
games=games,
|
|
profile=content,
|
|
url=url,
|
|
)
|
|
|
|
def poll(
|
|
self,
|
|
game : str,
|
|
payload : dict[str, Any],
|
|
**kwargs
|
|
) -> Optional[dict[str, Any]]:
|
|
"""
|
|
Inquire a game to make a move.
|
|
|
|
:param game: The game the ServerAgent is asked to play.
|
|
:type game: str
|
|
:param payload: The JSON payload that represents the game's state.
|
|
:type payload: dict[str, Any]
|
|
:return: The server's response, or None if the server did not respond.
|
|
:rtype: Optional[dict[str, Any]]
|
|
"""
|
|
url = f"{self.url.rstrip('/')}/{game.lstrip('/')}"
|
|
timeout = float(kwargs.get("timeout", 1.0))
|
|
|
|
try:
|
|
response = requests.get(url, json=payload, timeout=timeout)
|
|
response.raise_for_status()
|
|
content = response.json()
|
|
except (requests.exceptions.RequestException, ValueError):
|
|
return None
|
|
|
|
if self.debug:
|
|
print(f"[DBG] Agent `{self.name}` returned:")
|
|
print(content)
|
|
|
|
return content if isinstance(content, dict) else None
|
|
|
|
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,
|
|
)
|