API Wrapper Documentation:

OGSApiException

Bases: Exception

Exception raised for errors in the OGS API.

Source code in src/ogsapi/ogs_api_exception.py

class OGSApiException(Exception):
    """Exception raised for errors in the OGS API."""
    # TODO: Handle exceptions lol
    pass

InterceptHandler

Bases: Handler

Intercepts the logs from SocketIO, EngineIO, and urllib and sends them to the logger

Source code in src/ogsapi/client.py

class InterceptHandler(logging.Handler):
    """Intercepts the logs from SocketIO, EngineIO, and urllib and sends them to the logger"""
    def emit(self, record) -> None:
        """Parse the log and emit to the logger"""
        # Get corresponding Loguru level if it exists.
        try:
            level = logger.level(record.levelname).name
        except ValueError:
            level = record.levelno

        # Find caller from where originated the logged message.
        frame, depth = sys._getframe(6), 6
        while frame and frame.f_code.co_filename == logging.__file__:
            frame = frame.f_back # type: ignore[assignment]
            depth += 1

        # If log is from engineio.client set to TRACE and if from socketio.client set to DEBUG
        if record.name == "engineio.client" and record.levelname == "INFO":
            level = "TRACE"
        elif record.name == "socketio.client" and record.levelname == "INFO":
            level = "DEBUG"

        logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())

emit(record)

Parse the log and emit to the logger

Source code in src/ogsapi/client.py

def emit(self, record) -> None:
    """Parse the log and emit to the logger"""
    # Get corresponding Loguru level if it exists.
    try:
        level = logger.level(record.levelname).name
    except ValueError:
        level = record.levelno

    # Find caller from where originated the logged message.
    frame, depth = sys._getframe(6), 6
    while frame and frame.f_code.co_filename == logging.__file__:
        frame = frame.f_back # type: ignore[assignment]
        depth += 1

    # If log is from engineio.client set to TRACE and if from socketio.client set to DEBUG
    if record.name == "engineio.client" and record.levelname == "INFO":
        level = "TRACE"
    elif record.name == "socketio.client" and record.levelname == "INFO":
        level = "DEBUG"

    logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())

OGSClient

Connect to and interact with the OGS REST API and SocketIO API.

Examples:

>>> from ogsapi.client import OGSClient
>>> ogs = OGSClient(
    client_id=client_id, 
    client_secret=client_secret, 
    username=username, 
    password=password,
    )
Connecting to Websocket
Connected to socket, authenticating

Parameters:

Name	Type	Description	Default
client_id	str	Client ID from OGS	required
client_secret	str	Client Secret from OGS	required
username	str	Username of OGS account	required
password	str	Password of OGS account	required
dev	bool	Use the development API. Defaults to False.	False

Attributes:

Name	Type	Description
credentials	OGSCredentials	Credentials object containing all credentials
api	OGSRestAPI	REST API connection to OGS
sock	OGSSocket	SocketIO connection to OGS

Source code in src/ogsapi/client.py

class OGSClient:
    """Connect to and interact with the OGS REST API and SocketIO API.

    Examples:
        >>> from ogsapi.client import OGSClient
        >>> ogs = OGSClient(
            client_id=client_id, 
            client_secret=client_secret, 
            username=username, 
            password=password,
            )
        Connecting to Websocket
        Connected to socket, authenticating

    Args:
        client_id (str): Client ID from OGS
        client_secret (str): Client Secret from OGS
        username (str): Username of OGS account
        password (str): Password of OGS account
        dev (bool, optional): Use the development API. Defaults to False.    

    Attributes:
        credentials (OGSCredentials): Credentials object containing all credentials
        api (OGSRestAPI): REST API connection to OGS
        sock (OGSSocket): SocketIO connection to OGS

    """
    def __init__(self, client_id, client_secret, username, password, dev: bool = False):

        self.credentials = OGSCredentials(client_id=client_id, client_secret=client_secret,
                                          username=username, password=password)
        self.api = OGSRestAPI(self.credentials,dev=dev)
        self.credentials.user_id = self.user_vitals()['id']

    def enable_logging(self) -> None:
        """Enable logging from ogsapi"""
        logger.enable("src.ogsapi")

    def disable_logging(self) -> None:
        """Disable logging from ogsapi"""
        logger.disable("src.ogsapi")

    # User Specific Resources: /me

    def user_vitals(self) -> dict:
        """Get the user's vitals.

        Returns:
            response (dict): JSON response from the endpoint
        """

        endpoint = '/me'
        logger.info("Getting user vitals")
        return self.api.call_rest_endpoint('GET', endpoint=endpoint).json()

    def user_settings(self) -> dict:
        """Get the user's settings.

        Returns:
            response (dict): JSON response from the endpoint
        """

        endpoint = '/me/settings/'
        logger.info("Getting user settings")
        return self.api.call_rest_endpoint('GET', endpoint=endpoint).json()

    def update_user_settings(
            self, username: str | None = None,
            first_name: str | None = None,
            last_name: str | None = None,
            private_name: bool | None = None,
            country: str | None = None,
            website: str | None = None,
            about: str | None = None
        ) -> dict:
        """Update the user's settings.

        Args:
            username (str, optional): New username. Defaults to None.
            first_name (str, optional): New first name. Defaults to None.
            last_name (str, optional): New last name. Defaults to None.
            private_name (bool, optional): Whether or not to make the name private. Defaults to None.
            country (str, optional): New country. Defaults to None.
            website (str, optional): New website. Defaults to None.
            about (str, optional): New about. Defaults to None.

        Returns:
            response (dict): JSON response from the endpoint
        """

        # This is a bit of a mess, but it works, should be refactored
        payload: dict[str, Any] = {}
        if username is not None:
            payload['username'] = username
        if first_name is not None:
            payload['first_name'] = first_name
        if last_name is not None:
            payload['last_name'] = last_name
        if private_name is not None:
            payload['real_name_is_private'] = private_name
        if country is not None:
            payload['country'] = country
        if website is not None:
            payload['website'] = website
        if about is not None:
            payload['about'] = about

        endpoint = f'/players/{self.credentials.user_id}'
        # Add the inputs to a payload, only if they are not None
        logger.info(f"Updating user settings with the following payload: {payload}")
        return self.api.call_rest_endpoint('PUT', endpoint=endpoint, payload=payload).json()

    def active_games(self, player_id: int | None = None) -> list[dict]:
        """
        Get the user's active games.

        Args:
            player_id (int, optional): ID of player whos active games we want to retrieve.

        Returns:
            response (list[dict]): JSON object containing players active games.
        """

        if player_id is None:
            endpoint = "/ui/overview"
        else:
            endpoint = f"/players/{player_id}/full"
        return self.api.call_rest_endpoint('GET', endpoint=endpoint).json()["active_games"]


    def user_games(self, page: int = 1, page_size: int = 10) -> dict:
        """Get the user's games.

        Args:
            page (int): Page number of user games. Defaults to page 1
            page_size (int): Number of games per page. Defaults to 10

        Returns:
            response (dict): JSON response from the endpoint
        """

        endpoint = '/me/games'
        params = { 'page': page, 'page_size': page_size }
        logger.info(f"Getting user games - page {page} with page size {page_size}")
        return self.api.call_rest_endpoint('GET', endpoint=endpoint, params=params).json()


    def user_friends(self, username: str | None = None) -> dict:
        """Get the user's friends.

        Args:
            username (str, optional): Username of the user to get friends of. Defaults to None.

        Returns:
            response (dict): JSON response from the endpoint
        """

        endpoint = '/me/friends'
        logger.info("Getting user friends")
        return self.api.call_rest_endpoint('GET', endpoint=endpoint, params={'username' : username}).json()

    def send_friend_request(self, username: str) -> dict:
        """Send a friend request to a user.

        Args:
            username (str): Username of the user to send a friend request to.

        Returns:
            response (dict): JSON response from the endpoint
        """

        endpoint = '/me/friends'
        player_id = self.get_player(username)['id']
        payload = {
            "player_id" : player_id
        }
        logger.info(f"Sending friend request to {username} - {player_id}")
        return self.api.call_rest_endpoint('POST', endpoint=endpoint, payload=payload).json()

    def remove_friend(self, username: str) -> dict:
        """Remove a friend.

        Args:
            username (str): Username of the user to remove as a friend.

        Returns:
            response (dict): JSON response from the endpoint
        """

        endpoint = '/me/friends/'
        player_id = self.get_player(username)['id']
        payload = {
            "delete": True,
            "player_id" : player_id
        }
        logger.info(f"Removing friend {username} - {player_id}")
        return self.api.call_rest_endpoint('POST', endpoint=endpoint, payload=payload).json()

    # Players: /players

    def get_player(self, player_username: str) -> dict:
        """Get a player by username.

        Args:
            player_username (str): Username of the player to get.

        Returns:
            player_data (dict): Player data returned from the endpoint
        """

        endpoint = '/players/'
        logger.info(f"Getting player {player_username}")
        return self.api.call_rest_endpoint('GET', endpoint=endpoint, params={'username' : player_username}).json()['results'][0]

    def get_player_games(self, player_username: str) -> dict:
        """Get a player's games by username.

        Args:
            player_username (str): Username of the player to get games of.

        Returns:
            player_games (dict): Player games returned from the endpoint
        """
        logger.info(f"Getting player {player_username}'s games")
        player_id = self.get_player(player_username)['id']
        endpoint = f'/players/{player_id}/games'
        return self.api.call_rest_endpoint('GET', endpoint=endpoint).json()

    # TODO: This needs to be using a dataclass to make challenge customization easier
    def create_challenge(self, player_username: str | None = None, **game_settings) -> tuple[int, int]:
        """Create either an open challenge or a challenge to a specific player.
        The time control settings are built depending on which time control is used.
        Make sure that you pass the correct time control settings for the time control you want to use.
        The other time control settings will be ignored.

        Examples:
            >>> ogs.create_challenge(player_username='test', main_time=300, byoyomi_time=30, byoyomi_stones=5)
            Challenging player: test - 1234567
            (20328495, 53331333)

        Args:
            player_username (str): Username of the player to challenge. 
                If used will issue the challenge to the player. Defaults to None.

        Keyword Args:
            min_rank (int): Minimum rank of the player to challenge. Defaults to 7.
            max_rank (int): Maximum rank of the player to challenge. Defaults to 18.
            challenger_color (str): Color of the challenger. Defaults to 'white'.
            aga_ranked (bool): Whether or not the game is AGA ranked. Defaults to False.
            invite_only (bool): Whether or not the game is invite only. Defaults to False.  
            game_name (str): Name of the game. Defaults to 'Friendly Game'.
            game_rules (str): Rules of the game. Defaults to 'japanese'.
            game_ranked (bool): Whether or not the game is ranked. Defaults to False.
            game_width (int): Width of the board. Defaults to 19.
            game_height (int): Height of the board. Defaults to 19.
            game_handicap (int): Handicap of the game. Defaults to 0.
            game_komi_auto (bool): Whether or not to use automatic komi. Defaults to True.
            game_komi (float): Komi of the game. Defaults to 6.5.
                Not needed if using auto komi.            
            game_disable_analysis (bool): Whether or not to disable analysis. Defaults to False.
            game_initial_state (str): Initial state of the game. Defaults to None.   
            game_private (bool): Whether or not the game is private. Defaults to False.
            game_time_control (str): Time control of the game. Defaults to 'byoyomi'.
            byoyomi_main_time (int): Main time of the game in seconds. Defaults to 2400.
                only used if byoyomi time control is used.
            byoyomi_period_time (int): Period time of the game in seconds. Defaults to 30.
                only used if byoyomi time control is used.
            byoyomi_periods (int): Number of periods in the game. Defaults to 5.
                only used if byoyomi time control is used.
            byoyomi_periods_min (int): Minimum periods of the game. Defaults to 5.
                only used if byoyomi time control is used.
            byoyomi_periods_max (int): Maximum periods of the game. Defaults to 5.
                only used if byoyomi time control is used.
            fischer_time_initial_time (int): Initial time of the game in seconds. Defaults to 900.
                only used if fischer time control is used.
            fischer_time_increment (int): Increment of the game in seconds. Defaults to 0.
                only used if fischer time control is used.
            fischer_time_max_time (int): Maximum time of the game in seconds. Defaults to 1800.
                only used if fischer time control is used.       

        Returns:
            challenge_id (int): ID of the challenge created
            game_id (int): ID of the game created
        """
        time_control = game_settings.get('time_control', 'byoyomi')
        # Set common parameters
        time_control_parameters = {}
        time_control_parameters['speed'] = game_settings.get('speed', 'correspondence')
        time_control_parameters['pause_on_weekends'] = game_settings.get('pause_on_weekends', False)
        time_control_parameters['time_control'] = time_control

        # Create time control paramters depending on time control used
        logger.debug(f"Matching time control: {time_control}")
        match time_control:
            case 'byoyomi':
                logger.debug("Using byoyomi time control")
                time_control_parameters = {
                    'system' : 'byoyomi',
                    'main_time' : game_settings.get('byoyomi_main_time', 2400),
                    'period_time' : game_settings.get('byoyomi_period_time', 30),
                    'periods' : game_settings.get('byoyomi_periods', 5),
                    'periods_min' : game_settings.get('byoyomi_periods_min', 1),
                    'periods_max' : game_settings.get('byoyomi_periods_max', 300),

                }
            case 'fischer':
                logger.debug("Using fischer time control")
                time_control_parameters = {
                    'system' : 'fischer',
                    'initial_time' : game_settings.get('fischer_initial_time', 2400),
                    'time_increment' : game_settings.get('fischer_time_increment', 30),
                    'max_time' : game_settings.get('fischer_max_time', 300),
                }
            case 'canadian':
                # TODO: Implement
                time_control_parameters = {}
            case 'absolute':
                # TODO: Implement
                time_control_parameters = {}
            case 'none':
                logger.debug("Using no time control")
                time_control_parameters = {
                    'system' : 'none',
                    'speed' : 'correspondence',
                    'time_control' : 'none',
                    'pause_on_weekends' : False
                }

        # Create challenge from kwargs
        challenge = {
            'initialized' : False,
            'min_ranking' : game_settings.get('min_ranking', 7),
            'max_ranking' : game_settings.get('max_ranking', 18),
            'challenger_color' : game_settings.get('challenger_color', 'white'),
            'game' : {
                'name' : game_settings.get('game_name', 'Friendly Game'),
                'rules' : game_settings.get('game_rules', 'japanese'),
                'ranked' : game_settings.get('game_ranked', False),
                'width' : game_settings.get('game_width', 19),
                'height' : game_settings.get('game_height', 19),
                'handicap' : game_settings.get('game_handicap', '0'),
                'komi_auto' : game_settings.get('game_komi_auto', True),
                'komi' : game_settings.get('game_komi', '6.5'),
                'disable_analysis' : game_settings.get('game_disable_analysis', False),
                'initial_state' : game_settings.get('game_initial_state', None),
                'private' : game_settings.get('game_private', False),
                'time_control' : time_control,
                'time_control_parameters' : time_control_parameters
            },
            'aga_ranked' : game_settings.get('aga_ranked', False),
            'invite_only' : game_settings.get('invite_only', False),
        }
        logger.info(f"Created challenge object with following parameters: {challenge}")

        if player_username is not None:
            player_id = self.get_player(player_username)['id']
            print(f"Challenging player: {player_username} - {player_id}")
            endpoint = f'/players/{player_id}/challenge/'
            logger.info(f"Sending challenge to {player_username} - {player_id}")
            response = self.api.call_rest_endpoint('POST', endpoint, challenge).json()
        else:
            endpoint = '/challenges/'
            logger.info("Sending open challenge")
            response = self.api.call_rest_endpoint('POST', endpoint, challenge).json()

        logger.debug(f"Challenge response - {response}")
        challenge_id = response['challenge']
        game_id = response['game']
        logger.success(f"Challenge created with challenge ID: {challenge_id} and game ID: {game_id}")
        return challenge_id, game_id

    # Challenges

    # TODO: Change these to use the 'challenger' parameter instead of looping through all challenges
    def received_challenges(self) -> list[dict]:
        """Get all received challenges.

        Returns:
            challenges (list[dict]): JSON response from the endpoint
        """

        endpoint = '/me/challenges/'
        received_challenges = []
        logger.info("Getting received challenges")
        all_challenges = self.api.call_rest_endpoint('GET', endpoint).json()['results']
        logger.debug(f"Got challenges: {all_challenges}")
        for challenge in all_challenges:
            if challenge['challenger']['id'] != self.credentials.user_id:
                received_challenges.append(challenge)
        return received_challenges

    # TODO: Same as above
    def sent_challenges(self) -> list[dict]:
        """Get all sent challenges.

        Returns:
            challenges (list[dict]): JSON response from the endpoint
        """
        endpoint = '/me/challenges'
        sent_challenges = []
        logger.info("Getting sent challenges")
        all_challenges = self.api.call_rest_endpoint('GET', endpoint).json()['results']
        logger.debug(f"Got challenges: {all_challenges}")
        for challenge in all_challenges:
            if challenge['challenger']['id'] == self.credentials.user_id:
                sent_challenges.append(challenge)
        return sent_challenges

    def accept_challenge(self, challenge_id: str) -> dict:
        """Accept a challenge.

        Args:
            challenge_id (str): ID of the challenge to accept.

        Returns:
            response (dict): JSON response from the endpoint
        """

        endpoint = f'/me/challenges/{challenge_id}/accept'
        logger.info(f"Accepting challenge {challenge_id}")
        return self.api.call_rest_endpoint('POST', endpoint=endpoint,payload={}).json()

    def decline_challenge(self, challenge_id: str) -> dict:
        """Decline a challenge.

        Args:
            challenge_id (str): ID of the challenge to decline.

        Returns:
            response (dict): JSON response from the endpoint
        """

        endpoint = f'/me/challenges/{challenge_id}/'
        logger.info(f"Declining challenge {challenge_id}")
        return self.api.call_rest_endpoint('DELETE', endpoint=endpoint, payload={}).json()

    def challenge_details(self, challenge_id: str) -> dict:
        """Get details of a challenge.

        Args:
            challenge_id (str): ID of the challenge to get details of.

        Returns:
            response (dict): JSON response from the endpoint
        """

        endpoint = f'/me/challenges/{challenge_id}'
        logger.info(f"Getting challenge details for {challenge_id}")
        return self.api.call_rest_endpoint('GET', endpoint=endpoint).json()

    def game_details(self, game_id: str) -> dict:
        """Get details of a game.

        Args:
            game_id (str): ID of the game to get details of.

        Returns:
            response (dict): JSON response from the endpoint
        """
        endpoint = f'/games/{game_id}'
        logger.info(f"Getting game details for {game_id}")
        return self.api.call_rest_endpoint('GET', endpoint).json()

    def game_reviews(self, game_id: str) -> dict:
        """Get reviews of a game.

        Args:
            game_id (str): ID of the game to get reviews of.

        Returns:
            response (dict): JSON response from the endpoint
        """
        endpoint = f'/games/{game_id}/reviews'
        logger.info(f"Getting game reviews for {game_id}")
        return self.api.call_rest_endpoint('GET', endpoint).json()

    def game_png(self, game_id: str) -> bytes:
        """Get PNG of a game.

        Args:
            game_id (str): ID of the game to get PNG of.

        Returns:
            response (bytes): PNG image of the game
        """
        endpoint = f'/games/{game_id}/png'
        logger.info(f"Getting game PNG for {game_id}")
        return self.api.call_rest_endpoint('GET', endpoint).content

    def game_sgf(self, game_id: str) -> str:
        """Get SGF of a game.

        Args:  
            game_id (str): ID of the game to get SGF of.

        Returns:
            response (str): SGF of the game
        """
        endpoint = f'/games/{game_id}/sgf'
        logger.info(f"Getting game SGF for {game_id}")
        return self.api.call_rest_endpoint('GET', endpoint).text

    def socket_connect(self, callback_handler: Callable) -> None:
        """Connect to the socket.

        Args:
            callback_handler (Callable): Callback function to send socket events to.
        """
        self.sock = OGSSocket(self.credentials)
        self.sock.callback_handler = callback_handler
        self.sock.connect()

    def socket_disconnect(self) -> None:
        """Disconnect from the socket. You will need to do this before exiting your program, Or else it will hang and require a keyboard interrupt."""
        self.sock.disconnect()
        del self.sock

