112 lines
4.1 KiB
Python
112 lines
4.1 KiB
Python
"""Public client entry points."""
|
|
|
|
from typing import Any, Generator, List, Optional
|
|
|
|
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"]
|