Update Battlefield Protocol

opengsq/protocols/battlefield.py

@@ -1,64 +1,87 @@
+from __future__ import annotations
+
+from opengsq.responses.battlefield import Info, VersionInfo
 from opengsq.binary_reader import BinaryReader
 from opengsq.protocol_base import ProtocolBase
 from opengsq.protocol_socket import TcpClient
 
 
 class Battlefield(ProtocolBase):
-    """Battlefield Protocol"""
-    full_name = 'Battlefield Protocol'
+    """
+    This class represents the Battlefield Protocol. It provides methods to interact with the Battlefield API.
+    """
+
+    full_name = "Battlefield Protocol"
 
-    _info = b'\x00\x00\x00\x21\x1b\x00\x00\x00\x01\x00\x00\x00\x0a\x00\x00\x00serverInfo\x00'
-    _version = b'\x00\x00\x00\x22\x18\x00\x00\x00\x01\x00\x00\x00\x07\x00\x00\x00version\x00'
-    _players = b'\x00\x00\x00\x23\x24\x00\x00\x00\x02\x00\x00\x00\x0b\x00\x00\x00listPlayers\x00\x03\x00\x00\x00all\x00'
+    _info = b"\x00\x00\x00\x21\x1b\x00\x00\x00\x01\x00\x00\x00\x0a\x00\x00\x00serverInfo\x00"
+    _version = (
+        b"\x00\x00\x00\x22\x18\x00\x00\x00\x01\x00\x00\x00\x07\x00\x00\x00version\x00"
+    )
+    _players = b"\x00\x00\x00\x23\x24\x00\x00\x00\x02\x00\x00\x00\x0b\x00\x00\x00listPlayers\x00\x03\x00\x00\x00all\x00"
 
-    async def get_info(self) -> dict:
+    async def get_info(self) -> Info:
+        """
+        Asynchronously retrieves the information of the game server.
+
+        :return: An Info object containing the information of the game server.
+        """
         data = await self.__get_data(self._info)
 
         info = {}
-        info['hostname'] = str(data.pop(0)).strip()
-        info['numplayers'] = int(data.pop(0))
-        info['maxplayers'] = int(data.pop(0))
-        info['gametype'] = data.pop(0)
-        info['map'] = data.pop(0)
-        info['roundsplayed'] = int(data.pop(0))
-        info['roundstotal'] = int(data.pop(0))
+        info["hostname"] = str(data.pop(0)).strip()
+        info["num_players"] = int(data.pop(0))
+        info["max_players"] = int(data.pop(0))
+        info["game_type"] = data.pop(0)
+        info["map"] = data.pop(0)
+        info["rounds_played"] = int(data.pop(0))
+        info["rounds_total"] = int(data.pop(0))
         num_teams = int(data.pop(0))
-        info['teams'] = [float(data.pop(0)) for _ in range(num_teams)]
-        info['targetscore'] = int(data.pop(0))
-        info['status'] = data.pop(0)
-        info['ranked'] = data.pop(0) == 'true'
-        info['punkbuster'] = data.pop(0) == 'true'
-        info['password'] = data.pop(0) == 'true'
-        info['uptime'] = int(data.pop(0))
-        info['roundtime'] = int(data.pop(0))
+        info["teams"] = [float(data.pop(0)) for _ in range(num_teams)]
+        info["target_score"] = int(data.pop(0))
+        info["status"] = data.pop(0)
+        info["ranked"] = data.pop(0) == "true"
+        info["punk_buster"] = data.pop(0) == "true"
+        info["password"] = data.pop(0) == "true"
+        info["uptime"] = int(data.pop(0))
+        info["round_time"] = int(data.pop(0))
 
         try:
-            if data[0] == 'BC2':
-                info['mod'] = data.pop(0)
+            if data[0] == "BC2":
+                info["mod"] = data.pop(0)
                 data.pop(0)
 
-            info['ip_port'] = data.pop(0)
-            info['punkbuster_version'] = data.pop(0)
-            info['join_queue'] = data.pop(0) == 'true'
-            info['region'] = data.pop(0)
-            info['pingsite'] = data.pop(0)
-            info['country'] = data.pop(0)
+            info["ip_port"] = data.pop(0)
+            info["punk_buster_version"] = data.pop(0)
+            info["join_queue"] = data.pop(0) == "true"
+            info["region"] = data.pop(0)
+            info["ping_site"] = data.pop(0)
+            info["country"] = data.pop(0)
 
             try:
-                info['blaze_player_count'] = int(data[0])
-                info['blaze_game_state'] = data[1]
+                info["blaze_player_count"] = int(data[0])
+                info["blaze_game_state"] = data[1]
             except Exception:
-                info['quickmatch'] = data.pop(0) == 'true'
+                info["quick_match"] = data.pop(0) == "true"
         except Exception:
             pass
 
-        return info
+        return Info(**info)
+
+    async def get_version(self) -> VersionInfo:
+        """
+        Asynchronously retrieves the version information of the game server.
 
-    async def get_version(self) -> dict:
+        :return: A VersionInfo object containing the version information of the game server.
+        """
         data = await self.__get_data(self._version)
