players.py

import io
from dataclasses import dataclass
from typing import Generic, Union, TypeVar, overload

from a2s.defaults import DEFAULT_TIMEOUT, DEFAULT_ENCODING
from a2s.a2s_sync import request_sync
from a2s.a2s_async import request_async
from a2s.byteio import ByteReader



A2S_PLAYER_RESPONSE = 0x44


StrType = TypeVar("StrType", str, bytes)  # str (default) or bytes if encoding=None is used

@dataclass
class Player(Generic[StrType]):
    index: int
    """Apparently an entry index, but seems to be always 0"""

    name: StrType
    """Name of the player"""

    score: int
    """Score of the player"""

    duration: float
    """Time the player has been connected to the server"""


@overload
def players(address: tuple[str, int], timeout: float, encoding: str) -> list[Player[str]]:
    ...

@overload
def players(address: tuple[str, int], timeout: float, encoding: None) -> list[Player[bytes]]:
    ...

def players(
    address: tuple[str, int],
    timeout: float = DEFAULT_TIMEOUT,
    encoding: Union[str, None] = DEFAULT_ENCODING
) -> Union[list[Player[str]], list[Player[bytes]]]:
    return request_sync(address, timeout, encoding, PlayersProtocol)

@overload
async def aplayers(address: tuple[str, int], timeout: float, encoding: str) -> list[Player[str]]:
    ...

@overload
async def aplayers(address: tuple[str, int], timeout: float, encoding: None) -> list[Player[bytes]]:
    ...

async def aplayers(
    address: tuple[str, int],
    timeout: float = DEFAULT_TIMEOUT,
    encoding: Union[str, None] = DEFAULT_ENCODING
) -> Union[list[Player[str]], list[Player[bytes]]]:
    return await request_async(address, timeout, encoding, PlayersProtocol)


class PlayersProtocol:
    @staticmethod
    def validate_response_type(response_type):
        return response_type == A2S_PLAYER_RESPONSE

    @staticmethod
    def serialize_request(challenge):
        return b"\x55" + challenge.to_bytes(4, "little")

    @staticmethod
    def deserialize_response(reader, response_type, ping):
        player_count = reader.read_uint8()
        resp = [
            Player(
                index=reader.read_uint8(),
                name=reader.read_cstring(),
                score=reader.read_int32(),
                duration=reader.read_float()
            )
            for player_num in range(player_count)
        ]
        return resp