How to implement a client¶

For implementing a client, utilize the comprl.client package.

Create a class that inherits from comprl.client.Agent. The only required implementation is the get_step() methods. Additionally, you may override other methods to receive and handle data about games being played or handle any errors that may occur.

You can then use comprl.client.launch_client() to launch the client in a standardized way.

A simple example can be found below.

Client API¶

class comprl.client.Agent¶

Agent used by the end-user connecting to the server.

This class represents an agent that interacts with the server. It provides methods for registering event handlers and connecting to the server.

final run(token: str, host: str = 'localhost', port: int = 65335) → None¶

Connects the client to the server.

This method connects the client to the server using the specified token, host, and port. It internally calls the run method of the base class to establish the connection.

Parameters:¶

token : str¶: The token used for authentication.
host : str¶: The host address of the server. Defaults to “localhost”.
port : int¶: The port number of the server. Defaults to 65335.

Returns:¶

None

abstractmethod get_step(obv: list[float]) → list[float]¶

Requests the agent’s action based on the current observation.

Parameters:¶

obv : list[float]¶: The current observation.

Returns:¶

The agent’s action.

Return type:¶

list[float]

is_ready() → bool¶

Returns if the agent is ready to play.

The default implementation always returns True. May be overridden to implement some method of non-disruptive disconnection (e.g. stop after 10 games).

Returns:¶: True if the agent is ready to play, False otherwise.
Return type:¶: bool

on_start_game(game_id: int) → None¶

Called when a new game starts.

The default implementation does nothing. Override to implement custom behavior.

Parameters:¶

game_id : int¶: The ID of the new game.

on_end_game(result: bool, stats: list[float]) → None¶

Called when a game ends.

The default implementation does nothing. Override to implement custom behavior.

Parameters:¶

result: bool¶: The result of the game (True if won, False if lost or draw).
stats: list[float]¶: The statistics of the game. The meaning of the values depends on the game implementation.

on_disconnect()¶

Called when the agent disconnects from the server.

Prints an error message to stdout. May be overridden to implement custom disconnect handling.

on_message(msg: str)¶

Called if a message is sent from the server.

Prints the message to stdout. May be overridden to implement custom message handling.

on_error(msg: str)¶

Called if an error occurred on the server side.

Prints the message to stdout. May be overridden to implement custom error handling.

comprl.client.launch_client(initialize_agent_func: collections.abc.Callable[[list[str]], Agent])¶

Launch the comprl client and connect to the server.

This function parses command line arguments to get the server connection information (url, port and access token). Alternatively, these arguments can also be set via environment variables COMPRL_SERVER_URL, COMPRL_SERVER_PORT and COMPRL_ACCESS_TOKEN. It then initializes an Agent instance using the given function.

Custom arguments for initialize_agent_func can be passed on the command line using --args.

Parameters:¶

initialize_agent_func: collections.abc.Callable[[list[str]], Agent]¶: Function that returns an initialized Agent instance. The function takes as argument a (possibly empty) list of command line arguments that are passed after --args. It is expected to return an instance of a subclass of Agent.

Example: Hockey client¶

from __future__ import annotations

import argparse
import uuid

import hockey.hockey_env as h_env
import numpy as np


from comprl.client import Agent, launch_client


class RandomAgent(Agent):
    """A hockey agent that simply uses random actions."""

    def get_step(self, observation: list[float]) -> list[float]:
        return np.random.uniform(-1, 1, 4).tolist()

    def on_start_game(self, game_id) -> None:
        print("game started")

    def on_end_game(self, result: bool, stats: list[float]) -> None:
        text_result = "won" if result else "lost"
        print(
            f"game ended: {text_result} with my score: "
            f"{stats[0]} against the opponent with score: {stats[1]}"
        )


class HockeyAgent(Agent):
    """A hockey agent that can be weak or strong."""

    def __init__(self, weak: bool) -> None:
        super().__init__()

        self.hockey_agent = h_env.BasicOpponent(weak=weak)

    def get_step(self, observation: list[float]) -> list[float]:
        # NOTE: If your agent is using discrete actions (0-7), you can use
        # HockeyEnv.discrete_to_continous_action to convert the action:
        #
        # from hockey.hockey_env import HockeyEnv
        # env = HockeyEnv()
        # continuous_action = env.discrete_to_continous_action(discrete_action)

        action = self.hockey_agent.act(observation).tolist()
        return action

    def on_start_game(self, game_id) -> None:
        game_id = uuid.UUID(int=int.from_bytes(game_id))
        print(f"Game started (id: {game_id})")

    def on_end_game(self, result: bool, stats: list[float]) -> None:
        text_result = "won" if result else "lost"
        print(
            f"Game ended: {text_result} with my score: "
            f"{stats[0]} against the opponent with score: {stats[1]}"
        )


# Function to initialize the agent.  This function is used with `launch_client` below,
# to lauch the client and connect to the server.
def initialize_agent(agent_args: list[str]) -> Agent:
    # Use argparse to parse the arguments given in `agent_args`.
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--agent",
        type=str,
        choices=["weak", "strong", "random"],
        default="weak",
        help="Which agent to use.",
    )
    args = parser.parse_args(agent_args)

    # Initialize the agent based on the arguments.
    agent: Agent
    if args.agent == "weak":
        agent = HockeyAgent(weak=True)
    elif args.agent == "strong":
        agent = HockeyAgent(weak=False)
    elif args.agent == "random":
        agent = RandomAgent()
    else:
        raise ValueError(f"Unknown agent: {args.agent}")

    # And finally return the agent.
    return agent


def main() -> None:
    launch_client(initialize_agent)


if __name__ == "__main__":
    main()

Running the client:

The client needs server URL, port and the users access token as arguments (handled by comprl.client.launch_client()). Any custom arguments of the agent can be passed in the end using --args:

python3 ./client.py --server-url <URL> --server-port <PORT> \
    --token <YOUR ACCESS TOKEN> \
    --args --agent=strong

The server information can also be provided via environment variables, then they don’t need to be provided via the command line:

# put this in your .bashrc or some other file that is sourced before running the agent
export COMPRL_SERVER_URL=<URL>
export COMPRL_SERVER_PORT=<PORT>
export COMPRL_ACCESS_TOKEN=<YOUR ACCESS TOKEN>

Then just call

python3 ./run_client.py --args --agent=strong