Compare commits
1 Commits
main
...
agent/naiv
| Author | SHA1 | Date |
|---|---|---|
 Bram van den Heuvel
|
fc8375e251 |
|
|
@ -13,10 +13,6 @@ spec/
|
|||
# Compiled elm files
|
||||
elo_tracker/static/elo_tracker.js
|
||||
|
||||
# Data files
|
||||
games.jsonl
|
||||
known_players.json
|
||||
|
||||
# ---> Elm
|
||||
# elm-package generated files
|
||||
elm-stuff
|
||||
|
|
|
|||
|
|
@ -5,10 +5,6 @@
|
|||
# Compiled elm files
|
||||
elo_tracker/static/elo_tracker.js
|
||||
|
||||
# Data files
|
||||
games.jsonl
|
||||
known_players.json
|
||||
|
||||
# ---> Elm
|
||||
# elm-package generated files
|
||||
elm-stuff
|
||||
|
|
|
|||
27
README.md
27
README.md
|
|
@ -1,24 +1,15 @@
|
|||
# Bot-Man-Toe
|
||||
|
||||
Write a script that plays tic-tac-toe, and see how well it performs against
|
||||
other programs!
|
||||
Bot-Man-Toe is an attempt to create a way for players to play games against
|
||||
themselves, other players, or self-trained AI players.
|
||||
|
||||
- [🚀 Write your own agent](agents/README.md)
|
||||
- [🖊 Compare how well your agent performs](elo_tracker/README.md)
|
||||
- [🌐 Publish your agent to the internet](pyserver/README.md)
|
||||
## Technology stack
|
||||
|
||||
## Get started
|
||||
Counterintuitively, the **servers** are participants to a game. The **clients**
|
||||
are programs or browsers that mediate matches between servers.
|
||||
|
||||
1. Clone this repository.
|
||||
2. Run `python client.py` in the terminal and let two random agents play
|
||||
against each other.
|
||||
3. Copy the example agent and [create your own](agents/README.md).
|
||||
4. Play against the AgentOfChaos while you improve your strategy.
|
||||
5. Publish your agent. _(optional)_
|
||||
6. Compare it against others with the Elo tracker. _(optional)_
|
||||
## More
|
||||
|
||||
|
||||
## Links
|
||||
|
||||
- [📜 API specification](pyserver/spec.md)
|
||||
- [🏆 Online ELO tracker](https://elo.noordstar.me/)
|
||||
- The discovery contract is documented in `spec/README.md`.
|
||||
- Python client helpers live under `pyclient/`.
|
||||
- Python server helpers live under `pyserver/`.
|
||||
|
|
|
|||
|
|
@ -1,70 +0,0 @@
|
|||
# 🚀 Write your own agent!
|
||||
|
||||
An **agent** is a Python class that knows how to play one or more games.
|
||||
|
||||
Don't worry about writing a perfect strategy. Start with something that works,
|
||||
print the incoming game state, and improve it one step at a time.
|
||||
|
||||
## Quick start
|
||||
|
||||
1. Copy [example.py](example.py).
|
||||
2. Rename the file and the `ExampleAgent` class.
|
||||
3. Update the agent's name, author and version.
|
||||
4. Implement `play_tic_tac_toe()`.
|
||||
5. Import your agent in `client.py` and let it play a game.
|
||||
|
||||
That's enough to get started.
|
||||
|
||||
## Understanding the game
|
||||
|
||||
Whenever your agent has to make a move, it receives the current game state as
|
||||
a Python dictionary.
|
||||
|
||||
Start by printing it:
|
||||
|
||||
```py
|
||||
print(payload)
|
||||
```
|
||||
|
||||
Run a few games and watch how the payload changes after every move. Once you
|
||||
understand what you're receiving, you can start writing your own strategy!
|
||||
|
||||
Your function should return:
|
||||
|
||||
```py
|
||||
{"move": 7}
|
||||
```
|
||||
|
||||
where the number is the square you want to play.
|
||||
|
||||
## Testing your agent
|
||||
|
||||
Open [client.py](../client.py) and replace one of the players with your own agent.
|
||||
For example:
|
||||
|
||||
```py
|
||||
from agents.my_agent import MyAgent
|
||||
from agents.chaos import AgentOfChaos
|
||||
|
||||
players = [
|
||||
MyAgent(),
|
||||
AgentOfChaos(),
|
||||
]
|
||||
```
|
||||
|
||||
The repository includes `AgentOfChaos`, a very simple opponent that plays
|
||||
random moves. It's useful for testing your own agent while you're developing it.
|
||||
|
||||
> ⚠️ Don't try to build the perfect player immediately! Agents are easy to
|
||||
> improve while you test against other agents. Plus, imperfect agents are
|
||||
> typically the most interesting.
|
||||
|
||||
Run the client, inspect the output, tweak your algorithm, and repeat. You don't
|
||||
need to understand the rest of the project before you can start experimenting.
|
||||
|
||||
## What's next?
|
||||
|
||||
Once you're happy with your agent, you can:
|
||||
|
||||
- [🖊 Compare your agent against other agents with the ELO tracker](../elo_tracker/README.md)
|
||||
- [🌐 Publish your agent so other people can play against it](../pyserver/README.md)
|
||||
|
|
@ -115,26 +115,6 @@ class Agent:
|
|||
required_actions=[""],
|
||||
)
|
||||
|
||||
def add_royal_game_of_ur(
|
||||
self,
|
||||
on_move : ActionFunction,
|
||||
profile : Payload = {},
|
||||
) -> None:
|
||||
"""
|
||||
Convenience registration for the royal game of Ur.
|
||||
|
||||
:param on_move: Called function when the player is to make a move.
|
||||
:type on_move: Callable[[dict[str, Any]], dict[str, Any]]
|
||||
:param profile: Custom details to share about the game.
|
||||
:type profile: dict[str, Any]
|
||||
"""
|
||||
return self.add_game(
|
||||
name="royal-game-of-ur",
|
||||
actions={"": on_move},
|
||||
profile=profile,
|
||||
required_actions=[""],
|
||||
)
|
||||
|
||||
def get_action(self, game : str, action : str | None) -> ActionFunction | None:
|
||||
"""
|
||||
Get an action function for a given game, if it exists.
|
||||
|
|
|
|||
|
|
@ -32,7 +32,7 @@ class AgentOfChaos(Agent):
|
|||
|
||||
def __init__(self) -> None:
|
||||
"""
|
||||
Agent that always tries to make a "random" move.
|
||||
Create a new instance of the agent of chaos.
|
||||
"""
|
||||
super().__init__(
|
||||
name="Agent of chaos",
|
||||
|
|
@ -44,7 +44,6 @@ class AgentOfChaos(Agent):
|
|||
)
|
||||
|
||||
self.add_tic_tac_toe(on_move=play_tic_tac_toe, profile={})
|
||||
self.add_royal_game_of_ur(on_move=play_royal_game_of_ur, profile={})
|
||||
|
||||
|
||||
def play_tic_tac_toe(payload : Payload) -> Payload:
|
||||
|
|
@ -63,16 +62,3 @@ def play_tic_tac_toe(payload : Payload) -> Payload:
|
|||
if k in "0123456789" and v == ""
|
||||
]
|
||||
return { "move": random.choice(options) }
|
||||
|
||||
def play_royal_game_of_ur(payload : Payload) -> Payload:
|
||||
"""
|
||||
The royal game of Ur already returns a set of valid moves.
|
||||
|
||||
We can lazily just pick one of these options and return it.
|
||||
|
||||
:param payload: The incoming game state.
|
||||
:type payload: dict[str, Any]
|
||||
:return: The agent of chaos' random choice
|
||||
:rtype: dict[str, Any]
|
||||
"""
|
||||
return { "move": random.choice(payload.get("valid_moves", [""])) }
|
||||
|
|
|
|||
|
|
@ -1,104 +0,0 @@
|
|||
"""
|
||||
This module contains an example agent that you can use to create your own!
|
||||
|
||||
Please copy this file, rename it, and then build it the way you see fit.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import random
|
||||
|
||||
from .agent import Agent, Payload
|
||||
|
||||
class ExampleAgent(Agent):
|
||||
"""
|
||||
Describe here what your agent does and how it behaves! This will help
|
||||
others understand how your agent works.
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""
|
||||
Create a custom instance of your agent. This function allows you
|
||||
to define which games your agent can play.
|
||||
|
||||
You may add parameters to the function if your agent requires more
|
||||
information to be able to operate.
|
||||
"""
|
||||
|
||||
super().__init__(
|
||||
# Give your bot a name to display in leaderboards
|
||||
name="MY super smart agent",
|
||||
|
||||
# Your name, to give you credit
|
||||
author="Unknown programmer",
|
||||
|
||||
# Update the version to indicate the agent behaves differently.
|
||||
# This will later allow you to compare different versions of your
|
||||
# agent against one another.
|
||||
version="1.0.0",
|
||||
|
||||
# Add extra custom information about the agent to this dictionary.
|
||||
profile={},
|
||||
)
|
||||
|
||||
# Indicate that you're willing to play tic-tac-toe
|
||||
# Remove this if you don't want your bot to participate there.
|
||||
self.add_tic_tac_toe(on_move=self.play_tic_tac_toe, profile={})
|
||||
|
||||
@staticmethod
|
||||
def play_tic_tac_toe(payload : Payload) -> Payload:
|
||||
"""
|
||||
Play a game of tic-tac-toe.
|
||||
|
||||
You receive a payload that looks like this:
|
||||
|
||||
{
|
||||
"1": "X", "2": "", "3": "O",
|
||||
"4": "X", "5": "O", "6": "",
|
||||
"7": "", "8": "", "9": "",
|
||||
"your_token": "X"
|
||||
}
|
||||
|
||||
And you're expected to return a response of which field you'd like to
|
||||
place your piece in. For example, if you wish to place your token in
|
||||
field 7, your response should look like this:
|
||||
|
||||
{ "move": 7 }
|
||||
|
||||
The board is arranged as follows:
|
||||
|
||||
1 | 2 | 3
|
||||
---+---+---
|
||||
4 | 5 | 6
|
||||
---+---+---
|
||||
7 | 8 | 9
|
||||
|
||||
:param payload: The incoming JSON that contains the game state.
|
||||
:type payload: dict[str, Any]
|
||||
:return: The move you wish to take.
|
||||
:rtype: dict[str, Any]
|
||||
"""
|
||||
|
||||
# Try printing the payload to see what it looks like!
|
||||
print(payload)
|
||||
|
||||
options = [ 1, 2, 3, 4, 5, 6, 7, 8, 9, ]
|
||||
|
||||
# 1. Try filtering out the impossible moves!
|
||||
# If an X or O was already placed at a field, remove it from the options
|
||||
|
||||
#
|
||||
|
||||
# 2. Try finding two in a row! If possible, you can try to place the third
|
||||
# item on the board and get 3 in a row.
|
||||
|
||||
#
|
||||
|
||||
# 3. Perhaps you can block the opponent from getting 3 in a row?
|
||||
|
||||
#
|
||||
|
||||
# Now, pick any of the remaining options.
|
||||
# This is just a simple implementation. Naturally, you're welcome to try
|
||||
# your own algorithm.
|
||||
return { "move": random.choice(options) }
|
||||
|
|
@ -42,9 +42,6 @@ class MuteAgent(Agent):
|
|||
},
|
||||
)
|
||||
|
||||
self.add_tic_tac_toe(on_move=respond_mute)
|
||||
self.add_royal_game_of_ur(on_move=respond_mute)
|
||||
|
||||
def get_action(self, game: str, action: str | None) -> ActionFunction:
|
||||
"""
|
||||
Return the mute response for any game or action.
|
||||
|
|
|
|||
|
|
@ -0,0 +1,117 @@
|
|||
"""
|
||||
The naive line counter aims to calculate the best move by comparing the
|
||||
lines and checking how many are filled.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import random
|
||||
|
||||
from .agent import Agent, Payload
|
||||
|
||||
|
||||
class NaiveLineCounter(Agent):
|
||||
"""
|
||||
Tic-tac-toe has 8 possible lines where you can get 3 in a row.
|
||||
|
||||
For all 8 lines, count how many would benefit from the addition of a
|
||||
single sign. In priority, count how many lines are added.
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
super().__init__(
|
||||
name="Naive line counter",
|
||||
author="Bram",
|
||||
version="1.0.0",
|
||||
profile={},
|
||||
)
|
||||
|
||||
self.add_tic_tac_toe(on_move=self.play_tic_tac_toe, profile={})
|
||||
|
||||
def play_tic_tac_toe(self, payload : Payload) -> Payload:
|
||||
"""
|
||||
For each line:
|
||||
|
||||
[X][X][_] -> 3 in a row
|
||||
[O][O][_] -> ruin an opponent's 3 in a row
|
||||
[X][_][ ] -> 2 in a row
|
||||
[O][_][ ] -> ruin an opponent's 2 in a row
|
||||
[_][ ][ ] -> 1 in a row
|
||||
|
||||
Count how many they occur, then sort based on priority.
|
||||
"""
|
||||
lines = [
|
||||
[ 1, 2, 3, ],
|
||||
[ 4, 5, 6, ],
|
||||
[ 7, 8, 9, ],
|
||||
|
||||
[ 1, 4, 7, ],
|
||||
[ 2, 5, 8, ],
|
||||
[ 3, 6, 9, ],
|
||||
|
||||
[ 1, 5, 9, ],
|
||||
[ 3, 5, 7, ],
|
||||
]
|
||||
|
||||
my_piece = payload["your_token"]
|
||||
moves, value = [], ( 0, 0, 0, 0, 0 )
|
||||
|
||||
for i in range(1, 10):
|
||||
i_val = field_score(payload=payload, field=i, my_piece=my_piece)
|
||||
|
||||
if i_val > value:
|
||||
moves, value = [i], i_val
|
||||
elif i_val == value:
|
||||
moves.append(i)
|
||||
|
||||
return dict(move=random.choice(moves))
|
||||
|
||||
def field_score(payload : Payload, field : int, my_piece : str) -> tuple[int, int, int, int, int]:
|
||||
"""
|
||||
Determine the score of a specific field in the grid.
|
||||
"""
|
||||
a, b, c, d, e = 0, 0, 0, 0, 0
|
||||
lines = [
|
||||
[ 1, 2, 3, ],
|
||||
[ 4, 5, 6, ],
|
||||
[ 7, 8, 9, ],
|
||||
|
||||
[ 1, 4, 7, ],
|
||||
[ 2, 5, 8, ],
|
||||
[ 3, 6, 9, ],
|
||||
|
||||
[ 1, 5, 9, ],
|
||||
[ 3, 5, 7, ],
|
||||
]
|
||||
|
||||
if payload[str(field)] != "":
|
||||
return ( a, b, c, d, e )
|
||||
|
||||
for line in lines:
|
||||
if field not in line:
|
||||
continue
|
||||
|
||||
values = [ payload[str(f)] for f in line if field != f ]
|
||||
x = values.count("X")
|
||||
o = values.count("O")
|
||||
|
||||
s = ( x, o ) if my_piece == "X" else ( o, x )
|
||||
match s:
|
||||
case ( 2, 0 ):
|
||||
a += 1
|
||||
case ( 0, 2 ):
|
||||
b += 1
|
||||
case ( 1, 0 ):
|
||||
c += 1
|
||||
case ( 0, 1 ):
|
||||
d += 1
|
||||
case ( 0, 0 ):
|
||||
e += 1
|
||||
case ( 1, 1 ):
|
||||
pass
|
||||
case _:
|
||||
raise ValueError(
|
||||
f"Unknown number of items: {s}"
|
||||
)
|
||||
|
||||
return ( a, b, c, d, e )
|
||||
|
|
@ -20,7 +20,6 @@ class RemoteAgent(Agent):
|
|||
"""
|
||||
|
||||
url : str
|
||||
games : set[str]
|
||||
|
||||
def __init__(self, url : str, timeout : float = 1.0) -> None:
|
||||
"""
|
||||
|
|
@ -58,8 +57,6 @@ class RemoteAgent(Agent):
|
|||
profile=content,
|
||||
)
|
||||
|
||||
self.games = set()
|
||||
|
||||
games = content.get("games")
|
||||
if isinstance(games, dict):
|
||||
if "tic-tac-toe" in games:
|
||||
|
|
@ -67,13 +64,6 @@ class RemoteAgent(Agent):
|
|||
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:
|
||||
"""
|
||||
|
|
@ -103,7 +93,7 @@ class RemoteAgent(Agent):
|
|||
response = requests.get(url, json=payload, timeout=self.timeout)
|
||||
response.raise_for_status()
|
||||
content = response.json()
|
||||
except (requests.ConnectTimeout, requests.RequestException, requests.HTTPError, ValueError):
|
||||
except (requests.RequestException, requests.HTTPError, ValueError):
|
||||
return {}
|
||||
else:
|
||||
return content if isinstance(content, dict) else {}
|
||||
|
|
|
|||
27
client.py
27
client.py
|
|
@ -23,14 +23,9 @@ from __future__ import annotations
|
|||
import json
|
||||
import pyclient
|
||||
|
||||
# Import your game(s) here
|
||||
from agents import Agent, RemoteAgent
|
||||
from pyclient import PyClient
|
||||
from pyclient.games import TicTacToe
|
||||
# ...
|
||||
|
||||
# Import your agent(s) here
|
||||
from agents import Agent, AgentOfChaos
|
||||
# from agents.my_agent import MyAgent
|
||||
# ...
|
||||
|
||||
def main() -> int:
|
||||
"""
|
||||
|
|
@ -39,29 +34,22 @@ def main() -> int:
|
|||
:return: Exit code
|
||||
:rtype: int
|
||||
"""
|
||||
c = PyClient(debug=True)
|
||||
|
||||
# Start the game engine
|
||||
c = pyclient.PyClient(debug=True)
|
||||
|
||||
# Mix and match any agents you'd like.
|
||||
# During development it's usually easiest to play against AgentOfChaos().
|
||||
players : list[Agent] = [
|
||||
AgentOfChaos(),
|
||||
AgentOfChaos(),
|
||||
RemoteAgent(url="https://bmt001.noordstar.me/"),
|
||||
RemoteAgent(url="https://bmt002.noordstar.me/"),
|
||||
]
|
||||
|
||||
# Play a given game with your players
|
||||
result = c.play_game(
|
||||
out = c.play_game(
|
||||
players=players,
|
||||
start=TicTacToe.empty(),
|
||||
)
|
||||
|
||||
# Print the game results to the terminal!
|
||||
inspect_game(result)
|
||||
inspect_game(out)
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
def inspect_game(game : pyclient.GameReplay) -> None:
|
||||
"""
|
||||
Print a diagnostic of a played game to the terminal.
|
||||
|
|
@ -101,6 +89,5 @@ def inspect_game(game : pyclient.GameReplay) -> None:
|
|||
print(f"Total turns taken: {len(game.turns)}")
|
||||
print(f"Result: {final_state.winner()}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
|
|
|
|||
|
|
@ -30,7 +30,6 @@ type alias EloStat =
|
|||
, losses : Int
|
||||
, name : String
|
||||
, url : String
|
||||
, version : Maybe String
|
||||
, wins : Int
|
||||
}
|
||||
|
||||
|
|
@ -40,7 +39,7 @@ type alias HealthCheck =
|
|||
|
||||
|
||||
type alias Leaderboard =
|
||||
List { name : String, players : Players }
|
||||
{ name : String, players : Players }
|
||||
|
||||
|
||||
type alias Match =
|
||||
|
|
@ -64,13 +63,12 @@ type alias Players =
|
|||
|
||||
eloStatDecoder : D.Decoder EloStat
|
||||
eloStatDecoder =
|
||||
D.map7 EloStat
|
||||
D.map6 EloStat
|
||||
(D.field "draws" D.int)
|
||||
(D.field "elo" D.int)
|
||||
(D.field "losses" D.int)
|
||||
(D.field "name" D.string)
|
||||
(D.field "url" D.string)
|
||||
(D.maybe <| D.field "version" D.string)
|
||||
(D.field "wins" D.int)
|
||||
|
||||
|
||||
|
|
@ -131,10 +129,9 @@ healthCheckDecoder =
|
|||
|
||||
leaderboardDecoder : D.Decoder Leaderboard
|
||||
leaderboardDecoder =
|
||||
D.map2 (\name players -> { name = name, players = players })
|
||||
D.map2 Leaderboard
|
||||
(D.field "name" D.string)
|
||||
(D.field "players" playersDecoder)
|
||||
|> D.list
|
||||
|
||||
|
||||
matchDecoder : D.Decoder Match
|
||||
|
|
|
|||
|
|
@ -2,17 +2,17 @@ module EloTracker exposing (main)
|
|||
|
||||
import Api.Elo as Elo
|
||||
import Element exposing (Element)
|
||||
import Element.Background
|
||||
import Element.Font
|
||||
import Http
|
||||
import Json.Decode as D
|
||||
import Layout
|
||||
import Pixels
|
||||
import Program
|
||||
import Screen.Leaderboard
|
||||
import Theme
|
||||
import Time
|
||||
import Widget.Icon
|
||||
import Layout
|
||||
import Theme
|
||||
import Screen.Leaderboard
|
||||
import Element.Font
|
||||
import Pixels
|
||||
import Element.Background
|
||||
|
||||
|
||||
main =
|
||||
|
|
@ -35,14 +35,12 @@ type alias Model =
|
|||
{ baseUrl : String
|
||||
, leaderboard : Result Int Elo.Leaderboard
|
||||
, matches : Result Int (List Elo.Match)
|
||||
, selectedName : String
|
||||
}
|
||||
|
||||
|
||||
type Msg
|
||||
= OnLeaderboard (Result Http.Error Elo.Leaderboard)
|
||||
| OnMatches (Result Http.Error (List Elo.Match))
|
||||
| OnSelectLeaderboardName String
|
||||
| RefreshLeaderboard
|
||||
| RefreshMatches
|
||||
|
||||
|
|
@ -62,7 +60,6 @@ init flag =
|
|||
{ baseUrl = baseUrl
|
||||
, leaderboard = Err 0
|
||||
, matches = Err 0
|
||||
, selectedName = "tic-tac-toe"
|
||||
}
|
||||
in
|
||||
( model
|
||||
|
|
@ -102,9 +99,6 @@ update msg model =
|
|||
OnMatches (Ok v) ->
|
||||
( { model | matches = Ok v }, Cmd.none )
|
||||
|
||||
OnSelectLeaderboardName name ->
|
||||
( { model | selectedName = name }, Cmd.none )
|
||||
|
||||
RefreshLeaderboard ->
|
||||
( model, getLeaderboard model )
|
||||
|
||||
|
|
@ -187,7 +181,7 @@ view data =
|
|||
, Element.spacing 10
|
||||
]
|
||||
[ Element.text "Loading leaderboard..."
|
||||
, Element.el [ Element.centerX, Element.Font.color (Theme.redUI data.flavor) ] (Element.text ("Failed " ++ String.fromInt n ++ " times"))
|
||||
, Element.el [ Element.centerX, Element.Font.color (Theme.redUI data.flavor) ] (Element.text ("Failed " ++ (String.fromInt n) ++ " times"))
|
||||
, Layout.loadingIndicator { color = Theme.lavender data.flavor }
|
||||
|> Element.el [ Element.centerX ]
|
||||
]
|
||||
|
|
@ -195,11 +189,5 @@ view data =
|
|||
|
||||
Ok leaderboard ->
|
||||
Screen.Leaderboard.view
|
||||
{ flavor = data.flavor
|
||||
, model =
|
||||
{ board = leaderboard
|
||||
, onSelect = OnSelectLeaderboardName
|
||||
, selectedName = data.model.selectedName
|
||||
}
|
||||
, size = data.size
|
||||
}
|
||||
{ flavor = data.flavor, model = leaderboard, size = data.size }
|
||||
|
||||
|
|
|
|||
|
|
@ -3,88 +3,26 @@ module Screen.Leaderboard exposing (..)
|
|||
{-| Widget module for a leaderboard page.
|
||||
-}
|
||||
|
||||
import Api.Elo exposing (EloStat, Leaderboard, Players)
|
||||
import Color
|
||||
import Api.Elo exposing (EloStat, Leaderboard)
|
||||
import Element exposing (Element)
|
||||
import Element.Background
|
||||
import Element.Border
|
||||
import Element.Events
|
||||
import Element.Font
|
||||
import Layout
|
||||
import Pixels exposing (Pixels)
|
||||
import Program exposing (ViewBox)
|
||||
import Quantity exposing (Quantity)
|
||||
import Theme
|
||||
import Element.Background
|
||||
import Element.Border
|
||||
import Color
|
||||
|
||||
|
||||
view : ViewBox { board : Leaderboard, onSelect : String -> msg, selectedName : String } -> Element msg
|
||||
view : ViewBox Leaderboard -> Element msg
|
||||
view data =
|
||||
Element.column
|
||||
[]
|
||||
[ selectBoard
|
||||
{ flavor = data.flavor
|
||||
, leaderboard = data.model.board
|
||||
, onSelect = data.model.onSelect
|
||||
, selectedName = data.model.selectedName
|
||||
, width = data.size.width
|
||||
}
|
||||
, data.model.board
|
||||
|> List.map
|
||||
(\l ->
|
||||
if l.name == data.model.selectedName then
|
||||
viewSingleBoard
|
||||
{ flavor = data.flavor
|
||||
, model = l
|
||||
, size = data.size
|
||||
}
|
||||
|
||||
else
|
||||
Element.none
|
||||
)
|
||||
|> Element.column []
|
||||
]
|
||||
|
||||
|
||||
selectBoard :
|
||||
{ flavor : Theme.Flavor
|
||||
, leaderboard : Leaderboard
|
||||
, onSelect : String -> msg
|
||||
, selectedName : String
|
||||
, width : Quantity Int Pixels
|
||||
}
|
||||
-> Element msg
|
||||
selectBoard data =
|
||||
data.leaderboard
|
||||
|> List.map .name
|
||||
|> List.map
|
||||
(\s ->
|
||||
Element.el
|
||||
[ if data.selectedName == s then
|
||||
Element.Background.color (Theme.subtext0UI data.flavor)
|
||||
|
||||
else
|
||||
Element.Background.color (Theme.surface1UI data.flavor)
|
||||
, Element.Border.rounded 25
|
||||
, Element.centerX
|
||||
, Element.Events.onClick (data.onSelect s)
|
||||
, Element.padding 10
|
||||
]
|
||||
(Element.text s)
|
||||
)
|
||||
|> Element.wrappedRow
|
||||
[ Element.padding 20
|
||||
, Element.spacing 10
|
||||
, Element.width <| Element.px <| Pixels.inPixels data.width
|
||||
]
|
||||
|
||||
|
||||
viewSingleBoard : ViewBox { name : String, players : Players } -> Element msg
|
||||
viewSingleBoard data =
|
||||
let
|
||||
boardWidth =
|
||||
data.size.width
|
||||
|> Quantity.toFloatQuantity
|
||||
|> Quantity.multiplyBy (2 / 3)
|
||||
|> Quantity.multiplyBy (2/3)
|
||||
|> Quantity.floor
|
||||
|> Quantity.clamp (Pixels.pixels 600) (Pixels.pixels 1800)
|
||||
|> Pixels.inPixels
|
||||
|
|
@ -95,10 +33,10 @@ viewSingleBoard data =
|
|||
|> List.map (viewEloStat { flavor = data.flavor })
|
||||
|> Element.column [ Element.centerX, boardWidth ]
|
||||
|> List.singleton
|
||||
|> (::)
|
||||
(Element.text data.model.name
|
||||
|> Element.el
|
||||
[ Element.centerX, Element.Font.size 50 ]
|
||||
|> (::) (
|
||||
Element.text data.model.name
|
||||
|> Element.el
|
||||
[ Element.centerX, Element.Font.size 50 ]
|
||||
)
|
||||
|> Element.column
|
||||
[ Element.padding 15
|
||||
|
|
@ -124,13 +62,7 @@ viewEloStat data stat =
|
|||
[ Element.Font.color (Theme.subtext0UI data.flavor)
|
||||
, Element.Font.size 15
|
||||
]
|
||||
(case stat.version of
|
||||
Nothing ->
|
||||
Element.text stat.url
|
||||
|
||||
Just v ->
|
||||
Element.text (stat.url ++ " | " ++ v)
|
||||
)
|
||||
(Element.text stat.url)
|
||||
]
|
||||
|
||||
-- Bar to push the rest to the right
|
||||
|
|
@ -142,7 +74,6 @@ viewEloStat data stat =
|
|||
, field { text = "LOSSES", value = stat.losses, color = Theme.redUI data.flavor }
|
||||
]
|
||||
|
||||
|
||||
field : { text : String, value : Int, color : Element.Color } -> Element msg
|
||||
field { text, value, color } =
|
||||
Element.column
|
||||
|
|
|
|||
9
elo.py
9
elo.py
|
|
@ -3,7 +3,7 @@
|
|||
"""
|
||||
|
||||
from elo_tracker import EloTracker
|
||||
from pyclient.games import RoyalGameOfUr, TicTacToe
|
||||
from pyclient.games import TicTacToe
|
||||
|
||||
GAME_FILE = "games.jsonl"
|
||||
PLAYER_FILE = "known_players.json"
|
||||
|
|
@ -21,13 +21,6 @@ def main() -> int:
|
|||
player_count=2,
|
||||
)
|
||||
|
||||
tracker.start_periodic_matches(
|
||||
game=RoyalGameOfUr.empty(),
|
||||
interval_seconds=9 * 60,
|
||||
max_turns=1_500,
|
||||
player_count=2,
|
||||
)
|
||||
|
||||
try:
|
||||
tracker.start_server(import_name=__name__)
|
||||
except KeyboardInterrupt:
|
||||
|
|
|
|||
|
|
@ -1,49 +0,0 @@
|
|||
# ELO tracker
|
||||
|
||||
The ELO tracker lets your agent play many games and assigns every participant
|
||||
an Elo rating.
|
||||
|
||||
## Before you start
|
||||
|
||||
The tracker only plays against **remote agents**. Remote agents are available
|
||||
through a URL, and usually run on the internet.
|
||||
|
||||
If your agent only exists as a local Python class,
|
||||
[publish it first](../pyserver/README.md).
|
||||
|
||||
## Add players
|
||||
|
||||
Open `known_players.json` and add the URLs you want to include.
|
||||
|
||||
For example:
|
||||
|
||||
```json
|
||||
{
|
||||
"players": [
|
||||
"https://my-agent.example",
|
||||
"https://bmt001.noordstar.me"
|
||||
]
|
||||
}
|
||||
```
|
||||
|
||||
Every listed URL is considered a participant. The ELO tracker will compare
|
||||
these players with one another.
|
||||
|
||||
## Start the tracker
|
||||
|
||||
Run:
|
||||
|
||||
```sh
|
||||
python elo.py
|
||||
```
|
||||
|
||||
The tracker will continuously schedule new matches until you stop it.
|
||||
|
||||
## Results
|
||||
|
||||
Two files are updated while the tracker runs:
|
||||
|
||||
* `games.jsonl` stores the played games.
|
||||
* `known_players.json` stores the list of participants.
|
||||
|
||||
You can stop the tracker with <kbd>Ctrl</kbd>+<kbd>C</kbd>.
|
||||
|
|
@ -14,9 +14,8 @@ from dataclasses import dataclass
|
|||
from datetime import datetime, timezone
|
||||
from typing import Any, Sequence
|
||||
|
||||
import agents.remote
|
||||
import agents
|
||||
import pyclient
|
||||
import time
|
||||
|
||||
from pyclient.games import FinishState, Game
|
||||
|
||||
|
|
@ -39,7 +38,7 @@ class PlayerIdentifier:
|
|||
return (self.name, self.url, self.version)
|
||||
|
||||
@classmethod
|
||||
def from_server_agent(cls, agent : agents.remote.RemoteAgent) -> "PlayerIdentifier":
|
||||
def from_server_agent(cls, agent : agents.RemoteAgent) -> "PlayerIdentifier":
|
||||
"""
|
||||
Gain a player identifier from an agent.
|
||||
"""
|
||||
|
|
@ -177,7 +176,7 @@ class Match:
|
|||
@classmethod
|
||||
def from_replay(
|
||||
cls,
|
||||
players : list[agents.remote.RemoteAgent],
|
||||
players : list[agents.RemoteAgent],
|
||||
replay : pyclient.GameReplay,
|
||||
timestamp : str | None,
|
||||
) -> "Match":
|
||||
|
|
@ -281,7 +280,7 @@ class EloTracker:
|
|||
# Threading variables
|
||||
self.__lock = threading.RLock()
|
||||
self.__scheduler_stop = threading.Event()
|
||||
self.__scheduler_threads: list[threading.Thread] = []
|
||||
self.__scheduler_thread: threading.Thread | None = None
|
||||
|
||||
# Immutable values
|
||||
self.debug: bool = debug
|
||||
|
|
@ -291,9 +290,9 @@ class EloTracker:
|
|||
|
||||
# Thread-unsafe variables
|
||||
# Please use a lock while doing CRUD operations on them
|
||||
self.players: list[agents.remote.RemoteAgent] = []
|
||||
self.players: list[agents.RemoteAgent] = []
|
||||
self.__matches: list[Match] = []
|
||||
self.__stats: dict[str, dict[PlayerIdentifier, EloStat]] = {}
|
||||
self.__stats: dict[PlayerIdentifier, EloStat] = {}
|
||||
|
||||
# Initialize tracker
|
||||
self.__load_matches()
|
||||
|
|
@ -310,30 +309,25 @@ class EloTracker:
|
|||
with self.__lock:
|
||||
print(f"[EloTracker] {message}")
|
||||
|
||||
def __get_stat(self, game_name : str, player_id : PlayerIdentifier) -> EloStat:
|
||||
def __get_stat(self, player_id : PlayerIdentifier) -> EloStat:
|
||||
"""
|
||||
Get a player's statistics based on their player identifier.
|
||||
|
||||
If the player wasn't known, the function returns a newly
|
||||
initialized record in the database for them.
|
||||
|
||||
:param game_name: Name of the relevant game
|
||||
:type game_name: str
|
||||
:param player_id: Unique player identifier.
|
||||
:type player_id: PlayerIdentifier
|
||||
:return: Elo statistics
|
||||
"""
|
||||
with self.__lock:
|
||||
stat = self.__stats.get(game_name, {}).get(player_id, None)
|
||||
stat = self.__stats.get(player_id, None)
|
||||
|
||||
if stat is not None:
|
||||
return stat
|
||||
|
||||
if game_name not in self.__stats:
|
||||
self.__stats[game_name] = {}
|
||||
|
||||
stat = EloStat.new(player_id=player_id)
|
||||
self.__stats[game_name][player_id] = stat
|
||||
self.__stats[player_id] = stat
|
||||
|
||||
return stat
|
||||
|
||||
|
|
@ -393,13 +387,13 @@ class EloTracker:
|
|||
# Do not apply them yet, in order to guarantee fair ELO shifts
|
||||
for player_id1, result1 in m.participants:
|
||||
total_k = 0.0
|
||||
rating_1 = self.__get_stat(game_name=m.game_name, player_id=player_id1).elo
|
||||
rating_1 = self.__get_stat(player_id=player_id1).elo
|
||||
|
||||
for player_id2, result2 in m.participants:
|
||||
if player_id1 == player_id2:
|
||||
continue
|
||||
|
||||
rating_2 = self.__get_stat(game_name=m.game_name, player_id=player_id2).elo
|
||||
rating_2 = self.__get_stat(player_id=player_id2).elo
|
||||
|
||||
expected_score = 1 / (1 + 10 ** ((rating_2 - rating_1) / STD_DEV_DIFF))
|
||||
|
||||
|
|
@ -420,7 +414,7 @@ class EloTracker:
|
|||
|
||||
# Then, apply the ELO score update + count the wins, draws & losses
|
||||
for player_id, result in m.participants:
|
||||
player = self.__get_stat(game_name=m.game_name, player_id=player_id)
|
||||
player = self.__get_stat(player_id=player_id)
|
||||
player.elo += scores[player_id]
|
||||
|
||||
match result:
|
||||
|
|
@ -436,7 +430,6 @@ 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.
|
||||
|
|
@ -447,20 +440,13 @@ 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:
|
||||
self.load_players()
|
||||
|
||||
with self.__lock:
|
||||
players = [
|
||||
player for player in self.players
|
||||
if game.game_name() in player.games
|
||||
]
|
||||
|
||||
available = len(players)
|
||||
available = len(self.players)
|
||||
|
||||
if available < player_count:
|
||||
self.__debug(
|
||||
|
|
@ -469,16 +455,12 @@ class EloTracker:
|
|||
)
|
||||
else:
|
||||
self.__debug(
|
||||
f"Playing a new scheduled match of {game.game_name()}"
|
||||
)
|
||||
self.play_random_match(
|
||||
game=game,
|
||||
max_turns=max_turns,
|
||||
player_count=player_count,
|
||||
"Playing a new scheduled match"
|
||||
)
|
||||
self.play_random_match(game=game, player_count=player_count)
|
||||
except Exception as exc:
|
||||
self.__debug(f"Scheduled match failed: {exc}")
|
||||
# raise exc
|
||||
raise exc
|
||||
|
||||
if self.__scheduler_stop.wait(interval_seconds):
|
||||
break
|
||||
|
|
@ -491,7 +473,7 @@ class EloTracker:
|
|||
:type import_name: str
|
||||
"""
|
||||
try:
|
||||
from flask import Flask, Response, jsonify, request
|
||||
from flask import Flask, Response, jsonify
|
||||
except ImportError as exc:
|
||||
raise ImportError(
|
||||
"Flask is required to host the EloTracker server. "
|
||||
|
|
@ -544,15 +526,8 @@ class EloTracker:
|
|||
return jsonify(self.get_json_matches())
|
||||
|
||||
@app.get("/players")
|
||||
def players() -> tuple[Response, int]:
|
||||
payload = request.get_json(silent=True) or {}
|
||||
|
||||
game = payload.get("game", None)
|
||||
|
||||
if isinstance(game, str):
|
||||
return jsonify(self.get_json_players(game_name=game)), 200
|
||||
else:
|
||||
return jsonify({ "error": "No `game` parameter was given" }), 400
|
||||
def players() -> Response:
|
||||
return jsonify(self.get_json_players())
|
||||
|
||||
return app
|
||||
|
||||
|
|
@ -564,11 +539,7 @@ class EloTracker:
|
|||
:rtype: bool
|
||||
"""
|
||||
with self.__lock:
|
||||
for thread in self.__scheduler_threads:
|
||||
if thread.is_alive():
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
return self.__scheduler_thread is not None and self.__scheduler_thread.is_alive()
|
||||
|
||||
def load_players(self) -> None:
|
||||
"""
|
||||
|
|
@ -589,13 +560,13 @@ class EloTracker:
|
|||
"Expected `players` field to be a list of strings."
|
||||
)
|
||||
|
||||
players : list[agents.remote.RemoteAgent] = []
|
||||
players : list[agents.RemoteAgent] = []
|
||||
for url in urls:
|
||||
if not isinstance(url, str):
|
||||
continue
|
||||
|
||||
try:
|
||||
agent = agents.remote.RemoteAgent(url=url)
|
||||
agent = agents.RemoteAgent(url=url)
|
||||
except ValueError:
|
||||
pass # Not an available player right now
|
||||
else:
|
||||
|
|
@ -605,71 +576,49 @@ class EloTracker:
|
|||
self.players = players
|
||||
|
||||
for agent in players:
|
||||
for game_name in self.__stats:
|
||||
self.__get_stat(
|
||||
game_name=game_name,
|
||||
player_id=PlayerIdentifier.from_server_agent(agent)
|
||||
)
|
||||
self.__get_stat(PlayerIdentifier.from_server_agent(agent))
|
||||
|
||||
def play_match(
|
||||
self,
|
||||
players: list[str],
|
||||
game: Game,
|
||||
max_turns : int | None = None,
|
||||
) -> pyclient.GameReplay | None:
|
||||
def play_match(self, players: list[str], game: Game) -> pyclient.GameReplay:
|
||||
"""
|
||||
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.remote.RemoteAgent(url=url) for url in players ]
|
||||
ags : list[Any] = [ agents.RemoteAgent(url=url) for url in players ]
|
||||
|
||||
try:
|
||||
replay = pyclient.PyClient(debug=self.debug).play_game(
|
||||
players=ags,
|
||||
start=game,
|
||||
max_turns=max_turns,
|
||||
)
|
||||
replay = pyclient.PyClient(debug=self.debug).play_game(
|
||||
players=ags,
|
||||
start=game,
|
||||
)
|
||||
|
||||
except ValueError:
|
||||
return None # Exceeded turn limit
|
||||
m = Match.from_replay(players=ags, replay=replay, timestamp=Match.now())
|
||||
|
||||
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 | None:
|
||||
) -> pyclient.GameReplay:
|
||||
"""
|
||||
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.
|
||||
"""
|
||||
with self.__lock:
|
||||
players = [agent.url for agent in self.players if game.game_name() in agent.games]
|
||||
players = [agent.url for agent in self.players]
|
||||
|
||||
random.shuffle(players)
|
||||
|
||||
|
|
@ -682,7 +631,6 @@ class EloTracker:
|
|||
self,
|
||||
game: Game,
|
||||
interval_seconds: float = 300,
|
||||
max_turns : int | None = None,
|
||||
player_count: int = 2,
|
||||
) -> None:
|
||||
"""
|
||||
|
|
@ -692,8 +640,6 @@ 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
|
||||
"""
|
||||
|
|
@ -703,51 +649,40 @@ class EloTracker:
|
|||
raise ValueError("player_count must be greater than one.")
|
||||
|
||||
with self.__lock:
|
||||
if self.is_running_periodic_matches():
|
||||
self.stop_periodic_matches()
|
||||
|
||||
self.__scheduler_stop.clear()
|
||||
thread = threading.Thread(
|
||||
self.__scheduler_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)
|
||||
thread.start()
|
||||
self.__scheduler_thread.start()
|
||||
|
||||
def stop_periodic_matches(self) -> None:
|
||||
"""
|
||||
Stop the periodic match scheduler and wait briefly for it to exit.
|
||||
"""
|
||||
max_wait_time_s = 5
|
||||
now = time.time()
|
||||
time_left = lambda : max(0.1, now + max_wait_time_s - time.time())
|
||||
to_remove : list[threading.Thread] = []
|
||||
thread: threading.Thread | None
|
||||
|
||||
# Announce that all threads need to be stopped
|
||||
with self.__lock:
|
||||
self.__scheduler_stop.set()
|
||||
to_remove = [ thread for thread in self.__scheduler_threads ]
|
||||
thread = self.__scheduler_thread
|
||||
|
||||
# Then, quietly remove all existing threads
|
||||
for thread in self.__scheduler_threads:
|
||||
if thread is not None:
|
||||
thread.join(timeout=time_left())
|
||||
if thread is not None:
|
||||
thread.join(timeout=5)
|
||||
|
||||
# Remove all stopped threads
|
||||
# Keep in mind that another threads might've added new threads in the
|
||||
# meantime.
|
||||
with self.__lock:
|
||||
for thread in to_remove:
|
||||
self.__scheduler_threads.remove(thread)
|
||||
if self.__scheduler_thread is thread:
|
||||
self.__scheduler_thread = None
|
||||
|
||||
def get_json_players(self, game_name : str) -> list[dict[str, Any]]:
|
||||
def get_json_players(self) -> list[dict[str, Any]]:
|
||||
"""
|
||||
Return known currently available players as notebook-friendly dicts.
|
||||
"""
|
||||
with self.__lock:
|
||||
players : list[EloStat] = [
|
||||
player for player in self.__stats.get(game_name, {}).values()
|
||||
if abs(player.wins + player.draws + player.losses) > 0
|
||||
]
|
||||
players : list[EloStat] = list(self.__stats.values())
|
||||
|
||||
players.sort(
|
||||
key=lambda player: (-int(player.elo), player.player_id.name)
|
||||
|
|
@ -774,17 +709,14 @@ class EloTracker:
|
|||
|
||||
return [ m.to_json() for m in matches ]
|
||||
|
||||
def get_json_leaderboard(self) -> list[dict[str, Any]]:
|
||||
def get_json_leaderboard(self) -> dict[str, Any]:
|
||||
"""
|
||||
Return aggregate player statistics for local use or JSON APIs.
|
||||
"""
|
||||
return [
|
||||
{
|
||||
"name": game_name,
|
||||
"players": self.get_json_players(game_name=game_name),
|
||||
}
|
||||
for game_name in self.__stats
|
||||
]
|
||||
return {
|
||||
"name": self.name,
|
||||
"players": self.get_json_players(),
|
||||
}
|
||||
|
||||
def start_server(
|
||||
self,
|
||||
|
|
|
|||
|
|
@ -1,7 +1,6 @@
|
|||
{
|
||||
"players": [
|
||||
"https://bmt001.noordstar.me",
|
||||
"https://bmt002.noordstar.me",
|
||||
"https://bmt003.noordstar.me"
|
||||
"https://bmt002.noordstar.me"
|
||||
]
|
||||
}
|
||||
|
|
@ -19,12 +19,7 @@ class PyClient:
|
|||
|
||||
debug : bool
|
||||
|
||||
def gen_game(
|
||||
self,
|
||||
players : list[Agent],
|
||||
start : Game,
|
||||
max_turns : int | None = None,
|
||||
) -> Generator[Turn, None, None]:
|
||||
def gen_game(self, players : list[Agent], start : Game) -> Generator[Turn, None, None]:
|
||||
"""
|
||||
Generate a game by polling the players.
|
||||
|
||||
|
|
@ -32,22 +27,12 @@ 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:
|
||||
|
|
@ -67,21 +52,13 @@ class PyClient:
|
|||
if on_move is None:
|
||||
payload = {}
|
||||
else:
|
||||
try:
|
||||
payload = on_move(current_state.as_seen_by(player=player))
|
||||
except Exception:
|
||||
payload = {}
|
||||
payload = on_move(current_state.as_seen_by(player=player))
|
||||
|
||||
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,
|
||||
max_turns : int | None = None,
|
||||
) -> GameReplay:
|
||||
def play_game(self, players : list[Agent], start : Game) -> GameReplay:
|
||||
"""
|
||||
Generate a game by polling the players. Collect all moves in a
|
||||
summary.
|
||||
|
|
@ -90,16 +67,11 @@ 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, max_turns=max_turns
|
||||
)),
|
||||
turns=list(self.gen_game(players=players, start=start)),
|
||||
)
|
||||
|
|
|
|||
|
|
@ -4,11 +4,9 @@
|
|||
|
||||
from .game import FinishState, Game
|
||||
from .tic_tac_toe import TicTacToe
|
||||
from .ur import RoyalGameOfUr
|
||||
|
||||
__all__ = [
|
||||
"FinishState",
|
||||
"Game",
|
||||
"RoyalGameOfUr",
|
||||
"TicTacToe",
|
||||
]
|
||||
|
|
|
|||
|
|
@ -1,235 +0,0 @@
|
|||
"""
|
||||
The Royal game of Ur is supposedly one of the world's oldest known board
|
||||
games. Players race 7 horses to the finish line while trying to sabotage
|
||||
each other.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
from typing import Any, Generator
|
||||
|
||||
import random
|
||||
|
||||
from .game import FinishState, Game
|
||||
from dataclasses import dataclass
|
||||
|
||||
# On these fields, you can safely place your token without getting it removed
|
||||
# if the opponent plays a token on their respective spot.
|
||||
SAFE_FIELDS = set([ 0, 1, 2, 3, 4, 13, 14, 15 ])
|
||||
|
||||
# Whenever you land a token on this piece, you may play another turn.
|
||||
STAR_FIELDS = set([ 4, 8, 14 ])
|
||||
|
||||
# You may place multiple pieces on these fields.
|
||||
STACK_FIELDS = set([ 0, 15 ])
|
||||
|
||||
# Length of the board
|
||||
BOARD_LENGTH = 1 + 4 + (4 + 2 + 2) + 2 + 1
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class RoyalGameOfUr(Game):
|
||||
"""
|
||||
The Royal Game of Ur is a game where players roll dice to get to the
|
||||
finish line.
|
||||
"""
|
||||
board : list[tuple[int, int]]
|
||||
mover : int
|
||||
roll : int
|
||||
|
||||
def __updated_board(self, start : int) -> list[tuple[int, int]]:
|
||||
new_board = []
|
||||
|
||||
assert self.roll > 0
|
||||
assert 0 <= start < BOARD_LENGTH
|
||||
assert start + self.roll < BOARD_LENGTH
|
||||
|
||||
elim = False
|
||||
|
||||
for i, (a, b) in enumerate(self.board):
|
||||
if i == start:
|
||||
if self.player_to_move() == 1:
|
||||
new_board.append(( a - 1, b ))
|
||||
else:
|
||||
new_board.append(( a, b - 1 ))
|
||||
elif i == start + self.roll:
|
||||
if self.player_to_move() == 1:
|
||||
if b > 0 and i not in SAFE_FIELDS:
|
||||
elim = True
|
||||
b = 0
|
||||
new_board.append(( a + 1, b ))
|
||||
else:
|
||||
if a > 0 and i not in SAFE_FIELDS:
|
||||
elim = True
|
||||
a = 0
|
||||
new_board.append(( a, b + 1 ))
|
||||
else:
|
||||
new_board.append(( a, b ))
|
||||
|
||||
if elim:
|
||||
if self.player_to_move() == 1:
|
||||
new_board[0] = ( new_board[0][0], new_board[0][1] + 1)
|
||||
else:
|
||||
new_board[0] = ( new_board[0][0] + 1, new_board[0][1] )
|
||||
|
||||
return new_board
|
||||
|
||||
def __valid_moves(self) -> Generator[int, None, None]:
|
||||
for i in range(BOARD_LENGTH - self.roll):
|
||||
dest = i + self.roll
|
||||
|
||||
if self.board[i][self.mover] > 0:
|
||||
if dest in STACK_FIELDS or self.board[dest][self.mover] == 0:
|
||||
yield i
|
||||
|
||||
def action_name(self) -> str:
|
||||
return "" # Players can only move in this game
|
||||
|
||||
def as_seen_by(self, player: int) -> dict[str, Any]:
|
||||
my_index = player - 1
|
||||
ot_index = 1 - my_index
|
||||
|
||||
d : dict[str, Any] = dict(
|
||||
enemies_at_start=self.board[0][ot_index],
|
||||
enemies_at_finish=self.board[15][ot_index],
|
||||
safe_fields=list(SAFE_FIELDS),
|
||||
stack_fields=list(STACK_FIELDS),
|
||||
star_fields=list(STAR_FIELDS),
|
||||
tokens_at_start=self.board[0][my_index],
|
||||
tokens_at_finish=self.board[15][my_index],
|
||||
valid_moves=list(self.__valid_moves()),
|
||||
)
|
||||
|
||||
for i in range(BOARD_LENGTH):
|
||||
d[str(i)] = (
|
||||
"YOU" if self.board[i][my_index] > 0 else (
|
||||
"ENEMY" if my_index not in SAFE_FIELDS and self.board[i][ot_index] > 0 else (
|
||||
""
|
||||
)))
|
||||
|
||||
for i in SAFE_FIELDS:
|
||||
d[f"{i}_enemy"] = (
|
||||
"ENEMY" if self.board[i][ot_index] > 0 else ""
|
||||
)
|
||||
|
||||
return d
|
||||
|
||||
@classmethod
|
||||
def empty(cls) -> RoyalGameOfUr:
|
||||
"""
|
||||
Create a new game.
|
||||
|
||||
:return: Start state of the board
|
||||
:rtype: RoyalGameOfUr
|
||||
"""
|
||||
times, roll = cls.roll_repeatedly()
|
||||
|
||||
return cls(
|
||||
board=[ (7, 7) ] + ((BOARD_LENGTH - 1) * [ (0, 0 ) ]),
|
||||
mover=(1 + times) % 2,
|
||||
roll=roll,
|
||||
)
|
||||
|
||||
def game_name(self) -> str:
|
||||
return "royal-game-of-ur"
|
||||
|
||||
def move_default(self) -> Game:
|
||||
"""
|
||||
Fallback move option for when a user isn't available or cannot
|
||||
decide. In this case, the player attempts to move whatever is the
|
||||
furthest in the back.
|
||||
"""
|
||||
player = self.player_to_move()
|
||||
|
||||
for i in self.__valid_moves():
|
||||
new_board = self.__updated_board(start=i)
|
||||
new_mover = self.mover
|
||||
if i + self.roll not in STAR_FIELDS:
|
||||
new_mover += 1
|
||||
|
||||
return RoyalGameOfUr(
|
||||
board=new_board, mover=new_mover % 2, roll=self.roll_once(),
|
||||
).skip_if_possible()
|
||||
else:
|
||||
raise ValueError(
|
||||
"Player has no valid moves even though those should've been erased."
|
||||
)
|
||||
|
||||
def move(self, payload : dict[str, Any] | None = None):
|
||||
if not isinstance(payload, dict):
|
||||
return self.move_default()
|
||||
|
||||
key = str(payload.get("move", ""))
|
||||
|
||||
for i in self.__valid_moves():
|
||||
if str(i) == key:
|
||||
new_board = self.__updated_board(start=i)
|
||||
new_mover = self.mover
|
||||
if i + self.roll not in STAR_FIELDS:
|
||||
new_mover += 1
|
||||
|
||||
return RoyalGameOfUr(
|
||||
board=new_board, mover=new_mover % 2, roll=self.roll_once(),
|
||||
).skip_if_possible()
|
||||
else:
|
||||
return self.move_default()
|
||||
|
||||
def player_to_move(self) -> int:
|
||||
return self.mover + 1
|
||||
|
||||
@staticmethod
|
||||
def roll_once() -> int:
|
||||
tot = 0
|
||||
for _ in range(4):
|
||||
tot += random.randint(0, 1)
|
||||
return tot
|
||||
|
||||
@staticmethod
|
||||
def roll_repeatedly() -> tuple[int, int]:
|
||||
"""
|
||||
Roll until you don't roll a zero.
|
||||
"""
|
||||
n, roll = 0, 0
|
||||
|
||||
while roll == 0:
|
||||
n += 1
|
||||
roll = RoyalGameOfUr.roll_once()
|
||||
|
||||
return n, roll
|
||||
|
||||
def skip_if_possible(self) -> RoyalGameOfUr:
|
||||
"""
|
||||
Skip this turn. Allow the next player to make a move.
|
||||
Players are not allowed to skip if they can make a valid move.
|
||||
"""
|
||||
# Exit if at least one valid move exists
|
||||
if self.roll > 0:
|
||||
for _ in self.__valid_moves():
|
||||
return self
|
||||
|
||||
# Exit if game has already finished
|
||||
if self.winner() is not None:
|
||||
return self
|
||||
|
||||
return RoyalGameOfUr(
|
||||
board=self.board,
|
||||
mover=(self.mover + 1) % 2,
|
||||
roll=self.roll_once()
|
||||
).skip_if_possible()
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
return self.as_seen_by(player=1)
|
||||
|
||||
def winner(self) -> dict[int, FinishState] | None:
|
||||
a, b = self.board[-1]
|
||||
|
||||
if a == 7:
|
||||
return {
|
||||
1 : FinishState.win,
|
||||
2 : FinishState.loss,
|
||||
}
|
||||
elif b == 7:
|
||||
return {
|
||||
1 : FinishState.loss,
|
||||
2 : FinishState.win,
|
||||
}
|
||||
else:
|
||||
return None
|
||||
|
|
@ -1,36 +0,0 @@
|
|||
# Publishing your agent
|
||||
|
||||
Once your agent behaves the way you want, you can expose it over HTTP so other clients can play against it.
|
||||
|
||||
You only need this if you want other programs to connect to your agent.
|
||||
|
||||
## Use this repository
|
||||
|
||||
1. Open `server.py`.
|
||||
2. Replace the agent passed to `PyServer` with your own. For example:
|
||||
|
||||
```py
|
||||
from agents.my_agent import MyAgent
|
||||
|
||||
player = PyServer(MyAgent())
|
||||
```
|
||||
|
||||
3. Start the server. Run:
|
||||
|
||||
```sh
|
||||
python server.py
|
||||
```
|
||||
|
||||
By default, the server listens on port `5000`.
|
||||
|
||||
4. Open your browser and visit `http://localhost:5000/`. You should receive
|
||||
a JSON document describing your agent and the games it supports.
|
||||
If that works, your agent is ready to receive game requests.
|
||||
|
||||
## Writing your own server
|
||||
|
||||
The included server is the easiest way to publish a Python agent.
|
||||
|
||||
If you want to implement the protocol yourself
|
||||
_(for example in another language)_ you can use the specification in
|
||||
[the protocol specification](./spec.md).
|
||||
|
|
@ -6,6 +6,7 @@ from __future__ import annotations
|
|||
|
||||
import agents
|
||||
|
||||
from agents.naive_line_counter import NaiveLineCounter
|
||||
from pyserver import PyServer
|
||||
|
||||
def main() -> int:
|
||||
|
|
@ -15,7 +16,7 @@ def main() -> int:
|
|||
:return: Exit code
|
||||
:rtype: int
|
||||
"""
|
||||
player = PyServer(agents.AgentOfChaos())
|
||||
player = PyServer(NaiveLineCounter())
|
||||
|
||||
# Start listening for games
|
||||
player.start(
|
||||
|
|
|
|||
|
|
@ -3,10 +3,6 @@
|
|||
This document describes how a Bot-Man-Toe server or client is supposed to
|
||||
behave.
|
||||
|
||||
**This document requires familiarity with setting up an HTTP server.** Please
|
||||
use [the easy implementation](../pyserver/README.md) if you're building
|
||||
[a simple agent](../agents/README.md) in this repository.
|
||||
|
||||
## Terminology
|
||||
|
||||
A **server** is a REST API server that hosts a player willing to play games.
|
||||
|
|
@ -69,13 +65,3 @@ package namespace guidelines.
|
|||
}
|
||||
```
|
||||
|
||||
There's a few optional data fields that are specified here:
|
||||
|
||||
| Field | Type | Description |
|
||||
| ----- | ---- | ----------- |
|
||||
| author | string | The name of the person who designed the agent. |
|
||||
| version | string | The version of the agent, in case there's an update. |
|
||||
| me.noordstar.peanuts.containerized | bool | Whether the agent is running in a container. |
|
||||
| me.noordstar.peanuts.is_ai | bool | Whether the agent runs on a trained deep learning model or some artificial intelligence. |
|
||||
| me.noordstar.peanuts.agent.version | string | **Deprecated.** Experimental value to demonstrate an agent's version. Please use `version` instead. |
|
||||
| me.noordstar.peanuts.author | string | **Deprecated.** Experimental value to demonstrate an agent's author. Please use `author` instead. |
|
||||
Loading…
Reference in New Issue