How to implement a new game

Implementing the IGame Interface

The Game Class only implements game logic and does not have to handle networking (e.g. collecting actions, disconnected players), matchmaking or database tasks. In general, a Game can also implement a game with more than two players, but the Matchmaking and the Database has to be updated for this.

Start implementing the interface with

from comprl.server.interfaces import IGame, IPlayer

class MyGame(IGame):
    def __init__(self, players: list[IPlayer]) -> None:
        super().__init__(players)

    ...

and implement all the necessary methods listed below:

  • _update()

    • Updates the environment

    • Input: All the collected actions in a dictionary

    • Returns True if the game should end False otherwise

  • _validate_action()

    • Returns True if the action is valid

    • Disconnects the player if the action is invalid which ends the game

  • _get_observation()

    • Returns the observation that should be sent to this player

    • Observations can be sent to each player individually (e.g. for left and right players)

  • _player_won()

    • Returns True if the player has won, False otherwise

    • Should return False for all players in case of a draw

  • _player_stats()

    • Is called at the end of the game

    • Can be used to communicate Statistics at the end of the game

    • e.g. scores, disconnects, rounds, …

For the implementation of these methods the following instance variables can be helpful:

  • players: a dictionary with all players of the game

  • scores: a dictionary with all scores (defaults to all scores 0.0)

Stored game actions

The game implementation can store arbitrary information about the game in the dictionary comprl.server.interfaces.IGame.game_info. This dictionary is automatically saved to a pickle file at the end of the game. The file is written to {data_dir}/game_actions/{game_id}.pkl (see comprl.server.config.Config.data_dir).

Example: Hockey game

For a real-life example on how to implement a server for a specific game, see the implementation in comprl-hockey-game (in the comprl repository).

API

class comprl.server.interfaces.IGame(players: list[IPlayer])

Interface for a game.

id

The unique identifier of the game.

Type:

GameID

players

A dictionary of players participating in the game.

Type:

dict[PlayerID, IPlayer]

start_time

The start time of the game.

Type:

datetime

finish_callbacks

A list of callbacks to be executed when the game ends.

Type:

list[Callable[[“IGame”], None]]

scores

Scores of the players.

Type:

dict[PlayerID, float]

_end(reason='unknown')

Notifies all players that the game has ended and writes all actions in a file.

Parameters:
reason : str

The reason why the game has ended. Defaults to “unknown”.

abstractmethod _get_observation(id: UUID) → list[float]

Returns the observation for the player.

Parameters:
id : PlayerID

The ID of the player for which the observation is requested.

Returns:

The observation for the player.

Return type:

list[float]

abstractmethod _player_stats(id: UUID) → list[float]

Returns all stats that should be sent to the player if the game has ended.

Parameters:
id : PlayerID

The ID of the player.

Returns:

The result of the player.

Return type:

int

abstractmethod _player_won(id: UUID) → bool

Checks whether the player has won.

Parameters:
id : PlayerID

The ID of the player to be checked.

Returns:

True if the player has won, False otherwise.

Return type:

bool

_run()

Collects all actions and puts them in the current_actions list.

abstractmethod _update(actions: dict[UUID, list[float]]) → bool

Updates the game with the players’ actions.

Returns:

True if the game is over, False otherwise.

Return type:

bool

abstractmethod _validate_action(action) → bool

Checks whether an action is valid.

Parameters:
action

The action to be validated.

Returns:

True if the action is valid, False otherwise.

Return type:

bool

add_finish_callback(callback: Callable[[IGame], None]) → None

Adds a callback function to be executed when the game ends.

Parameters:
callback : Callable[["IGame"], None]

The callback function to be added.

force_end(player_id: UUID)

forces the end of the game. Should be used when a player disconnects.

Parameters:
player_id : PlayerID

the player that caused the forced end (disconnected)

get_result() → GameResult | None

Returns the result of the game.

Returns:

The result of the game.

Return type:

GameResult

start()

Notifies all players that the game has started and starts the game cycle.

class comprl.server.interfaces.IPlayer

Interface for a player

abstractmethod authenticate(result_callback)

authenticates player

Parameters:
result_callback : Callable

callback

abstractmethod disconnect(reason: str)

disconnect the player

abstractmethod get_action(obv, result_callback) → IAction

gets an action from the player

Parameters:
obv : Any

observation

result_callback : Callable

callback

Returns:

action

Return type:

IAction

abstractmethod is_ready(result_callback) → bool

checks if the player is ready to play

Returns:

returns true if the player is ready to play

Return type:

bool

abstractmethod notify_end(result, stats)

notifies player that the game has ended

abstractmethod notify_error(error: str)

notifies the player of an error

abstractmethod notify_info(msg: str)

notifies the player of an information

abstractmethod notify_start(game_id)

notifies player that the game has started