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

Compare commits

base: Bram:2ace9688ffe81de2eb40edcc59b86d5304655136
Branches Tags
Bram:bram/chatgpt
Bram:bram/ur-heuristic
Bram:elo
Bram:bram/greedy
Bram:main
Bram:bram/mute
Bram:sam/sneaky
Bram:bram/deployment
Bram:web-client
Bram:agent/mute-1.0.1
Bram:elo-tracker-1.0.3
Bram:agent/ur-heuristic-1.0.0
Bram:elo-tracker-1.0.2
Bram:elo-tracker-1.0.1
Bram:agent/agent-of-chaos-1.0.1
Bram:agent/sneaky-1.0.0
Bram:agent/naive-line-counter-1.0.0
Bram:elo-tracker-1.0.0
Bram:agent/mute-1.0.0
Bram:agent/agent-of-chaos-1.0.0
..
compare: Bram:bb3bb366651adfc2d9017a0cb03095e9df6bf4db
Branches Tags
Bram:bram/chatgpt
Bram:bram/ur-heuristic
Bram:elo
Bram:bram/greedy
Bram:main
Bram:bram/mute
Bram:sam/sneaky
Bram:bram/deployment
Bram:web-client
Bram:agent/mute-1.0.1
Bram:elo-tracker-1.0.3
Bram:agent/ur-heuristic-1.0.0
Bram:elo-tracker-1.0.2
Bram:elo-tracker-1.0.1
Bram:agent/agent-of-chaos-1.0.1
Bram:agent/sneaky-1.0.0
Bram:agent/naive-line-counter-1.0.0
Bram:elo-tracker-1.0.0
Bram:agent/mute-1.0.0
Bram:agent/agent-of-chaos-1.0.0

No commits in common. "2ace9688ffe81de2eb40edcc59b86d5304655136" and "bb3bb366651adfc2d9017a0cb03095e9df6bf4db" have entirely different histories.
2ace9688ff ... bb3bb36665

21 changed files with 454 additions and 1002 deletions
Show Stats Expand all files Collapse all files

3
.dockerignore
View File

@ -10,9 +10,6 @@ spec/
.venv-pyserver
.venv-pyclient
# Compiled elm files
elo_tracker/static/elo_tracker.js
# ---> Elm
# elm-package generated files
elm-stuff

3
.gitignore vendored
View File

@ -2,9 +2,6 @@
.venv-pyserver
.venv-pyclient
# Compiled elm files
elo_tracker/static/elo_tracker.js
# ---> Elm
# elm-package generated files
elm-stuff

20
agents/__init__.py
View File

@ -1,20 +0,0 @@
"""
Agents are players that can participate in a game. THey contain a unified
API that allows other systems to apply to them for communication.
For example:
1. The PyClient can locally run a game with agents as participants.
2. A PyServer can host a web server and asks a specific agent to respond.
"""
from .agent import Agent
from .chaos import AgentOfChaos
from .mute import MuteAgent
from .remote import RemoteAgent
__all__ = [
"Agent",
"AgentOfChaos",
"MuteAgent",
"RemoteAgent",
]

134
agents/agent.py
View File

@ -1,134 +0,0 @@
"""
The agent module hosts the base class of all agents. It dictates how
agents are expected to behave.
"""
from __future__ import annotations
from typing import Any, Callable
# Communication type that the players communicate in
Payload = dict[str, Any]
# Function that the agent is expected to communicate with.
# The ActionFunction takes a Payload as input and returns a Payload of its own.
ActionFunction = Callable[[Payload], Payload]
GameInfo = tuple[Payload, dict[str, ActionFunction]]
class Agent:
"""
Base class for all agents to communicate in.
"""
# Local variables
author : str
name : str
profile : Payload
registered_games : dict[str, tuple[Payload, dict[str, ActionFunction]]]
version : str | None
def __init__(
self,
name : str,
author : str,
version : str | None = None,
profile : Payload = {},
) -> None:
"""
Create an instance of the agent.
Note that the base class expects quite a lot of variables for the
initialization, whereas the subclasses are expected to require
no inputs for initialization. It is the subclasses' responsibility
to override the current function by inserting its own values.
:param name: String value giving the player a name.
:type name: str
:param author: String value representing the player's designer or programmer.
:type author: str
:param version: Version of the player, in case an update changes its behavior.
:type version: str | None
:param profile: Custom profile pertaining to the player.
:type profile: dict[str, Any]
"""
# Register basic (required) variables
self.author = author
self.name = name
self.profile = profile
self.version = version
# Have the subclass register games
self.registered_games = {}
def add_game(
self,
name : str,
actions : dict[str, ActionFunction],
profile : Payload,
required_actions : list[str] = [],
) -> None:
"""
Add a new game to the player. Replaces any prior definition for the
game.
:param name: Game name
:type name: str
:param actions: Dictionary containing all actions the player is willing to take.
:type actions: dict[str, Callable[[dict[str, Any]], dict[str, Any]]]
:param profile: Custom profile describing some details about a game.
:type profile: dict[str, Any]
:param required_actions: Optional completeness check. If provided,
all required action names must exist in `actions` and map to
callables.
:type required_actions: list[str]
:raises AssertionError: Some of the required actions aren't present.
"""
# Verify that all required actions exist
for ra in required_actions:
if ra not in actions.keys():
raise AssertionError(
f"Missing required action `{ra}` in action mapping."
)
self.registered_games[name] = ( profile, actions )
def add_tic_tac_toe(
self,
on_move : ActionFunction,
profile : Payload = {},
) -> None:
"""
Convenience registration for the tic-tac-toe game.
: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="tic-tac-toe",
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.
: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
"""
_, actions = self.registered_games.get(game, ({}, {}))
if action is None:
return list(actions.values())[0] if len(actions) == 1 else None
else:
return actions.get(action, None)

