diff --git a/opengsq/protocols/gamespy2.py b/opengsq/protocols/gamespy2.py index 8a402cb..c6aa02f 100644 --- a/opengsq/protocols/gamespy2.py +++ b/opengsq/protocols/gamespy2.py @@ -1,58 +1,95 @@ from enum import Flag, auto +from opengsq.responses.gamespy2 import Status from opengsq.binary_reader import BinaryReader from opengsq.protocol_base import ProtocolBase from opengsq.protocol_socket import UdpClient class GameSpy2(ProtocolBase): - """GameSpy Protocol version 2""" - full_name = 'GameSpy Protocol version 2' + """ + This class represents the GameSpy Protocol version 2. It provides methods to interact with the GameSpy API. + """ + + full_name = "GameSpy Protocol version 2" class Request(Flag): INFO = auto() PLAYERS = auto() TEAMS = auto() - async def get_status(self, request: Request = Request.INFO | Request.PLAYERS | Request.TEAMS) -> dict: - """Retrieves information about the server including, Info, Players, and Teams.""" - data = b'\xFE\xFD\x00\x04\x05\x06\x07' + self.__get_request_bytes(request) + async def get_status( + self, request: Request = Request.INFO | Request.PLAYERS | Request.TEAMS + ) -> Status: + """ + Asynchronously retrieves the status of the game server. + + :param request: A Request object indicating the type of information to retrieve. + :return: A Status object containing the status of the game server. + """ + data = b"\xFE\xFD\x00\x04\x05\x06\x07" + self.__get_request_bytes(request) response = await UdpClient.communicate(self, data) # Remove the first 5 bytes { 0x00, 0x04, 0x05, 0x06, 0x07 } br = BinaryReader(response[5:]) - status = {} - - if self.__has_flag(request, self.Request.INFO): - status['info'] = self.__get_info(br) - - if self.__has_flag(request, self.Request.PLAYERS): - status['players'] = self.__get_players(br) - - if self.__has_flag(request, self.Request.TEAMS): - status['teams'] = self.__get_teams(br) + info = ( + self.__get_info(br) if self.__has_flag(request, self.Request.INFO) else {} + ) + players = ( + self.__get_players(br) + if self.__has_flag(request, self.Request.PLAYERS) + else [] + ) + teams = ( + self.__get_teams(br) if self.__has_flag(request, self.Request.TEAMS) else [] + ) - return status + return Status(info, players, teams) def __get_request_bytes(self, request: Request): - request_bytes = self.__has_flag(request, self.Request.INFO) and b'\xFF' or b'\x00' - request_bytes += self.__has_flag(request, self.Request.PLAYERS) and b'\xFF' or b'\x00' - request_bytes += self.__has_flag(request, self.Request.TEAMS) and b'\xFF' or b'\x00' + """ + Gets the request bytes for the given request. + + :param request: The request to get the request bytes for. + :return: The request bytes. + """ + request_bytes = ( + self.__has_flag(request, self.Request.INFO) and b"\xFF" or b"\x00" + ) + request_bytes += ( + self.__has_flag(request, self.Request.PLAYERS) and b"\xFF" or b"\x00" + ) + request_bytes += ( + self.__has_flag(request, self.Request.TEAMS) and b"\xFF" or b"\x00" + ) return request_bytes def __has_flag(self, request, flag) -> bool: + """ + Checks if the given request has the specified flag. + + :param request: The request to check. + :param flag: The flag to check for. + :return: True if the request has the flag, False otherwise. + """ return request & flag == flag def __get_info(self, br: BinaryReader) -> dict: + """ + Gets the information from the given BinaryReader object. + + :param br: The BinaryReader object to get the information from. + :return: A dictionary containing the information. + """ info = {} # Read all key values while br.remaining_bytes() > 0: key = br.read_string() - if key == '': + if key == "": break info[key] = br.read_string().strip() @@ -60,6 +97,12 @@ def __get_info(self, br: BinaryReader) -> dict: return info def __get_players(self, br: BinaryReader) -> list: + """ + Gets the players from the given BinaryReader object. + + :param br: The BinaryReader object to get the players from. + :return: A list containing the players. + """ players = [] # Skip a byte @@ -74,10 +117,10 @@ def __get_players(self, br: BinaryReader) -> list: while br.remaining_bytes() > 0: key = br.read_string() - if key == '': + if key == "": break - keys.append(key.rstrip('_')) + keys.append(key.rstrip("_")) # Set all keys and values for _ in range(player_count): @@ -86,6 +129,12 @@ def __get_players(self, br: BinaryReader) -> list: return players def __get_teams(self, br: BinaryReader) -> list: + """ + Gets the teams from the given BinaryReader object. + + :param br: The BinaryReader object to get the teams from. + :return: A list containing the teams. + """ teams = [] # Skip a byte @@ -100,10 +149,10 @@ def __get_teams(self, br: BinaryReader) -> list: while br.remaining_bytes() > 0: key = br.read_string() - if key == '': + if key == "": break - keys.append(key.rstrip('t').rstrip('_')) + keys.append(key.rstrip("t").rstrip("_")) # Set all keys and values for _ in range(team_count): @@ -112,14 +161,15 @@ def __get_teams(self, br: BinaryReader) -> list: return teams -if __name__ == '__main__': +if __name__ == "__main__": import asyncio import json + from dataclasses import asdict async def main_async(): # bfv - gs2 = GameSpy2(host='108.61.236.22', port=23000, timeout=5.0) + gs2 = GameSpy2(host="108.61.236.22", port=23000, timeout=5.0) status = await gs2.get_status() - print(json.dumps(status, indent=None) + '\n') + print(json.dumps(asdict(status), indent=None) + "\n") asyncio.run(main_async()) diff --git a/opengsq/responses/gamespy2/__init__.py b/opengsq/responses/gamespy2/__init__.py new file mode 100644 index 0000000..2f91dae --- /dev/null +++ b/opengsq/responses/gamespy2/__init__.py @@ -0,0 +1 @@ +from .status import Status diff --git a/opengsq/responses/gamespy2/status.py b/opengsq/responses/gamespy2/status.py new file mode 100644 index 0000000..c644c2b --- /dev/null +++ b/opengsq/responses/gamespy2/status.py @@ -0,0 +1,19 @@ +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass +class Status: + """ + Represents the status of the server. + """ + + info: dict[str, str] + """The server information.""" + + players: list[dict[str, str]] + """The list of players.""" + + teams: list[dict[str, str]] + """The list of teams."""