Update Scum Protocol

docs/conf.py

@@ -3,6 +3,8 @@
 # For the full list of built-in configuration values, see the documentation:
 # https://www.sphinx-doc.org/en/master/usage/configuration.html
 
+import os
+
 # -- Project information -----------------------------------------------------
 # https://www.sphinx-doc.org/en/master/usage/configuration.html#project-information
 
@@ -14,12 +16,11 @@
 # -- General configuration ---------------------------------------------------
 # https://www.sphinx-doc.org/en/master/usage/configuration.html#general-configuration
 
-extensions = ['sphinx_rtd_theme', 'sphinx.ext.autodoc', 'sphinxcontrib.googleanalytics']
+extensions = ['sphinx_rtd_theme', 'sphinx.ext.autodoc']
 
 templates_path = ['_templates']
 exclude_patterns = ['_build', 'Thumbs.db', '.DS_Store']
 autodoc_member_order = 'bysource'
-googleanalytics_id = "G-GLNNDPSR1B"
 
 
 # -- Options for HTML output -------------------------------------------------
@@ -28,3 +29,8 @@
 html_theme = 'sphinx_rtd_theme'
 html_favicon = 'favicon.ico'
 html_static_path = ['_static']
+
+# Enabling the extension only when building on GitHub Actions
+if os.getenv("GITHUB_ACTIONS"):
+    extensions.append("sphinxcontrib.googleanalytics")
+    googleanalytics_id = "G-GLNNDPSR1B"

opengsq/protocols/quake2.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import re
 
 from opengsq.responses.quake2 import Player, Status

opengsq/protocols/scum.py

@@ -1,53 +1,59 @@
+from __future__ import annotations
+
+from opengsq.responses.scum import Status
 from opengsq.binary_reader import BinaryReader
 from opengsq.exceptions import ServerNotFoundException
 from opengsq.protocol_base import ProtocolBase
 from opengsq.protocol_socket import Socket, TcpClient
 
 
 class Scum(ProtocolBase):
-    """Scum Protocol"""
-    full_name = 'Scum Protocol'
+    """
+    This class represents the Scum Protocol. It provides methods to interact with the Scum API.
+    """
+
+    full_name = "Scum Protocol"
 
     _master_servers = [
         ("176.57.138.2", 1040),
         ("172.107.16.215", 1040),
-        ("206.189.248.133", 1040)
+        ("206.189.248.133", 1040),
     ]
 
-    async def get_status(self, master_servers: list = None) -> dict:
+    async def get_status(self, master_servers: list[Status] = None) -> Status:
         """
-        Retrieves information about the server
-        Notice: this method calls Scum.query_master_servers()
-        function everytime if master_servers is not passed,
-        you may need to cache the master servers if you had
-        lots of servers to query.
+        Asynchronously retrieves the status of the game server. If the master_servers parameter is not passed, this method calls the Scum.query_master_servers() function every time it is invoked. You may need to cache the master servers if you have a lot of servers to query.
+
+        :param master_servers: A list of master servers. Defaults to None.
+        :return: A Status object containing the status of the game server.
         """
         ip = await Socket.gethostbyname(self._host)
 
         if master_servers is None:
             master_servers = await Scum.query_master_servers()
 
         for server in master_servers:
-            if server['ip'] == ip and server['port'] == self._port:
+            if server.ip == ip and server.port == self._port:
                 return server
 
         raise ServerNotFoundException()
 
     @staticmethod
-    async def query_master_servers() -> list:
-        """
-        Query SCUM Master-Server list
+    async def query_master_servers() -> list[Status]:
         """
+        Asynchronously queries the SCUM Master-Server list.
 