64
agents/chaos.py
View File

@ -1,64 +0,0 @@
"""
The agent of chaos is an agent that always attempts to return a random
response.
In most games, the Agent of chaos makes two considerations:
1. Which moves are valid? Since invalid moves fallback to "default" moves,
this can destabilize the probability distribution, making some moves
more likely to be chosen than others. As a result, we refrain from
"invalid" moves and take the effort to filter them out.
2. What's a reasonable probability distribution? Sometimes, a uniform
distribution across all options doesn't make a lot of sense. For
example, imagine asking the agent every turn whether they want to
use their super duper special single-use ability, and leaving that to
a 50/50 call every turn. It feels much more random if such an ability
is more randomly used throughout the GAME, than to have every choice
be a uniformly distributed decision.
"""
from __future__ import annotations
import random
from .agent import Agent, Payload
class AgentOfChaos(Agent):
"""
The agent of chaos always aims to deliver a random response.
"""
def __init__(self) -> None:
"""
Create a new instance of the agent of chaos.
"""
super().__init__(
name="Agent of chaos",
author="Bram",
version="1.0.1",
profile={
"me.noordstar.peanuts.is_ai": False,
},
)
self.add_tic_tac_toe(on_move=play_tic_tac_toe, profile={})
def play_tic_tac_toe(payload : Payload) -> Payload:
"""
In tic-tac-toe, the agent of chaos makes uniformly distributed choices
on unclaimed tiles.
:param payload: The incoming game state.
:type payload: dict[str, Any]
:return: The agent of chaos' random choice
:rtype: dict[str, Any]
"""
options = [
int(k)
for k, v in dict(payload).items()
if k in "0123456789" and v == ""
]
return { "move": random.choice(options) }

54
agents/mute.py
View File

@ -1,54 +0,0 @@
"""
The mute agent is a proof-of-concept agent of an agent that simply doesn't
respond - it always responds with an empty object.
This is the simplest agent to implement, and it can be used as a test to
make sure that the "default" move can be picked properly each turn.
"""
from __future__ import annotations
from .agent import ActionFunction, Agent, Payload
def respond_mute(payload : Payload) -> Payload:
"""
Standard response from the mute. Returns an empty object, always.
:param payload: Incoming game state.
:type payload: dict[str, Any]
:return: The empty object.
:rtype: dict[str, Any]
"""
return {}
class MuteAgent(Agent):
"""
The mute agent class refuses to respond to any incoming game states,
and simply responds with an empty dictionary. For most games,
this means that the mute player takes the "default" option as a result
of not responding.
"""
def __init__(self) -> None:
"""
Create a new instance of the mute agent.
"""
super().__init__(
name="Mute",
author="Bram",
version="1.0.1",
profile={
"me.noordstar.peanuts.is_ai": False,
},
)
def get_action(self, game: str, action: str | None) -> ActionFunction:
"""
Return the mute response for any game or action.
:param game: The game that the mute ignores.
:type game: str
:param action: The action that the mute ignores.
:type action: str
"""
return respond_mute

99
agents/remote.py
View File

