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

Introduce maximum number of turns

bram/mute
Bram van den Heuvel 2026-07-01 22:45:13 +02:00
parent d7ac20a6dd
commit 194f962746
3 changed files with 77 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
elo.py
View File

@ -24,6 +24,7 @@ def main() -> int:
tracker.start_periodic_matches(
game=RoyalGameOfUr.empty(),
interval_seconds=9 * 60,
max_turns=1_500,
player_count=2,
)

67
elo_tracker/app.py
View File

@ -14,7 +14,7 @@ from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any, Sequence
import agents
import agents.remote
import pyclient
import time
@ -39,7 +39,7 @@ class PlayerIdentifier:
return (self.name, self.url, self.version)
@classmethod
def from_server_agent(cls, agent : agents.RemoteAgent) -> "PlayerIdentifier":
def from_server_agent(cls, agent : agents.remote.RemoteAgent) -> "PlayerIdentifier":
"""
Gain a player identifier from an agent.
"""
@ -177,7 +177,7 @@ class Match:
@classmethod
def from_replay(
cls,
players : list[agents.RemoteAgent],
players : list[agents.remote.RemoteAgent],
replay : pyclient.GameReplay,
timestamp : str | None,
) -> "Match":
@ -291,7 +291,7 @@ class EloTracker:
# Thread-unsafe variables
# Please use a lock while doing CRUD operations on them
self.players: list[agents.RemoteAgent] = []
self.players: list[agents.remote.RemoteAgent] = []
self.__matches: list[Match] = []
self.__stats: dict[str, dict[PlayerIdentifier, EloStat]] = {}
@ -436,6 +436,7 @@ class EloTracker:
game: Game,
interval_seconds: float,
player_count: int,
max_turns : int | None = None,
) -> None:
"""
Perform a schedule in which you play a random game.
@ -446,6 +447,8 @@ class EloTracker:
:type interval_seconds: float
:param player_count: The number of players that are supposed to participate
:type player_count: int
:param max_turns: Optional maximum number of turns to allow
:type max_turns: int | None
"""
while not self.__scheduler_stop.is_set():
try:
@ -468,7 +471,11 @@ class EloTracker:
self.__debug(
f"Playing a new scheduled match of {game.game_name()}"
)
self.play_random_match(game=game, player_count=player_count)
self.play_random_match(
game=game,
max_turns=max_turns,
player_count=player_count,
)
except Exception as exc:
self.__debug(f"Scheduled match failed: {exc}")
# raise exc
@ -582,13 +589,13 @@ class EloTracker:
"Expected `players` field to be a list of strings."
)
players : list[agents.RemoteAgent] = []
players : list[agents.remote.RemoteAgent] = []
for url in urls:
if not isinstance(url, str):
continue
try:
agent = agents.RemoteAgent(url=url)
agent = agents.remote.RemoteAgent(url=url)
except ValueError:
pass # Not an available player right now
else:
@ -604,41 +611,59 @@ class EloTracker:
player_id=PlayerIdentifier.from_server_agent(agent)
)
def play_match(self, players: list[str], game: Game) -> pyclient.GameReplay:
def play_match(
self,
players: list[str],
game: Game,
max_turns : int | None = None,
) -> pyclient.GameReplay | None:
"""
Play a single match with appointed players.
:param players: List of URLs that participate.
:type players: list[str]
:param game: The game to start playing
:type game: Game
:param max_turns: Optional maximum number of turns to allow
:type max_turns: int | None
:return: A summary of the game.
:rtype: pyclient.GameReplay
:raises ValueError: One of the URLs could not be accessed.
"""
ags : list[Any] = [ agents.RemoteAgent(url=url) for url in players ]
ags : list[Any] = [ agents.remote.RemoteAgent(url=url) for url in players ]
replay = pyclient.PyClient(debug=self.debug).play_game(
players=ags,
start=game,
)
try:
replay = pyclient.PyClient(debug=self.debug).play_game(
players=ags,
start=game,
max_turns=max_turns,
)
m = Match.from_replay(players=ags, replay=replay, timestamp=Match.now())
except ValueError:
return None # Exceeded turn limit
else:
m = Match.from_replay(players=ags, replay=replay, timestamp=Match.now())
# Record match
m.log(self.game_file_name)
self.__register_match(m)
# Record match
m.log(self.game_file_name)
self.__register_match(m)
return replay
return replay
def play_random_match(
self,
game: Game,
max_turns : int | None = None,
player_count: int | None = None,
) -> pyclient.GameReplay:
) -> pyclient.GameReplay | None:
"""
Play a game with any known players.
:param game: The game to start playing
:type game: Game
:param max_turns: Optional maximum number of turns to allow
:type max_turns: int | None
:param player_count: Optional number of players to select.
:type player_count: int | None
:raises ValueError: One of the randomly chosen URLs could not be accessed.
@ -657,6 +682,7 @@ class EloTracker:
self,
game: Game,
interval_seconds: float = 300,
max_turns : int | None = None,
player_count: int = 2,
) -> None:
"""
@ -666,6 +692,8 @@ class EloTracker:
:type game: pyclient.Game
:param interval_seconds: Number of seconds to sleep between games
:type interval_seconds: float
:param max_turns: Optional maximum number of turns to allow
:type max_turns: int | None
:param player_count: The number of players that are supposed to participate
:type player_count: int
"""
@ -679,6 +707,7 @@ class EloTracker:
thread = threading.Thread(
target=self.__scheduler_loop,
args=(game, interval_seconds, player_count),
kwargs={"max_turns": max_turns},
daemon=True,
)
self.__scheduler_threads.append(thread)

31
pyclient/client.py
View File

@ -19,7 +19,12 @@ class PyClient:
debug : bool
def gen_game(self, players : list[Agent], start : Game) -> Generator[Turn, None, None]:
def gen_game(
self,
players : list[Agent],
start : Game,
max_turns : int | None = None,
) -> Generator[Turn, None, None]:
"""
Generate a game by polling the players.
@ -27,12 +32,22 @@ class PyClient:
:type players: list[Agent]
:param start: The start state of the game.
:type start: Game
:param max_turns: The maximum number of turns a game may take
:type max_turns: int
:return: A generator that yields turns.
:rtype: Generator[Turn, None, None]
:raises ValueError: The game is taking too long.
"""
current_state = start
turns_taken = 0
while current_state.winner() is None:
if max_turns is not None and turns_taken >= max_turns:
raise ValueError(
f"Turn limit exceeded: played for {turns_taken} turns and still needed more."
)
turns_taken += 1
player = current_state.player_to_move()
if len(players) < player:
@ -61,7 +76,12 @@ class PyClient:
yield Turn(action=payload, player=player, state=current_state)
def play_game(self, players : list[Agent], start : Game) -> GameReplay:
def play_game(
self,
players : list[Agent],
start : Game,
max_turns : int | None = None,
) -> GameReplay:
"""
Generate a game by polling the players. Collect all moves in a
summary.
@ -70,11 +90,16 @@ class PyClient:
:type players: list[Agent]
:param start: The start state of the game.
:type start: Game
:param max_turns: The maximum number of turns a game may take
:type max_turns: int
:return: Summary describing how the game went.
:rtype: GameReplay
:raises ValueError: The game is taking too long.
"""
return GameReplay(
game_name=start.game_name(),
start=start,
turns=list(self.gen_game(players=players, start=start)),
turns=list(self.gen_game(
players=players, start=start, max_turns=max_turns
)),
)