110 lines
3.8 KiB
Python
110 lines
3.8 KiB
Python
"""
|
|
The remote agent is the agent that is accessible on the internet.
|
|
|
|
The common implementation of the back-end is the PyClient, although the
|
|
remote agent communicates through an API so the agent is
|
|
implementation-agnostic.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
from typing import Any, Callable
|
|
|
|
import requests
|
|
|
|
from .agent import Agent, Payload
|
|
|
|
class RemoteAgent(Agent):
|
|
"""
|
|
The RemoteAgent is the agent that is remotely accessible on an external
|
|
URL through the specified protocol.
|
|
"""
|
|
|
|
url : str
|
|
games : set[str]
|
|
|
|
def __init__(self, url : str, timeout : float = 1.0) -> None:
|
|
"""
|
|
Create a new instance of the remote agent.
|
|
|
|
:param url: URL at which the agent can be accessed.
|
|
:type url: str
|
|
:param timeout: The maximum time the client is allowed to respond.
|
|
:type timeout: float
|
|
:raises requests.HTTPError: The server could not be reached.
|
|
:raises ValueError: The server returned an invalid response.
|
|
"""
|
|
self.url = url.strip("/")
|
|
self.timeout = timeout
|
|
|
|
# Get discovery from URL
|
|
response = requests.get(self.url.strip("/") + "/", timeout=self.timeout)
|
|
response.raise_for_status()
|
|
content = response.json()
|
|
|
|
if not isinstance(content, dict):
|
|
raise ValueError(
|
|
"Remote agent's discovery response must be a JSON object."
|
|
)
|
|
|
|
raw_author = content.get("author", content.get("me.noordstar.peanuts.author", ""))
|
|
raw_name = content.get("name", "")
|
|
raw_version = content.get("version", content.get("me.noordstar.peanuts.agent.version", None))
|
|
|
|
# Initialize agent based on given information
|
|
super().__init__(
|
|
author="" if raw_author is None else str(raw_author),
|
|
name="" if raw_name is None else str(raw_name),
|
|
version=None if raw_version is None else str(raw_version),
|
|
profile=content,
|
|
)
|
|
|
|
self.games = set()
|
|
|
|
games = content.get("games")
|
|
if isinstance(games, dict):
|
|
if "tic-tac-toe" in games:
|
|
self.add_tic_tac_toe(
|
|
on_move=lambda x : self.poll("tic-tac-toe", "", x),
|
|
profile=games["tic-tac-toe"],
|
|
)
|
|
if "royal-game-of-ur" in games:
|
|
self.add_royal_game_of_ur(
|
|
on_move=lambda x : self.poll("royal-game-of-ur", "", x),
|
|
profile=games["royal-game-of-ur"],
|
|
)
|
|
|
|
self.games = set(games.keys())
|
|
|
|
def get_action(self, game: str, action: str | None) -> Callable[[dict[str, Any]], dict[str, Any]] | None:
|
|
"""
|
|
Get an action function for any given game.
|
|
|
|
:param game: The name of the game.
|
|
:type game: str
|
|
:param action: The name of the action, or None if the game only has one action.
|
|
:type action: str | None
|
|
:return: The action with which the agent would like to act, if at all.
|
|
:rtype: Callable[[dict[str, Any]], dict[str, Any]] | None
|
|
"""
|
|
return lambda x : self.poll(game=game, action=action or "", payload=x)
|
|
|
|
def poll(
|
|
self,
|
|
game : str,
|
|
action : str,
|
|
payload : Payload
|
|
) -> Payload:
|
|
"""
|
|
Inquire the remote agent for a response.
|
|
"""
|
|
url = "/".join([self.url, game.strip("/"), action.strip("/")]).strip("/")
|
|
|
|
try:
|
|
response = requests.get(url, json=payload, timeout=self.timeout)
|
|
response.raise_for_status()
|
|
content = response.json()
|
|
except (requests.ConnectTimeout, requests.RequestException, requests.HTTPError, ValueError):
|
|
return {}
|
|
else:
|
|
return content if isinstance(content, dict) else {}
|