accept_challenge(challenge_id)

Accept a challenge.

Parameters:

Name	Type	Description	Default
challenge_id	str	ID of the challenge to accept.	required

Returns:

Name	Type	Description
response	dict	JSON response from the endpoint

Source code in src/ogsapi/client.py

def accept_challenge(self, challenge_id: str) -> dict:
    """Accept a challenge.

    Args:
        challenge_id (str): ID of the challenge to accept.

    Returns:
        response (dict): JSON response from the endpoint
    """

    endpoint = f'/me/challenges/{challenge_id}/accept'
    logger.info(f"Accepting challenge {challenge_id}")
    return self.api.call_rest_endpoint('POST', endpoint=endpoint,payload={}).json()

active_games(player_id=None)

Get the user's active games.

Parameters:

Name	Type	Description	Default
player_id	int	ID of player whos active games we want to retrieve.	None

Returns:

Name	Type	Description
response	list[dict]	JSON object containing players active games.

Source code in src/ogsapi/client.py

def active_games(self, player_id: int | None = None) -> list[dict]:
    """
    Get the user's active games.

    Args:
        player_id (int, optional): ID of player whos active games we want to retrieve.

    Returns:
        response (list[dict]): JSON object containing players active games.
    """

    if player_id is None:
        endpoint = "/ui/overview"
    else:
        endpoint = f"/players/{player_id}/full"
    return self.api.call_rest_endpoint('GET', endpoint=endpoint).json()["active_games"]

challenge_details(challenge_id)

Get details of a challenge.

Parameters:

Name	Type	Description	Default
challenge_id	str	ID of the challenge to get details of.	required

Returns:

Name	Type	Description
response	dict	JSON response from the endpoint

Source code in src/ogsapi/client.py

def challenge_details(self, challenge_id: str) -> dict:
    """Get details of a challenge.

    Args:
        challenge_id (str): ID of the challenge to get details of.

    Returns:
        response (dict): JSON response from the endpoint
    """

    endpoint = f'/me/challenges/{challenge_id}'
    logger.info(f"Getting challenge details for {challenge_id}")
    return self.api.call_rest_endpoint('GET', endpoint=endpoint).json()

create_challenge(player_username=None, **game_settings)

Create either an open challenge or a challenge to a specific player. The time control settings are built depending on which time control is used. Make sure that you pass the correct time control settings for the time control you want to use. The other time control settings will be ignored.

Examples:

>>> ogs.create_challenge(player_username='test', main_time=300, byoyomi_time=30, byoyomi_stones=5)
Challenging player: test - 1234567
(20328495, 53331333)

Parameters:

Name	Type	Description	Default
player_username	str	Username of the player to challenge. If used will issue the challenge to the player. Defaults to None.	None

Other Parameters:

