Merge pull request #35 from Hornochs/add-renegade-x

Adding Renegade LAN Support

Author: BattlefieldDuck

docs/tests/protocols/index.rst

@@ -12,6 +12,7 @@ Protocols Tests
 	test_minecraft/index
 	test_raknet/index
 	test_eos/index
+	test_renegadex/index
 	test_kaillera/index
 	test_ase/index
 	test_quake1/index
@@ -32,3 +33,4 @@ Protocols Tests
 	test_vcmp/index
 	test_satisfactory/index
 	test_gamespy3/index
+	test_renegadex/index

docs/tests/protocols/test_renegadex/index.rst

@@ -0,0 +1,7 @@
+.. _test_renegadex:
+
+test_renegadex
+==============
+
+.. toctree::
+	test_renegadex_status

docs/tests/protocols/test_renegadex/test_renegadex_status.rst

@@ -0,0 +1,46 @@
+test_renegadex_status
+=====================
+
+Here are the results for the test method.
+
+.. code-block:: json
+
+	{
+	    "name": "Ich bin ein ziemlich langer Server der nicht weis wie lang das geht",
+	    "current_map": "CNC-Field",
+	    "port": 7777,
+	    "players": 0,
+	    "game_version": "Open Beta 5.85.815",
+	    "variables": {
+	        "player_limit": 64,
+	        "vehicle_limit": 20,
+	        "mine_limit": 24,
+	        "time_limit": 50,
+	        "passworded": false,
+	        "steam_required": false,
+	        "team_mode": 6,
+	        "spawn_crates": true,
+	        "game_type": 1,
+	        "ranked": false
+	    },
+	    "raw": {
+	        "Current Map": "CNC-Field",
+	        "Players": 0,
+	        "Port": 7777,
+	        "Name": "Ich bin ein ziemlich langer Server der nicht weis wie lang das geht",
+	        "IP": "10.13.37.149",
+	        "Game Version": "Open Beta 5.85.815",
+	        "Variables": {
+	            "Player Limit": 64,
+	            "Vehicle Limit": 20,
+	            "Mine Limit": 24,
+	            "Time Limit": 50,
+	            "bPassworded": false,
+	            "bSteamRequired": false,
+	            "Team Mode": 6,
+	            "bSpawnCrates": true,
+	            "Game Type": 1,
+	            "bRanked": false
+	        }
+	    }
+	}

opengsq/protocols/__init__.py

@@ -16,6 +16,7 @@
 from opengsq.protocols.quake2 import Quake2
 from opengsq.protocols.quake3 import Quake3
 from opengsq.protocols.raknet import RakNet
+from opengsq.protocols.renegadex import RenegadeX
 from opengsq.protocols.samp import Samp
 from opengsq.protocols.satisfactory import Satisfactory
 from opengsq.protocols.scum import Scum

opengsq/protocols/renegadex.py

@@ -0,0 +1,49 @@
+import json
+import asyncio
+from opengsq.protocol_base import ProtocolBase
+from opengsq.responses.renegadex import Status
+
+class RenegadeX(ProtocolBase):
+    full_name = "Renegade X Protocol"
+    BROADCAST_PORT = 45542
+
+    def __init__(self, host: str, port: int = 7777, timeout: float = 5.0):
+        super().__init__(host, port, timeout)
+        
+    async def get_status(self) -> Status:
+        loop = asyncio.get_running_loop()
+        queue = asyncio.Queue()
+        
+        class BroadcastProtocol(asyncio.DatagramProtocol):
+            def __init__(self, queue, host):
+                self.queue = queue
+                self.target_host = host
+                
+            def datagram_received(self, data, addr):
+                if addr[0] == self.target_host:
+                    self.queue.put_nowait(data)
+        
+        transport, _ = await loop.create_datagram_endpoint(
+            lambda: BroadcastProtocol(queue, self._host),
+            local_addr=('0.0.0.0', self.BROADCAST_PORT)
+        )
+        
+        try:
+            complete_data = bytearray()
+            while True:
+                try:
+                    data = await asyncio.wait_for(queue.get(), timeout=self._timeout)
+                    complete_data.extend(data)
+                    
+                    try:
+                        json_str = complete_data.decode('utf-8')
+                        server_info = json.loads(json_str)
+                        return Status.from_dict(server_info)
+                    except (UnicodeDecodeError, json.JSONDecodeError):
+                        continue
+                        
+                except asyncio.TimeoutError:
+                    raise TimeoutError("No broadcast received from the specified server")
+                    
+        finally:
+            transport.close()

opengsq/responses/renegadex/__init__.py

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

opengsq/responses/renegadex/status.py

@@ -0,0 +1,52 @@
+from dataclasses import dataclass
+from typing import Any
+
+@dataclass
+class Variables:
+    player_limit: int
+    vehicle_limit: int
+    mine_limit: int
+    time_limit: int
+    passworded: bool
+    steam_required: bool
+    team_mode: int
+    spawn_crates: bool
+    game_type: int
+    ranked: bool
+    
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> 'Variables':
+        return cls(
+            player_limit=data["Player Limit"],
+            vehicle_limit=data["Vehicle Limit"],
+            mine_limit=data["Mine Limit"],
+            time_limit=data["Time Limit"],
+            passworded=data["bPassworded"],
+            steam_required=data["bSteamRequired"],
+            team_mode=data["Team Mode"],
+            spawn_crates=data["bSpawnCrates"],
+            game_type=data["Game Type"],
+            ranked=data["bRanked"]
+        )
+
+@dataclass
+class Status:
+    name: str
+    current_map: str
+    port: int
+    players: int
+    game_version: str
+    variables: Variables
+    raw: dict[str, Any]
+    
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> 'Status':
+        return cls(
+            name=data["Name"],
+            current_map=data["Current Map"],
+            port=data["Port"],
+            players=data["Players"],
+            game_version=data["Game Version"],
+            variables=Variables.from_dict(data["Variables"]),
+            raw=data
+        )

tests/protocols/test_renegadex.py

@@ -0,0 +1,33 @@
+import pytest
+from opengsq.protocols.renegadex import RenegadeX
+from ..result_handler import ResultHandler
+
+handler = ResultHandler(__file__)
+handler.enable_save = True
+
+@pytest.mark.asyncio
+async def test_renegadex_status():
+    rx = RenegadeX(host="10.13.37.149")
+    result = await rx.get_status()
+
+    print("\nRenegade X Server Details:")
+    print(f"Server Name: {result.name}")
+    print(f"Current Map: {result.current_map}")
+    print(f"Game Version: {result.game_version}")
+    print(f"Players: {result.players}/{result.variables.player_limit}")
+    print(f"Port: {result.port}")
+
+    print("\nServer Settings:")
+    print(f"Vehicle Limit: {result.variables.vehicle_limit}")
+    print(f"Mine Limit: {result.variables.mine_limit}")
+    print(f"Time Limit: {result.variables.time_limit}")
+    print(f"Team Mode: {result.variables.team_mode}")
+    print(f"Game Type: {result.variables.game_type}")
+
+    print("\nServer Flags:")
+    print(f"Password Protected: {'Yes' if result.variables.passworded else 'No'}")
+    print(f"Steam Required: {'Yes' if result.variables.steam_required else 'No'}")
+    print(f"Spawn Crates: {'Yes' if result.variables.spawn_crates else 'No'}")
+    print(f"Ranked: {'Yes' if result.variables.ranked else 'No'}")
+
+    await handler.save_result("test_renegadex_status", result)