Update Vice City Multiplayer Protocol

opengsq/protocols/vcmp.py

@@ -1,4 +1,7 @@
+from __future__ import annotations
+
 import struct
+from opengsq.responses.vcmp import Player, Status
 
 from opengsq.binary_reader import BinaryReader
 from opengsq.exceptions import InvalidPacketException
@@ -7,71 +10,99 @@
 
 
 class Vcmp(ProtocolBase):
-    """Vice City Multiplayer Protocol"""
-    full_name = 'Vice City Multiplayer Protocol'
+    """
+    This class represents the Vice City Multiplayer Protocol. It provides methods to interact with the Vice City Multiplayer API.
+    """
+
+    full_name = "Vice City Multiplayer Protocol"
 
-    _request_header = b'VCMP'
-    _response_header = b'MP04'
+    _request_header = b"VCMP"
+    _response_header = b"MP04"
 
-    async def get_status(self):
-        response = await self.__send_and_receive(b'i')
+    async def get_status(self) -> Status:
+        """
+        Asynchronously retrieves the status of the game server.
+
+        :return: A Status object containing the status of the game server.
+        """
+        response = await self.__send_and_receive(b"i")
 
         br = BinaryReader(response)
-        result = {}
-        result['version'] = str(br.read_bytes(12).strip(b'\x00'), encoding='utf-8', errors='ignore')
-        result['password'] = br.read_byte()
-        result['numplayers'] = br.read_short()
-        result['maxplayers'] = br.read_short()
-        result['servername'] = self.__read_string(br, 4)
-        result['gametype'] = self.__read_string(br, 4)
-        result['language'] = self.__read_string(br, 4)
-
-        return result
-
-    async def get_players(self):
+
+        return Status(
+            version=str(
+                br.read_bytes(12).strip(b"\x00"), encoding="utf-8", errors="ignore"
+            ),
+            password=br.read_byte(),
+            num_players=br.read_short(),
+            max_players=br.read_short(),
+            server_name=self.__read_string(br, 4),
+            game_type=self.__read_string(br, 4),
+            language=self.__read_string(br, 4),
+        )
+
+    async def get_players(self) -> list[Player]:
+        """
+        Asynchronously retrieves the list of players on the game server.
+
+        :return: A list of Player objects representing the players on the game server.
+        """
         """Server may not response when numplayers > 100"""
-        response = await self.__send_and_receive(b'c')
+        response = await self.__send_and_receive(b"c")
 
         br = BinaryReader(response)
-        players = []
         numplayers = br.read_short()
-
-        for _ in range(numplayers):
-            player = {}
-            player['name'] = self.__read_string(br)
-            players.append(player)
+        players = [Player(self.__read_string(br)) for _ in range(numplayers)]
 
         return players
 
     async def __send_and_receive(self, data: bytes):
+        """
+        Asynchronously sends the given data to the game server and receives the response.
+
+        :param data: The data to send to the game server.
+        :return: The response from the game server.
+        """
         # Format the address
         host = await Socket.gethostbyname(self._host)
-        packet_header = struct.pack('BBBBH', *map(int, host.split('.') + [self._port])) + data
+        packet_header = (
+            struct.pack("BBBBH", *map(int, host.split(".") + [self._port])) + data
+        )
         request = self._request_header + packet_header
 
         # Validate the response
         response = await UdpClient.communicate(self, request)
-        header = response[:len(self._response_header)]
+        header = response[: len(self._response_header)]
 
         if header != self._response_header:
-            raise InvalidPacketException(f'Packet header mismatch. Received: {header}. Expected: {self._response_header}.')
+            raise InvalidPacketException(
+                f"Packet header mismatch. Received: {header}. Expected: {self._response_header}."
+            )
 
-        return response[len(self._response_header) + len(packet_header):]
+        return response[len(self._response_header) + len(packet_header) :]
 
     def __read_string(self, br: BinaryReader, read_offset=1):
+        """
+        Reads a string from the given BinaryReader object.
+
+        :param br: The BinaryReader object to read the string from.
+        :param read_offset: The offset to start reading from.
+        :return: The string read from the BinaryReader object.
+        """
         length = br.read_byte() if read_offset == 1 else br.read_long()
-        return str(br.read_bytes(length), encoding='utf-8', errors='ignore')
+        return str(br.read_bytes(length), encoding="utf-8", errors="ignore")
 
 
-if __name__ == '__main__':
+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)
+        vcmp = Vcmp(host="51.178.65.136", port=8114, timeout=5.0)
         status = await vcmp.get_status()
-        print(json.dumps(status, indent=None) + '\n')
+        print(json.dumps(asdict(status), indent=None) + "\n")
         players = await vcmp.get_players()
-        print(json.dumps(players, indent=None) + '\n')
+        print(json.dumps([asdict(player) for player in players], indent=None) + "\n")
 
     asyncio.run(main_async())

opengsq/responses/vcmp/__init__.py

@@ -0,0 +1,2 @@
+from .player import Player
+from .status import Status

opengsq/responses/vcmp/player.py

@@ -0,0 +1,11 @@
+from dataclasses import dataclass
+
+
+@dataclass
+class Player:
+    """
+    Represents a player in the game.
+    """
+
+    name: str
+    """The player's name."""

opengsq/responses/vcmp/status.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+from dataclasses import dataclass
+
+
+@dataclass
+class Status:
+    """
+    Represents the status response from a server.
+    """
+
+    version: str
+    """The version of the server."""
+
+    password: bool
+    """Indicates whether a password is required to connect to the 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."""
+
+    game_type: str
+    """The type of game being played on the server."""
+
+    language: str
+    """The language of the server."""

tests/protocols/result_handler.py

@@ -22,6 +22,9 @@ async def save_result(self, function_name, result, is_json=True):
             if is_json:
                 if is_dataclass(result):
                     result = asdict(result)
+                elif isinstance(result, list):
+                    # set asdict to all items
+                    result = [asdict(item) for item in result if is_dataclass(item)]
 
                 result = json.dumps(result, indent=4, ensure_ascii=False)
 

tests/protocols/test_vcmp.py

@@ -9,14 +9,16 @@
 # handler.enable_save = True
 
 # Vice City Multiplayer
-test = Vcmp(host='51.178.65.136', port=8114)
+test = Vcmp(host="51.178.65.136", port=8114)
+
 
 @pytest.mark.asyncio
 async def test_get_status():
     result = await test.get_status()
-    await handler.save_result('test_get_status', result)
+    await handler.save_result("test_get_status", result)
+
 
 @pytest.mark.asyncio
 async def test_get_players():
     result = await test.get_players()
-    await handler.save_result('test_get_players', result)
+    await handler.save_result("test_get_players", result)

tests/results/test_vcmp/test_get_status.json

@@ -1,9 +1,9 @@
 {
     "version": "04rel006",
     "password": 1,
-    "numplayers": 16,
-    "maxplayers": 30,
-    "servername": "[0.4] Vice Paradise - Roleplay - OneVice Hosting",
-    "gametype": "Roleplay EN/SPA",
+    "num_players": 16,
+    "max_players": 30,
+    "server_name": "[0.4] Vice Paradise - Roleplay - OneVice Hosting",
+    "game_type": "Roleplay EN/SPA",
     "language": "Vice City"
 }