Name	Type	Description
min_rank	int	Minimum rank of the player to challenge. Defaults to 7.
max_rank	int	Maximum rank of the player to challenge. Defaults to 18.
challenger_color	str	Color of the challenger. Defaults to 'white'.
aga_ranked	bool	Whether or not the game is AGA ranked. Defaults to False.
invite_only	bool	Whether or not the game is invite only. Defaults to False.
game_name	str	Name of the game. Defaults to 'Friendly Game'.
game_rules	str	Rules of the game. Defaults to 'japanese'.
game_ranked	bool	Whether or not the game is ranked. Defaults to False.
game_width	int	Width of the board. Defaults to 19.
game_height	int	Height of the board. Defaults to 19.
game_handicap	int	Handicap of the game. Defaults to 0.
game_komi_auto	bool	Whether or not to use automatic komi. Defaults to True.
game_komi	float	Komi of the game. Defaults to 6.5. Not needed if using auto komi.
game_disable_analysis	bool	Whether or not to disable analysis. Defaults to False.
game_initial_state	str	Initial state of the game. Defaults to None.
game_private	bool	Whether or not the game is private. Defaults to False.
game_time_control	str	Time control of the game. Defaults to 'byoyomi'.
byoyomi_main_time	int	Main time of the game in seconds. Defaults to 2400. only used if byoyomi time control is used.
byoyomi_period_time	int	Period time of the game in seconds. Defaults to 30. only used if byoyomi time control is used.
byoyomi_periods	int	Number of periods in the game. Defaults to 5. only used if byoyomi time control is used.
byoyomi_periods_min	int	Minimum periods of the game. Defaults to 5. only used if byoyomi time control is used.
byoyomi_periods_max	int	Maximum periods of the game. Defaults to 5. only used if byoyomi time control is used.
fischer_time_initial_time	int	Initial time of the game in seconds. Defaults to 900. only used if fischer time control is used.
fischer_time_increment	int	Increment of the game in seconds. Defaults to 0. only used if fischer time control is used.
fischer_time_max_time	int	Maximum time of the game in seconds. Defaults to 1800. only used if fischer time control is used.

Returns:

Name	Type	Description
challenge_id	int	ID of the challenge created
game_id	int	ID of the game created

Source code in src/ogsapi/client.py

def create_challenge(self, player_username: str | None = None, **game_settings) -> tuple[int, int]:
    """Create either an open challenge or a challenge to a specific player.
    The time control settings are built depending on which time control is used.
    Make sure that you pass the correct time control settings for the time control you want to use.
    The other time control settings will be ignored.

    Examples:
        >>> ogs.create_challenge(player_username='test', main_time=300, byoyomi_time=30, byoyomi_stones=5)
        Challenging player: test - 1234567
        (20328495, 53331333)

    Args:
        player_username (str): Username of the player to challenge. 
            If used will issue the challenge to the player. Defaults to None.

    Keyword Args:
        min_rank (int): Minimum rank of the player to challenge. Defaults to 7.
        max_rank (int): Maximum rank of the player to challenge. Defaults to 18.
        challenger_color (str): Color of the challenger. Defaults to 'white'.
        aga_ranked (bool): Whether or not the game is AGA ranked. Defaults to False.
        invite_only (bool): Whether or not the game is invite only. Defaults to False.  
        game_name (str): Name of the game. Defaults to 'Friendly Game'.
        game_rules (str): Rules of the game. Defaults to 'japanese'.
        game_ranked (bool): Whether or not the game is ranked. Defaults to False.
        game_width (int): Width of the board. Defaults to 19.
        game_height (int): Height of the board. Defaults to 19.
        game_handicap (int): Handicap of the game. Defaults to 0.
        game_komi_auto (bool): Whether or not to use automatic komi. Defaults to True.
        game_komi (float): Komi of the game. Defaults to 6.5.
            Not needed if using auto komi.            
        game_disable_analysis (bool): Whether or not to disable analysis. Defaults to False.
        game_initial_state (str): Initial state of the game. Defaults to None.   
        game_private (bool): Whether or not the game is private. Defaults to False.
        game_time_control (str): Time control of the game. Defaults to 'byoyomi'.
        byoyomi_main_time (int): Main time of the game in seconds. Defaults to 2400.
            only used if byoyomi time control is used.
        byoyomi_period_time (int): Period time of the game in seconds. Defaults to 30.
            only used if byoyomi time control is used.
        byoyomi_periods (int): Number of periods in the game. Defaults to 5.
            only used if byoyomi time control is used.
        byoyomi_periods_min (int): Minimum periods of the game. Defaults to 5.
            only used if byoyomi time control is used.
        byoyomi_periods_max (int): Maximum periods of the game. Defaults to 5.
            only used if byoyomi time control is used.
        fischer_time_initial_time (int): Initial time of the game in seconds. Defaults to 900.
            only used if fischer time control is used.
        fischer_time_increment (int): Increment of the game in seconds. Defaults to 0.
            only used if fischer time control is used.
        fischer_time_max_time (int): Maximum time of the game in seconds. Defaults to 1800.
            only used if fischer time control is used.       

    Returns:
        challenge_id (int): ID of the challenge created
        game_id (int): ID of the game created
    """
    time_control = game_settings.get('time_control', 'byoyomi')
    # Set common parameters
    time_control_parameters = {}
    time_control_parameters['speed'] = game_settings.get('speed', 'correspondence')
    time_control_parameters['pause_on_weekends'] = game_settings.get('pause_on_weekends', False)
    time_control_parameters['time_control'] = time_control

    # Create time control paramters depending on time control used
    logger.debug(f"Matching time control: {time_control}")
    match time_control:
        case 'byoyomi':
            logger.debug("Using byoyomi time control")
            time_control_parameters = {
                'system' : 'byoyomi',
                'main_time' : game_settings.get('byoyomi_main_time', 2400),
                'period_time' : game_settings.get('byoyomi_period_time', 30),
                'periods' : game_settings.get('byoyomi_periods', 5),
                'periods_min' : game_settings.get('byoyomi_periods_min', 1),
                'periods_max' : game_settings.get('byoyomi_periods_max', 300),

            }
        case 'fischer':
            logger.debug("Using fischer time control")
            time_control_parameters = {
                'system' : 'fischer',
                'initial_time' : game_settings.get('fischer_initial_time', 2400),
                'time_increment' : game_settings.get('fischer_time_increment', 30),
                'max_time' : game_settings.get('fischer_max_time', 300),
            }
        case 'canadian':
            # TODO: Implement
            time_control_parameters = {}
        case 'absolute':
            # TODO: Implement
            time_control_parameters = {}
        case 'none':
            logger.debug("Using no time control")
            time_control_parameters = {
                'system' : 'none',
                'speed' : 'correspondence',
                'time_control' : 'none',
                'pause_on_weekends' : False
            }

    # Create challenge from kwargs
    challenge = {
        'initialized' : False,
        'min_ranking' : game_settings.get('min_ranking', 7),
        'max_ranking' : game_settings.get('max_ranking', 18),
        'challenger_color' : game_settings.get('challenger_color', 'white'),
        'game' : {
            'name' : game_settings.get('game_name', 'Friendly Game'),
            'rules' : game_settings.get('game_rules', 'japanese'),
            'ranked' : game_settings.get('game_ranked', False),
            'width' : game_settings.get('game_width', 19),
            'height' : game_settings.get('game_height', 19),
            'handicap' : game_settings.get('game_handicap', '0'),
            'komi_auto' : game_settings.get('game_komi_auto', True),
            'komi' : game_settings.get('game_komi', '6.5'),
            'disable_analysis' : game_settings.get('game_disable_analysis', False),
            'initial_state' : game_settings.get('game_initial_state', None),
            'private' : game_settings.get('game_private', False),
            'time_control' : time_control,
            'time_control_parameters' : time_control_parameters
        },
        'aga_ranked' : game_settings.get('aga_ranked', False),
        'invite_only' : game_settings.get('invite_only', False),
    }
    logger.info(f"Created challenge object with following parameters: {challenge}")

    if player_username is not None:
        player_id = self.get_player(player_username)['id']
        print(f"Challenging player: {player_username} - {player_id}")
        endpoint = f'/players/{player_id}/challenge/'
        logger.info(f"Sending challenge to {player_username} - {player_id}")
        response = self.api.call_rest_endpoint('POST', endpoint, challenge).json()
    else:
        endpoint = '/challenges/'
        logger.info("Sending open challenge")
        response = self.api.call_rest_endpoint('POST', endpoint, challenge).json()

    logger.debug(f"Challenge response - {response}")
    challenge_id = response['challenge']
    game_id = response['game']
    logger.success(f"Challenge created with challenge ID: {challenge_id} and game ID: {game_id}")
    return challenge_id, game_id

decline_challenge(challenge_id)

Decline a challenge.

Parameters:

Name	Type	Description	Default
challenge_id	str	ID of the challenge to decline.	required

Returns:

Name	Type	Description
response	dict	JSON response from the endpoint

Source code in src/ogsapi/client.py

def decline_challenge(self, challenge_id: str) -> dict:
    """Decline a challenge.

    Args:
        challenge_id (str): ID of the challenge to decline.

    Returns:
        response (dict): JSON response from the endpoint
    """

    endpoint = f'/me/challenges/{challenge_id}/'
    logger.info(f"Declining challenge {challenge_id}")
    return self.api.call_rest_endpoint('DELETE', endpoint=endpoint, payload={}).json()

disable_logging()

Disable logging from ogsapi

Source code in src/ogsapi/client.py

def disable_logging(self) -> None:
    """Disable logging from ogsapi"""
    logger.disable("src.ogsapi")

enable_logging()

Enable logging from ogsapi

Source code in src/ogsapi/client.py

def enable_logging(self) -> None:
    """Enable logging from ogsapi"""
    logger.enable("src.ogsapi")

game_details(game_id)

Get details of a game.

Parameters:

Name	Type	Description	Default
game_id	str	ID of the game to get details of.	required

Returns:

Name	Type	Description
response	dict	JSON response from the endpoint

Source code in src/ogsapi/client.py

def game_details(self, game_id: str) -> dict:
    """Get details of a game.

    Args:
        game_id (str): ID of the game to get details of.

    Returns:
        response (dict): JSON response from the endpoint
    """
    endpoint = f'/games/{game_id}'
    logger.info(f"Getting game details for {game_id}")
    return self.api.call_rest_endpoint('GET', endpoint).json()

game_png(game_id)

Get PNG of a game.

Parameters:

Name	Type	Description	Default
game_id	str	ID of the game to get PNG of.	required

Returns:

Name	Type	Description
response	bytes	PNG image of the game

Source code in src/ogsapi/client.py

def game_png(self, game_id: str) -> bytes:
    """Get PNG of a game.

    Args:
        game_id (str): ID of the game to get PNG of.

    Returns:
        response (bytes): PNG image of the game
    """
    endpoint = f'/games/{game_id}/png'
    logger.info(f"Getting game PNG for {game_id}")
    return self.api.call_rest_endpoint('GET', endpoint).content

game_reviews(game_id)

Get reviews of a game.

Parameters:

Name	Type	Description	Default
game_id	str	ID of the game to get reviews of.	required

Returns:

Name	Type	Description
response	dict	JSON response from the endpoint

Source code in src/ogsapi/client.py

def game_reviews(self, game_id: str) -> dict:
    """Get reviews of a game.

    Args:
        game_id (str): ID of the game to get reviews of.

    Returns:
        response (dict): JSON response from the endpoint
    """
    endpoint = f'/games/{game_id}/reviews'
    logger.info(f"Getting game reviews for {game_id}")
    return self.api.call_rest_endpoint('GET', endpoint).json()

game_sgf(game_id)

Get SGF of a game.

Parameters:

Name	Type	Description	Default
game_id	str	ID of the game to get SGF of.	required

Returns:

Name	Type	Description
response	str	SGF of the game

Source code in src/ogsapi/client.py

def game_sgf(self, game_id: str) -> str:
    """Get SGF of a game.

    Args:  
        game_id (str): ID of the game to get SGF of.

    Returns:
        response (str): SGF of the game
    """
    endpoint = f'/games/{game_id}/sgf'
    logger.info(f"Getting game SGF for {game_id}")
    return self.api.call_rest_endpoint('GET', endpoint).text

get_player(player_username)

Get a player by username.

Parameters:

Name	Type	Description	Default
player_username	str	Username of the player to get.	required

Returns:

Name	Type	Description
player_data	dict	Player data returned from the endpoint

Source code in src/ogsapi/client.py

