diff --git a/opengsq/cli.py b/opengsq/cli.py index d5b0aa9..c535634 100644 --- a/opengsq/cli.py +++ b/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() diff --git a/opengsq/protocol_base.py b/opengsq/protocol_base.py index 391e04c..001aab4 100644 --- a/opengsq/protocol_base.py +++ b/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): diff --git a/opengsq/protocols/__init__.py b/opengsq/protocols/__init__.py index 2efacbd..e6ae6d2 100644 --- a/opengsq/protocols/__init__.py +++ b/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 diff --git a/opengsq/protocols/ase.py b/opengsq/protocols/ase.py index 6ee48bf..a831da0 100644 --- a/opengsq/protocols/ase.py +++ b/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()) diff --git a/opengsq/protocols/battlefield.py b/opengsq/protocols/battlefield.py index d94cfcc..317f330 100644 --- a/opengsq/protocols/battlefield.py +++ b/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()) diff --git a/opengsq/protocols/doom3.py b/opengsq/protocols/doom3.py index 0dee5fb..0bb53fb 100644 --- a/opengsq/protocols/doom3.py +++ b/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()) diff --git a/opengsq/protocols/fivem.py b/opengsq/protocols/fivem.py index efbc15d..e246f42 100644 --- a/opengsq/protocols/fivem.py +++ b/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()) diff --git a/opengsq/protocols/gamespy1.py b/opengsq/protocols/gamespy1.py index 56c2085..7e2ec0f 100644 --- a/opengsq/protocols/gamespy1.py +++ b/opengsq/protocols/gamespy1.py @@ -212,8 +212,6 @@ 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 @@ -221,6 +219,6 @@ async def main_async(): # 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()) diff --git a/opengsq/protocols/gamespy2.py b/opengsq/protocols/gamespy2.py index c6aa02f..4381223 100644 --- a/opengsq/protocols/gamespy2.py +++ b/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()) diff --git a/opengsq/protocols/gamespy3.py b/opengsq/protocols/gamespy3.py index 666aea4..4ae6de1 100644 --- a/opengsq/protocols/gamespy3.py +++ b/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()) diff --git a/opengsq/protocols/gamespy4.py b/opengsq/protocols/gamespy4.py index e53143a..2b0fd06 100644 --- a/opengsq/protocols/gamespy4.py +++ b/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()) diff --git a/opengsq/protocols/killingfloor.py b/opengsq/protocols/killingfloor.py index 046181c..9a92d48 100644 --- a/opengsq/protocols/killingfloor.py +++ b/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()) diff --git a/opengsq/protocols/minecraft.py b/opengsq/protocols/minecraft.py index e8694ae..89a367b 100644 --- a/opengsq/protocols/minecraft.py +++ b/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()) diff --git a/opengsq/protocols/quake1.py b/opengsq/protocols/quake1.py index 628034d..368dbf3 100644 --- a/opengsq/protocols/quake1.py +++ b/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()) diff --git a/opengsq/protocols/quake2.py b/opengsq/protocols/quake2.py index f8b65da..f6b6990 100644 --- a/opengsq/protocols/quake2.py +++ b/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()) diff --git a/opengsq/protocols/quake3.py b/opengsq/protocols/quake3.py index f480429..2269ebd 100644 --- a/opengsq/protocols/quake3.py +++ b/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()) diff --git a/opengsq/protocols/raknet.py b/opengsq/protocols/raknet.py index a483cf2..b39fc92 100644 --- a/opengsq/protocols/raknet.py +++ b/opengsq/protocols/raknet.py @@ -79,12 +79,10 @@ async def get_status(self) -> Status: if __name__ == "__main__": import asyncio - import json - from dataclasses import asdict async def main_async(): raknet = RakNet(host="mc.advancius.net", port=19132, timeout=5.0) status = await raknet.get_status() - print(json.dumps(asdict(status), indent=None) + "\n") + print(status) asyncio.run(main_async()) diff --git a/opengsq/protocols/samp.py b/opengsq/protocols/samp.py index 2501b8f..85a7f78 100644 --- a/opengsq/protocols/samp.py +++ b/opengsq/protocols/samp.py @@ -107,16 +107,18 @@ def __read_string(self, br: BinaryReader, read_offset=1): if __name__ == "__main__": import asyncio - import json - from dataclasses import asdict async def main_async(): samp = Samp(host="51.254.178.238", port=7777, timeout=5.0) status = await samp.get_status() - print(json.dumps(asdict(status), indent=None) + "\n") + print(status) + + await asyncio.sleep(5) players = await samp.get_players() - print(json.dumps([asdict(player) for player in players], indent=None) + "\n") + print(players) + + await asyncio.sleep(5) rules = await samp.get_rules() - print(json.dumps(rules, indent=None) + "\n") + print(rules) asyncio.run(main_async()) diff --git a/opengsq/protocols/satisfactory.py b/opengsq/protocols/satisfactory.py index f6d4e7e..d096a38 100644 --- a/opengsq/protocols/satisfactory.py +++ b/opengsq/protocols/satisfactory.py @@ -49,12 +49,10 @@ async def get_status(self) -> Status: if __name__ == "__main__": import asyncio - import json - from dataclasses import asdict async def main_async(): satisfactory = Satisfactory(host="79.136.0.124", port=15777, timeout=5.0) status = await satisfactory.get_status() - print(json.dumps(asdict(status), indent=None) + "\n") + print(status) asyncio.run(main_async()) diff --git a/opengsq/protocols/scum.py b/opengsq/protocols/scum.py index dce2d6f..0d0b8ed 100644 --- a/opengsq/protocols/scum.py +++ b/opengsq/protocols/scum.py @@ -118,13 +118,13 @@ async def query_master_servers() -> list[Status]: if __name__ == "__main__": import asyncio - import json - from dataclasses import asdict async def main_async(): + master_servers = await Scum.query_master_servers() + print(master_servers) + 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(asdict(server), indent=None) + "\n") + print(server) asyncio.run(main_async()) diff --git a/opengsq/protocols/source.py b/opengsq/protocols/source.py index 0e3c776..3c1c616 100644 --- a/opengsq/protocols/source.py +++ b/opengsq/protocols/source.py @@ -373,17 +373,18 @@ async def __parse_gold_source_packet(self, udpClient: UdpClient, packets: list): if __name__ == "__main__": import asyncio - import json async def main_async(): source = Source(host="45.62.160.71", port=27015, timeout=5.0) info = await source.get_info() - print(json.dumps(info, ensure_ascii=False, default=lambda o: o.__dict__) + "\n") + print(info) + + await asyncio.sleep(1) players = await source.get_players() - print( - json.dumps(players, ensure_ascii=False, default=lambda o: o.__dict__) + "\n" - ) + print(players) + + await asyncio.sleep(1) rules = await source.get_rules() - print(json.dumps(rules, ensure_ascii=False) + "\n") + print(rules) asyncio.run(main_async()) diff --git a/opengsq/protocols/teamspeak3.py b/opengsq/protocols/teamspeak3.py index 9703714..4443259 100644 --- a/opengsq/protocols/teamspeak3.py +++ b/opengsq/protocols/teamspeak3.py @@ -118,10 +118,10 @@ async def main_async(): host="145.239.200.2", port=10011, voice_port=9987, timeout=5.0 ) info = await teamspeak3.get_info() - print(json.dumps(info, indent=None) + "\n") + print(info) clients = await teamspeak3.get_clients() - print(json.dumps(clients, indent=None) + "\n") + print(clients) channels = await teamspeak3.get_channels() - print(json.dumps(channels, indent=None) + "\n") + print(channels) asyncio.run(main_async()) diff --git a/opengsq/protocols/unreal2.py b/opengsq/protocols/unreal2.py index 99fe0f0..403b82b 100644 --- a/opengsq/protocols/unreal2.py +++ b/opengsq/protocols/unreal2.py @@ -165,17 +165,15 @@ def _read_string(self, br: BinaryReader): if __name__ == "__main__": import asyncio - import json - from dataclasses import asdict async def main_async(): # ut2004 unreal2 = Unreal2(host="109.230.224.189", port=6970) details = await unreal2.get_details() - print(json.dumps(asdict(details), indent=None) + "\n") + print(details) rules = await unreal2.get_rules() - print(json.dumps(rules, indent=None) + "\n") + print(rules) players = await unreal2.get_players() - print(json.dumps([asdict(player) for player in players], indent=None) + "\n") + print(players) asyncio.run(main_async()) diff --git a/opengsq/protocols/vcmp.py b/opengsq/protocols/vcmp.py index 58e9524..bf16d6b 100644 --- a/opengsq/protocols/vcmp.py +++ b/opengsq/protocols/vcmp.py @@ -95,14 +95,12 @@ def __read_string(self, br: BinaryReader, read_offset=1): if __name__ == "__main__": import asyncio - import json - from dataclasses import asdict async def main_async(): vcmp = Vcmp(host="51.178.65.136", port=8114, timeout=5.0) status = await vcmp.get_status() - print(json.dumps(asdict(status), indent=None) + "\n") + print(status) players = await vcmp.get_players() - print(json.dumps([asdict(player) for player in players], indent=None) + "\n") + print(players) asyncio.run(main_async()) diff --git a/opengsq/protocols/won.py b/opengsq/protocols/won.py index 356e3bd..859d40e 100644 --- a/opengsq/protocols/won.py +++ b/opengsq/protocols/won.py @@ -3,24 +3,24 @@ class WON(Source): """World Opponent Network (WON) Query Protocol""" - full_name = 'World Opponent Network (WON) Protocol' - _A2S_INFO = b'details\0' - _A2S_PLAYER = b'players' - _A2S_RULES = b'rules' + full_name = "World Opponent Network (WON) Protocol" + _A2S_INFO = b"details\0" + _A2S_PLAYER = b"players" + _A2S_RULES = b"rules" -if __name__ == '__main__': + +if __name__ == "__main__": import asyncio - import json async def main_async(): - won = WON(host='212.227.190.150', port=27020, timeout=5.0) + won = WON(host="212.227.190.150", port=27020, timeout=5.0) info = await won.get_info() + print(info) players = await won.get_players() + print(players) rules = await won.get_rules() - print(json.dumps(info, indent=None, ensure_ascii=False) + '\n') - print(json.dumps(players, indent=None, ensure_ascii=False) + '\n') - print(json.dumps(rules, indent=None, ensure_ascii=False) + '\n') + print(rules) asyncio.run(main_async())