putnam.py

from asyncio import sleep
from enum import Enum, auto
from random import choice, shuffle
from time import monotonic


class GameStates(Enum):
    NOT_STARTED = auto()
    CHOOSING_SETUP = auto()
    ALICE_TURN = auto()
    BOB_TURN = auto()
    GAME_OVER = auto()
    ABORTED = auto()


BOT = "Bot Player"


class Putnam:
    TEMPLATE_NAME = "putnam.html"

    def __init__(self, connections, when_started=None, when_finished=None):
        self.connections = connections  # mutable reference outside of this scope.
        self.players = ()  # should be immutable or owned by this class
        self.lobby = set()
        self.ready = set()
        self.when_started = when_started
        self.when_finished = when_finished
        self.state = GameStates.NOT_STARTED
        self.alice = self.bob = None
        self.holes = []
        self.last_communication = monotonic()

        self.STATES = {
            GameStates.NOT_STARTED: self.not_started,
            GameStates.CHOOSING_SETUP: self.choosing_setup,
            GameStates.ALICE_TURN: self.alice_turn,
            GameStates.BOB_TURN: self.bob_turn,
            GameStates.GAME_OVER: self.game_is_over,
            GameStates.ABORTED: lambda player_id, move: 0,  # do nothing
        }

    async def player_move(self, player_id, move, queue):
        self.communicated()
        if type(move) is not dict:
            return
        if move.get("kind") == "catch_up":
            return await self.catch_up(queue)
        handler = self.STATES[self.state]
        await handler(player_id, move)

    async def not_started(self, player_id, move):
        kind = move.get("kind")
        if kind == "start":
            await self.mark_ready(player_id)
        if kind == "add_bot":
            await self.join(BOT)
            await self.mark_ready(BOT)

    async def choosing_setup(self, player_id, move):
        if move.get("kind") == "choose_setup" and player_id == self.bob:
            try:
                num_pegs = int(move.get("num_pegs"))
                num_holes = int(move.get("num_holes"))
            except ValueError:
                return
            if num_pegs >= num_holes or num_pegs <= 0 or num_holes <= 0:
                return
            self.holes = [True] * num_pegs + [False] * (num_holes - num_pegs)
            self.state = GameStates.ALICE_TURN
            await self.broadcast(self.board_state_message())
            await self.possible_bot_move()

    async def alice_turn(self, player_id, move):
        if move.get("kind") == "take_turn" and player_id == self.alice:
            await self.take_turn(move, GameStates.BOB_TURN)

    async def bob_turn(self, player_id, move):
        if move.get("kind") == "take_turn" and player_id == self.bob:
            await self.take_turn(move, GameStates.ALICE_TURN)

    async def game_is_over(self, player_id, move):
        pass

    async def broadcast(self, message):
        for client, _ in self.connections:
            await client.put(message)
        self.communicated()

    def can_join(self, player_id):
        return True

    async def join(self, player_id):
        if self.state == GameStates.NOT_STARTED:
            self.lobby.add(player_id)
            await self.broadcast({"kind": "lobby_update", "players": list(self.lobby)})

    async def discard(self, player_id):
        # since we allow duplicate connections, we can't go off of player_id alone.
        if self.state == GameStates.NOT_STARTED:
            lobby = set(name for _, name in self.connections)
            if not lobby:
                self.state = GameStates.ABORTED
                if self.when_finished is not None:
                    self.when_finished()
            if lobby != self.lobby:
                self.lobby = lobby
                await self.broadcast(
                    {"kind": "lobby_update", "players": list(self.lobby)}
                )
                await self.start()

    async def mark_ready(self, player_id):
        self.ready.add(player_id)
        await self.broadcast(
            {"kind": "ready_update", "waiting": list(self.lobby - self.ready)}
        )
        await self.start()

    async def start(self):
        players = set(name for _, name in self.connections if name in self.ready)
        if BOT in self.ready:
            players.add(BOT)

        if len(players) != 2:
            return  # client made a bad request

        self.state = GameStates.CHOOSING_SETUP
        self.players = tuple(players)

        if self.when_started is not None:
            self.when_started()

        self.choose_player_order()

        await self.broadcast(self.start_message())
        await self.broadcast({"kind": "want_game_parameters"})

    def start_message(self):
        return {
            "kind": "game_start",
            "players": self.players,
            "alice": self.alice,
            "bob": self.bob,
        }

    def choose_player_order(self):
        players = list(self.players)
        shuffle(players)
        self.alice, self.bob = players
        if self.bob == BOT:
            self.alice, self.bob = self.bob, self.alice

    async def take_turn(self, move, new_state):
        try:
            start = int(move.get("from"))
            end = int(move.get("to"))
        except ValueError:
            return
        if (
            0 <= start < end < len(self.holes)
            and self.holes[start]
            and not self.holes[end]
        ):
            self.holes[start] = False
            self.holes[end] = True
            if self.game_ended():
                await self.end_game()
            else:
                self.state = new_state
                await self.broadcast(self.board_state_message())
                await self.possible_bot_move()

    async def end_game(self):
        winner = self.current_player()
        self.state = GameStates.GAME_OVER
        message = self.board_state_message()
        message["kind"] = "game_over"
        message["winner"] = winner
        await self.broadcast(message)
        if self.when_finished is not None:
            self.when_finished()

    def game_ended(self):
        found_peg = False
        for hole_has_peg in self.holes:
            found_peg = found_peg or hole_has_peg
            if found_peg and not hole_has_peg:
                return False
        return True

    async def possible_bot_move(self):
        if self.current_player() == BOT:
            from_, to = putnam_bot(self.holes)
            next_state = (
                GameStates.ALICE_TURN
                if self.state == GameStates.BOB_TURN
                else GameStates.BOB_TURN
            )
            await sleep(1)
            await self.take_turn({"from": from_, "to": to}, next_state)

    def abort(self):
        """Provide the ability to forcibly abort the game from an external caller."""
        self.state = GameStates.ABORTED
        if self.when_finished is not None:
            self.when_finished()

    async def catch_up(self, connection):
        if self.state in (GameStates.NOT_STARTED, GameStates.GAME_OVER):
            return

        await connection.put(self.start_message())

        if self.state == GameStates.CHOOSING_SETUP:
            await connection.put({"kind": "want_game_parameters"})
        else:
            await connection.put(self.board_state_message())

    def board_state_message(self):
        return {
            "kind": "board_state",
            "holes": self.holes,
            "current_player": self.current_player(),
        }

    def current_player(self):
        if self.state == GameStates.ALICE_TURN:
            return self.alice
        if self.state == GameStates.BOB_TURN:
            return self.bob

    def communicated(self):
        self.last_communication = monotonic()