def get_player(self, player_username: str) -> dict:
    """Get a player by username.

    Args:
        player_username (str): Username of the player to get.

    Returns:
        player_data (dict): Player data returned from the endpoint
    """

    endpoint = '/players/'
    logger.info(f"Getting player {player_username}")
    return self.api.call_rest_endpoint('GET', endpoint=endpoint, params={'username' : player_username}).json()['results'][0]

get_player_games(player_username)

Get a player's games by username.

Parameters:

Name	Type	Description	Default
player_username	str	Username of the player to get games of.	required

Returns:

Name	Type	Description
player_games	dict	Player games returned from the endpoint

Source code in src/ogsapi/client.py

def get_player_games(self, player_username: str) -> dict:
    """Get a player's games by username.

    Args:
        player_username (str): Username of the player to get games of.

    Returns:
        player_games (dict): Player games returned from the endpoint
    """
    logger.info(f"Getting player {player_username}'s games")
    player_id = self.get_player(player_username)['id']
    endpoint = f'/players/{player_id}/games'
    return self.api.call_rest_endpoint('GET', endpoint=endpoint).json()

received_challenges()

Get all received challenges.

Returns:

Name	Type	Description
challenges	list[dict]	JSON response from the endpoint

Source code in src/ogsapi/client.py

def received_challenges(self) -> list[dict]:
    """Get all received challenges.

    Returns:
        challenges (list[dict]): JSON response from the endpoint
    """

    endpoint = '/me/challenges/'
    received_challenges = []
    logger.info("Getting received challenges")
    all_challenges = self.api.call_rest_endpoint('GET', endpoint).json()['results']
    logger.debug(f"Got challenges: {all_challenges}")
    for challenge in all_challenges:
        if challenge['challenger']['id'] != self.credentials.user_id:
            received_challenges.append(challenge)
    return received_challenges

remove_friend(username)

Remove a friend.

Parameters:

Name	Type	Description	Default
username	str	Username of the user to remove as a friend.	required

Returns:

Name	Type	Description
response	dict	JSON response from the endpoint

Source code in src/ogsapi/client.py

def remove_friend(self, username: str) -> dict:
    """Remove a friend.

    Args:
        username (str): Username of the user to remove as a friend.

    Returns:
        response (dict): JSON response from the endpoint
    """

    endpoint = '/me/friends/'
    player_id = self.get_player(username)['id']
    payload = {
        "delete": True,
        "player_id" : player_id
    }
    logger.info(f"Removing friend {username} - {player_id}")
    return self.api.call_rest_endpoint('POST', endpoint=endpoint, payload=payload).json()

send_friend_request(username)

Send a friend request to a user.

Parameters:

Name	Type	Description	Default
username	str	Username of the user to send a friend request to.	required

Returns:

Name	Type	Description
response	dict	JSON response from the endpoint

Source code in src/ogsapi/client.py

def send_friend_request(self, username: str) -> dict:
    """Send a friend request to a user.

    Args:
        username (str): Username of the user to send a friend request to.

    Returns:
        response (dict): JSON response from the endpoint
    """

    endpoint = '/me/friends'
    player_id = self.get_player(username)['id']
    payload = {
        "player_id" : player_id
    }
    logger.info(f"Sending friend request to {username} - {player_id}")
    return self.api.call_rest_endpoint('POST', endpoint=endpoint, payload=payload).json()

sent_challenges()

Get all sent challenges.

Returns:

Name	Type	Description
challenges	list[dict]	JSON response from the endpoint

Source code in src/ogsapi/client.py

def sent_challenges(self) -> list[dict]:
    """Get all sent challenges.

    Returns:
        challenges (list[dict]): JSON response from the endpoint
    """
    endpoint = '/me/challenges'
    sent_challenges = []
    logger.info("Getting sent challenges")
    all_challenges = self.api.call_rest_endpoint('GET', endpoint).json()['results']
    logger.debug(f"Got challenges: {all_challenges}")
    for challenge in all_challenges:
        if challenge['challenger']['id'] == self.credentials.user_id:
            sent_challenges.append(challenge)
    return sent_challenges

socket_connect(callback_handler)

Connect to the socket.

Parameters:

Name	Type	Description	Default
callback_handler	Callable	Callback function to send socket events to.	required

Source code in src/ogsapi/client.py

def socket_connect(self, callback_handler: Callable) -> None:
    """Connect to the socket.

    Args:
        callback_handler (Callable): Callback function to send socket events to.
    """
    self.sock = OGSSocket(self.credentials)
    self.sock.callback_handler = callback_handler
    self.sock.connect()

socket_disconnect()

Disconnect from the socket. You will need to do this before exiting your program, Or else it will hang and require a keyboard interrupt.

Source code in src/ogsapi/client.py

def socket_disconnect(self) -> None:
    """Disconnect from the socket. You will need to do this before exiting your program, Or else it will hang and require a keyboard interrupt."""
    self.sock.disconnect()
    del self.sock

update_user_settings(username=None, first_name=None, last_name=None, private_name=None, country=None, website=None, about=None)

Update the user's settings.

Parameters:

Name	Type	Description	Default
username	str	New username. Defaults to None.	None
first_name	str	New first name. Defaults to None.	None
last_name	str	New last name. Defaults to None.	None
private_name	bool	Whether or not to make the name private. Defaults to None.	None
country	str	New country. Defaults to None.	None
website	str	New website. Defaults to None.	None
about	str	New about. Defaults to None.	None

Returns:

Name	Type	Description
response	dict	JSON response from the endpoint

Source code in src/ogsapi/client.py

def update_user_settings(
        self, username: str | None = None,
        first_name: str | None = None,
        last_name: str | None = None,
        private_name: bool | None = None,
        country: str | None = None,
        website: str | None = None,
        about: str | None = None
    ) -> dict:
    """Update the user's settings.

    Args:
        username (str, optional): New username. Defaults to None.
        first_name (str, optional): New first name. Defaults to None.
        last_name (str, optional): New last name. Defaults to None.
        private_name (bool, optional): Whether or not to make the name private. Defaults to None.
        country (str, optional): New country. Defaults to None.
        website (str, optional): New website. Defaults to None.
        about (str, optional): New about. Defaults to None.

    Returns:
        response (dict): JSON response from the endpoint
    """

    # This is a bit of a mess, but it works, should be refactored
    payload: dict[str, Any] = {}
    if username is not None:
        payload['username'] = username
    if first_name is not None:
        payload['first_name'] = first_name
    if last_name is not None:
        payload['last_name'] = last_name
    if private_name is not None:
        payload['real_name_is_private'] = private_name
    if country is not None:
        payload['country'] = country
    if website is not None:
        payload['website'] = website
    if about is not None:
        payload['about'] = about

    endpoint = f'/players/{self.credentials.user_id}'
    # Add the inputs to a payload, only if they are not None
    logger.info(f"Updating user settings with the following payload: {payload}")
    return self.api.call_rest_endpoint('PUT', endpoint=endpoint, payload=payload).json()

user_friends(username=None)

Get the user's friends.

Parameters:

Name	Type	Description	Default
username	str	Username of the user to get friends of. Defaults to None.	None

Returns:

Name	Type	Description
response	dict	JSON response from the endpoint

Source code in src/ogsapi/client.py

def user_friends(self, username: str | None = None) -> dict:
    """Get the user's friends.

    Args:
        username (str, optional): Username of the user to get friends of. Defaults to None.

    Returns:
        response (dict): JSON response from the endpoint
    """

    endpoint = '/me/friends'
    logger.info("Getting user friends")
    return self.api.call_rest_endpoint('GET', endpoint=endpoint, params={'username' : username}).json()

user_games(page=1, page_size=10)

Get the user's games.

Parameters:

Name	Type	Description	Default
page	int	Page number of user games. Defaults to page 1	1
page_size	int	Number of games per page. Defaults to 10	10

Returns:

Name	Type	Description
response	dict	JSON response from the endpoint

Source code in src/ogsapi/client.py

def user_games(self, page: int = 1, page_size: int = 10) -> dict:
    """Get the user's games.

    Args:
        page (int): Page number of user games. Defaults to page 1
        page_size (int): Number of games per page. Defaults to 10

    Returns:
        response (dict): JSON response from the endpoint
    """

    endpoint = '/me/games'
    params = { 'page': page, 'page_size': page_size }
    logger.info(f"Getting user games - page {page} with page size {page_size}")
    return self.api.call_rest_endpoint('GET', endpoint=endpoint, params=params).json()

user_settings()

Get the user's settings.

Returns:

Name	Type	Description
response	dict	JSON response from the endpoint

Source code in src/ogsapi/client.py

def user_settings(self) -> dict:
    """Get the user's settings.

    Returns:
        response (dict): JSON response from the endpoint
    """

    endpoint = '/me/settings/'
    logger.info("Getting user settings")
    return self.api.call_rest_endpoint('GET', endpoint=endpoint).json()

user_vitals()

Get the user's vitals.

Returns:

Name	Type	Description
response	dict	JSON response from the endpoint

Source code in src/ogsapi/client.py

def user_vitals(self) -> dict:
    """Get the user's vitals.

    Returns:
        response (dict): JSON response from the endpoint
    """

    endpoint = '/me'
    logger.info("Getting user vitals")
    return self.api.call_rest_endpoint('GET', endpoint=endpoint).json()

OGSRestAPI

OGS Rest API Class for handling REST connections to OGS

Parameters:

Name	Type	Description	Default
credentials	OGSCredentials	The credentials to use for authentication	required
dev	bool	Whether to connect to beta OGS instance. Defaults to False.	False

Attributes:

Name	Type	Description
credentials	OGSCredentials	The credentials used for authentication
api_ver	str	The API version to use
base_url	str	The base URL to use for API calls

Source code in src/ogsapi/ogsrestapi.py

class OGSRestAPI:
    """OGS Rest API Class for handling REST connections to OGS

    Args:
        credentials (OGSCredentials): The credentials to use for authentication
        dev (bool, optional): Whether to connect to beta OGS instance. Defaults to False.

    Attributes:
        credentials (OGSCredentials): The credentials used for authentication
        api_ver (str): The API version to use
        base_url (str): The base URL to use for API calls
    """

    def __init__(self, credentials: OGSCredentials, dev: bool = False):

        self.credentials = credentials
        self.api_ver = "v1"
        if dev:
            self.base_url = 'https://beta.online-go.com/'
            logger.debug("Connecting to beta OGS instance")
        else:
            self.base_url = 'https://online-go.com/'
            logger.debug("Connecting to production OGS instance")

        # TODO: Maybe implement some form of token caching
        self.authenticate()
        self.get_auth_data()

    # TODO: All these internal functions should be moved into private functions
    @logger.catch
    def authenticate(self) -> None:
        """Authenticate with the OGS API and save the access token and user ID."""

        endpoint = f'{self.base_url}/oauth2/token/'
        logger.info("Authenticating with OGS API")
        try:
            response = requests.post(endpoint, data={
                'client_id': self.credentials.client_id,
                'grant_type': 'password',
                'username': self.credentials.username,
                'password': self.credentials.password,
            },
            headers={'Content-Type': 'application/x-www-form-urlencoded'},
            timeout=20
            )
        except requests.exceptions.RequestException as e:
            raise OGSApiException("Authentication Failed") from e

        if 299 >= response.status_code >= 200:
            # Save Access Token, Refresh Token, and User ID
            # TODO: This should probably be made into a user object that has token and ID info
            self.credentials.access_token = response.json()['access_token']
            self.credentials.refresh_token = response.json()['refresh_token']
        else:
            raise OGSApiException(f"{response.status_code}: {response.reason}")

    @logger.catch
    def call_rest_endpoint(self, method: str, endpoint: str, params: dict | None = None, payload: dict | None = None) -> requests.Response:
        """Make a request to the OGS REST API.

        Args:
            method (str): HTTP method to use. Accepts GET, POST, PUT, DELETE
            endpoint (str): Endpoint to make request to
            params (dict, optional): Parameters to pass to the endpoint. Defaults to None.
            payload (dict, optional): Payload to pass to the endpoint. Defaults to None.

        Returns:
            response (Callable): Returns the request response
        """
        method = method.upper()
        url = f'{self.base_url}api/{self.api_ver}{endpoint}'
        headers = {
            'Authorization' : f'Bearer {self.credentials.access_token}',
            'Content-Type': 'application/json'
        }

        # Bail if method is invalid
        if method not in ['GET', 'POST', 'PUT', 'DELETE']:
            raise OGSApiException(f"Invalid HTTP Method, Got: {method}. Expected: GET, POST, PUT, DELETE")

        # Add payload if method is POST or PUT
        logger.debug(f"Making {method} request to {url}")
        if method in ['POST', 'PUT']:
            try:
                response = requests.request(method, url, headers=headers, params=params, json=payload, timeout=20)
            except requests.exceptions.RequestException as e:
                raise OGSApiException(f"{method} Failed") from e
        else:
            try:
                response = requests.request(method, url, headers=headers, params=params, timeout=20)
            except requests.exceptions.RequestException as e:
                raise OGSApiException(f"{method} Failed") from e

        if 299 >= response.status_code >= 200:
            return response

        raise OGSApiException(f"{response.status_code}: {response.reason}")

    @logger.catch
    def get_auth_data(self) -> None:
        """Get the auth data from the OGS API and save it to the credentials object for use in the socket connection."""
        logger.info("Getting auth data from OGS API")
        auth_data = self.call_rest_endpoint('GET', '/ui/config').json()
        self.credentials.chat_auth = auth_data['chat_auth']
        self.credentials.user_jwt = auth_data['user_jwt']

