Update protocols

opengsq/protocols/doom3.py

@@ -7,38 +7,49 @@
 
 
 class Doom3(ProtocolBase):
-    """Doom3 Protocol"""
-    full_name = 'Doom3 Protocol'
+    """
+    This class represents the Doom3 Protocol. It provides methods to interact with the Doom3 API.
+    """
+
+    full_name = "Doom3 Protocol"
 
     _player_fields = {
-        'doom': ['id', 'ping', 'rate', 'name'],
-        'quake4': ['id', 'ping', 'rate', 'name', 'clantag'],
-        'etqw': ['id', 'ping', 'name', 'clantag_pos', 'clantag', 'typeflag']
+        "doom": ["id", "ping", "rate", "name"],
+        "quake4": ["id", "ping", "rate", "name", "clantag"],
+        "etqw": ["id", "ping", "name", "clantag_pos", "clantag", "typeflag"],
     }
 
-    async def get_info(self, strip_color=True):
-        request = b'\xFF\xFFgetInfo\x00ogsq\x00'
+    async def get_info(self, strip_color=True) -> dict:
+        """
+        Asynchronously retrieves the information of the game server.
+
+        :param strip_color: A boolean indicating whether to strip color codes from the information.
+        :return: A dictionary containing the information of the game server.
+        """
+        request = b"\xFF\xFFgetInfo\x00ogsq\x00"
         response = await UdpClient.communicate(self, request)
 
         # Remove the first two 0xFF
         br = BinaryReader(response[2:])
         header = br.read_string()
 
-        if header != 'infoResponse':
-            raise InvalidPacketException(f'Packet header mismatch. Received: {header}. Expected: infoResponse.')
+        if header != "infoResponse":
+            raise InvalidPacketException(
+                f"Packet header mismatch. Received: {header}. Expected: infoResponse."
+            )
 
         # Read challenge
         br.read_bytes(4)
 
-        if br.read_bytes(4) != b'\xff\xff\xff\xff':
+        if br.read_bytes(4) != b"\xff\xff\xff\xff":
             br.stream_position -= 4
 
         info = {}
 
         # Read protocol version
         minor = br.read_short()
         major = br.read_short()
-        info['version'] = f'{major}.{minor}'
+        info["version"] = f"{major}.{minor}"
 
         # Read packet size
         if br.read_long() != br.remaining_bytes():
@@ -49,7 +60,7 @@ async def get_info(self, strip_color=True):
             key = br.read_string().strip()
             val = br.read_string().strip()
 
-            if key == '' and val == '':
+            if key == "" and val == "":
                 break
 
             info[key] = Doom3.strip_colors(val) if strip_color else val
@@ -59,23 +70,33 @@ async def get_info(self, strip_color=True):
         # Try parse the fields
         for mod in self._player_fields.keys():
             try:
-                info['players'] = self.__parse_player(br, self._player_fields[mod], strip_color)
+                info["players"] = self.__parse_player(
+                    br, self._player_fields[mod], strip_color
+                )
                 break
             except Exception:
-                info['players'] = []
+                info["players"] = []
                 br.stream_position = stream_position
 
         return info
 
     def __parse_player(self, br: BinaryReader, fields: list, strip_color: bool):
+        """
+        Parses the player information from the BinaryReader object.
+
+        :param br: The BinaryReader object containing the player information.
+        :param fields: A list of fields to parse from the BinaryReader object.
+        :param strip_color: A boolean indicating whether to strip color codes from the player information.
+        :return: A list of dictionaries containing the player information.
+        """
         players = []
 
         def value_by_field(field: str, br: BinaryReader):
-            if field == 'id' or field == 'clantag_pos' or field == 'typeflag':
+            if field == "id" or field == "clantag_pos" or field == "typeflag":
                 return br.read_byte()
-            elif field == 'ping':
+            elif field == "ping":
                 return br.read_short()
-            elif field == 'rate':
+            elif field == "rate":
                 return br.read_long()
 
             string = br.read_string()
@@ -85,37 +106,42 @@ def value_by_field(field: str, br: BinaryReader):
         while True:
             player = {field: value_by_field(field, br) for field in fields}
 
-            if player['id'] == 32:
+            if player["id"] == 32:
                 break
 
             players.append(player)
 
         return players
 
     @staticmethod
