Update Quake1, Quake2 Protocol

opengsq/protocols/__init__.py

@@ -1,3 +1,34 @@
+"""
+This module provides interfaces to various game server query protocols.
+
+The following protocols are supported:
+- ASE (All-Seeing Eye)
+- Battlefield
+- Doom3
+- EOS (Epic Online Services)
+- FiveM
+- GameSpy1
+- GameSpy2
+- GameSpy3
+- GameSpy4
+- KillingFloor
+- Minecraft
+- Quake1
+- Quake2
+- Quake3
+- RakNet
+- Samp (San Andreas Multiplayer)
+- Satisfactory
+- Scum
+- Source (Source Engine)
+- TeamSpeak3
+- Unreal2
+- Vcmp (Vice City Multiplayer)
+- WON (World Opponent Network)
+
+Each protocol is implemented as a separate module and can be imported individually.
+"""
+
 from opengsq.protocols.ase import ASE
 from opengsq.protocols.battlefield import Battlefield
 from opengsq.protocols.doom3 import Doom3

opengsq/protocols/quake1.py

@@ -1,50 +1,74 @@
 import re
 
+from opengsq.responses.quake1 import Player, Status
 from opengsq.binary_reader import BinaryReader
 from opengsq.protocol_base import ProtocolBase
 from opengsq.protocol_socket import UdpClient
 
 
 class Quake1(ProtocolBase):
-    """Quake1 Protocol"""
-    full_name = 'Quake1 Protocol'
+    """
+    This class represents the Quake1 Protocol. It provides methods to interact with the Quake1 API.
+    """
+
+    full_name = "Quake1 Protocol"
 
     def __init__(self, host: str, port: int, timeout: float = 5.0):
-        """Quake1 Query Protocol"""
+        """
+        Initializes the Quake1 Query Protocol.
+
+        :param host: The host of the game server.
+        :param port: The port of the game server.
+        :param timeout: The timeout for the connection. Defaults to 5.0.
+        """
         super().__init__(host, port, timeout)
-        self._delimiter1 = b'\\'
-        self._delimiter2 = b'\n'
-        self._request_header = b'status'
-        self._response_header = 'n'
+        self._delimiter1 = b"\\"
+        self._delimiter2 = b"\n"
+        self._request_header = b"status"
+        self._response_header = "n"
 
-    async def get_status(self) -> dict:
-        """This returns server information and players."""
+    async def get_status(self) -> Status:
+        """
+        Asynchronously retrieves the status of the game server.
+
+        :return: A Status object containing the status of the game server.
+        """
         br = await self._get_response_binary_reader()
 
-        return {
-            'info': self._parse_info(br),
-            'players': self._parse_players(br),
-        }
+        return Status(info=self._parse_info(br), players=self._parse_players(br))
 
     async def _get_response_binary_reader(self) -> BinaryReader:
+        """
+        Asynchronously gets the response from the game server and returns a BinaryReader object.
+
+        :return: A BinaryReader object containing the response from the game server.
+        """
         response_data = await self._connect_and_send(self._request_header)
 
         br = BinaryReader(response_data)
         header = br.read_string(self._delimiter1)
 
         if header != self._response_header:
-            raise Exception(f'Packet header mismatch. Received: {header}. Expected: {self._response_header}.')
+            raise Exception(
+                f"Packet header mismatch. Received: {header}. Expected: {self._response_header}."
+            )
 
         return br
 
     def _parse_info(self, br: BinaryReader) -> dict:
+        """
+        Parses the information from the given BinaryReader object.
+
+        :param br: The BinaryReader object to parse the information from.
+        :return: A dictionary containing the information.
+        """
         info = {}
 
         # Read all key values until meet \n
         while br.remaining_bytes() > 0:
             key = br.read_string(self._delimiter1)
 
-            if key == '':
+            if key == "":
                 break
 
             info[key] = br.read_string([self._delimiter1, self._delimiter2])
@@ -57,25 +81,39 @@ def _parse_info(self, br: BinaryReader) -> dict:
         return info
 
     def _parse_players(self, br: BinaryReader) -> list:
+        """
+        Parses the players from the given BinaryReader object.
+
+        :param br: The BinaryReader object to parse the players from.
+        :return: A list containing the players.
+        """
         players = []
 
         for matches in self._get_player_match_collections(br):
             matches: list[re.Match] = [match.group() for match in matches]
 
-            players.append({
-                'id': int(matches[0]),
-                'score': int(matches[1]),
-                'time': int(matches[2]),
-                'ping': int(matches[3]),
-                'name': str(matches[4]).strip('"'),
-                'skin': str(matches[5]).strip('"'),
-                'color1': int(matches[6]),
-                'color2': int(matches[7]),
-            })
+            players.append(
+                Player(
+                    id=int(matches[0]),
+                    score=int(matches[1]),
+                    time=int(matches[2]),
+                    ping=int(matches[3]),
+                    name=str(matches[4]).strip('"'),
+                    skin=str(matches[5]).strip('"'),
+                    color1=int(matches[6]),
+                    color2=int(matches[7]),
+                )
+            )
 
         return players
 
     def _get_player_match_collections(self, br: BinaryReader):
+        """
+        Gets the player match collections from the given BinaryReader object.
+
+        :param br: The BinaryReader object to get the player match collections from.
+        :return: The player match collections.
+        """
         match_collections = []
 
         # Regex to split with whitespace and double quote
@@ -88,8 +126,14 @@ def _get_player_match_collections(self, br: BinaryReader):
         return match_collections
 
     async def _connect_and_send(self, data):
