Update Protocols

opengsq/cli.py

@@ -19,17 +19,24 @@ def __init__(self):
 
     def register(self, parser: argparse.ArgumentParser):
         # Add version argument
-        parser.add_argument('-V', '--version', action='store_true', help='print the opengsq version number and exit')
+        parser.add_argument(
+            "-V",
+            "--version",
+            action="store_true",
+            help="print the opengsq version number and exit",
+        )
 
         opengsq_path = os.path.abspath(os.path.dirname(__file__))
-        subparsers = parser.add_subparsers(dest='subparser_name')
-        pattern = re.compile(r'from\s+(\S+)\s+import\s+(.+,?\S)')
+        subparsers = parser.add_subparsers(dest="subparser_name")
+        pattern = re.compile(r"from\s+(\S+)\s+import\s+(.+,?\S)")
 
         # Load all protocols from __init__.py
-        with open(os.path.join(opengsq_path, 'protocols', '__init__.py')) as f:
-            for (protocol_path, protocol_classnames) in re.findall(pattern, f.read()):
-                for protocol_classname in protocol_classnames.split(','):
-                    name, fullpath, parameters = self.__extract(protocol_path, protocol_classname)
+        with open(os.path.join(opengsq_path, "protocols", "__init__.py")) as f:
+            for protocol_path, protocol_classnames in re.findall(pattern, f.read()):
+                for protocol_classname in protocol_classnames.split(","):
+                    name, fullpath, parameters = self.__extract(
+                        protocol_path, protocol_classname
+                    )
 
                     # Save to self.__paths dictionary
                     # Example: name = 'source', fullpath = 'opengsq.protocols.source.Source'
@@ -39,15 +46,26 @@ def register(self, parser: argparse.ArgumentParser):
                     obj: ProtocolBase = locate(fullpath)
                     sub = subparsers.add_parser(name, help=obj.full_name)
                     self.__add_arguments(sub, parameters)
-                    method_names = [func for func in dir(obj) if callable(getattr(obj, func)) and func.startswith('get_')]
-                    sub.add_argument('--function', default=method_names[0], type=str, help='(default: %(default)s)')
-                    sub.add_argument('--indent', default=None, type=int, nargs='?')
+
+                    method_names = [
+                        func
+                        for func in dir(obj)
+                        if callable(getattr(obj, func)) and func.startswith("get_")
+                    ]
+
+                    sub.add_argument(
+                        "--function",
+                        default=method_names[0],
+                        type=str,
+                        help="(default: %(default)s)",
+                    )
+                    sub.add_argument("--indent", default=None, type=int, nargs="?")
 
     # Get the query response in json format
     async def run(self, args: Sequence[str]) -> str:
         # Return version if -V or --version
         if args.version:
-            return 'v' + __version__
+            return "v" + __version__
         else:
             del args.version
 
@@ -69,28 +87,46 @@ async def run(self, args: Sequence[str]) -> str:
         protocol: ProtocolBase = obj()
         func = getattr(protocol, function)
 
-        return json.dumps(await func(), ensure_ascii=False, indent=indent)
+        return json.dumps(
+            await func(),
+            ensure_ascii=False,
+            indent=indent,
+            default=lambda o: o.__dict__,
+        )
 
     # Extract name, fullpath, parameters from path, classname
     def __extract(self, path: str, classname: str):
-        name = path.split('.')[-1]
-        fullpath = '{}.{}'.format(path, classname.strip())
+        name = path.split(".")[-1]
+        fullpath = "{}.{}".format(path, classname.strip())
         parameters = inspect.signature(locate(fullpath).__init__).parameters
 
         return name, fullpath, parameters
 
-    def __add_arguments(self, sub: argparse.ArgumentParser, parameters: Mapping[str, inspect.Parameter]):
+    def __add_arguments(
+        self, sub: argparse.ArgumentParser, parameters: Mapping[str, inspect.Parameter]
+    ):
         for key in parameters:
-            if parameters[key].name == 'self':
+            if parameters[key].name == "self":
                 continue
 
-            name_or_flags = '--{}'.format(parameters[key].name)
+            name_or_flags = "--{}".format(parameters[key].name)
             required = parameters[key].default == inspect._empty
             default = None if required else parameters[key].default
