Palworld REST API Support

README.md

@@ -27,6 +27,7 @@ from opengsq.protocols import (
     Kaillera,
     KillingFloor,
     Minecraft,
+    Palworld,
     Quake1,
     Quake2,
     Quake3,

docs/tests/protocols/test_palworld/index.rst

@@ -0,0 +1,7 @@
+.. _test_palworld:
+
+test_palworld
+=================
+
+.. toctree::
+	test_get_status

docs/tests/protocols/test_palworld/test_get_status.rst

@@ -0,0 +1,12 @@
+test_get_status
+===============
+
+Here are the results for the test method.
+
+.. code-block:: json
+
+	{
+	    "num_players": 3,
+	    "max_players": 32,
+	    "server_name": "A Palworld Server"
+	}

opengsq/protocols/__init__.py

@@ -10,6 +10,7 @@
 from opengsq.protocols.kaillera import Kaillera
 from opengsq.protocols.killingfloor import KillingFloor
 from opengsq.protocols.minecraft import Minecraft
+from opengsq.protocols.palworld import Palworld
 from opengsq.protocols.quake1 import Quake1
 from opengsq.protocols.quake2 import Quake2
 from opengsq.protocols.quake3 import Quake3

opengsq/protocols/palworld.py

@@ -0,0 +1,80 @@
+import struct
+import time
+import aiohttp
+
+from opengsq.responses.palworld import Status
+from opengsq.protocol_base import ProtocolBase
+
+
+class Palworld(ProtocolBase):
+    """
+    This class represents the Palworld Protocol. It provides methods to interact with the Palworld REST API.
+    """
+
+    full_name = "Palworld Protocol"
+
+    def __init__(self, host: str, port: int, api_username: str, api_password: str, timeout: float = 5):
+        """
+        Initializes the Palworld object with the given parameters.
+
+        :param host: The host of the server.
+        :param port: The port of the server.
+        :param api_username: The API username.
+        :param api_password: The API password.
+        :param timeout: The timeout for the server connection.
+        """
+
+        super().__init__(host, port, timeout)
+
+        if api_username is None:
+            raise ValueError("api_username must not be None")
+        if api_password is None:
+            raise ValueError("api_password must not be None")
+
+        self.api_url = f"http://{self._host}:{self._port}/v1/api"
+        self.api_username = api_username
+        self.api_password = api_password
+
+    async def api_request(self,url):
+        """
+        Asynchronously retrieves data from the game server through the REST API.
+        """
+        auth = aiohttp.BasicAuth(self.api_username,self.api_password)
+        async with aiohttp.ClientSession(auth=auth) as session:
+            async with session.get(url) as response:
+                data = await response.json()
+        return data
+
+    async def get_status(self) -> Status:
+        """
+        Retrieves the status of the game server. The status includes the server state, name, player count and max player count.
+        """
+        info_data = await self.api_request(f"{self.api_url}/info")
+        metrics_data = await self.api_request(f"{self.api_url}/metrics")
+
+        server_name = info_data["servername"]
+        server_cur_players = metrics_data["currentplayernum"]
+        server_max_players = metrics_data["maxplayernum"]
+
+        return Status(
+            server_name=server_name,
+            num_players=server_cur_players,
+            max_players=server_max_players,
+        )
+
+
+if __name__ == "__main__":
+    import asyncio
+
+    async def main_async():
+        palworld = Palworld(
+            host="79.136.0.124",
+            port=8212,
+            timeout=5.0,
+            api_username="admin",
+            api_password="",
+        )
+        status = await palworld.get_status()
+        print(status)
+
+    asyncio.run(main_async())

opengsq/responses/palworld/__init__.py

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

opengsq/responses/palworld/status.py

@@ -0,0 +1,17 @@
+from dataclasses import dataclass
+
+
+@dataclass
+class Status:
+    """
+    Represents the status response from a server.
+    """
+
+    num_players: int
+    """The number of players currently connected to the server."""
+
+    max_players: int
+    """The maximum number of players that can connect to the server."""
+
+    server_name: str
+    """The name of the server."""

tests/protocols/test_palworld.py

@@ -0,0 +1,21 @@
+import pytest
+from opengsq.protocols.palworld import Palworld
+
+from ..result_handler import ResultHandler
+
+handler = ResultHandler(__file__)
+# handler.enable_save = True
+
+# Palworld
+test = Palworld(
+    host="72.65.106.166",
+    port=8212,
+    api_username="admin",
+    api_password="admin",
+)
+
+
+@pytest.mark.asyncio
+async def test_get_status():
+    result = await test.get_status()
+    await handler.save_result("test_get_status", result)