diff --git a/opengsq/protocols/battlefield.py b/opengsq/protocols/battlefield.py index 4487a04..d94cfcc 100644 --- a/opengsq/protocols/battlefield.py +++ b/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()) diff --git a/opengsq/responses/battlefield/__init__.py b/opengsq/responses/battlefield/__init__.py new file mode 100644 index 0000000..2c5c03d --- /dev/null +++ b/opengsq/responses/battlefield/__init__.py @@ -0,0 +1,2 @@ +from .info import Info +from .version_info import VersionInfo diff --git a/opengsq/responses/battlefield/info.py b/opengsq/responses/battlefield/info.py new file mode 100644 index 0000000..d8d3d1b --- /dev/null +++ b/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.""" diff --git a/opengsq/responses/battlefield/version_info.py b/opengsq/responses/battlefield/version_info.py new file mode 100644 index 0000000..7e0e0c0 --- /dev/null +++ b/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."""