@ -1,99 +0,0 @@
"""
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
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,
)
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"],
)
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.RequestException, requests.HTTPError, ValueError):
return {}
else:
return content if isinstance(content, dict) else {}

9
client.py
View File

@ -23,8 +23,7 @@ from __future__ import annotations
import json
import pyclient
from agents import Agent, RemoteAgent
from pyclient import PyClient
from pyclient import Agent, PyClient
from pyclient.games import TicTacToe
def main() -> int:
@ -34,11 +33,11 @@ def main() -> int:
:return: Exit code
:rtype: int
"""
c = PyClient(debug=True)
c = PyClient(debug=False)
players : list[Agent] = [
RemoteAgent(url="https://bmt001.noordstar.me/"),
RemoteAgent(url="https://bmt002.noordstar.me/"),
Agent.from_url(url="https://bmt001.noordstar.me/"),
Agent.from_url(url="https://bmt001.noordstar.me/"),
]
out = c.play_game(

156
elm/Api/Elo.elm
View File

@ -1,156 +0,0 @@
module Api.Elo exposing
( EloStat
, HealthCheck
, Leaderboard
, Match
, Participant
, Players
, getHealth
, getLeaderboard
, getMatches
, getPlayers
)
{-|
# ELO API
This module contains functions to address the API of the ELO tracker.
-}
import Http
import Json.Decode as D
type alias EloStat =
{ draws : Int
, elo : Int
, losses : Int
, name : String
, url : String
, wins : Int
}
type alias HealthCheck =
{ ok : Bool, periodicMatches : Bool }
type alias Leaderboard =
{ name : String, players : Players }
type alias Match =
{ name : String
, participants : List Participant
, timestamp : String
}
type alias Participant =
{ name : String
, result : String
, url : String
, version : Maybe String
}
type alias Players =
List EloStat
eloStatDecoder : D.Decoder EloStat
eloStatDecoder =
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.field "wins" D.int)
getHealth :
{ baseUrl : String
, toMsg : Result Http.Error HealthCheck -> msg
}
-> Cmd msg
getHealth data =
Http.get
{ url = data.baseUrl ++ "/health"
, expect = Http.expectJson data.toMsg healthCheckDecoder
}
getLeaderboard :
{ baseUrl : String
, toMsg : Result Http.Error Leaderboard -> msg
}
-> Cmd msg
getLeaderboard data =
Http.get
{ url = data.baseUrl ++ "/leaderboard"
, expect = Http.expectJson data.toMsg leaderboardDecoder
}
getMatches :
{ baseUrl : String
, toMsg : Result Http.Error (List Match) -> msg
}
-> Cmd msg
getMatches data =
Http.get
{ url = data.baseUrl ++ "/matches"
, expect = Http.expectJson data.toMsg (D.list matchDecoder)
}
getPlayers :
{ baseUrl : String
, toMsg : Result Http.Error Players -> msg
}
-> Cmd msg
getPlayers data =
Http.get
{ url = data.baseUrl ++ "/players"
, expect = Http.expectJson data.toMsg playersDecoder
}
healthCheckDecoder : D.Decoder HealthCheck
healthCheckDecoder =
D.map2 HealthCheck
(D.field "ok" D.bool)
(D.field "periodic_matches" D.bool)
leaderboardDecoder : D.Decoder Leaderboard
leaderboardDecoder =
D.map2 Leaderboard
(D.field "name" D.string)
(D.field "players" playersDecoder)
matchDecoder : D.Decoder Match
matchDecoder =
D.map3 Match
(D.field "name" D.string)
(D.field "participants" <| D.list participantDecoder)
(D.field "timestamp" D.string)
participantDecoder : D.Decoder Participant
participantDecoder =
D.map4 Participant
(D.field "name" D.string)
(D.field "result" D.string)
(D.field "url" D.string)
(D.maybe <| D.field "version" D.string)
playersDecoder : D.Decoder Players
playersDecoder =
D.list eloStatDecoder

193
elm/EloTracker.elm
View File