authenticate()

Authenticate with the OGS API and save the access token and user ID.

Source code in src/ogsapi/ogsrestapi.py

@logger.catch
def authenticate(self) -> None:
    """Authenticate with the OGS API and save the access token and user ID."""

    endpoint = f'{self.base_url}/oauth2/token/'
    logger.info("Authenticating with OGS API")
    try:
        response = requests.post(endpoint, data={
            'client_id': self.credentials.client_id,
            'grant_type': 'password',
            'username': self.credentials.username,
            'password': self.credentials.password,
        },
        headers={'Content-Type': 'application/x-www-form-urlencoded'},
        timeout=20
        )
    except requests.exceptions.RequestException as e:
        raise OGSApiException("Authentication Failed") from e

    if 299 >= response.status_code >= 200:
        # Save Access Token, Refresh Token, and User ID
        # TODO: This should probably be made into a user object that has token and ID info
        self.credentials.access_token = response.json()['access_token']
        self.credentials.refresh_token = response.json()['refresh_token']
    else:
        raise OGSApiException(f"{response.status_code}: {response.reason}")

call_rest_endpoint(method, endpoint, params=None, payload=None)

Make a request to the OGS REST API.

Parameters:

Name	Type	Description	Default
method	str	HTTP method to use. Accepts GET, POST, PUT, DELETE	required
endpoint	str	Endpoint to make request to	required
params	dict	Parameters to pass to the endpoint. Defaults to None.	None
payload	dict	Payload to pass to the endpoint. Defaults to None.	None

Returns:

Name	Type	Description
response	Callable	Returns the request response

Source code in src/ogsapi/ogsrestapi.py

@logger.catch
def call_rest_endpoint(self, method: str, endpoint: str, params: dict | None = None, payload: dict | None = None) -> requests.Response:
    """Make a request to the OGS REST API.

    Args:
        method (str): HTTP method to use. Accepts GET, POST, PUT, DELETE
        endpoint (str): Endpoint to make request to
        params (dict, optional): Parameters to pass to the endpoint. Defaults to None.
        payload (dict, optional): Payload to pass to the endpoint. Defaults to None.

    Returns:
        response (Callable): Returns the request response
    """
    method = method.upper()
    url = f'{self.base_url}api/{self.api_ver}{endpoint}'
    headers = {
        'Authorization' : f'Bearer {self.credentials.access_token}',
        'Content-Type': 'application/json'
    }

    # Bail if method is invalid
    if method not in ['GET', 'POST', 'PUT', 'DELETE']:
        raise OGSApiException(f"Invalid HTTP Method, Got: {method}. Expected: GET, POST, PUT, DELETE")

    # Add payload if method is POST or PUT
    logger.debug(f"Making {method} request to {url}")
    if method in ['POST', 'PUT']:
        try:
            response = requests.request(method, url, headers=headers, params=params, json=payload, timeout=20)
        except requests.exceptions.RequestException as e:
            raise OGSApiException(f"{method} Failed") from e
    else:
        try:
            response = requests.request(method, url, headers=headers, params=params, timeout=20)
        except requests.exceptions.RequestException as e:
            raise OGSApiException(f"{method} Failed") from e

    if 299 >= response.status_code >= 200:
        return response

    raise OGSApiException(f"{response.status_code}: {response.reason}")

get_auth_data()

Get the auth data from the OGS API and save it to the credentials object for use in the socket connection.

Source code in src/ogsapi/ogsrestapi.py

@logger.catch
def get_auth_data(self) -> None:
    """Get the auth data from the OGS API and save it to the credentials object for use in the socket connection."""
    logger.info("Getting auth data from OGS API")
    auth_data = self.call_rest_endpoint('GET', '/ui/config').json()
    self.credentials.chat_auth = auth_data['chat_auth']
    self.credentials.user_jwt = auth_data['user_jwt']

OGSGame

OGSGame class for handling games connected via the OGSSocket.

Parameters:

Name	Type	Description	Default
game_socket	OGSSocket	OGSSocket object to connect to the game.	required
game_id	str	ID of the game to connect to.	required
credentials	OGSCredentials	OGSCredentials object containing tokens for authentication to the Socket	required
callback_handler	Callable	Callback handler function to send events to the user.	required

Attributes:

Name	Type	Description
socket	OGSSocket	OGSSocket object to connect to the game.
game_data	OGSGameData	OGSGameData object containing game data.
credentials	OGSCredentials	OGSCredentials object containing tokens for authentication to the Socket
callback_handler	Callable	Callback handler function to send events to the user.
callback_func	dict	Dictionary containing the callback functions.

Source code in src/ogsapi/ogsgame.py

class OGSGame:
    """OGSGame class for handling games connected via the OGSSocket.

    Args:
        game_socket (OGSSocket): OGSSocket object to connect to the game.
        game_id (str): ID of the game to connect to.
        credentials (OGSCredentials): OGSCredentials object containing tokens for authentication to the Socket
        callback_handler (Callable): Callback handler function to send events to the user.

    Attributes:
        socket (OGSSocket): OGSSocket object to connect to the game.
        game_data (OGSGameData): OGSGameData object containing game data.
        credentials (OGSCredentials): OGSCredentials object containing tokens for authentication to the Socket
        callback_handler (Callable): Callback handler function to send events to the user.
        callback_func (dict): Dictionary containing the callback functions.

    """

    def __init__(self, game_socket: socketio.Client, credentials: OGSCredentials, game_id, callback_handler: Callable):
        self.socket = game_socket
        self.game_data = OGSGameData(game_id=game_id)
        self.clock = OGSGameClock()
        # Define callback functions from the API
        self._game_call_backs()
        self.credentials = credentials
        self.callback_handler = callback_handler

        # Connect to the game
        self.connect()

        # Define relevant game data
    def __del__(self):
        self.disconnect()

    # def register_callback(self, event: str, callback: Callable):
    #     """Register a callback function for receiving data from the API.

    #     Args:
    #         event (str): Event to register the callback function for.
    #             Accepted events are:
    #                 - on_move
    #                 - on_clock
    #                 - on_phase_change
    #                 - on_undo_requested
    #                 - on_undo_accepted
    #                 - on_undo_canceled
    #         callback (Callable): Callback function to register.   
    #     """
    #     self.callback_func[event] = callback

    # Low level socket functions
    def _game_call_backs(self) -> None:

        # TODO: Need to create a game board state and have this update it
        @self.socket.on(f'game/{self.game_data.game_id}/move')
        def _on_game_move(data) -> None:
            logger.debug(f"Received move {data['move']} from game {self.game_data.game_id} - {data}")
            self.callback_handler(event_name='move', data=data)

        @self.socket.on(f'game/{self.game_data.game_id}/gamedata')
        def _on_game_data(data) -> None:
            logger.debug(f"Received game data from game {self.game_data.game_id} - {data}")
            # Set important game data
            self.game_data.update(data)
            if self.clock.system == None:
                self.clock.system = self.game_data.time_control.system
            self.callback_handler(event_name="gamedata", data=data)

        @self.socket.on(f'game/{self.game_data.game_id}/clock')
        def _on_game_clock(data) -> None:
            logger.debug(f"Received clock data from game {self.game_data.game_id} - {data}")
            #TODO: Need to create a game clock and sync clock with this event

            # Define clock parameters based on time control
            self.clock.update(data)

            # Call the on_clock callback
            self.callback_handler(event_name="clock", data=data)

        @self.socket.on(f'game/{self.game_data.game_id}/phase')
        def _on_game_phase(data) -> None:
            logger.debug(f"Received phase data from game {self.game_data.game_id} - {data}")
            self.game_data.phase = data
            self.callback_handler(event_name="phase", data=data)

        @self.socket.on(f'game/{self.game_data.game_id}/latency')
        def _on_game_latency(data) -> None:
            logger.debug(f"Received latency data from game {self.game_data.game_id} - {data}")
            self.game_data.latency = data['latency']
            self.callback_handler(event_name="latency", data=data)

        @self.socket.on(f'game/{self.game_data.game_id}/undo_requested')
        def _on_undo_requested(data) -> None:
            logger.debug(f"Received undo request from game {self.game_data.game_id} - {data}")
            #TODO: Handle This 
            self.callback_handler(event_name="undo_requested", data=data)

        @self.socket.on(f'game/{self.game_data.game_id}/undo_accepted')
        def _on_undo_accepted(data) -> None:
            logger.debug(f"Received undo accepted from game {self.game_data.game_id} - {data}")
            #TODO: Handle This
            self.callback_handler(event_name="undo_accepted", data=data)

        @self.socket.on(f'game/{self.game_data.game_id}/undo_canceled')
        def _on_undo_canceled(data) -> None:
            logger.debug(f"Received undo canceled from game {self.game_data.game_id} - {data}")
            self.callback_handler(event_name="undo_canceled", data=data)

    # Send functions
    def connect(self) -> None:
        """Connect to the game"""
        logger.info(f"Connecting to game {self.game_data.game_id}")
        self.socket.emit(event="game/connect", data={'game_id': self.game_data.game_id, 'player_id': self.credentials.user_id, 'chat': False})

    def disconnect(self) -> None:
        """Disconnect from the game"""
        logger.info(f"Disconnecting game {self.game_data.game_id}")
        self.socket.emit(event="game/disconnect", data={'game_id': self.game_data.game_id})

    def get_gamedata(self) -> None:
        """Get game data"""
        logger.info(f"Getting game data for game {self.game_data.game_id}")
        self.socket.emit(event=f"game/{self.game_data.game_id}/gamedata", data={})

    def pause(self) -> None:
        """Pause the game"""
        logger.info(f"Pausing game {self.game_data.game_id}")
        self.socket.emit(event="game/pause", data={'game_id': self.game_data.game_id})

    def resume(self) -> None:
        """Resume the game"""
        logger.info(f"Resuming game {self.game_data.game_id}")
        self.socket.emit(event="game/resume", data={'game_id': self.game_data.game_id})

    def move(self, move: str) -> None:
        """Submit a move to the game

        Args:
            move (str): The move to submit to the game. Accepts GTP format.

        Examples:
            >>> game.move('B2')
        """

        logger.info(f"Submitting move {move} to game {self.game_data.game_id}")
        self.socket.emit(event="game/move", data={'auth': self.credentials.chat_auth, 'player_id': self.credentials.user_id, 'game_id': self.game_data.game_id, 'move': move})

    def resign(self) -> None:
        """Resign the game"""
        logger.info(f"Resigning game {self.game_data.game_id}")
        self.socket.emit(event="game/resign", data={'auth': self.credentials.chat_auth, 'game_id': self.game_data.game_id})  

    def cancel(self) -> None:
        """Cancel the game if within the first few moves"""
        logger.info(f"Canceling game {self.game_data.game_id}")
        self.socket.emit(event="game/cancel", data={'auth': self.credentials.chat_auth, 'game_id': self.game_data.game_id})

    def undo(self, move: int) -> None:
        """Request an undo on the game

        Args:
            move (int): The move number to accept the undo at.        
        """
        logger.info(f"Requesting undo on game {self.game_data.game_id}")
        self.socket.emit(event="game/undo/request", data={'auth': self.credentials.chat_auth, 'game_id': self.game_data.game_id, 'move_number': move})

    def cancel_undo(self, move: int) -> None:
        """Cancel an undo request on the game

        Args:
            move (int): The move number to accept the undo at.        
        """
        logger.info(f"Canceling undo on game {self.game_data.game_id}")
        self.socket.emit(event="game/undo/cancel", data={'auth': self.credentials.chat_auth, 'game_id': self.game_data.game_id, 'move_number': move})

    def accept_undo(self, move: int) -> None:
        """Accept an undo request on the game

        Args:
            move (int): The move number to accept the undo at.
        """
        logger.info(f"Accepting undo on game {self.game_data.game_id}")
        self.socket.emit(event="game/undo/accept", data={'auth': self.credentials.chat_auth, 'game_id': self.game_data.game_id, 'move_number': move})

    def pass_turn(self) -> None:
        """Pass the turn in the game"""
        logger.info(f'Submitting move pass to game {self.game_data.game_id}')
        self.socket.emit(event="game/move", data={'auth': self.credentials.chat_auth, 'player_id': self.credentials.user_id, 'game_id': self.game_data.game_id, 'move': '..'})

    def send_chat(self, message: str, chat_type: str, move: int) -> None:
        """Send a chat message to the game

        Args:
            message (str): The message to send to the game.
            type (str): The type of message to send. Accepts 'main', 'malkovich', 'hidden', or 'personal'
            move (int): The move number to send the message at.

        Examples:
            >>> game.send_chat('Hello World', 'game')
        """
        logger.info(f'Sending chat message to game {self.game_data.game_id}')
        self.socket.emit(event="game/chat", data={'auth': self.credentials.chat_auth, 'game_id': self.game_data.game_id, 'body': message, 'type': chat_type, 'move_number': move})

