Update GameSpy Protocol version 2

opengsq/protocols/gamespy2.py

@@ -1,65 +1,108 @@
 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()
 
         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())

opengsq/responses/gamespy2/__init__.py

@@ -0,0 +1 @@
+from .status import Status

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."""