Update protocols

.vscode/extensions.json

@@ -3,6 +3,8 @@
         "ms-python.autopep8",
         "ms-python.vscode-pylance",
         "ms-python.python",
-        "donjayamanne.python-environment-manager"
+        "donjayamanne.python-environment-manager",
+        "njpwerner.autodocstring",
+        "ms-python.black-formatter"
     ]
 }

opengsq/binary_reader.py

@@ -67,3 +67,9 @@ def read_string(self, delimiters=[b'\x00'], encoding='utf-8', errors='ignore') -
             bytes_string += stream_byte
 
         return str(bytes_string, encoding=encoding, errors=errors)
+
+    def read_pascal_string(self, encoding='utf-8', errors='ignore'):
+        length = self.read_byte()
+        pascal_string = str(self.read_bytes(length - 1), encoding=encoding, errors=errors)
+        return pascal_string
+

opengsq/exceptions/__init__.py

@@ -0,0 +1,3 @@
+from .authentication_exception import AuthenticationException
+from .invalid_packet_exception import InvalidPacketException
+from .server_not_found_exception import ServerNotFoundException

opengsq/exceptions/authentication_exception.py

@@ -0,0 +1,2 @@
+class AuthenticationException(Exception):
+    pass

opengsq/exceptions/invalid_packet_exception.py

@@ -0,0 +1,54 @@
+class InvalidPacketException(Exception):
+    """Represents errors that occur during application execution when a packet is invalid."""
+
+    def __init__(self, message: str):
+        """
+        Initializes a new instance of the InvalidPacketException class with a specified error message.
+
+        Args:
+            message (str): The message that describes the error.
+        """
+        super().__init__(message)
+
+    @staticmethod
+    def throw_if_not_equal(received, expected):
+        """
+        Checks if the received value is equal to the expected value.
+
+        Args:
+            received: The received value.
+            expected: The expected value.
+
+        Raises:
+            InvalidPacketException: Thrown when the received value does not match the expected value.
+        """
+        if isinstance(received, bytes) and isinstance(expected, bytes):
+            if received != expected:
+                raise InvalidPacketException(
+                    InvalidPacketException.get_message(received, expected)
+                )
+        elif received != expected:
+            raise InvalidPacketException(
+                InvalidPacketException.get_message(received, expected)
+            )
+
+    @staticmethod
+    def get_message(received, expected):
+        """
+        Returns a formatted error message.
+
+        Args:
+            received: The received value.
+            expected: The expected value.
+
+        Returns:
+            str: The formatted error message.
+        """
+        if isinstance(received, bytes) and isinstance(expected, bytes):
+            received_str = " ".join(format(x, "02x") for x in received)
+            expected_str = " ".join(format(x, "02x") for x in expected)
+        else:
+            received_str = str(received)
+            expected_str = str(expected)
+
+        return f"Packet header mismatch. Received: {received_str}. Expected: {expected_str}."

opengsq/exceptions/server_not_found_exception.py

@@ -0,0 +1,2 @@
+class ServerNotFoundException(Exception):
+    pass

opengsq/protocols/ase.py

@@ -2,84 +2,105 @@
 from opengsq.exceptions import InvalidPacketException
 from opengsq.protocol_base import ProtocolBase
 from opengsq.protocol_socket import UdpClient
+from opengsq.responses.ase import Player, Status
 
 
 class ASE(ProtocolBase):
     """All-Seeing Eye Protocol"""
-    full_name = 'All-Seeing Eye Protocol'
 
-    _request = b's'
-    _response = b'EYE1'
+    full_name = "All-Seeing Eye Protocol"
 
-    async def get_status(self) -> dict:
+    _request = b"s"
+    _response = b"EYE1"
+
+    async def get_status(self) -> Status:
+        """
+        Asynchronously get the status of the game server.
+
+        Returns:
+            Status: The status of the game server.
+        """
         response = await UdpClient.communicate(self, self._request)
-        header = response[:4]
-
-        if header != self._response:
-            raise InvalidPacketException(
-                'Packet header mismatch. Received: {}. Expected: {}.'
-                .format(chr(header), chr(self._response))
-            )
-
-        br = BinaryReader(response[4:])
-
-        result = {}
-        result['gamename'] = self.__read_string(br)
-        result['gameport'] = self.__read_string(br)
-        result['hostname'] = self.__read_string(br)
-        result['gametype'] = self.__read_string(br)
-        result['map'] = self.__read_string(br)
-        result['version'] = self.__read_string(br)
-        result['password'] = self.__read_string(br)
-        result['numplayers'] = self.__read_string(br)
-        result['maxplayers'] = self.__read_string(br)
-        result['rules'] = self.__parse_rules(br)
-        result['players'] = self.__parse_players(br)
-
-        return result
-
-    def __parse_rules(self, br: BinaryReader):
+
+        br = BinaryReader(response)
+        header = br.read_bytes(4)
+        InvalidPacketException.throw_if_not_equal(header, self._response)
+
+        return Status(
+            gamename=br.read_pascal_string(),
+            gameport=int(br.read_pascal_string()),
+            hostname=br.read_pascal_string(),
+            gametype=br.read_pascal_string(),
+            map=br.read_pascal_string(),
+            version=br.read_pascal_string(),
+            password=br.read_pascal_string() != "0",
+            numplayers=int(br.read_pascal_string()),
+            maxplayers=int(br.read_pascal_string()),
+            rules=self.__parse_rules(br),
+            players=self.__parse_players(br),
+        )
+
+    def __parse_rules(self, br: BinaryReader) -> dict[str, str]:
         rules = {}
 
         while not br.is_end():