@ -1,193 +0,0 @@
module EloTracker exposing (main)
import Api.Elo as Elo
import Element exposing (Element)
import Http
import Json.Decode as D
import Program
import Time
import Widget.Icon
import Layout
import Theme
import Screen.Leaderboard
import Element.Font
import Pixels
import Element.Background
main =
Program.document
{ flagsDecoder = D.field "base_url" D.string
, headers = headers
, init = init
, subscriptions = subscriptions
, title = title
, update = update
, view = view
}
-- MODEL
type alias Model =
{ baseUrl : String
, leaderboard : Result Int Elo.Leaderboard
, matches : Result Int (List Elo.Match)
}
type Msg
= OnLeaderboard (Result Http.Error Elo.Leaderboard)
| OnMatches (Result Http.Error (List Elo.Match))
| RefreshLeaderboard
| RefreshMatches
init : Result D.Error String -> ( Model, Cmd Msg )
init flag =
let
baseUrl =
case flag of
Err _ ->
"http://localhost:5000"
Ok v ->
v
model =
{ baseUrl = baseUrl
, leaderboard = Err 0
, matches = Err 0
}
in
( model
, Cmd.batch
[ getLeaderboard model
, getMatches model
]
)
-- UPDATE
update : Msg -> Model -> ( Model, Cmd Msg )
update msg model =
case msg of
OnLeaderboard (Err _) ->
case model.leaderboard of
Err n ->
( { model | leaderboard = Err (n + 1) }, Cmd.none )
Ok _ ->
( model, Cmd.none )
OnLeaderboard (Ok v) ->
( { model | leaderboard = Ok v }, Cmd.none )
OnMatches (Err _) ->
case model.matches of
Err n ->
( { model | matches = Err (n + 1) }, Cmd.none )
Ok _ ->
( model, Cmd.none )
OnMatches (Ok v) ->
( { model | matches = Ok v }, Cmd.none )
RefreshLeaderboard ->
( model, getLeaderboard model )
RefreshMatches ->
( model, getMatches model )
getLeaderboard : Model -> Cmd Msg
getLeaderboard model =
Elo.getLeaderboard { baseUrl = model.baseUrl, toMsg = OnLeaderboard }
getMatches : Model -> Cmd Msg
getMatches model =
Elo.getMatches { baseUrl = model.baseUrl, toMsg = OnMatches }
-- SUBSCRIPTIONS
subscriptions : Model -> Sub Msg
subscriptions _ =
Sub.batch
[ Time.every (5 * 1000) (always RefreshLeaderboard)
, Time.every (1 * 60 * 1000) (always RefreshMatches)
]
-- VIEW
headers : Model -> List { icon : Widget.Icon.Icon msg, onPress : msg }
headers _ =
[]
title : Model -> String
title model =
case model.leaderboard of
Err _ ->
"EloTracker | Loading..."
Ok _ ->
"EloTracker | Leaderboard"
view : Program.ViewBox Model -> Element Msg
view data =
case data.model.leaderboard of
Err 0 ->
Element.column
[ Element.width <| Element.px <| Pixels.inPixels data.size.width
, Element.height <| Element.px <| Pixels.inPixels data.size.height
]
[ Element.column
[ Element.centerX
, Element.centerY
, Element.Background.color (Theme.mantleUI data.flavor)
, Element.padding 15
, Element.spacing 10
]
[ Element.text "Loading leaderboard..."
, Layout.loadingIndicator { color = Theme.lavender data.flavor }
|> Element.el [ Element.centerX ]
]
]
Err n ->
Element.column
[ Element.width <| Element.px <| Pixels.inPixels data.size.width
, Element.height <| Element.px <| Pixels.inPixels data.size.height
]
[ Element.column
[ Element.centerX
, Element.centerY
, Element.Background.color (Theme.mantleUI data.flavor)
, Element.padding 15
, 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"))
, Layout.loadingIndicator { color = Theme.lavender data.flavor }
|> Element.el [ Element.centerX ]
]
]
Ok leaderboard ->
Screen.Leaderboard.view
{ flavor = data.flavor, model = leaderboard, size = data.size }

29
elm/Program.elm
View File

