diff --git a/docs/conf.py b/docs/conf.py index 471cef8..bacce40 100644 --- a/docs/conf.py +++ b/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" diff --git a/opengsq/protocols/quake2.py b/opengsq/protocols/quake2.py index 891041b..f8b65da 100644 --- a/opengsq/protocols/quake2.py +++ b/opengsq/protocols/quake2.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import re from opengsq.responses.quake2 import Player, Status diff --git a/opengsq/protocols/scum.py b/opengsq/protocols/scum.py index a6f6dcd..dce2d6f 100644 --- a/opengsq/protocols/scum.py +++ b/opengsq/protocols/scum.py @@ -1,3 +1,6 @@ +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 @@ -5,22 +8,24 @@ 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) @@ -28,26 +33,27 @@ async def get_status(self, master_servers: list = None) -> dict: 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()) diff --git a/opengsq/responses/scum/__init__.py b/opengsq/responses/scum/__init__.py new file mode 100644 index 0000000..2f91dae --- /dev/null +++ b/opengsq/responses/scum/__init__.py @@ -0,0 +1 @@ +from .status import Status diff --git a/opengsq/responses/scum/status.py b/opengsq/responses/scum/status.py new file mode 100644 index 0000000..2c2aa66 --- /dev/null +++ b/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."""