-            key = self.__read_string(br)
+            key = br.read_pascal_string()
 
             if not key:
                 break
 
-            rules[key] = self.__read_string(br)
+            rules[key] = br.read_pascal_string()
 
         return rules
 
-    def __parse_players(self, br: BinaryReader):
-        players = []
-        keys = {1: 'name', 2: 'team', 4: 'skin', 8: 'score', 16: 'ping', 32: 'time'}
+    def __parse_players(self, br: BinaryReader) -> list[Player]:
+        players: list[Player] = []
 
         while not br.is_end():
             flags = br.read_byte()
             player = {}
 
-            for key, value in keys.items():
-                if flags & key == key:
-                    player[value] = self.__read_string(br)
+            if flags & 1 == 1:
+                player["name"] = br.read_pascal_string()
+
+            if flags & 2 == 2:
+                player["team"] = br.read_pascal_string()
+
+            if flags & 4 == 4:
+                player["skin"] = br.read_pascal_string()
+
+            if flags & 8 == 8:
+                try:
+                    player["score"] = int(br.read_pascal_string())
+                except ValueError:
+                    player["score"] = 0
+
+            if flags & 16 == 16:
+                try:
+                    player["ping"] = int(br.read_pascal_string())
+                except ValueError:
+                    player["ping"] = 0
 
-            players.append(player)
+            if flags & 32 == 32:
+                try:
+                    player["time"] = int(br.read_pascal_string())
+                except ValueError:
+                    player["time"] = 0
 
-        return players
+            players.append(Player(**player))
 
-    def __read_string(self, br: BinaryReader):
-        length = br.read_byte()
-        return str(br.read_bytes(length - 1), encoding='utf-8', errors='ignore')
+        return player
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     import asyncio
     import json
+    from dataclasses import asdict
 
     async def main_async():
         # mtasa
-        ase = ASE(host='79.137.97.3', port=22126, timeout=10.0)
+        ase = ASE(host="79.137.97.3", port=22126, timeout=10.0)
         status = await ase.get_status()
-        print(json.dumps(status, indent=None) + '\n')
+        print(json.dumps(asdict(status), indent=None) + "\n")
 
     asyncio.run(main_async())

opengsq/protocols/minecraft.py

@@ -138,7 +138,7 @@ def _unpack_varint(self, br: BinaryReader):
     import asyncio
 
     async def main_async():
-        minecraft = Minecraft(host='valistar.site', port=25565, timeout=5.0)
+        minecraft = Minecraft(host='mc.goldcraft.ir', port=25565, timeout=5.0)
         status = await minecraft.get_status(47, strip_color=True)
         print(json.dumps(status, indent=None, ensure_ascii=False) + '\n')
         status = await minecraft.get_status_pre17()

opengsq/responses/ase/__init__.py

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

opengsq/responses/ase/player.py

@@ -0,0 +1,27 @@
+from dataclasses import asdict, dataclass
+
+
+@dataclass
+class Player:
+    """
+    Represents a player in the game.
+
+    Attributes:
+        name (str): The name of the player.
+        team (str): The team of the player.
+        skin (str): The skin of the player.
+        score (int): The score of the player.
+        ping (int): The ping of the player.
+        time (int): The time of the player.
+    """
+
+    name: str
+    team: str
+    skin: str
+    score: int
+    ping: int
+    time: int
+
+    @property
+    def __dict__(self):
+        return asdict(self)

opengsq/responses/ase/status.py

@@ -0,0 +1,39 @@
+from dataclasses import asdict, dataclass, field
+
+from opengsq.responses.ase.player import Player
+
+
+@dataclass
+class Status:
+    """
+    Represents the status of a game server.
+
+    Attributes:
+        gamename (str): The name of the game.
+        gameport (int): The port number of the game server.
+        hostname (str): The hostname of the game server.
+        gametype (str): The type of the game.
+        map (str): The current map of the game.
+        version (str): The version of the game.
+        password (bool): Whether a password is required to join the game.
+        numplayers (int): The number of players currently in the game.
+        maxplayers (int): The maximum number of players allowed in the game.
+        rules (dict[str, str]): The rules of the game. Defaults to an empty dictionary.
+        players (list[Player]): The players currently in the game. Defaults to an empty list.
+    """
+
+    gamename: str
+    gameport: int
+    hostname: str
+    gametype: str
+    map: str
+    version: str
+    password: bool
+    numplayers: int
+    maxplayers: int
+    rules: dict[str, str] = field(default_factory=dict)
+    players: list[Player] = field(default_factory=list)
+
+    @property
+    def __dict__(self):
+        return asdict(self)