@ -156,43 +156,32 @@ view :
-> Html.Html (Msg msg)
view data model =
let
navBarIconHeight =
preferredNavBarHeight =
Pixels.pixels 40
navBarHeight =
Quantity.twice navBarIconHeight
showNavBar =
navBarHeight
preferredNavBarHeight
|> Quantity.multiplyBy 6
|> Quantity.lessThanOrEqualTo model.size.height
contentHeight =
if showNavBar then
model.size.height |> Quantity.minus navBarHeight
model.size.height |> Quantity.minus preferredNavBarHeight
else
model.size.height
in
[ if showNavBar then
viewNavBar
{ headers = data.headers model.content
, iconHeight = navBarIconHeight
, model = model
}
else
Element.none
[ viewNavBar
{ headers = data.headers model.content
, iconHeight = preferredNavBarHeight
, model = model
}
, data.body
{ flavor = model.flavor
, model = model.content
, size =
{ height = contentHeight --|> Quantity.minus (Pixels.pixels 25)
, width = model.size.width --|> Quantity.minus (Pixels.pixels 25)
}
, size = { height = contentHeight, width = model.size.width }
}
|> Element.map OnContent
|> Element.el []
]
|> Element.column [ Element.width Element.fill ]
|> Element.layout

20
elm/Screen/CreateGame.elm
View File

@ -2,23 +2,18 @@ module Screen.CreateGame exposing (..)
-- MODEL
type alias Model =
{ baseUrl : String
, players : List String
}
type Msg
= OnBaseUrl String
| OnPlayer Int String
| RemovePlayer Int
-- UPDATE
update : Msg -> Model -> ( Model, Cmd Msg )
update msg model =
case msg of
@ -27,35 +22,32 @@ update msg model =
OnPlayer n p ->
let
newIndex =
List.length model.players == n
newIndex = List.length model.players == n
newPlayers =
if newIndex && mayCreateNewPlayer model.players then
List.append model.players [ p ]
else
List.indexedMap
(\i player ->
if n == i then
p
else
player
)
model.players
in
( { model | players = newPlayers }, Cmd.none )
( { model | players = newPlayers }, Cmd.none )
RemovePlayer n ->
( { model
| players =
| players =
model.players
|> List.indexedMap
(\i player ->
if n == i then
Nothing
else
Just player
)
@ -64,11 +56,9 @@ update msg model =
, Cmd.none
)
-- SUBSCRIPTIONS
-- VIEW
-- VIEW
mayCreateNewPlayer : List String -> Bool
mayCreateNewPlayer =

90
elm/Screen/Leaderboard.elm
View File

@ -1,90 +0,0 @@
module Screen.Leaderboard exposing (..)
{-| Widget module for a leaderboard page.
-}
import Api.Elo exposing (EloStat, Leaderboard)
import Element exposing (Element)
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 Leaderboard -> Element msg
view data =
let
boardWidth =
data.size.width
|> Quantity.toFloatQuantity
|> Quantity.multiplyBy (2/3)
|> Quantity.floor
|> Quantity.clamp (Pixels.pixels 600) (Pixels.pixels 1800)
|> Pixels.inPixels
|> Element.px
|> Element.width
in
data.model.players
|> 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.column
[ Element.padding 15
, Element.spacing 30
, Element.width <| Element.px <| Pixels.inPixels data.size.width
]
viewEloStat : { flavor : Theme.Flavor } -> EloStat -> Element msg
viewEloStat data stat =
Element.row
[ Element.Background.color (Theme.mantleUI data.flavor)
, Element.Border.rounded 5
, Element.padding 10
, Element.spacing 5
, Element.width Element.fill
]
[ Element.column
[ Element.centerY, Element.spacing 5, Element.width Element.fill ]
[ Element.text stat.name
|> Element.el [ Element.Font.size 30 ]
, Element.el
[ Element.Font.color (Theme.subtext0UI data.flavor)
, Element.Font.size 15
]
(Element.text stat.url)
]
-- Bar to push the rest to the right
, Element.column [ Element.width Element.fill ] []
, field { text = "ELO", value = stat.elo, color = Theme.lavenderUI data.flavor }
|> Element.el [ Element.paddingEach { top = 0, bottom = 0, left = 0, right = 15 } ]
, field { text = "WINS", value = stat.wins, color = Theme.greenUI data.flavor }
, field { text = "DRAWS", value = stat.draws, color = Theme.subtext0UI data.flavor }
, 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
[ Element.Font.color color
, Element.spacing 3
, Element.width (Element.px 70)
]
[ String.fromInt value
|> Element.text
|> Element.el [ Element.centerX, Element.Font.size 25 ]
, text
|> Element.text
|> Element.el [ Element.centerX, Element.Font.size 15 ]
]

60
elo_tracker/Containerfile
View File

@ -1,60 +0,0 @@
FROM alpine:3.10 AS elm-builder
ARG ELM_VERSION=0.19.1
ARG ELM_URL=https://github.com/elm/compiler/releases/download/${ELM_VERSION}/binary-for-linux-64-bit.gz
# Download internet packages + Node & npm
RUN apk add --no-cache ca-certificates curl gzip bash nodejs npm
# Download & install Elm
RUN curl -L ${ELM_URL} \
| gunzip > /usr/local/bin/elm \
&& chmod +x /usr/local/bin/elm
# Install uglify-js globally
RUN npm install -g uglify-js
WORKDIR /app
# Copy Elm code
COPY elm.json .
COPY elm/ elm/
# Compile module
RUN elm make --output=/app/elm.js --optimize elm/EloTracker.elm
# Optimize & Minify compiled JS
RUN uglifyjs elm.js \
--compress "pure_funcs=[F2,F3,F4,F5,F6,F7,F8,F9,A2,A3,A4,A5,A6,A7,A8,A9],pure_getters,keep_fargs=false,unsafe_comps,unsafe" \
| uglifyjs --mangle --output elo_tracker.js
FROM python:3.10-alpine AS python-builder
WORKDIR /app
# Install build dependencies
RUN apk add --no-cache gcc musl-dev python3-dev
COPY requirements-elo.txt .
# Create wheels for faster installation
RUN pip wheel --no-cache-dir --wheel-dir /wheels -r requirements-elo.txt
FROM python:3.10-alpine
WORKDIR /app
# Install from pre-built wheels
COPY --from=python-builder /wheels /wheels
COPY requirements-elo.txt .
RUN pip install --no-cache-dir --no-index --find-links=/wheels \
-r requirements-elo.txt && rm -rf /wheels
# Install Elm front-end JS
COPY --from=elm-builder /app/elo_tracker.js /app/elo_tracker/static/elo_tracker.js
# Install ELO tracker code
COPY agents/ agents/
COPY elo_tracker/ elo_tracker/
COPY pyclient/ pyclient/
COPY pyserver/ pyserver/
COPY elo.py .
CMD ["python", "elo.py"]

41
elo_tracker/app.py
View File

@ -14,7 +14,6 @@ from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any, Sequence
import agents
import pyclient
from pyclient.games import FinishState, Game
@ -38,11 +37,17 @@ 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 : pyclient.ServerAgent) -> "PlayerIdentifier":
"""
Gain a player identifier from an agent.
"""
return cls(name=agent.name, url=agent.url, version=agent.version)
return cls(
name=agent.name,
url=agent.url,
version=agent.profile.get("version",
agent.profile.get("me.noordstar.peanuts.agent.version", None)
),
)
@dataclass()
class EloStat:
@ -176,7 +181,7 @@ class Match:
@classmethod
def from_replay(
cls,
players : list[agents.RemoteAgent],
players : list[pyclient.ServerAgent],
replay : pyclient.GameReplay,
timestamp : str | None,
) -> "Match":
@ -184,7 +189,7 @@ class Match:
Convert a GameReplay into a match.
:param players: The participants of the match.
:type players: list[agents.RemoteAgent]
:type players: list[pyclient.ServerAgent]
:param replay: Game summary.
:type replay: pyclient.GameReplay
:param timestamp: ISO formatted timestamp of when the game was planned.
@ -290,7 +295,7 @@ class EloTracker:
# Thread-unsafe variables
# Please use a lock while doing CRUD operations on them
self.players: list[agents.RemoteAgent] = []
self.players: list[pyclient.ServerAgent] = []
self.__matches: list[Match] = []
self.__stats: dict[PlayerIdentifier, EloStat] = {}
@ -491,20 +496,13 @@ class EloTracker:
<head>
<meta charset="utf-8">
<title>Bot-Man-Toe Elo Tracker</title>
<script src="/static/elo_tracker.js"></script>
</head>
<body>
<main id="main-block">
<main>
<h1>Bot-Man-Toe Elo Tracker</h1>
<p>The JSON API is available at /leaderboard, /matches, /players, and /health.</p>
</main>
</body>
<script>
var app = Elm.EloTracker.init({
node: document.body,
flag: { baseUrl: "http://127.0.0.1:5000" }
});
</script>
</html>
""".strip(),
mimetype="text/html",
@ -560,13 +558,13 @@ class EloTracker:
"Expected `players` field to be a list of strings."
)
players : list[agents.RemoteAgent] = []
players : list[pyclient.ServerAgent] = []
for url in urls:
if not isinstance(url, str):
continue
try:
agent = agents.RemoteAgent(url=url)
agent = pyclient.Agent.from_url(url, debug=self.debug)
except ValueError:
pass # Not an available player right now
else:
@ -588,14 +586,17 @@ class EloTracker:
:rtype: pyclient.GameReplay
:raises ValueError: One of the URLs could not be accessed.
"""
ags : list[Any] = [ agents.RemoteAgent(url=url) for url in players ]
agents : list[Any] = [
pyclient.Agent.from_url(url, debug=self.debug)
for url in players
]
replay = pyclient.PyClient(debug=self.debug).play_game(
players=ags,
players=agents,
start=game,
)
m = Match.from_replay(players=ags, replay=replay, timestamp=Match.now())
m = Match.from_replay(players=agents, replay=replay, timestamp=Match.now())
# Record match
m.log(self.game_file_name)
@ -720,7 +721,7 @@ class EloTracker:
def start_server(
self,
host: str = "0.0.0.0",
host: str = "127.0.0.1",
import_name : str = __name__,
port: int = 5000,
debug: bool = False,

3
pyclient/__init__.py
View File

@ -2,12 +2,15 @@
Entry points for developers who wish to use the PyClient module.
"""
from .agent import Agent, ServerAgent
from .client import PyClient
from .replay import GameReplay
from .transition import Transition
__all__ = [
"Agent",
"GameReplay",
"PyClient",
"ServerAgent",
"Transition",
]

178
pyclient/agent.py Normal file
View File

@ -0,0 +1,178 @@
"""
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
:raises ValueError: The server fails to reach out one of the URLs.
"""
try:
return ServerAgent.from_server_url(url=url, **kwargs)
except (ValueError, requests.RequestException, requests.HTTPError):
pass
raise ValueError(
"URL did not lead to a willing agent"
)
@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
def to_dict(self) -> dict[str, Any]:
"""
Represent the agent in the form of a dict.
:return: Dictionary representation of the ServerAgent
:rtype: dict[str, Any]
"""
return dict(
name=self.name,
games=self.games,
url=self.url,
profile=self.profile,
)

11
pyclient/client.py
View File

@ -5,7 +5,7 @@
from __future__ import annotations
from agents import Agent
from .agent import Agent
from .games import Game
from .replay import GameReplay, Turn
from dataclasses import dataclass
@ -43,17 +43,12 @@ class PyClient:
else:
agent = players[player - 1]
on_move = agent.get_action(
payload = agent.poll(
game=current_state.game_name(),
action=current_state.action_name(),
payload=current_state.as_seen_by(player=player),
)
# Calculate move
if on_move is None:
payload = {}
else:
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)

202
pyserver/server.py
View File

@ -2,39 +2,85 @@
from __future__ import annotations
from flask import Flask, jsonify, request
from agents import Agent
from typing import Any
from collections.abc import Callable, Mapping
from typing import Any, Optional
from flask import Flask, Response, jsonify, request
from functools import wraps
PayloadType = dict[str, Any]
GameHandler = Callable[[PayloadType], PayloadType]
class PyServer:
"""
A tiny stateless Flask app that serves discovery and routes
game behavior to an agent.
"""
"""A tiny stateless Flask app that serves discovery and game routes."""
def __init__(self, agent : Agent, import_name : str = __name__) -> None:
def __init__(self,
name: str,
import_name: str = __name__,
profile: Optional[PayloadType] = None,
subpath : str = "",
) -> None:
"""
Create a PyServer that serves the behavior of an Agent.
Create a PyServer.
:param agent: The agent that one can communicate with.
:type agent: Agent
:param import_name: Flask application import name
:param name: Preferred display name for discovery.
:type name: str
:param import_name: Flask import name.
:type import_name: str
:param profile: Additional root-level discovery metadata.
:type profile: Optional[Dict[str, Any]]
:type subpath: str
:raises ValueError: The input contains invalid information.
"""
self.agent = agent
self.name = name
self.profile = dict(profile or {})
if "name" in self.profile:
raise ValueError(
"Root profile metadata must not define 'name'."
)
if "games" in self.profile:
raise ValueError(
"Root profile metadata must not define 'games'."
)
self.app = Flask(import_name)
self.__games: dict[str, PayloadType] = {}
self.__registered_routes: set[str] = set()
@self.app.get("/")
def discovery():
return jsonify(self.__discovery())
# Register the root
self.__add_api_endpoint("", "", lambda _ : self.__discovery())
# self.app.add_url_rule("/",
# endpoint="botman_discovery",
# view_func=self.__discovery,
# methods=["GET"]
# )
@self.app.get("/<path:game>")
@self.app.get("/<path:game>/<path:action>")
def dispatch(game: str, action: str | None = None):
return jsonify(self.__dispatch(game=game, action=action))
def __add_api_endpoint(self, name : str, route : str, func : GameHandler) -> None:
"""
Create a new API endpoint.
:param name: The name of the action to undertake.
:type name: str
:param func: The player's function that determines what action to take.
:type func: Callable[[dict[str, Any]], dict[str, Any]]
:raises ValueError: The URL has already been registered.
"""
url = self.__make_url(name, route)
if url in self.__registered_routes:
raise ValueError(
f"Route already registered: {url}"
)
self.__registered_routes.add(url)
return self.app.add_url_rule(url,
endpoint="botman_" + name.replace("/", "_"),
view_func=self.__func_wrapper(func),
methods=["GET"],
)
def __discovery(self) -> PayloadType:
"""
@ -43,34 +89,104 @@ class PyServer:
:return: The personal discovery information.
:rtype: dict[str, Any]
"""
d = dict(name=self.agent.name, author=self.agent.author)
if self.agent.version is not None:
d["version"] = self.agent.version
return {
"name": self.name,
"games": dict(self.__games),
**self.profile,
}
d = { **self.agent.profile, **d }
d["games"] = {}
for game, (profile, _) in self.agent.registered_games.items():
d["games"][game] = profile
return d
def __dispatch(self, game: str, action: str | None) -> PayloadType:
def __func_wrapper(self, func : GameHandler) -> Callable[[], Response]:
"""
Resolve a request against the agent's action lookup.
Wrapper that catches an incoming request, parses it, and responds
with a player's action response.
"""
payload = request.get_json(silent=True) or {}
func = self.agent.get_action(
game=game.strip("/"),
action=action.strip("/") if isinstance(action, str) else None,
@wraps(func)
def exec():
payload = request.get_json(silent=True) or {}
result = func(payload) or {}
return jsonify(result)
return exec
def __make_url(self, name : str, route : str) -> str:
return "/" + "/".join([ name.strip("/"), route.strip("/") ]).strip("/")
def add_game(
self,
name: str,
profile: PayloadType,
actions: dict[str, GameHandler],
required_actions: Optional[list[str]] = None,
) -> None:
"""
Register a stateless game with one or more action routes.
:param name: Base route name for the game.
:type name: str
:param profile: Game-specific discovery metadata.
:type profile: dict[str, Any]
:param actions: Mapping of action subpaths to handlers.
:type actions: dict[str, Callable[[dict[str, Any]], dict[str, Any]]]
:param required_actions: Optional completeness check. If provided, all
required action names must exist in ``actions`` and map to callables.
:type required_actions: Optional[list[str]]
:raises AssertionError: Some of the required actions aren't present.
:raises ValueError: Some of the requested URL paths are already occupied.
"""
# Verify that all required actions are present
if required_actions is not None:
missing_actions = [ action for action in required_actions if action not in actions ]
if len(missing_actions) > 0:
raise AssertionError(
f"Missing required action handlers: {', '.join(sorted(missing_actions))}"
)
# Verify that there are no duplicate URLs being registered
# Even though this will automatically be checked later, checking this
# now ensures that the operation is atomic and doesn't add
# games partially.
new_routes : set[str] = set()
for route in actions:
url = self.__make_url(name, route)
if url in self.__registered_routes:
raise ValueError(
"Route {url} was already registered"
)
if url in new_routes:
raise ValueError(
"Route {url} was registered twice by the same function call"
)
new_routes.add(url)
# Register all actions
for route, func in actions.items():
self.__add_api_endpoint(name=name, route=route, func=func)
# Add profile data
self.__games[name] = profile
def add_tic_tac_toe(self, on_move: GameHandler, profile: PayloadType = {}) -> None:
"""
Convenience registration for tic-tac-toe.
The game is exposed at `/tic-tac-toe`.
:param profile: The player's custom profile.
:type profile: dict[str, Any]
:param on_move:
"""
self.add_game(
name="tic-tac-toe",
profile=profile,
actions={"": on_move},
required_actions=[""],
)
if func is None:
return {}
result = func(payload) or {}
return result if isinstance(result, dict) else {}
def start(self, host: str = "127.0.0.1", port: int = 5000, debug: bool = False, **kwargs: Any) -> None:
"""Start the Flask development server."""
self.app.run(host=host, port=port, debug=debug, **kwargs)

13
requirements-elo.txt
View File

@ -1,13 +0,0 @@
blinker==1.9.0
certifi==2026.6.17
charset-normalizer==3.4.7
click==8.4.1
colorama==0.4.6
Flask==3.1.3
idna==3.18
itsdangerous==2.2.0
Jinja2==3.1.6
MarkupSafe==3.0.3
requests==2.34.2
urllib3==2.7.0
Werkzeug==3.1.8

74
server.py
View File

@ -4,9 +4,10 @@
from __future__ import annotations
import agents
import random
from pyserver import PyServer
from typing import Any
def main() -> int:
"""
@ -15,7 +16,19 @@ def main() -> int:
:return: Exit code
:rtype: int
"""
player = PyServer(agents.RemoteAgent(url="https://bmt001.noordstar.me/"))
player = PyServer(
# Customize this to whatever you'd like to call your player
name="My super smart robot player",
# Custom information that you can use to tell people about this player
profile={},
# Unless you know what you're doing, don't touch this.
import_name=__name__,
)
# Register games! Comment out any you don't want your player to play.
player.add_tic_tac_toe(on_move=play_tic_tac_toe, profile={})
# Start listening for games
player.start(
@ -25,5 +38,62 @@ def main() -> int:
return 0
def play_tic_tac_toe(payload : dict[str, Any]) -> dict[str, Any]:
"""
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 logic.
return { "move": random.choice(options) }
if __name__ == "__main__":
raise SystemExit(main())