Support Satisfactory Protocol

Author: BattlefieldDuck

README.md

@@ -23,6 +23,7 @@ from opengsq.protocols.quake2 import Quake2
 from opengsq.protocols.quake3 import Quake3
 from opengsq.protocols.raknet import Raknet
 from opengsq.protocols.samp import Samp
+from opengsq.protocols.satisfactory import Satisfactory
 from opengsq.protocols.source import Source
 from opengsq.protocols.teamspeak3 import Teamspeak3
 from opengsq.protocols.unreal2 import Unreal2
@@ -66,6 +67,8 @@ asyncio.run(main())
 Rcon server using Source Remote Console, example output: [tests/results/test_source/test_remote_console.txt](/tests/results/test_source/test_remote_console.txt)
 ```py
 import asyncio
+
+from opengsq.exceptions import AuthenticationException
 from opengsq.protocols import Source
 
 async def main():
@@ -74,9 +77,9 @@ async def main():
             await rcon.authenticate('serverRconPassword')
             result = await rcon.send_command('cvarlist')
             print(result)
-        except:
+        except AuthenticationException:
             print('Fail to authenticate')
-        
+
 asyncio.run(main())
 ```
 

opengsq/cli.py

@@ -10,13 +10,17 @@
 from typing import Mapping, Sequence
 
 from opengsq.protocol_base import ProtocolBase
+from opengsq.version import __version__
 
 
 class CLI:
     def __init__(self):
         self.__paths = {}
 
     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')
+
         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)')
@@ -32,7 +36,7 @@ def register(self, parser: argparse.ArgumentParser):
                     self.__paths[name] = fullpath
 
                     # Add parser and arguments
-                    obj = locate(fullpath)
+                    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_')]
@@ -41,6 +45,12 @@ def register(self, parser: argparse.ArgumentParser):
 
     # 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 __version__
+        else:
+            del args.version
+
         # Load the obj from path
         obj = locate(self.__paths[args.subparser_name])
         del args.subparser_name

opengsq/protocols/__init__.py

@@ -11,6 +11,7 @@
 from opengsq.protocols.quake3 import Quake3
 from opengsq.protocols.raknet import Raknet
 from opengsq.protocols.samp import Samp
+from opengsq.protocols.satisfactory import Satisfactory
 from opengsq.protocols.source import Source
 from opengsq.protocols.teamspeak3 import Teamspeak3
 from opengsq.protocols.unreal2 import Unreal2

opengsq/protocols/satisfactory.py

@@ -0,0 +1,49 @@
+import struct
+
+from opengsq.binary_reader import BinaryReader
+from opengsq.exceptions import InvalidPacketException
+from opengsq.protocol_base import ProtocolBase
+from opengsq.socket_async import SocketAsync
+
+
+class Satisfactory(ProtocolBase):
+    """Satisfactory Protocol"""
+    full_name = 'Satisfactory Protocol'
+
+    async def get_status(self) -> dict:
+        """
+        Retrieves information about the server including state, version, and beacon port
+        Server state: 1 - Idle (no game loaded), 2 - currently loading or creating a game, 3 - currently in game
+        """
+        # Credit: https://github.com/dopeghoti/SF-Tools/blob/main/Protocol.md
+
+        # Send message id, protocol version
+        request = struct.pack('2b', 0, 0) + 'opengsq'.encode()
+        response = await SocketAsync.send_and_receive(self._address, self._query_port, self._timeout, request)
+        br = BinaryReader(response)
+        header = br.read_byte()
+
+        if header != 1:
+            raise InvalidPacketException('Packet header mismatch. Received: {}. Expected: {}.'.format(chr(header), chr(1)))
+
+        br.read_byte()  # Protocol version
+        br.read_bytes(8)  # Request data
+
+        result = {}
+        result['State'] = br.read_byte()
+        result['Version'] = br.read_long()
+        result['BeaconPort'] = br.read_short()
+
+        return result
+
+
+if __name__ == '__main__':
+    import asyncio
+    import json
+
+    async def main_async():
+        satisfactory = Satisfactory(address='delta3.ptse.host', query_port=15777, timeout=5.0)
+        status = await satisfactory.get_status()
+        print(json.dumps(status, indent=None) + '\n')
+
+    asyncio.run(main_async())

opengsq/version.py

@@ -1 +1 @@
-__version__ = '1.4.5'
+__version__ = '1.5.0'

tests/protocols/test_satisfactory.py

@@ -0,0 +1,18 @@
+import os
+
+import pytest
+
+from opengsq.protocols.satisfactory import Satisfactory
+
+from .result_handler import ResultHandler
+
+handler = ResultHandler(os.path.basename(__file__)[:-3])
+handler.enable_save = True
+
+# Satisfactory
+test = Satisfactory(address='delta3.ptse.host', query_port=15777)
+
+@pytest.mark.asyncio
+async def test_get_status():
+    result = await test.get_status()
+    await handler.save_result('test_get_status', result)

tests/results/test_satisfactory/test_get_status.json

@@ -0,0 +1,5 @@
+{
+    "State": 3,
+    "Version": 211839,
+    "BeaconPort": 15000
+}