-            type = parameters[key].annotation
-            help = None if required else '(default: %(default)s)'
 
-            sub.add_argument(name_or_flags, default=default, required=required, type=type, help=help)
+            type_mapping = {"str": str, "int": int, "float": float}
+            type_func = type_mapping.get(
+                parameters[key].annotation, parameters[key].annotation
+            )
+
+            help = None if required else "(default: %(default)s)"
+
+            sub.add_argument(
+                name_or_flags,
+                default=default,
+                required=required,
+                type=type_func,
+                help=help,
+            )
 
 
 def main():
@@ -114,11 +150,11 @@ async def main_async():
         result = await cli.run(args)
         sys.stdout.write(result)
     except asyncio.exceptions.TimeoutError:
-        sys.stderr.write('opengsq: error: timed out\n')
+        sys.stderr.write("opengsq: error: timed out\n")
         sys.exit(-2)
 
     sys.exit(0)
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     main()

opengsq/protocol_base.py

@@ -4,7 +4,7 @@
 class ProtocolBase(abc.ABC):
     @property
     @abc.abstractmethod
-    def full_name(self):
+    def full_name(self) -> str:
         pass
 
     def __init__(self, host: str, port: int, timeout: float = 5.0):

opengsq/protocols/__init__.py

@@ -1,34 +1,3 @@
-"""
-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/ase.py

@@ -99,12 +99,11 @@ def __parse_players(self, br: BinaryReader) -> list[Player]:
 
 if __name__ == "__main__":
     import asyncio
-    import json
 
     async def main_async():
         # mtasa
         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(status)
 
     asyncio.run(main_async())

opengsq/protocols/battlefield.py

@@ -126,8 +126,6 @@ def __decode(self, response: bytes):
 
 if __name__ == "__main__":
     import asyncio
-    import json
-    from dataclasses import asdict
 
     async def main_async():
         entries = [
@@ -140,10 +138,10 @@ async def main_async():
         for address, query_port in entries:
             battlefield = Battlefield(address, query_port, timeout=10.0)
             info = await battlefield.get_info()
-            print(json.dumps(asdict(info), indent=None) + "\n")
+            print(info)
             version = await battlefield.get_version()
-            print(json.dumps(asdict(version), indent=None) + "\n")
+            print(version)
             players = await battlefield.get_players()
-            print(json.dumps(players, indent=None) + "\n")
+            print(players)
 
     asyncio.run(main_async())

opengsq/protocols/doom3.py

@@ -126,22 +126,21 @@ def strip_colors(text: str) -> str:
 
 if __name__ == "__main__":
     import asyncio
-    import json
 
     async def main_async():
         # doom3
         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(info)
 
         # etqw
         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(info)
 
         # quake4
         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(info)
 
     asyncio.run(main_async())

opengsq/protocols/fivem.py

@@ -51,15 +51,14 @@ async def get_dynamic(self) -> dict:
 
 if __name__ == "__main__":
     import asyncio
-    import json
 
     async def main_async():
         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(info)
         players = await fivem.get_players()
-        print(json.dumps(players, indent=None) + "\n")
+        print(players)
         dynamic = await fivem.get_dynamic()
-        print(json.dumps(dynamic, indent=None) + "\n")
+        print(dynamic)
 
     asyncio.run(main_async())

opengsq/protocols/gamespy1.py

@@ -212,15 +212,13 @@ def __parse_as_object(self, br: BinaryReader, is_player=False):
 
 if __name__ == "__main__":
     import asyncio
-    import json
-    from dataclasses import asdict
 
     async def main_async():
         gs1 = GameSpy1(host="51.81.48.224", port=23000, timeout=5.0)  # bfield1942
         # gs1 = GameSpy1(address='139.162.235.20', query_port=7778, timeout=5.0)  # ut
         # gs1 = GameSpy1(address='192.223.24.6', query_port=7778, timeout=5.0)  # ut
         gs1 = GameSpy1(host="141.94.205.35", port=12300, timeout=5.0)  # mohaa
         status = await gs1.get_status()
-        print(json.dumps(asdict(status), indent=None) + "\n")
+        print(status)
 
     asyncio.run(main_async())

opengsq/protocols/gamespy2.py

@@ -163,13 +163,11 @@ def __get_teams(self, br: BinaryReader) -> list:
 
 if __name__ == "__main__":
     import asyncio
-    import json
-    from dataclasses import asdict
 
     async def main_async():
         # bfv
         gs2 = GameSpy2(host="108.61.236.22", port=23000, timeout=5.0)
         status = await gs2.get_status()
-        print(json.dumps(asdict(status), indent=None) + "\n")
+        print(status)
 
     asyncio.run(main_async())

opengsq/protocols/gamespy3.py

@@ -152,11 +152,10 @@ async def __read(self, udpClient: UdpClient) -> bytes:
 
 if __name__ == "__main__":
     import asyncio
-    import json
 
     async def main_async():
-        gs3 = GameSpy3(host="185.107.96.59", port=29900, timeout=5.0)
+        gs3 = GameSpy3(host="95.172.92.116", port=29900, timeout=5.0)
         server = await gs3.get_status()
-        print(json.dumps(server, indent=None))
+        print(server)
 
     asyncio.run(main_async())

opengsq/protocols/gamespy4.py

@@ -12,11 +12,10 @@ class GameSpy4(GameSpy3):
 
 if __name__ == "__main__":
     import asyncio
-    import json
 
     async def main_async():
         gs4 = GameSpy4(host="play.avengetech.me", port=19132, timeout=5.0)
         server = await gs4.get_status()
-        print(json.dumps(server, indent=None))
+        print(server)
 
     asyncio.run(main_async())

opengsq/protocols/killingfloor.py

@@ -53,17 +53,15 @@ async def get_details(self) -> Status:
 
 if __name__ == "__main__":
     import asyncio
-    import json
-    from dataclasses import asdict
 
     async def main_async():
         # killingfloor
         killingFloor = KillingFloor(host="185.80.128.168", port=7708, timeout=10.0)
         details = await killingFloor.get_details()
-        print(json.dumps(asdict(details), indent=None) + "\n")
+        print(details)
         rules = await killingFloor.get_rules()
-        print(json.dumps(rules, indent=None) + "\n")
+        print(rules)
         players = await killingFloor.get_players()
-        print(json.dumps([asdict(player) for player in players], indent=None) + "\n")
+        print(players)
 
     asyncio.run(main_async())

opengsq/protocols/minecraft.py

@@ -165,13 +165,12 @@ def _unpack_varint(self, br: BinaryReader):
 
 if __name__ == "__main__":
     import asyncio
-    from dataclasses import asdict
 
     async def main_async():
         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")
+        print(status)
         status_pre17 = await minecraft.get_status_pre17()
-        print(json.dumps(asdict(status_pre17), indent=None, ensure_ascii=False) + "\n")
+        print(status_pre17)
 
     asyncio.run(main_async())

opengsq/protocols/quake1.py

@@ -151,12 +151,10 @@ async def _connect_and_send(self, data):
 
 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)
         status = await quake1.get_status()
-        print(json.dumps(asdict(status), indent=None) + "\n")
+        print(status)
 
     asyncio.run(main_async())

opengsq/protocols/quake2.py

@@ -60,12 +60,10 @@ def _parse_players(self, br: BinaryReader):
 
 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)
         status = await quake2.get_status()
-        print(json.dumps(asdict(status), indent=None) + "\n")
+        print(status)
 
     asyncio.run(main_async())

opengsq/protocols/quake3.py

@@ -83,13 +83,12 @@ def strip_colors(text: str) -> str:
 
 if __name__ == "__main__":
     import asyncio
-    import json
 
     async def main_async():
         quake3 = Quake3(host="108.61.18.110", port=27960, timeout=5.0)
         info = await quake3.get_info()
+        print(info)
         status = await quake3.get_status()
-        print(json.dumps(info, indent=None, default=lambda o: o.__dict__) + "\n")
-        print(json.dumps(status, indent=None, default=lambda o: o.__dict__) + "\n")
+        print(status)
 
     asyncio.run(main_async())