-        header = b'\xFF\xFF\xFF\xFF'
-        response_data = await UdpClient.communicate(self, header + data + b'\x00')
+        """
+        Asynchronously connects to the game server and sends the given data.
+
+        :param data: The data to send to the game server.
+        :return: The response from the game server.
+        """
+        header = b"\xFF\xFF\xFF\xFF"
+        response_data = await UdpClient.communicate(self, header + data + b"\x00")
 
         # Remove the last 0x00 if exists (Only if Quake1)
         if response_data[-1] == 0:
@@ -100,16 +144,17 @@ async def _connect_and_send(self, data):
             response_data += self._delimiter2
 
         # Remove the first four 0xFF
-        return response_data[len(header):]
+        return response_data[len(header) :]
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     import asyncio
     import json
+    from dataclasses import asdict
 
     async def main_async():
-        quake1 = Quake1(host='35.185.44.174', port=27500, timeout=5.0)
+        quake1 = Quake1(host="35.185.44.174", port=27500, timeout=5.0)
         status = await quake1.get_status()
-        print(json.dumps(status, indent=None) + '\n')
+        print(json.dumps(asdict(status), indent=None) + "\n")
 
     asyncio.run(main_async())

opengsq/protocols/quake2.py

@@ -1,47 +1,69 @@
 import re
 
+from opengsq.responses.quake2 import Player, Status
 from opengsq.binary_reader import BinaryReader
 from opengsq.protocols.quake1 import Quake1
 
 
 class Quake2(Quake1):
-    """Quake2 Protocol"""
-    full_name = 'Quake2 Protocol'
+    """
+    This class represents the Quake2 Protocol. It provides methods to interact with the Quake2 API.
+    """
+
+    full_name = "Quake2 Protocol"
 
     def __init__(self, host: str, port: int, timeout: float = 5.0):
-        """Quake2 Query Protocol"""
+        """
+        Initializes the Quake2 Query Protocol.
+
+        :param host: The host of the game server.
+        :param port: The port of the game server.
+        :param timeout: The timeout for the connection. Defaults to 5.0.
+        """
         super().__init__(host, port, timeout)
-        self._response_header = 'print\n'
+        self._response_header = "print\n"
+
+    async def get_status(self) -> Status:
+        """
+        Asynchronously retrieves the status of the game server.
+
+        :return: A Status object containing the status of the game server.
+        """
+        br = await self._get_response_binary_reader()
+        return Status(info=self._parse_info(br), players=self._parse_players(br))
 
-    def _parse_players(self, br: BinaryReader) -> list:
+    def _parse_players(self, br: BinaryReader):
+        """
+        Parses the players from the given BinaryReader object.
+
+        :param br: The BinaryReader object to parse the players from.
+        :return: A list containing the players.
+        """
         players = []
 
         for matches in self._get_player_match_collections(br):
             matches: list[re.Match] = [match.group() for match in matches]
 
-            player = {
-                'frags': int(matches[0]),
-                'ping': int(matches[1]),
-            }
-
-            if len(matches) > 2:
-                player['name'] = str(matches[2]).strip('"')
-
-            if len(matches) > 3:
-                player['address'] = str(matches[3]).strip('"')
+            player = Player(
+                frags=int(matches[0]),
+                ping=int(matches[1]),
+                name=str(matches[2]).strip('"') if len(matches) > 2 else "",
+                address=str(matches[3]).strip('"') if len(matches) > 3 else "",
+            )
 
             players.append(player)
 
         return players
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     import asyncio
     import json
+    from dataclasses import asdict
 
     async def main_async():
-        quake2 = Quake2(host='46.165.236.118', port=27910, timeout=5.0)
+        quake2 = Quake2(host="46.165.236.118", port=27910, timeout=5.0)
         status = await quake2.get_status()
-        print(json.dumps(status, indent=None) + '\n')
+        print(json.dumps(asdict(status), indent=None) + "\n")
 
     asyncio.run(main_async())

opengsq/responses/quake1/__init__.py

@@ -0,0 +1,2 @@
+from .player import Player
+from .status import Status

opengsq/responses/quake1/player.py

@@ -0,0 +1,32 @@
+from dataclasses import dataclass
+
+
+@dataclass
+class Player:
+    """
+    Represents a player in the game.
+    """
+
+    id: int
+    """The player's ID."""
+
+    score: int
+    """The player's score."""
+
+    time: int
+    """The player's time."""
+
+    ping: int
+    """The player's ping."""
+
+    name: str
+    """The player's name."""
+
+    skin: str
+    """The player's skin."""
+
+    color1: int
+    """The player's first color."""
+
+    color2: int
+    """The player's second color."""

opengsq/responses/quake1/status.py

@@ -0,0 +1,15 @@
+from dataclasses import dataclass
+from .player import Player
+
+
+@dataclass
+class Status:
+    """
+    Represents the status of the server.
+    """
+
+    info: dict[str, str]
+    """The server information."""
+
+    players: list[Player]
+    """The list of players."""

opengsq/responses/quake2/__init__.py

@@ -0,0 +1,2 @@
+from .player import Player
+from .status import Status

opengsq/responses/quake2/player.py

@@ -0,0 +1,20 @@
+from dataclasses import dataclass
+
+
+@dataclass
+class Player:
+    """
+    Represents a player in the game.
+    """
+
+    frags: int
+    """The player's frags."""
+
+    ping: int
+    """The player's ping."""
+
+    name: str
+    """The player's name."""
+
+    address: str
+    """The player's address."""

opengsq/responses/quake2/status.py

@@ -0,0 +1,15 @@
+from dataclasses import dataclass
+from .player import Player
+
+
+@dataclass
+class Status:
+    """
+    Represents the status of the server.
+    """
+
+    info: dict[str, str]
+    """The server information."""
+
+    players: list[Player]
+    """The list of players."""