Adding CoD5 Support

Author: Hornochs

opengsq/protocols/__init__.py

@@ -6,6 +6,7 @@
 from opengsq.protocols.battlefield2 import Battlefield2
 from opengsq.protocols.cod1 import CoD1
 from opengsq.protocols.cod4 import CoD4
+from opengsq.protocols.cod5 import CoD5
 from opengsq.protocols.directplay import DirectPlay
 from opengsq.protocols.doom3 import Doom3
 from opengsq.protocols.eldewrito import ElDewrito

opengsq/protocols/cod5.py

@@ -0,0 +1,172 @@
+from __future__ import annotations
+
+from opengsq.binary_reader import BinaryReader
+from opengsq.exceptions import InvalidPacketException
+from opengsq.protocol_base import ProtocolBase
+from opengsq.protocol_socket import UdpClient
+from opengsq.responses.cod5 import Info, Status, Cod5Status
+
+
+class CoD5(ProtocolBase):
+    """
+    This class represents the Call of Duty 5: World at War Protocol. It provides methods to interact with CoD5 servers.
+    """
+
+    full_name = "Call of Duty 5: World at War Protocol"
+
+    def __init__(self, host: str, port: int = 28960, timeout: float = 5.0):
+        """
+        Initializes the CoD5 object with the given parameters.
+
+        :param host: The host of the server.
+        :param port: The port of the server (default: 28960).
+        :param timeout: The timeout for the server connection.
+        """
+        super().__init__(host, port, timeout)
+        self._source_port = 28960  # CoD5 requires source port 28960
+
+    async def get_info(self, challenge: str = "xxx") -> Info:
+        """
+        Asynchronously retrieves the server information.
+
+        :param challenge: The challenge string to send (default: "xxx").
+        :return: An Info object containing the server information.
+        """
+        # Construct the getinfo payload: ffffffff676574696e666f20787878
+        payload = b"\xFF\xFF\xFF\xFF" + b"getinfo " + challenge.encode('ascii')
+        
+        response_data = await UdpClient.communicate(self, payload, source_port=self._source_port)
+        
+        # Parse the response
+        br = BinaryReader(response_data)
+        
+        # Skip the header (4 bytes of 0xFF)
+        header = br.read_bytes(4)
+        if header != b"\xFF\xFF\xFF\xFF":
+            raise InvalidPacketException(
+                f"Invalid packet header. Expected: \\xFF\\xFF\\xFF\\xFF. Received: {header.hex()}"
+            )
+        
+        # Read the response type
+        response_type = br.read_string([b'\n'])
+        if response_type != "infoResponse":
+            raise InvalidPacketException(
+                f"Unexpected response type. Expected: infoResponse. Received: {response_type}"
+            )
+        
+        # Parse the key-value pairs
+        info_data = self._parse_key_value_pairs(br)
+        
+        return Info(info_data)
+
+    async def get_status(self) -> Status:
+        """
+        Asynchronously retrieves the server status.
+
+        :return: A Status object containing the server status.
+        """
+        # Construct the getstatus payload: ffffffff676574737461747573
+        payload = b"\xFF\xFF\xFF\xFF" + b"getstatus"
+        
+        response_data = await UdpClient.communicate(self, payload, source_port=self._source_port)
+        
+        # Parse the response
+        br = BinaryReader(response_data)
+        
+        # Skip the header (4 bytes of 0xFF)
+        header = br.read_bytes(4)
+        if header != b"\xFF\xFF\xFF\xFF":
+            raise InvalidPacketException(
+                f"Invalid packet header. Expected: \\xFF\\xFF\\xFF\\xFF. Received: {header.hex()}"
+            )
+        
+        # Read the response type
+        response_type = br.read_string([b'\n'])
+        if response_type != "statusResponse":
+            raise InvalidPacketException(
+                f"Unexpected response type. Expected: statusResponse. Received: {response_type}"
+            )
+        
+        # Parse the key-value pairs
+        status_data = self._parse_key_value_pairs(br)
+        
+        return Status(status_data)
+
+    async def get_full_status(self, challenge: str = "xxx") -> Cod5Status:
+        """
+        Asynchronously retrieves both server info and status.
+
+        :param challenge: The challenge string to send (default: "xxx").
+        :return: A Cod5Status object containing both info and status.
+        """
+        import asyncio
+        
+        # Add a small delay between requests to avoid socket conflicts
+        info = await self.get_info(challenge)
+        await asyncio.sleep(0.1)  # 100ms delay
+        status = await self.get_status()
+        
+        return Cod5Status(info=info, status=status)
+
+    def _parse_key_value_pairs(self, br: BinaryReader) -> dict[str, str]:
+        """
+        Parses key-value pairs from the binary reader.
+        CoD5 uses backslash (\) as delimiter between keys and values.
+
+        :param br: The BinaryReader object to parse from.
+        :return: A dictionary containing the parsed key-value pairs.
+        """
+        data = {}
+        
+        # Read the remaining data as string
+        remaining_data = br.read().decode('ascii', errors='ignore')
+        
+        # Split by backslash and process pairs
+        parts = remaining_data.split('\\')
+        
+        # Remove empty first element if it exists (starts with \)
+        if parts and parts[0] == '':
+            parts = parts[1:]
+        
+        # Process pairs (key, value, key, value, ...)
+        for i in range(0, len(parts) - 1, 2):
+            if i + 1 < len(parts):
+                key = parts[i].strip()
+                value = parts[i + 1].strip()
+                if key:  # Only add non-empty keys
+                    data[key] = value
+        
+        return data
+
+
+if __name__ == "__main__":
+    import asyncio
+
+    async def main_async():
+        # Test with the provided server
+        cod5 = CoD5(host="172.29.100.29", port=28960, timeout=5.0)
+        
+        try:
+            print("Getting server info...")
+            info = await cod5.get_info()
+            print(f"Info: {info}")
+            print(f"Hostname: {info.hostname}")
+            print(f"Map: {info.mapname}")
+            print(f"Gametype: {info.gametype}")
+            print(f"Players: {info.clients}/{info.sv_maxclients}")
+            
+            print("\n" + "="*50)
+            print("Getting server status...")
+            await asyncio.sleep(0.2)  # Wait a bit before next request
+            status = await cod5.get_status()
+            print(f"Status: {status}")
+            print(f"Server Name: {status.sv_hostname}")
+            print(f"Game: {status.gamename}")
+            print(f"Map: {status.mapname}")
+            
+        except Exception as e:
+            print(f"Error: {e}")
+            import traceback
+            traceback.print_exc()
+
+    asyncio.run(main_async())