accept_undo(move)

Accept an undo request on the game

Parameters:

Name	Type	Description	Default
move	int	The move number to accept the undo at.	required

Source code in src/ogsapi/ogsgame.py

def accept_undo(self, move: int) -> None:
    """Accept an undo request on the game

    Args:
        move (int): The move number to accept the undo at.
    """
    logger.info(f"Accepting undo on game {self.game_data.game_id}")
    self.socket.emit(event="game/undo/accept", data={'auth': self.credentials.chat_auth, 'game_id': self.game_data.game_id, 'move_number': move})

cancel()

Cancel the game if within the first few moves

Source code in src/ogsapi/ogsgame.py

def cancel(self) -> None:
    """Cancel the game if within the first few moves"""
    logger.info(f"Canceling game {self.game_data.game_id}")
    self.socket.emit(event="game/cancel", data={'auth': self.credentials.chat_auth, 'game_id': self.game_data.game_id})

cancel_undo(move)

Cancel an undo request on the game

Parameters:

Name	Type	Description	Default
move	int	The move number to accept the undo at.	required

Source code in src/ogsapi/ogsgame.py

def cancel_undo(self, move: int) -> None:
    """Cancel an undo request on the game

    Args:
        move (int): The move number to accept the undo at.        
    """
    logger.info(f"Canceling undo on game {self.game_data.game_id}")
    self.socket.emit(event="game/undo/cancel", data={'auth': self.credentials.chat_auth, 'game_id': self.game_data.game_id, 'move_number': move})

connect()

Connect to the game

Source code in src/ogsapi/ogsgame.py

def connect(self) -> None:
    """Connect to the game"""
    logger.info(f"Connecting to game {self.game_data.game_id}")
    self.socket.emit(event="game/connect", data={'game_id': self.game_data.game_id, 'player_id': self.credentials.user_id, 'chat': False})

disconnect()

Disconnect from the game

Source code in src/ogsapi/ogsgame.py

def disconnect(self) -> None:
    """Disconnect from the game"""
    logger.info(f"Disconnecting game {self.game_data.game_id}")
    self.socket.emit(event="game/disconnect", data={'game_id': self.game_data.game_id})

get_gamedata()

Get game data

Source code in src/ogsapi/ogsgame.py

def get_gamedata(self) -> None:
    """Get game data"""
    logger.info(f"Getting game data for game {self.game_data.game_id}")
    self.socket.emit(event=f"game/{self.game_data.game_id}/gamedata", data={})

move(move)

Submit a move to the game

Parameters:

Name	Type	Description	Default
move	str	The move to submit to the game. Accepts GTP format.	required

Examples:

>>> game.move('B2')

Source code in src/ogsapi/ogsgame.py

def move(self, move: str) -> None:
    """Submit a move to the game

    Args:
        move (str): The move to submit to the game. Accepts GTP format.

    Examples:
        >>> game.move('B2')
    """

    logger.info(f"Submitting move {move} to game {self.game_data.game_id}")
    self.socket.emit(event="game/move", data={'auth': self.credentials.chat_auth, 'player_id': self.credentials.user_id, 'game_id': self.game_data.game_id, 'move': move})

pass_turn()

Pass the turn in the game

Source code in src/ogsapi/ogsgame.py

def pass_turn(self) -> None:
    """Pass the turn in the game"""
    logger.info(f'Submitting move pass to game {self.game_data.game_id}')
    self.socket.emit(event="game/move", data={'auth': self.credentials.chat_auth, 'player_id': self.credentials.user_id, 'game_id': self.game_data.game_id, 'move': '..'})

pause()

Pause the game

Source code in src/ogsapi/ogsgame.py

def pause(self) -> None:
    """Pause the game"""
    logger.info(f"Pausing game {self.game_data.game_id}")
    self.socket.emit(event="game/pause", data={'game_id': self.game_data.game_id})

resign()

Resign the game

Source code in src/ogsapi/ogsgame.py

def resign(self) -> None:
    """Resign the game"""
    logger.info(f"Resigning game {self.game_data.game_id}")
    self.socket.emit(event="game/resign", data={'auth': self.credentials.chat_auth, 'game_id': self.game_data.game_id})  

resume()

Resume the game

Source code in src/ogsapi/ogsgame.py

def resume(self) -> None:
    """Resume the game"""
    logger.info(f"Resuming game {self.game_data.game_id}")
    self.socket.emit(event="game/resume", data={'game_id': self.game_data.game_id})

send_chat(message, chat_type, move)

Send a chat message to the game

Parameters:

Name	Type	Description	Default
message	str	The message to send to the game.	required
type	str	The type of message to send. Accepts 'main', 'malkovich', 'hidden', or 'personal'	required
move	int	The move number to send the message at.	required

Examples:

>>> game.send_chat('Hello World', 'game')

Source code in src/ogsapi/ogsgame.py

def send_chat(self, message: str, chat_type: str, move: int) -> None:
    """Send a chat message to the game

    Args:
        message (str): The message to send to the game.
        type (str): The type of message to send. Accepts 'main', 'malkovich', 'hidden', or 'personal'
        move (int): The move number to send the message at.

    Examples:
        >>> game.send_chat('Hello World', 'game')
    """
    logger.info(f'Sending chat message to game {self.game_data.game_id}')
    self.socket.emit(event="game/chat", data={'auth': self.credentials.chat_auth, 'game_id': self.game_data.game_id, 'body': message, 'type': chat_type, 'move_number': move})

undo(move)

Request an undo on the game

Parameters:

Name	Type	Description	Default
move	int	The move number to accept the undo at.	required

Source code in src/ogsapi/ogsgame.py

def undo(self, move: int) -> None:
    """Request an undo on the game

    Args:
        move (int): The move number to accept the undo at.        
    """
    logger.info(f"Requesting undo on game {self.game_data.game_id}")
    self.socket.emit(event="game/undo/request", data={'auth': self.credentials.chat_auth, 'game_id': self.game_data.game_id, 'move_number': move})

OGSSocket

OGS Socket Class for handling SocketIO connections to OGS

Parameters:

Name	Type	Description	Default
credentials	OGSCredentials	OGSCredentials object containing tokens for authentication to the Socket	required
debug	bool	Enable debug logging. Defaults to False.	required

Attributes:

Name	Type	Description
clock_drift	float	The clock drift of the socket
clock_latency	float	The clock latency of the socket
last_ping	int	The last ping time of the socket
last_issued_ping	int	The last time a ping was issued
games	dict[OGSGame]	A dict of connected game objects
client_callbacks	dict	A dict of socket level callbacks
credentials	OGSCredentials	OGSCredentials object containing tokens for authentication to the Socket
socket	Client	The socketio client object

Source code in src/ogsapi/ogssocket.py