-        return {'mod': data[0], 'version': data[1]}
+        return VersionInfo(data[0], data[1])
 
     async def get_players(self) -> list:
+        """
+        Asynchronously retrieves the list of players on the game server.
+
+        :return: A list of dictionaries containing the player information.
+        """
         data = await self.__get_data(self._players)
         count = int(data.pop(0))  # field count
         fields, data = data[:count], data[count:]
@@ -72,10 +95,22 @@ async def get_players(self) -> list:
         return players
 
     async def __get_data(self, request: bytes):
+        """
+        Asynchronously sends the given request to the game server and receives the response.
+
+        :param request: The request to send to the game server.
+        :return: A list containing the response data from the game server.
+        """
         response = await TcpClient.communicate(self, request)
         return self.__decode(response)
 
     def __decode(self, response: bytes):
+        """
+        Decodes the given response from the game server.
+
+        :param response: The response to decode.
+        :return: A list containing the decoded response data.
+        """
         br = BinaryReader(response)
         br.read_long()  # header
         br.read_long()  # packet length
@@ -89,25 +124,26 @@ def __decode(self, response: bytes):
         return data[1:]
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     import asyncio
     import json
+    from dataclasses import asdict
 
     async def main_async():
         entries = [
-            ('91.206.15.69', 48888),  # bfbc2
-            ('94.250.199.214', 47200),  # bf3
-            ('74.91.124.140', 47200),  # bf4
-            ('185.189.255.240', 47600),  # bfh
+            ("91.206.15.69", 48888),  # bfbc2
+            ("94.250.199.214", 47200),  # bf3
+            ("74.91.124.140", 47200),  # bf4
+            ("185.189.255.240", 47600),  # bfh
         ]
 
         for address, query_port in entries:
             battlefield = Battlefield(address, query_port, timeout=10.0)
             info = await battlefield.get_info()
-            print(json.dumps(info, indent=None) + '\n')
+            print(json.dumps(asdict(info), indent=None) + "\n")
             version = await battlefield.get_version()
-            print(json.dumps(version, indent=None) + '\n')
+            print(json.dumps(asdict(version), indent=None) + "\n")
             players = await battlefield.get_players()
-            print(json.dumps(players, indent=None) + '\n')
+            print(json.dumps(players, indent=None) + "\n")
 
     asyncio.run(main_async())

opengsq/responses/battlefield/__init__.py

@@ -0,0 +1,2 @@
+from .info import Info
+from .version_info import VersionInfo

opengsq/responses/battlefield/info.py

@@ -0,0 +1,85 @@
+from __future__ import annotations
+from typing import Optional
+from dataclasses import dataclass
+
+
+@dataclass
+class Info:
+    """
+    Represents the info of a game.
+    """
+
+    hostname: str
+    """The hostname of the game server."""
+
+    num_players: int
+    """The number of players in the game."""
+
+    max_players: int
+    """The maximum number of players allowed in the game."""
+
+    game_type: str
+    """The type of the game."""
+
+    map: str
+    """The current map of the game."""
+
+    rounds_played: int
+    """The number of rounds played."""
+
+    rounds_total: int
+    """The total number of rounds."""
+
+    teams: list[float]
+    """The list of teams."""
+
+    target_score: int
+    """The target score."""
+
+    status: str
+    """The status of the game."""
+
+    ranked: bool
+    """Whether the game is ranked."""
+
+    punk_buster: bool
+    """Whether PunkBuster is enabled."""
+
+    password: bool
+    """Whether a password is required."""
+
+    uptime: int
+    """The uptime of the game server."""
+
+    round_time: int
+    """The round time."""
+
+    mod: Optional[str] = None
+    """The game mod. This property is optional."""
+
+    ip_port: Optional[str] = None
+    """The IP port of the game server. This property is optional."""
+
+    punk_buster_version: Optional[str] = None
+    """The version of PunkBuster. This property is optional."""
+
+    join_queue: Optional[bool] = None
+    """Whether the join queue is enabled. This property is optional."""
+
+    region: Optional[str] = None
+    """The region of the game server. This property is optional."""
+
+    ping_site: Optional[str] = None
+    """The ping site of the game server. This property is optional."""
+
+    country: Optional[str] = None
+    """The country of the game server. This property is optional."""
+
+    blaze_player_count: Optional[int] = None
+    """The number of players in the Blaze game state. This property is optional."""
+
+    blaze_game_state: Optional[str] = None
+    """The Blaze game state. This property is optional."""
+
+    quick_match: Optional[bool] = None
+    """Whether quick match is enabled. This property is optional."""

opengsq/responses/battlefield/version_info.py

@@ -0,0 +1,16 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass
+class VersionInfo:
+    """
+    Represents the version of a game mod.
+    """
+
+    mod: str
+    """The mod of the game."""
+
+    version: str
+    """The version of the mod."""