opengsq/responses/cod5/__init__.py

@@ -0,0 +1,3 @@
+from .info import Info
+from .status import Status
+from .cod5_status import Cod5Status

opengsq/responses/cod5/cod5_status.py

@@ -0,0 +1,17 @@
+from dataclasses import dataclass
+from .info import Info
+from .status import Status
+
+
+@dataclass
+class Cod5Status:
+    """
+    Represents the combined status information from a Call of Duty 5: World at War server.
+    Contains both info and status responses.
+    """
+
+    info: Info
+    """The server info response."""
+
+    status: Status
+    """The server status response."""

opengsq/responses/cod5/info.py

@@ -0,0 +1,95 @@
+from dataclasses import dataclass
+
+
+def translate_gametype(gametype_code: str) -> str:
+    """
+    Translate CoD5 gametype codes to German display names.
+    
+    :param gametype_code: The gametype code from the server
+    :return: German display name for the gametype
+    """
+    gametype_translations = {
+        'dm': 'Death Match',
+        'war': 'Team Death Match',
+        'dom': 'Domination',
+        'koth': 'HQ',
+        'sab': 'Sabotage',
+        'sd': 'Search and Destroy',
+        'ctf': 'Capture the Flag'
+    }
+    
+    return gametype_translations.get(gametype_code.lower(), gametype_code)
+
+
+@dataclass
+class Info:
+    """
+    Represents the info response from a Call of Duty 5: World at War server.
+    """
+
+    challenge: str = ""
+    """Challenge string."""
+
+    protocol: str = ""
+    """Protocol version."""
+
+    hostname: str = ""
+    """Server hostname."""
+
+    mapname: str = ""
+    """Current map name."""
+
+    clients: str = ""
+    """Current clients."""
+
+    sv_maxclients: str = ""
+    """Maximum clients."""
+
+    gametype: str = ""
+    """Game type."""
+
+    pure: str = ""
+    """Pure server."""
+
+    hw: str = ""
+    """Hardware information."""
+
+    mod: str = ""
+    """Mod information."""
+
+    voice: str = ""
+    """Voice chat enabled."""
+
+    pb: str = ""
+    """PunkBuster enabled."""
+
+    def __init__(self, data: dict[str, str]):
+        """
+        Initialize Info object from parsed data dictionary.
+        
+        :param data: Dictionary containing server information
+        """
+        for key, value in data.items():
+            if hasattr(self, key):
+                setattr(self, key, value)
+    
+    @property
+    def gametype_translated(self) -> str:
+        """
+        Get the translated gametype name.
+        
+        :return: German display name for the gametype
+        """
+        return translate_gametype(self.gametype)
+    
+    def __getattribute__(self, name):
+        if name == '__dict__':
+            # Create a custom dict that includes properties
+            result = {}
+            # Get the original __dict__ first
+            original_dict = object.__getattribute__(self, '__dict__')
+            result.update(original_dict)
+            # Add the translated gametype
+            result['gametype_translated'] = self.gametype_translated
+            return result
+        return object.__getattribute__(self, name)