+        :return: A list of Status objects containing the status of the master servers.
+        """
         for host, port in Scum._master_servers:
             try:
                 with TcpClient() as tcpClient:
                     tcpClient.settimeout(5)
                     await tcpClient.connect((host, port))
-                    tcpClient.send(b'\x04\x03\x00\x00')
+                    tcpClient.send(b"\x04\x03\x00\x00")
 
                     total = -1
-                    response = b''
+                    response = b""
                     servers = []
 
                     while total == -1 or len(servers) < total:
@@ -61,19 +67,44 @@ async def query_master_servers() -> list:
                         # server bytes length always 127
                         while br.remaining_bytes() >= 127:
                             server = {}
-                            server['ip'] = '.'.join(map(str, reversed([br.read_byte(), br.read_byte(), br.read_byte(), br.read_byte()])))
-                            server['port'] = br.read_short()
-                            server['name'] = str(br.read_bytes(100).rstrip(b'\x00'), encoding='utf-8', errors='ignore')
+                            server["ip"] = ".".join(
+                                map(
+                                    str,
+                                    reversed(
+                                        [
+                                            br.read_byte(),
+                                            br.read_byte(),
+                                            br.read_byte(),
+                                            br.read_byte(),
+                                        ]
+                                    ),
+                                )
+                            )
+                            server["port"] = br.read_short()
+                            server["name"] = str(
+                                br.read_bytes(100).rstrip(b"\x00"),
+                                encoding="utf-8",
+                                errors="ignore",
+                            )
                             br.read_byte()  # skip
-                            server['numplayers'] = br.read_byte()
-                            server['maxplayers'] = br.read_byte()
-                            server['time'] = br.read_byte()
+                            server["num_players"] = br.read_byte()
+                            server["max_players"] = br.read_byte()
+                            server["time"] = br.read_byte()
                             br.read_byte()  # skip
-                            server['password'] = ((br.read_byte() >> 1) & 1) == 1
+                            server["password"] = ((br.read_byte() >> 1) & 1) == 1
                             br.read_bytes(7)  # skip
-                            v = list(reversed([hex(br.read_byte())[2:].rjust(2, '0') for _ in range(8)]))
-                            server['version'] = f'{int(v[0], 16)}.{int(v[1], 16)}.{int(v[2] + v[3], 16)}.{int(v[4] + v[5] + v[6] + v[7], 16)}'
-                            servers.append(server)
+                            v = list(
+                                reversed(
+                                    [
+                                        hex(br.read_byte())[2:].rjust(2, "0")
+                                        for _ in range(8)
+                                    ]
+                                )
+                            )
+                            server[
+                                "version"
+                            ] = f"{int(v[0], 16)}.{int(v[1], 16)}.{int(v[2] + v[3], 16)}.{int(v[4] + v[5] + v[6] + v[7], 16)}"
+                            servers.append(Status(**server))
 
                         # if the length is less than 127, save the unused bytes for next loop
                         response = br.read()
@@ -82,17 +113,18 @@ async def query_master_servers() -> list:
             except TimeoutError:
                 pass
 
-        raise Exception('Failed to connect to any of the master servers')
+        raise Exception("Failed to connect to any of the master servers")
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     import asyncio
     import json
+    from dataclasses import asdict
 
     async def main_async():
-        scum = Scum(host='15.235.181.19', port=7042, timeout=5.0)
+        scum = Scum(host="15.235.181.19", port=7042, timeout=5.0)
         master_servers = await scum.query_master_servers()
         server = await scum.get_status(master_servers)
-        print(json.dumps(server, indent=None) + '\n')
+        print(json.dumps(asdict(server), indent=None) + "\n")
 
     asyncio.run(main_async())

opengsq/responses/scum/__init__.py

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

opengsq/responses/scum/status.py

@@ -0,0 +1,32 @@
+from dataclasses import dataclass
+
+
+@dataclass
+class Status:
+    """
+    Represents the response status of a server.
+    """
+
+    ip: str
+    """The IP address of the server."""
+
+    port: int
+    """The port number of the server."""
+
+    name: str
+    """The name of the server."""
+
+    num_players: int
+    """The number of players currently connected to the server."""
+
+    max_players: int
+    """The maximum number of players that can connect to the server."""
+
+    time: int
+    """The server time."""
+
+    password: bool
+    """A value indicating whether a password is required to connect to the server."""
+
+    version: str
+    """The version of the server."""