def putnam_bot(board):
    """
    Make a bot move based on the current game state.

    If the game is winnable, make the (one and only) optimal move.
    Otherwise, make a random move to confuse the human.

    Make the move that preserves "even alignment" on a board with an even
    number of pegs and an even number of holes.

    This even alignment is that we want all pegs to be grouped in pairs such
    that the first peg has an even index and the second peg has an odd index.
    If we're in a winnable state, there's exactly one move we can make to
    create alignment; otherwise, we are already aligned and so any single
    move will disrupt the alignment.

    Note that if we shrink the problem by moving away the leftmost peg, then
    the evenness/oddness of our pegs swaps, so we need to consider the other
    pairing as well.

    We want to take a move when there are 2 (mod 4) parity errors, because we
    want our opponent to have 0 (mod 4) parity errors on their turn.

    :param board: A list of bools, where True means there's a peg in the hole
    :return: (from, to) tuple of the index of the peg that is being moved and
             the index of the whole it's being moved to
    """
    # 0-based pairings (0 w/ 1, 2 w/ 3, etc.)
    parity_errors = [i for i in range(0, len(board), 2) if parity_error(i, board)]
    if len(parity_errors) % 4 == 2:
        return peg(parity_errors[0], board), hole(parity_errors[1], board)
    # 1-based pairings (-1 w/ 0, 1 w/ 2, etc.)
    parity_errors = [i for i in range(-1, len(board), 2) if parity_error(i, board)]
    if len(parity_errors) % 4 == 2:
        return peg(parity_errors[0], board), hole(parity_errors[1], board)
    # The game is unwinnable. Make a random move.
    return random_move(board)


def parity_error(i, board):
    """
    Check if the pair (i, i+1) in the board is a parity error.
    :param i: an index (possibly outside of board)
    :param board: The board
    :return: a bool that says whether this is a parity error.
    """
    return safe_board_lookup(i, board) != safe_board_lookup(i + 1, board)


def safe_board_lookup(i, board):
    """
    Safely look up values from the board.

    The left side of the board is infinitely many holes, and the right side
    is infinitely many pegs.

    :param i: any (possibly invalid) index
    :param board: The board
    :return: True for peg or False for hole
    """
    if i < 0:
        return False
    if i >= len(board):
        return True
    return board[i]


def peg(i, board):
    """Return the peg from a bad-parity pair."""
    return i if safe_board_lookup(i, board) else i + 1


def hole(i, board):
    """Return the peg from a bad-parity pair."""
    return i + 1 if safe_board_lookup(i, board) else i


def random_move(board):
    """
    Make a random move.

    :param board: A list of bools, where True means there's a peg in the hole.
                  All pegs are assumed to be movable.
    :return: (from, to) tuple of the index of the peg that is being moved and
             the index of the whole it's being moved to
    """
    # can't move these pegs right
    while board[-1]:
        board = board[:-1]
    peg_positions = [i for i, p in enumerate(board) if p]
    random_peg = choice(peg_positions)
    hole_positions = [i for i, p in enumerate(board) if i > random_peg and not p]
    random_hole = choice(hole_positions)
    return random_peg, random_hole