Refactor PyClient module
parent
78c461d9b2
commit
04e21632ea
32
client.py
32
client.py
|
|
@ -21,8 +21,9 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import pyclient
|
||||
|
||||
from pyclient import PyClient
|
||||
from pyclient import Agent, PyClient
|
||||
from pyclient.games.tic_tac_toe import TicTacToe
|
||||
from typing import Any
|
||||
|
||||
|
|
@ -33,23 +34,28 @@ def main() -> int:
|
|||
:return: Exit code
|
||||
:rtype: int
|
||||
"""
|
||||
c = PyClient([
|
||||
"https://bmt001.noordstar.me",
|
||||
"https://bmt001.noordstar.me",
|
||||
], debug=False)
|
||||
c = PyClient(debug=False)
|
||||
|
||||
out = c.play_game("tic-tac-toe", TicTacToe)
|
||||
players : list[Agent] = [
|
||||
Agent.from_url(url="https://bmt001.noordstar.me/"),
|
||||
Agent.from_url(url="https://bmt001.noordstar.me/"),
|
||||
]
|
||||
|
||||
out = c.play_game(
|
||||
players=players,
|
||||
start=TicTacToe.empty(),
|
||||
)
|
||||
|
||||
inspect_game(out)
|
||||
|
||||
return 0
|
||||
|
||||
def inspect_game(game : list[tuple[int, dict[str, Any], Any]]) -> None:
|
||||
def inspect_game(game : pyclient.GameReplay) -> None:
|
||||
"""
|
||||
Print a diagnostic of a played game to the terminal.
|
||||
|
||||
:param game: The results of a played game.
|
||||
:type game: list[typle[int, dict[str, Any], Any]]
|
||||
:type game: pyclient.GameReplay
|
||||
"""
|
||||
max_width = 40
|
||||
hbar = max_width * '%'
|
||||
|
|
@ -69,19 +75,19 @@ def inspect_game(game : list[tuple[int, dict[str, Any], Any]]) -> None:
|
|||
|
||||
# Show all moves made throughout the game
|
||||
title("Turns taken")
|
||||
for player, action, _ in game:
|
||||
print(f"Player {player} : " + json.dumps(action)[:max_width-12])
|
||||
for turn in game.turns:
|
||||
print(f"Player {turn.player} : " + json.dumps(turn.action)[:max_width-12])
|
||||
|
||||
# Show all remaining variables in the finishing state of the game
|
||||
title("Final state")
|
||||
final_state = game[-1][2]
|
||||
final_state = game.turns[-1].state
|
||||
for k, v in final_state.to_dict().items():
|
||||
print(f"{k} => {json.dumps(v)}")
|
||||
|
||||
# Some final (usually the most relevant) statistics
|
||||
title("Result")
|
||||
print(f"Total turns taken: {len(game)}")
|
||||
print("Winner: " + ("0 (tie)" if final_state.winner() == 0 else f"player {final_state.winner()}"))
|
||||
print(f"Total turns taken: {len(game.turns)}")
|
||||
print(f"Result: {final_state.winner()}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
|
|
|
|||
|
|
@ -1,111 +1,16 @@
|
|||
"""Public client entry points."""
|
||||
"""
|
||||
Entry points for developers who wish to use the PyClient module.
|
||||
"""
|
||||
|
||||
from typing import Any, Generator, List, Optional
|
||||
from .agent import Agent, ServerAgent
|
||||
from .client import PyClient
|
||||
from .replay import GameReplay
|
||||
from .transition import Transition
|
||||
|
||||
import requests
|
||||
|
||||
from .poll import ServerAgent
|
||||
|
||||
|
||||
class PyClient:
|
||||
"""Host games between discovered server agents."""
|
||||
|
||||
def __init__(self, hosts: List[str], debug : bool = False) -> None:
|
||||
"""
|
||||
Create a PyClient.
|
||||
|
||||
:param hosts: URLs of servers that can participate.
|
||||
:type hosts: List[str]
|
||||
:raises ValueError: If no reachable servers are provided.
|
||||
"""
|
||||
self.agents: List[ServerAgent] = []
|
||||
self.debug = debug
|
||||
self.hosts: List[str] = []
|
||||
|
||||
for host in hosts:
|
||||
agent = self.__discover_host(host)
|
||||
if agent is not None:
|
||||
self.agents.append(agent)
|
||||
self.hosts.append(agent.url)
|
||||
|
||||
if len(self.hosts) <= 0:
|
||||
raise ValueError(
|
||||
"No valid hosts found! Check your internet connection or verify the URLs are correct."
|
||||
)
|
||||
|
||||
def __discover_host(self, url: str) -> Optional[ServerAgent]:
|
||||
try:
|
||||
return ServerAgent.from_url(url, debug=self.debug)
|
||||
except (requests.exceptions.RequestException, ValueError):
|
||||
return None
|
||||
|
||||
def gen_game(self, name : str, game : Any, urls : list[str], timeout : float = 1.0, move_default_if_nonexistent : bool = False) -> Generator[tuple[int, dict[str, Any], Any], None, None]:
|
||||
"""
|
||||
Play game. Return the results as they've been calculated.
|
||||
|
||||
:param name: Unique identifier string name for the game
|
||||
:type name: str
|
||||
:param game: Game to be played
|
||||
:type game: Any
|
||||
:param urls: List of URLs where players are located
|
||||
:type urls: list[str]
|
||||
:param timeout: Maximum time in seconds to wait for a player to respond
|
||||
:type timeout: float
|
||||
:param move_default_if_nonexistent: If a required player doesn't exist, raise an error. When this is set to true, however, the program will instead assume a player that always takes the default option.
|
||||
:type move_default_if_nonexistent: bool
|
||||
:return: Generator of each turn consisting of a player, what action it took and what that resulted in.
|
||||
:rtype: Generator[tuple[int, dict[str, Any], Any], None, None]
|
||||
:raises KeyError: The number of URLs provided is too low, the game requires more players.
|
||||
"""
|
||||
agents = [
|
||||
ServerAgent(url=url, name="", games={}, debug=self.debug)
|
||||
for url in urls
|
||||
]
|
||||
|
||||
current_state = game.empty()
|
||||
|
||||
while current_state.winner() is None:
|
||||
player : int = current_state.player_to_move()
|
||||
|
||||
if len(agents) < player:
|
||||
if not move_default_if_nonexistent:
|
||||
raise KeyError(
|
||||
f"Game requires at least {player} players to exist, found only {len(agents)}"
|
||||
)
|
||||
|
||||
current_state = current_state.move_default()
|
||||
|
||||
yield player, {}, current_state
|
||||
|
||||
else:
|
||||
agent = agents[player-1]
|
||||
payload = agent.poll(
|
||||
game=(name + "/" + current_state.action_name()).strip("/"),
|
||||
payload=current_state.as_seen_by(player),
|
||||
timeout=timeout,
|
||||
)
|
||||
|
||||
# Calculate move
|
||||
current_state = current_state.move(payload)
|
||||
|
||||
yield player, payload or {}, current_state
|
||||
|
||||
def play_game(self, name : str, game, timeout : float = 1.0) -> list[tuple[int, dict[str, Any], Any]]:
|
||||
"""
|
||||
Play a given game. Ask the registered agents to participate.
|
||||
"""
|
||||
urls = [ agent.url for agent in self.agents if name in agent.games ]
|
||||
|
||||
return list(self.gen_game(
|
||||
name=name, game=game, urls=urls, timeout=timeout,
|
||||
move_default_if_nonexistent=False,
|
||||
))
|
||||
|
||||
def verify_host(self, url: str) -> bool:
|
||||
"""
|
||||
Verify whether the URL seems to contain a link to a playable server.
|
||||
"""
|
||||
return self.__discover_host(url) is not None
|
||||
|
||||
|
||||
__all__ = ["PyClient", "ServerAgent"]
|
||||
__all__ = [
|
||||
"Agent",
|
||||
"GameReplay",
|
||||
"PyClient",
|
||||
"ServerAgent",
|
||||
"Transition",
|
||||
]
|
||||
|
|
|
|||
|
|
@ -0,0 +1,156 @@
|
|||
"""
|
||||
This module hosts various agents that can participate in games.
|
||||
|
||||
Examples of possible agents could be:
|
||||
|
||||
- An online server
|
||||
- A locally running neural network
|
||||
- A hacker who participates from the terminal
|
||||
- A user interface that allows users to play against their own creations
|
||||
"""
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Dict, Optional, Tuple, Union
|
||||
|
||||
import requests
|
||||
import time
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Agent:
|
||||
"""
|
||||
Base class of a game participant. Mostly used to inherit from.
|
||||
"""
|
||||
|
||||
debug : bool
|
||||
games : dict[str, dict[str, Any]]
|
||||
profile : dict[str, Any]
|
||||
|
||||
@classmethod
|
||||
def from_url(cls, url : str, **kwargs) -> "ServerAgent":
|
||||
"""
|
||||
Create an agent based on a URL.
|
||||
|
||||
:param url: The URL where the agent can be accessed.
|
||||
:type url: str
|
||||
:return: An agent that contacts a server when polled.
|
||||
:rtype: ServerAgent
|
||||
"""
|
||||
return ServerAgent.from_server_url(url=url, **kwargs)
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
"""
|
||||
The name by which the agent calls itself.
|
||||
"""
|
||||
return str(self.profile.get("name", "Nameless agent"))
|
||||
|
||||
def poll(
|
||||
self,
|
||||
game : str,
|
||||
payload : dict[str, Any],
|
||||
**kwargs
|
||||
) -> Optional[dict[str, Any]]:
|
||||
"""
|
||||
Ask the agent to make a move.
|
||||
|
||||
:param game: The game the ServerAgent is asked to play.
|
||||
:type game: str
|
||||
:param payload: The JSON payload that represents the game's state.
|
||||
:type payload: dict[str, Any]
|
||||
:return: The agent's response, or None if the agent doesn't respond.
|
||||
:rtype: Optional[dict[str, Any]]
|
||||
"""
|
||||
print(f"WARNING: The `poll` method is not defined on the {self.__class__.__name__} class")
|
||||
return None
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ServerAgent(Agent):
|
||||
"""
|
||||
Agent that reaches out to the internet to poll for moves.
|
||||
"""
|
||||
|
||||
url : str
|
||||
|
||||
@classmethod
|
||||
def from_server_url(
|
||||
cls,
|
||||
url: str,
|
||||
timeout: float | tuple[float, float] | None = 10.0,
|
||||
debug : bool = False
|
||||
) -> "ServerAgent":
|
||||
"""
|
||||
Create a server agent by polling its discovery endpoint.
|
||||
|
||||
:param url: The URL that the server can be reached at.
|
||||
:type url: str
|
||||
:param timeout: Request timeout.
|
||||
:type timeout: float | tuple[float, float] | None
|
||||
:param debug: Enables debug mode
|
||||
:type debug: bool
|
||||
:return: The server's representation as an agent.
|
||||
:rtype: ServerAgent
|
||||
:raises requests.exceptions.HTTPError: If the server returns a
|
||||
non-success HTTP status code.
|
||||
:raises requests.exceptions.RequestException: If the request fails
|
||||
before a response is received.
|
||||
:raises ValueError: If the response body is not a JSON object or if the
|
||||
payload contains malformed discovery fields.
|
||||
"""
|
||||
response = requests.get(url.rstrip("/") + "/", timeout=timeout)
|
||||
response.raise_for_status()
|
||||
content = response.json()
|
||||
|
||||
if not isinstance(content, dict):
|
||||
raise ValueError("Server discovery responses must be JSON objects.")
|
||||
|
||||
raw_name = content.get("name", "")
|
||||
name = "" if raw_name is None else str(raw_name)
|
||||
|
||||
games: dict[str, dict[str, Any]] = {}
|
||||
raw_games = content.get("games", {})
|
||||
if raw_games is not None:
|
||||
if not isinstance(raw_games, dict):
|
||||
raise ValueError("The 'games' field must be a JSON object when provided.")
|
||||
|
||||
for game_name, profile in raw_games.items():
|
||||
if isinstance(profile, dict):
|
||||
games[str(game_name)] = profile
|
||||
|
||||
return cls(
|
||||
debug=debug,
|
||||
games=games,
|
||||
profile=content,
|
||||
url=url,
|
||||
)
|
||||
|
||||
def poll(
|
||||
self,
|
||||
game : str,
|
||||
payload : dict[str, Any],
|
||||
**kwargs
|
||||
) -> Optional[dict[str, Any]]:
|
||||
"""
|
||||
Inquire a game to make a move.
|
||||
|
||||
:param game: The game the ServerAgent is asked to play.
|
||||
:type game: str
|
||||
:param payload: The JSON payload that represents the game's state.
|
||||
:type payload: dict[str, Any]
|
||||
:return: The server's response, or None if the server did not respond.
|
||||
:rtype: Optional[dict[str, Any]]
|
||||
"""
|
||||
url = f"{self.url.rstrip('/')}/{game.lstrip('/')}"
|
||||
timeout = float(kwargs.get("timeout", 1.0))
|
||||
|
||||
try:
|
||||
response = requests.get(url, json=payload, timeout=timeout)
|
||||
response.raise_for_status()
|
||||
content = response.json()
|
||||
except (requests.exceptions.RequestException, ValueError):
|
||||
return None
|
||||
|
||||
if self.debug:
|
||||
print(f"[DBG] Agent `{self.name}` returned:")
|
||||
print(content)
|
||||
|
||||
return content if isinstance(content, dict) else None
|
||||
|
|
@ -0,0 +1,71 @@
|
|||
"""
|
||||
This module hosts the PyClient class. You can use this class to simulate
|
||||
games among mulltiple agents.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from .agent import Agent
|
||||
from .games import Game
|
||||
from .replay import GameReplay, Turn
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Generator
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class PyClient:
|
||||
"""
|
||||
Host games among multiple agents.
|
||||
"""
|
||||
|
||||
debug : bool
|
||||
|
||||
def gen_game(self, players : list[Agent], start : Game) -> Generator[Turn, None, None]:
|
||||
"""
|
||||
Generate a game by polling the players.
|
||||
|
||||
:param players: All players that wish to participate.
|
||||
:type players: list[Agent]
|
||||
:param start: The start state of the game.
|
||||
:type start: Game
|
||||
:return: A generator that yields turns.
|
||||
:rtype: Generator[Turn, None, None]
|
||||
"""
|
||||
current_state = start
|
||||
|
||||
while current_state.winner() is None:
|
||||
player = current_state.player_to_move()
|
||||
|
||||
if len(players) < player:
|
||||
# Player not found! Make a default move.
|
||||
current_state = current_state.move_default()
|
||||
|
||||
yield Turn(action={}, player=player, state=current_state)
|
||||
|
||||
else:
|
||||
agent = players[player - 1]
|
||||
payload = agent.poll(
|
||||
game=current_state.game_name(),
|
||||
payload=current_state.as_seen_by(player=player),
|
||||
)
|
||||
|
||||
# Calculate move
|
||||
current_state = current_state.move(payload=payload)
|
||||
|
||||
yield Turn(action=payload, player=player, state=current_state)
|
||||
|
||||
def play_game(self, players : list[Agent], start : Game) -> GameReplay:
|
||||
"""
|
||||
Generate a game by polling the players. Collect all moves in a
|
||||
summary.
|
||||
|
||||
:param players: All players that wish to participate.
|
||||
:type players: list[Agent]
|
||||
:param start: The start state of the game.
|
||||
:type start: Game
|
||||
:return: Summary describing how the game went.
|
||||
:rtype: GameReplay
|
||||
"""
|
||||
return GameReplay(
|
||||
start=start,
|
||||
turns=list(self.gen_game(players=players, start=start)),
|
||||
)
|
||||
|
|
@ -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",
|
||||
]
|
||||
|
|
@ -0,0 +1,138 @@
|
|||
"""
|
||||
This module contains the base class for any game.
|
||||
|
||||
If you'd like to create a new game, please copy this module and fill in
|
||||
all the necessary methods in order to have a functioning game.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from enum import Enum, auto
|
||||
from typing import Any
|
||||
|
||||
class FinishState(Enum):
|
||||
draw = auto()
|
||||
loss = auto()
|
||||
win = auto()
|
||||
|
||||
class Game:
|
||||
"""
|
||||
Base class for all games.
|
||||
|
||||
A game is always a snapshot of a game that is paused because it waits
|
||||
for a user to make a move.
|
||||
"""
|
||||
|
||||
def action_name(self) -> str:
|
||||
"""
|
||||
Return the type of action to take. Please only use alphanumeric
|
||||
characters.
|
||||
|
||||
This helps the player understand what they are supposed to do.
|
||||
For example, in Risk, you can use this to explain whether they are
|
||||
expected to "recruit" or "attack" or "move" with their troops.
|
||||
|
||||
If all moves require the same action (such as with tic-tac-toe),
|
||||
you can leave this to return an empty string.
|
||||
"""
|
||||
return ""
|
||||
|
||||
def as_seen_by(self, player : int) -> dict[str, Any]:
|
||||
"""
|
||||
From the perspective of a given player, return the game's state.
|
||||
|
||||
Note that it is very common in games for players to not have ALL
|
||||
the details available. For example, in card games, you are
|
||||
generally unable to see the cards in your opponents' hands.
|
||||
|
||||
The game's state is formatted in a way that makes it easy to
|
||||
convert it to JSON or a tensor.
|
||||
|
||||
:param player: Player. Counting starts from one.
|
||||
:type player: int
|
||||
:return: The game's state from the perspective of the player.
|
||||
:rtype: dict[str, Any]
|
||||
"""
|
||||
return {}
|
||||
|
||||
@classmethod
|
||||
def empty(cls) -> "Game":
|
||||
"""
|
||||
Create a new game.
|
||||
|
||||
Use this method to shuffle the cards, to arrange pieces on the
|
||||
board, and set up the state in a way where the first player can
|
||||
start to make its move.
|
||||
|
||||
:return: A game that's ready to start.
|
||||
:rtype: Game
|
||||
"""
|
||||
return cls()
|
||||
|
||||
def game_name(self) -> str:
|
||||
"""
|
||||
Return a non-empty string that uniquely identifies the game.
|
||||
When creating a new unspecified game, please respect the Java
|
||||
package naming convention.
|
||||
"""
|
||||
return "default-game"
|
||||
|
||||
def move_default(self) -> "Game":
|
||||
"""
|
||||
Fallback move option for when a user isn't available or cannot
|
||||
decide. Program this move to be predictable, and resemble an
|
||||
attempted skip as much as possible. For example:"
|
||||
|
||||
- In Go, place a tile in an empty space near the top-left corner
|
||||
- In Monopoly, choose not to buy anything
|
||||
- In poker, check. (Or fold after a raise.)
|
||||
- In a maze, choose the left-most path
|
||||
"""
|
||||
# NOTE: Games are meant to be immutable values!
|
||||
# As such, you should always consider creating a deep copy of oneself,
|
||||
# do not simply return self if it might update.
|
||||
return self
|
||||
|
||||
def move(self, payload : dict[str, Any] | None = None) -> "Game":
|
||||
"""
|
||||
Make a move on behalf of whose turn it is.
|
||||
|
||||
:param payload: Dictionary containing the player's response.
|
||||
:type payload: dict[str, Any] | None
|
||||
:return: A new instance of the game where a new action is required.
|
||||
:rtype: Game
|
||||
"""
|
||||
if not isinstance(payload, dict):
|
||||
return self.move_default()
|
||||
|
||||
# NOTE: Games are meant to be immutable values!
|
||||
# As such, you should always consider creating a deep copy of oneself,
|
||||
# do not simply return self if it might update.
|
||||
return self
|
||||
|
||||
def player_to_move(self) -> int:
|
||||
"""
|
||||
Return which player is currently meant to move.
|
||||
|
||||
:return: Which player to move. Counting starts from one.
|
||||
:rtype: int
|
||||
"""
|
||||
return 1
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
"""
|
||||
Return a dictionary containing the full game state.
|
||||
|
||||
:return: The current game's state.
|
||||
:rtype: dict[str, Any]
|
||||
"""
|
||||
return {}
|
||||
|
||||
def winner(self) -> dict[int, FinishState] | None:
|
||||
"""
|
||||
Determine whether the game has ended.
|
||||
|
||||
:return: A list detailing which players have won, lost or drawn - or None if the game hasn't finished.
|
||||
:rtype: dict[int, FinishState] | None
|
||||
"""
|
||||
return {}
|
||||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
102
pyclient/poll.py
102
pyclient/poll.py
|
|
@ -1,102 +0,0 @@
|
|||
"""Discovery and polling helpers for Bot-Man-Toe servers."""
|
||||
|
||||
from typing import Any, Dict, Optional, Tuple, Union
|
||||
|
||||
import requests
|
||||
import time
|
||||
|
||||
class ServerAgent:
|
||||
"""Representation of a server that can host one or more games."""
|
||||
|
||||
def __init__(self, url: str, name: str, games: Dict[str, Dict[str, Any]], debug : bool = False) -> None:
|
||||
"""
|
||||
Create a server representation.
|
||||
|
||||
:param url: The URL used to discover the server.
|
||||
:type url: str
|
||||
:param name: Name of the server's player.
|
||||
:type name: str
|
||||
:param games: Games the server is willing to play.
|
||||
:type games: Dict[str, Dict[str, Any]]
|
||||
"""
|
||||
self.debug = debug
|
||||
self.games = games
|
||||
self.name = name
|
||||
self.url = url.strip("/")
|
||||
|
||||
@classmethod
|
||||
def from_url(
|
||||
cls,
|
||||
url: str,
|
||||
timeout: Optional[Union[float, Tuple[float, float]]] = 10.0,
|
||||
debug : bool = False
|
||||
) -> "ServerAgent":
|
||||
"""
|
||||
Create a server agent by polling its discovery endpoint.
|
||||
|
||||
The root endpoint is expected to return a JSON object with an optional
|
||||
``name`` field and an optional ``games`` mapping.
|
||||
|
||||
:param url: The URL that the server can be reached at.
|
||||
:type url: str
|
||||
:param timeout: Request timeout passed to ``requests.get``.
|
||||
:type timeout: Optional[Union[float, Tuple[float, float]]]
|
||||
:return: The server's representation.
|
||||
:rtype: ServerAgent
|
||||
:raises requests.exceptions.HTTPError: If the server returns a
|
||||
non-success HTTP status code.
|
||||
:raises requests.exceptions.RequestException: If the request fails
|
||||
before a response is received.
|
||||
:raises ValueError: If the response body is not a JSON object or if the
|
||||
payload contains malformed discovery fields.
|
||||
"""
|
||||
response = requests.get(url.rstrip("/") + "/", timeout=timeout)
|
||||
response.raise_for_status()
|
||||
content = response.json()
|
||||
|
||||
if not isinstance(content, dict):
|
||||
raise ValueError("Server discovery responses must be JSON objects.")
|
||||
|
||||
raw_name = content.get("name", "")
|
||||
name = "" if raw_name is None else str(raw_name)
|
||||
|
||||
games: Dict[str, Dict[str, Any]] = {}
|
||||
raw_games = content.get("games", {})
|
||||
if raw_games is not None:
|
||||
if not isinstance(raw_games, dict):
|
||||
raise ValueError("The 'games' field must be a JSON object when provided.")
|
||||
|
||||
for game_name, profile in raw_games.items():
|
||||
if isinstance(profile, dict):
|
||||
games[str(game_name)] = profile
|
||||
|
||||
return cls(url=url, name=name, games=games, debug=debug)
|
||||
|
||||
def poll(self, game: str, payload: Dict[str, Any], timeout: float = 1.0) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
Inquire a game to make a move.
|
||||
|
||||
:param game: The game the ServerAgent is asked to play.
|
||||
:type game: str
|
||||
:param payload: The JSON payload that represents the game's state.
|
||||
:type payload: Dict[str, Any]
|
||||
:param timeout: Maximum number of seconds to wait for a move.
|
||||
:type timeout: float
|
||||
:return: The server's response, or None if the server did not respond
|
||||
in time or returned an invalid response.
|
||||
:rtype: Optional[Dict[str, Any]]
|
||||
"""
|
||||
url = f"{self.url.rstrip('/')}/{game.lstrip('/')}"
|
||||
|
||||
try:
|
||||
response = requests.get(url, json=payload, timeout=timeout)
|
||||
response.raise_for_status()
|
||||
content = response.json()
|
||||
except (requests.exceptions.RequestException, ValueError):
|
||||
return None
|
||||
|
||||
if self.debug:
|
||||
print(f"[DBG] Agent `{self.name}` returned:")
|
||||
print(content)
|
||||
|
||||
return content if isinstance(content, dict) else None
|
||||
|
|
@ -0,0 +1,92 @@
|
|||
"""
|
||||
This module helps describe & evaluate previously played games.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from .games import Game
|
||||
from .transition import Transition
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Generator
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class GameReplay:
|
||||
"""
|
||||
Game replay detailing the events of what happened in a previously
|
||||
played game.
|
||||
"""
|
||||
|
||||
start : Game
|
||||
turns : list[Turn]
|
||||
|
||||
def gen_transitions(self) -> Generator[Transition, None, None]:
|
||||
"""
|
||||
Generate transitions from the game's summary.
|
||||
|
||||
:return: Transitions that describe consequences throughout the game.
|
||||
:rtype: Generator[Transition, None, None]
|
||||
"""
|
||||
prev_state : Game = self.start
|
||||
seen_participants : dict[int, tuple[Game, dict[str, Any]]] = {}
|
||||
|
||||
for turn in self.turns:
|
||||
# If a player has made a move before, show the current state as
|
||||
# the result of their previous turn's action.
|
||||
state, action = seen_participants.get(turn.player, ( None, {} ))
|
||||
if state is not None:
|
||||
yield Transition(
|
||||
full_state_after=prev_state,
|
||||
full_state_before=state,
|
||||
move=action,
|
||||
player=turn.player,
|
||||
)
|
||||
|
||||
# Save the player's current action for the future.
|
||||
# We'll see how this action worked out!
|
||||
seen_participants[turn.player] = ( prev_state, turn.action )
|
||||
prev_state = turn.state
|
||||
|
||||
else:
|
||||
# Record to all players that the game has finished.
|
||||
for player, ( state, action ) in seen_participants.items():
|
||||
yield Transition(
|
||||
full_state_after=prev_state,
|
||||
full_state_before=state,
|
||||
move=action,
|
||||
player=player,
|
||||
)
|
||||
|
||||
def to_transitions(self, player : int | None = None) -> list[Transition]:
|
||||
"""
|
||||
Convert the GameSummary into a list of transitions that can be
|
||||
fed to a system that can reflect on past moves.
|
||||
|
||||
:param player: Filter to only show transitions of a given player.
|
||||
:type player: int | None
|
||||
:return: A list of transitions extracted from a game.
|
||||
:rtype: list[Transition]
|
||||
"""
|
||||
if player is None:
|
||||
return list(self.gen_transitions())
|
||||
else:
|
||||
return [ t for t in self.gen_transitions() if t.player == player ]
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Turn:
|
||||
"""
|
||||
A turn is a snapshot of a moment where a user was prompted to take an
|
||||
action. A turn consists of three values:
|
||||
|
||||
1. The player (number) that was prompted for an action
|
||||
2. The action that that player has decided to take
|
||||
3. The game's state after the action was prompted and before the next
|
||||
prompted move.
|
||||
|
||||
Note that turns might not correlate with in-game turns. A user might
|
||||
take multiple actions in a Risk game (recruit, attack, move) and each
|
||||
action is registered as an independent move.
|
||||
"""
|
||||
|
||||
player : int
|
||||
action : dict[str, Any]
|
||||
state : Game
|
||||
|
|
@ -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)
|
||||
Loading…
Reference in New Issue