class OGSSocket:
    """OGS Socket Class for handling SocketIO connections to OGS

    Args:
        credentials (OGSCredentials): OGSCredentials object containing tokens for authentication to the Socket
        debug (bool, optional): Enable debug logging. Defaults to False.

    Attributes:
        clock_drift (float): The clock drift of the socket
        clock_latency (float): The clock latency of the socket
        last_ping (int): The last ping time of the socket
        last_issued_ping (int): The last time a ping was issued
        games (dict[OGSGame]): A dict of connected game objects
        client_callbacks (dict): A dict of socket level callbacks
        credentials (OGSCredentials): OGSCredentials object containing tokens for authentication to the Socket
        socket (socketio.Client): The socketio client object

    """

    def __init__(self, credentials: OGSCredentials):
        # Clock Settings
        self.clock_drift = 0.0
        self.clock_latency = 0.0
        self.last_ping = 0.0
        self.last_issued_ping = 0
        # Dict of connected game objects
        self.games: dict[int, OGSGame] = {}
        # Socket level callbacks
        self.callback_handler = lambda event_name, data: None
        self.credentials = credentials
        self.socket = socketio.Client()

    def __del__(self):
        self.disconnect()

    def enable_logging(self) -> None:
        """Enable logging from the socket"""
        logger.enable("engineio.client")
        logger.enable("socketio.client")

    def disable_logging(self) -> None:
        """Disable logging from the socket"""
        logger.disable("engineio.client")
        logger.disable("socketio.client")

    @logger.catch
    def connect(self) -> None:
        """Connect to the socket"""
        self.socket_callbacks()
        logger.info("Connecting to Websocket")
        try:
            self.socket.connect('https://online-go.com/socket.io/?EIO=4', transports='websocket', headers={"Authorization" : f"Bearer {self.credentials.access_token}"})
        except Exception as e:
            raise OGSApiException("Failed to connect to OGS Websocket") from e

    # def register_callback(self, event: str, callback: Callable):
    #     """Register a callback function for receiving data from the API.

    #     Args:
    #         event (str): Event to register the callback function for.
    #             Accepted events are:
    #                 - notification
    #                 - chat
    #                 - error
    #         callback (Callable): Callback function to register.   
    #     """
    #     self.client_callbacks[event]: OGSGame = callback

    # Listens to events received from the socket via the decorators, and calls the appropriate function
    def socket_callbacks(self) -> None:
        """Set the callback functions for the socket"""

        @self.socket.on('connect')
        def authenticate() -> None:
            """Authenticate to the socket"""
            logger.success("Connected to Websocket, authenticating")
            self.socket.emit(event="authenticate", data={"auth": self.credentials.chat_auth, "player_id": self.credentials.user_id, "username": self.credentials.username, "jwt": self.credentials.user_jwt})
            sleep(1)
            logger.success("Authenticated to Websocket")

        @self.socket.on('hostinfo')
        def on_hostinfo(data) -> None:
            """Called when hostinfo is received on the socket"""
            logger.debug(f"Got Hostinfo: {data}")

        @self.socket.on('net/pong')
        def on_pong(data) -> None:
            """Called when a pong is received on the socket"""
            now = time() * 1000
            latency = now - data["client"]
            drift = ((now - latency / 2) - data["server"])
            self.clock_latency = latency / 1000
            self.clock_drift = drift / 1000
            self.last_ping = now / 1000
            logger.debug(f"Got Pong: {data}")

        @self.socket.on('active_game')
        def on_active_game(data) -> None:
            """Called when an active game is received on the socket"""
            logger.debug(f"Got Active Game: {data}")
            self.callback_handler(event_name="active_game", data=data)

        @self.socket.on('notification')
        def on_notification(data) -> None:
            """Called when a notification is received on the socket"""
            logger.debug(f"Got Notification: {data}")
            self.callback_handler(event_name="notification", data=data)

        @self.socket.on('ERROR')
        def on_error(data) -> None:
            """Called when an error is received from the server"""
            logger.error(f"Got Error: {data}")
            self.callback_handler(event_name="ERROR", data=data)

        @self.socket.on('*')
        def catch_all(event, data) -> None:
            """Catch all for events"""
            logger.debug(f"Got Event: {event} with data: {data}")
            self.callback_handler(event_name=event, data=data)

    # Get info on connected server
    def host_info(self) -> None:
        """Get the host info of the socket"""
        logger.info("Getting Host Info")
        self.socket.emit(event="hostinfo", namespace='/')


    def ping(self) -> None:
        """Ping the socket"""
        logger.info("Pinging Websocket")
        self.socket.emit(event="net/ping", data={"client": int(time() * 1000), "drift": self.clock_drift, "latency": self.clock_latency})

    def notification_connect(self) -> None:
        """Connect to the notification socket"""
        logger.info("Connecting to Notification Websocket")
        self.socket.emit(event="notification/connect", data={"auth": self.credentials.notification_auth, "player_id": self.credentials.user_id, "username": self.credentials.username})

    def chat_connect(self) -> None:
        """Connect to the chat socket"""
        logger.info("Connecting to Chat Websocket")
        self.socket.emit(event="chat/connect", data={"auth": self.credentials.chat_auth, "player_id": self.credentials.user_id, "username": self.credentials.username})

    def game_connect(self, game_id: int, callback_handler: Callable | None = None) -> OGSGame:
        """Connect to a game

        Args:
            game_id (int): The id of the game to connect to
            callback_handler (Callable, optional): The callback handler for the game. Defaults to the callback_handler of the socket.

        Returns:
            OGSGame (OGSGame): The game object
        """
        logger.info(f"Connecting to Game {game_id}")
        if callback_handler is None:
            callback_handler = self.callback_handler
        self.games[game_id] = OGSGame(game_socket=self.socket, game_id=game_id, credentials=self.credentials, callback_handler=callback_handler)
        logger.success(f"Connected to Game {game_id}")
        logger.debug(f"{self.games[game_id]}")

        return self.games[game_id]

    def game_disconnect(self, game_id: int) -> None:
        """Disconnect from a game

        Args:
            game_id (int): The id of the game to disconnect from
        """
        logger.info(f"Disconnecting from Game {game_id}")
        del self.games[game_id]

    def disconnect(self) -> None:
        """Disconnect from the socket"""
        logger.info("Disconnecting from Websocket")
        self.socket.disconnect()

chat_connect()

Connect to the chat socket

Source code in src/ogsapi/ogssocket.py

def chat_connect(self) -> None:
    """Connect to the chat socket"""
    logger.info("Connecting to Chat Websocket")
    self.socket.emit(event="chat/connect", data={"auth": self.credentials.chat_auth, "player_id": self.credentials.user_id, "username": self.credentials.username})

connect()

Connect to the socket

Source code in src/ogsapi/ogssocket.py

@logger.catch
def connect(self) -> None:
    """Connect to the socket"""
    self.socket_callbacks()
    logger.info("Connecting to Websocket")
    try:
        self.socket.connect('https://online-go.com/socket.io/?EIO=4', transports='websocket', headers={"Authorization" : f"Bearer {self.credentials.access_token}"})
    except Exception as e:
        raise OGSApiException("Failed to connect to OGS Websocket") from e

disable_logging()

Disable logging from the socket

Source code in src/ogsapi/ogssocket.py

def disable_logging(self) -> None:
    """Disable logging from the socket"""
    logger.disable("engineio.client")
    logger.disable("socketio.client")

disconnect()

Disconnect from the socket

Source code in src/ogsapi/ogssocket.py

def disconnect(self) -> None:
    """Disconnect from the socket"""
    logger.info("Disconnecting from Websocket")
    self.socket.disconnect()

enable_logging()

Enable logging from the socket

Source code in src/ogsapi/ogssocket.py

def enable_logging(self) -> None:
    """Enable logging from the socket"""
    logger.enable("engineio.client")
    logger.enable("socketio.client")

game_connect(game_id, callback_handler=None)

Connect to a game

Parameters:

Name	Type	Description	Default
game_id	int	The id of the game to connect to	required
callback_handler	Callable	The callback handler for the game. Defaults to the callback_handler of the socket.	None

Returns:

Name	Type	Description
OGSGame	OGSGame	The game object

Source code in src/ogsapi/ogssocket.py

def game_connect(self, game_id: int, callback_handler: Callable | None = None) -> OGSGame:
    """Connect to a game

    Args:
        game_id (int): The id of the game to connect to
        callback_handler (Callable, optional): The callback handler for the game. Defaults to the callback_handler of the socket.

    Returns:
        OGSGame (OGSGame): The game object
    """
    logger.info(f"Connecting to Game {game_id}")
    if callback_handler is None:
        callback_handler = self.callback_handler
    self.games[game_id] = OGSGame(game_socket=self.socket, game_id=game_id, credentials=self.credentials, callback_handler=callback_handler)
    logger.success(f"Connected to Game {game_id}")
    logger.debug(f"{self.games[game_id]}")

    return self.games[game_id]

game_disconnect(game_id)

Disconnect from a game

Parameters:

Name	Type	Description	Default
game_id	int	The id of the game to disconnect from	required

Source code in src/ogsapi/ogssocket.py

def game_disconnect(self, game_id: int) -> None:
    """Disconnect from a game

    Args:
        game_id (int): The id of the game to disconnect from
    """
    logger.info(f"Disconnecting from Game {game_id}")
    del self.games[game_id]

host_info()

Get the host info of the socket

Source code in src/ogsapi/ogssocket.py

def host_info(self) -> None:
    """Get the host info of the socket"""
    logger.info("Getting Host Info")
    self.socket.emit(event="hostinfo", namespace='/')

notification_connect()

Connect to the notification socket

Source code in src/ogsapi/ogssocket.py

def notification_connect(self) -> None:
    """Connect to the notification socket"""
    logger.info("Connecting to Notification Websocket")
    self.socket.emit(event="notification/connect", data={"auth": self.credentials.notification_auth, "player_id": self.credentials.user_id, "username": self.credentials.username})

ping()

Ping the socket

Source code in src/ogsapi/ogssocket.py

def ping(self) -> None:
    """Ping the socket"""
    logger.info("Pinging Websocket")
    self.socket.emit(event="net/ping", data={"client": int(time() * 1000), "drift": self.clock_drift, "latency": self.clock_latency})

socket_callbacks()

Set the callback functions for the socket

Source code in src/ogsapi/ogssocket.py

def socket_callbacks(self) -> None:
    """Set the callback functions for the socket"""

    @self.socket.on('connect')
    def authenticate() -> None:
        """Authenticate to the socket"""
        logger.success("Connected to Websocket, authenticating")
        self.socket.emit(event="authenticate", data={"auth": self.credentials.chat_auth, "player_id": self.credentials.user_id, "username": self.credentials.username, "jwt": self.credentials.user_jwt})
        sleep(1)
        logger.success("Authenticated to Websocket")

    @self.socket.on('hostinfo')
    def on_hostinfo(data) -> None:
        """Called when hostinfo is received on the socket"""
        logger.debug(f"Got Hostinfo: {data}")

    @self.socket.on('net/pong')
    def on_pong(data) -> None:
        """Called when a pong is received on the socket"""
        now = time() * 1000
        latency = now - data["client"]
        drift = ((now - latency / 2) - data["server"])
        self.clock_latency = latency / 1000
        self.clock_drift = drift / 1000
        self.last_ping = now / 1000
        logger.debug(f"Got Pong: {data}")

    @self.socket.on('active_game')
    def on_active_game(data) -> None:
        """Called when an active game is received on the socket"""
        logger.debug(f"Got Active Game: {data}")
        self.callback_handler(event_name="active_game", data=data)

    @self.socket.on('notification')
    def on_notification(data) -> None:
        """Called when a notification is received on the socket"""
        logger.debug(f"Got Notification: {data}")
        self.callback_handler(event_name="notification", data=data)

    @self.socket.on('ERROR')
    def on_error(data) -> None:
        """Called when an error is received from the server"""
        logger.error(f"Got Error: {data}")
        self.callback_handler(event_name="ERROR", data=data)

    @self.socket.on('*')
    def catch_all(event, data) -> None:
        """Catch all for events"""
        logger.debug(f"Got Event: {event} with data: {data}")
        self.callback_handler(event_name=event, data=data)

OGSCredentials dataclass

OGS REST API Credentials dataclass

Attributes:

Name	Type	Description
client_id	str	OGS Client ID
client_secret	str	OGS Client Secret
username	str	Case sensitive OGS Username
password	str	OGS Password
access_token	str	Access token to use for authentication. Defaults to None.
refresh_token	str	The refresh token to use for authentication. Defaults to None.
user_id	str	The user ID to use for authentication. Defaults to None.
chat_auth	str	The chat auth token to use for authentication. Defaults to None.
user_jwt	str	The user JWT to use for authentication. Defaults to None.
notification_auth	str	The notification auth token to use for authentication. Defaults to None.

Source code in src/ogsapi/ogscredentials.py

@dataclasses.dataclass
class OGSCredentials:
    """OGS REST API Credentials dataclass

    Attributes:
        client_id (str): OGS Client ID
        client_secret (str): OGS Client Secret
        username (str): Case sensitive OGS Username
        password (str): OGS Password
        access_token (str, optional): Access token to use for authentication. Defaults to None.
        refresh_token (str, optional): The refresh token to use for authentication. Defaults to None.
        user_id (str, optional): The user ID to use for authentication. Defaults to None.
        chat_auth (str, optional): The chat auth token to use for authentication. Defaults to None.
        user_jwt (str, optional): The user JWT to use for authentication. Defaults to None.
        notification_auth (str, optional): The notification auth token to use for authentication. Defaults to None.
    """
    client_id: str
    client_secret: str
    username: str
    password: str
    access_token: str | None = None
    refresh_token: str | None = None
    user_id: str | None = None
    chat_auth: str | None = None
    user_jwt: str | None = None
    notification_auth: str | None = None

OGSGameData dataclass

OGS Game Dataclass

Attributes:

Name	Type	Description
game_id	int	ID of the game.
game_name	str	Name of the game.
private	bool	Whether the game is private or not.
white_player	Player	Player object containing information about the white player.
black_player	Player	Player object containing information about the black player.
ranked	bool	Whether the game is ranked or not.
handicap	int	Handicap of the game.
komi	float	Komi of the game.
width	int	Width of the board.
height	int	Height of the board.
rules	str	Ruleset of the game. EX: "japanese", "chinese", "aga"
time_control	dict	Dictionary containing information about the time control.
phase	str	Phase of the game.
move_list	list[str]	List of moves in the game.
initial_state	dict	Initial state of the game.
start_time	int	Start time of the game.
clock	dict	Dictionary containing the clock data.
latency	int	Latency of the game.