-    def strip_colors(text: str):
-        """Strip color codes"""
-        return re.compile('\\^(X.{6}|.)').sub('', text)
+    def strip_colors(text: str) -> str:
+        """
+        Strips color codes from the given text.
+
+        :param text: The text to strip color codes from.
+        :return: The text with color codes stripped.
+        """
+        return re.compile("\\^(X.{6}|.)").sub("", text)
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     import asyncio
     import json
 
     async def main_async():
         # doom3
-        doom3 = Doom3(host='66.85.14.240', port=27666, timeout=5.0)
+        doom3 = Doom3(host="66.85.14.240", port=27666, timeout=5.0)
         info = await doom3.get_info()
-        print(json.dumps(info, indent=None) + '\n')
+        print(json.dumps(info, indent=None) + "\n")
 
         # etqw
-        doom3 = Doom3(host='178.162.135.83', port=27735, timeout=5.0)
+        doom3 = Doom3(host="178.162.135.83", port=27735, timeout=5.0)
         info = await doom3.get_info()
-        print(json.dumps(info, indent=None) + '\n')
+        print(json.dumps(info, indent=None) + "\n")
 
         # quake4
-        doom3 = Doom3(host='88.99.0.7', port=28007, timeout=5.0)
+        doom3 = Doom3(host="88.99.0.7", port=28007, timeout=5.0)
         info = await doom3.get_info()
-        print(json.dumps(info, indent=None) + '\n')
+        print(json.dumps(info, indent=None) + "\n")
 
     asyncio.run(main_async())

opengsq/protocols/eos.py

@@ -21,9 +21,9 @@ def __init__(
         self,
         host: str,
         port: int,
+        deployment_id: str,
+        access_token: str,
         timeout: float = 5,
-        deployment_id: str = None,
-        access_token: str = None,
     ):
         """
         Initializes the EOS object with the given parameters.
@@ -234,9 +234,9 @@ async def main_async():
         eos = EOS(
             host="5.62.115.46",
             port=7783,
-            timeout=5.0,
             deployment_id=deployment_id,
             access_token=access_token,
+            timeout=5.0,
         )
 
         info = await eos.get_info()

opengsq/protocols/fivem.py

@@ -5,37 +5,61 @@
 
 
 class FiveM(ProtocolBase):
-    """FiveM Protocol (https://docs.fivem.net/docs/server-manual/proxy-setup/)"""
-    full_name = 'FiveM Protocol'
+    """
+    This class represents the FiveM Protocol (https://docs.fivem.net/docs/server-manual/proxy-setup/). It provides methods to interact with the FiveM API.
+    """
+
+    full_name = "FiveM Protocol"
 
     async def _get(self, filename: str) -> dict:
-        url = f'http://{self._host}:{self._port}/{filename}.json?v={int(time.time())}'
+        """
+        Asynchronously retrieves the JSON data from the given filename on the server.
+
+        :param filename: The filename to retrieve data from.
+        :return: A dictionary containing the JSON data.
+        """
+        url = f"http://{self._host}:{self._port}/{filename}.json?v={int(time.time())}"
 
         async with aiohttp.ClientSession() as session:
             async with session.get(url) as response:
                 return await response.json(content_type=None)
 
     async def get_info(self) -> dict:
-        return await self._get('info')
+        """
+        Asynchronously retrieves the information of the game server.
+
+        :return: A dictionary containing the information of the game server.
+        """
+        return await self._get("info")
 
     async def get_players(self) -> list:
-        return await self._get('players')
+        """
+        Asynchronously retrieves the list of players on the game server.
+
+        :return: A list of players on the game server.
+        """
+        return await self._get("players")
 
     async def get_dynamic(self) -> dict:
-        return await self._get('dynamic')
+        """
+        Asynchronously retrieves the dynamic data of the game server.
+
+        :return: A dictionary containing the dynamic data of the game server.
+        """
+        return await self._get("dynamic")
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     import asyncio
     import json
 
     async def main_async():
-        fivem = FiveM(host='144.217.10.12', port=30120, timeout=5.0)
+        fivem = FiveM(host="144.217.10.12", port=30120, timeout=5.0)
         info = await fivem.get_info()
-        print(json.dumps(info, indent=None) + '\n')
+        print(json.dumps(info, indent=None) + "\n")
         players = await fivem.get_players()
-        print(json.dumps(players, indent=None) + '\n')
+        print(json.dumps(players, indent=None) + "\n")
         dynamic = await fivem.get_dynamic()
-        print(json.dumps(dynamic, indent=None) + '\n')
+        print(json.dumps(dynamic, indent=None) + "\n")
 
     asyncio.run(main_async())