Source code in src/ogsapi/ogsgamedata.py

@dataclasses.dataclass
class OGSGameData:
  """OGS Game Dataclass

  Attributes:
    game_id (int): ID of the game.
    game_name (str): Name of the game.
    private (bool): Whether the game is private or not.
    white_player (Player): Player object containing information about the white player.
    black_player (Player): Player object containing information about the black player.
    ranked (bool): Whether the game is ranked or not.
    handicap (int): Handicap of the game.
    komi (float): Komi of the game.
    width (int): Width of the board.
    height (int): Height of the board.
    rules (str): Ruleset of the game. EX: "japanese", "chinese", "aga"
    time_control (dict): Dictionary containing information about the time control.
    phase (str): Phase of the game.
    move_list (list[str]): List of moves in the game.
    initial_state (dict): Initial state of the game.
    start_time (int): Start time of the game.
    clock (dict): Dictionary containing the clock data.
    latency (int): Latency of the game.

  """

  game_id: int
  game_name: str | None = None
  private: bool | None = None
  white_player: Player = dataclasses.field(default_factory=Player)
  black_player: Player = dataclasses.field(default_factory=Player)
  ranked: bool | None = None
  handicap: int | None = None
  komi: float | None = None
  width: int | None = None
  height: int | None = None
  rules: str | None = None
  time_control: TimeControl = dataclasses.field(default_factory=TimeControl)
  phase: str | None = None
  moves: list[str] = dataclasses.field(default_factory=list)
  initial_state: dict = dataclasses.field(default_factory= lambda: {
    "black": None,
    "white": None
  })
  start_time: int | None = None
  latency: int | None = None

  def update(self, new_values: dict) -> None:
    """Update the game data with new values

    Args:
      new_values (dict): Dictionary containing the new values to update the game data with.
    """
    for key, value in new_values.items():
      if key == "players":
        self.white_player.update(value['white'])
        self.black_player.update(value['black'])
      elif key == "time_control":
        self.time_control.update(value)
      elif hasattr(self, key):
        setattr(self, key, value)
    logger.debug(f"Updated game data: {self}")

update(new_values)

Update the game data with new values

Parameters:

Name	Type	Description	Default
new_values	dict	Dictionary containing the new values to update the game data with.	required

Source code in src/ogsapi/ogsgamedata.py

def update(self, new_values: dict) -> None:
  """Update the game data with new values

  Args:
    new_values (dict): Dictionary containing the new values to update the game data with.
  """
  for key, value in new_values.items():
    if key == "players":
      self.white_player.update(value['white'])
      self.black_player.update(value['black'])
    elif key == "time_control":
      self.time_control.update(value)
    elif hasattr(self, key):
      setattr(self, key, value)
  logger.debug(f"Updated game data: {self}")

Player dataclass

OGS Player Dataclass

Attributes:

Name	Type	Description
username	str	Username of the player.
rank	str	Rank of the player.
professional	bool	Whether the player is a professional or not.
id	int	ID of the player.

Source code in src/ogsapi/ogsgamedata.py

@dataclasses.dataclass
class Player:
  """OGS Player Dataclass

  Attributes:
    username (str): Username of the player.
    rank (str): Rank of the player.
    professional (bool): Whether the player is a professional or not.
    id (int): ID of the player.
  """
  username: str | None = None
  rank: str | None = None
  professional: bool | None = None
  id: int | None = None

  def update(self, new_values: dict) -> None:
    """Update the player data with new values"""
    for key, value in new_values.items():
      if hasattr(self, key):
        setattr(self, key, value)
    logger.debug(f"Updated player data: {self}")

update(new_values)

Update the player data with new values

Source code in src/ogsapi/ogsgamedata.py

def update(self, new_values: dict) -> None:
  """Update the player data with new values"""
  for key, value in new_values.items():
    if hasattr(self, key):
      setattr(self, key, value)
  logger.debug(f"Updated player data: {self}")

TimeControl dataclass

OGS Time Control Dataclass

Attributes:

Name	Type	Description
system	str	Timecontrol system used in the game. EX: "byoyomi", "fischer"
time_control	str	Time control used in the game. EX: "simple", "absolute", "canadian"
speed	str	Speed of the game. EX: "correspondence", "live"
pause_on_weekends	bool	Whether the game pauses on weekends or not.
time_increment	int	Time added to the clock after each move.
initial_time	int	Initial time on the clock.
max_time	int	Maximum time on the clock.

Source code in src/ogsapi/ogsgamedata.py

@dataclasses.dataclass
class TimeControl:
  """OGS Time Control Dataclass

  Attributes:
    system (str): Timecontrol system used in the game. EX: "byoyomi", "fischer"
    time_control (str): Time control used in the game. EX: "simple", "absolute", "canadian"
    speed (str): Speed of the game. EX: "correspondence", "live"
    pause_on_weekends (bool): Whether the game pauses on weekends or not.
    time_increment (int): Time added to the clock after each move.
    initial_time (int): Initial time on the clock.
    max_time (int): Maximum time on the clock.
  """

  system: str | None = None
  time_control: str | None = None
  speed: str | None = None
  pause_on_weekends: bool | None = None
  time_increment: int | None = None
  initial_time: int | None = None
  max_time: int | None = None

  def update(self, new_values: dict) -> None:
    """Update the player data with new values"""
    for key, value in new_values.items():
      if hasattr(self, key):
        setattr(self, key, value)
    logger.debug(f"Updated TimeControl data: {self}")

update(new_values)

Update the player data with new values

Source code in src/ogsapi/ogsgamedata.py

def update(self, new_values: dict) -> None:
  """Update the player data with new values"""
  for key, value in new_values.items():
    if hasattr(self, key):
      setattr(self, key, value)
  logger.debug(f"Updated TimeControl data: {self}")

ByoyomiTime dataclass

OGS Byoyomi Time Data

Attributes:

Name	Type	Description
thinking_time	int	Main time left on the clock.
periods	int	Number of periods left.
period_time	int	Time of each period.

Source code in src/ogsapi/ogsgameclock.py

@dataclasses.dataclass
class ByoyomiTime:
  """OGS Byoyomi Time Data

  Attributes:
    thinking_time (int): Main time left on the clock.
    periods (int): Number of periods left.
    period_time (int): Time of each period.
  """

  thinking_time: int | None = None
  periods: int | None = None
  period_time: int | None = None

  def update(self, new_values: dict) -> None:
    """Update the Byoyomi time data with new values"""
    for key, value in new_values.items():
      if hasattr(self, key):
        setattr(self, key, value)
    logger.debug(f"Updated time data: {self}")

update(new_values)

Update the Byoyomi time data with new values

Source code in src/ogsapi/ogsgameclock.py

def update(self, new_values: dict) -> None:
  """Update the Byoyomi time data with new values"""
  for key, value in new_values.items():
    if hasattr(self, key):
      setattr(self, key, value)
  logger.debug(f"Updated time data: {self}")

FischerTime dataclass

OGS Fischer Time Data

Attributes:

Name	Type	Description
thinking_time	int	Time left on the clock.
increment	int	Time added to the clock after each move.

Source code in src/ogsapi/ogsgameclock.py

@dataclasses.dataclass
class FischerTime:
  """OGS Fischer Time Data

  Attributes:
    thinking_time (int): Time left on the clock.
    increment (int): Time added to the clock after each move.
  """

  thinking_time: int | None = None
  skip_bonus: int | None = None

  def update(self, new_values: dict) -> None:
    """Update the Fischer time data with new values

    Args:
      new_values (dict): New values to update the Fischer time data with."""
    for key, value in new_values.items():
      if hasattr(self, key):
        setattr(self, key, value)
    logger.debug(f"Updated time data: {self}")

update(new_values)

Update the Fischer time data with new values

Parameters:

Name	Type	Description	Default
new_values	dict	New values to update the Fischer time data with.	required

Source code in src/ogsapi/ogsgameclock.py

def update(self, new_values: dict) -> None:
  """Update the Fischer time data with new values

  Args:
    new_values (dict): New values to update the Fischer time data with."""
  for key, value in new_values.items():
    if hasattr(self, key):
      setattr(self, key, value)
  logger.debug(f"Updated time data: {self}")

OGSGameClock dataclass

OGS Game Clock Dataclass

Attributes:

Name	Type	Description
system	str	Timecontrol system used in the game. EX: "byoyomi", "fischer"
current_player	str	Which players turn is it
last_move	str	Last move made in the game.
expiration	int	Time when the game will expire.
received	int	Time when the game clock data was received.
latency_when_received	int	Latency when the game clock data was received.
white_time	ByoyomiTime or FischerTime	White players time control data
black_time	ByoyomiTime or FischerTime	Black players time control data

Source code in src/ogsapi/ogsgameclock.py

@dataclasses.dataclass
class OGSGameClock:
  """OGS Game Clock Dataclass

  Attributes:
    system (str): Timecontrol system used in the game. EX: "byoyomi", "fischer"
    current_player (str): Which players turn is it
    last_move (str): Last move made in the game.
    expiration (int): Time when the game will expire.
    received (int): Time when the game clock data was received.
    latency_when_received (int): Latency when the game clock data was received.
    white_time (ByoyomiTime or FischerTime): White players time control data
    black_time (ByoyomiTime or FischerTime): Black players time control data
    """

  system: str | None = None
  current_player: str | None = None
  last_move: str | None = None
  expiration: int | None = None
  received: int | None = None
  latency_when_received: int | None = None
  white_time: ByoyomiTime | FischerTime | None = None
  black_time: ByoyomiTime | FischerTime | None = None

  def __post_init__(self) -> None:
    # TODO: This expects us to receive the clock AFTER the game data, 
    # which may not always the case  
    self.set_timecontrol()

  def update(self, new_values: dict) -> None:
    """Update the game clock data with new values

    Args:
      new_values (dict): New values to update the game clock data with.
    """
    for key, value in new_values.items():
      if key == "white_time" and self.white_time is not None:
        self.white_time.update(value)
      elif key == "black_time" and self.black_time is not None:
        self.black_time.update(value)
      elif hasattr(self, key):
        setattr(self, key, value)
    # Update the ruleset if it has changed
    logger.debug(f"Updated game clock data: {self}")
    self.set_timecontrol()

  def set_timecontrol(self) -> None:
    """Set the time control attributes based on the time control system"""
    for player in [self.white_time, self.black_time]:
      if self.system == "byoyomi" and player is not ByoyomiTime:
        player = ByoyomiTime()
        logger.debug("Set time control to Byoyomi")
      elif self.system == "fischer" and player is not FischerTime:
        player = FischerTime()
        logger.debug("Set time control to Fischer")
      else:
        player = None
        logger.debug("Set time control to None")

set_timecontrol()

Set the time control attributes based on the time control system

Source code in src/ogsapi/ogsgameclock.py

def set_timecontrol(self) -> None:
  """Set the time control attributes based on the time control system"""
  for player in [self.white_time, self.black_time]:
    if self.system == "byoyomi" and player is not ByoyomiTime:
      player = ByoyomiTime()
      logger.debug("Set time control to Byoyomi")
    elif self.system == "fischer" and player is not FischerTime:
      player = FischerTime()
      logger.debug("Set time control to Fischer")
    else:
      player = None
      logger.debug("Set time control to None")

update(new_values)

Update the game clock data with new values

Parameters:

Name	Type	Description	Default
new_values	dict	New values to update the game clock data with.	required

Source code in src/ogsapi/ogsgameclock.py

def update(self, new_values: dict) -> None:
  """Update the game clock data with new values

  Args:
    new_values (dict): New values to update the game clock data with.
  """
  for key, value in new_values.items():
    if key == "white_time" and self.white_time is not None:
      self.white_time.update(value)
    elif key == "black_time" and self.black_time is not None:
      self.black_time.update(value)
    elif hasattr(self, key):
      setattr(self, key, value)
  # Update the ruleset if it has changed
  logger.debug(f"Updated game clock data: {self}")
  self.set_timecontrol()