diff --git a/bot/__main__.py b/bot/__main__.py index 6f33094..5c1ae22 100644 --- a/bot/__main__.py +++ b/bot/__main__.py @@ -2,6 +2,7 @@ from importlib import import_module from os import listdir, path +from pathlib import Path from time import time from apscheduler.schedulers.asyncio import AsyncIOScheduler @@ -45,6 +46,6 @@ async def on_disconnect(self: "Bot") -> None: server_manager = ServerManager(bot, servers) -bot.load_extensions("bot\\exts", manager=server_manager, scheduler=scheduler) +bot.load_extensions(str(Path("bot", "exts")), manager=server_manager, scheduler=scheduler) bot.start(CONFIG.SECRET) diff --git a/bot/exts/card.py b/bot/exts/card.py index 912dd80..01173a6 100644 --- a/bot/exts/card.py +++ b/bot/exts/card.py @@ -33,8 +33,17 @@ from matplotlib import pyplot as plt from PIL import Image -from bot.util import TYPE_COLORS, Card, EffectCard, HermitCard, Server, ServerManager, probability -from bot.util.datagen import ItemCard +from bot.util import ( + TYPE_COLORS, + Card, + EffectCard, + HermitCard, + ItemCard, + Server, + ServerManager, + probability, + rgb_to_int, +) def take(items: int, iterable: Iterable) -> list: @@ -45,16 +54,6 @@ def take(items: int, iterable: Iterable) -> list: beige = (226, 202, 139) -def rgb_to_int(rgb: tuple[int, int, int]) -> int: - """Convert an rgb tuple to an integer. - - Args: - ---- - rgb (tuple): RGB color to convert - """ - return (rgb[0] << 16) + (rgb[1] << 8) + rgb[2] - - def count(s: str) -> str: """Count the number of items required.""" final = [] diff --git a/bot/exts/stats.py b/bot/exts/stats.py new file mode 100644 index 0000000..e089450 --- /dev/null +++ b/bot/exts/stats.py @@ -0,0 +1,318 @@ +"""Get information about cards and decks.""" + +from __future__ import annotations + +from datetime import datetime, timezone +from io import BytesIO +from math import floor + +from apscheduler.schedulers.asyncio import AsyncIOScheduler +from interactions import ( + Client, + Embed, + Extension, + File, + OptionType, + SlashContext, + slash_command, + slash_option, +) +from matplotlib import pyplot as plt +from matplotlib.offsetbox import AnnotationBbox, OffsetImage +from numpy import ndarray +from PIL import Image, ImageDraw + +from bot.util import TYPE_COLORS, Server, ServerManager, rgb_to_int + +LOSS = (198, 43, 43) +WIN = (126, 196, 96) +TIE = (255, 234, 132) + + +def get_type_color(types: list[str]) -> tuple[float, float, float]: + """Mix several type colors from a list together.""" + r = 0 + g = 0 + b = 0 + for hermit_type in types: + try: + color = TYPE_COLORS[hermit_type] + except KeyError: + color = (0, 0, 0) # Black if type not found + r += color[0] + g += color[1] + b += color[2] + + return (round(r / len(types)), round(g / len(types)), round(b / len(types))) + +def reduce_rgb(color: tuple[int, int, int]) -> tuple[float, float, float]: + """Convert a 0->255 rgb tuple into a 0->1 rgb tuple.""" + r = color[0] / 255 + g = color[1] / 255 + b = color[2] / 255 + return (r, g, b) + + +class StatsFailureError(Exception): + """Failure in generating stats.""" + + +class StatsExt(Extension): + """Get game and player stats.""" + + def __init__( + self: StatsExt, + client: Client, + manager: ServerManager, + _scheduler: AsyncIOScheduler, + ) -> None: + """Get game and player stats. + + Args: + ---- + client (Client): The discord bot client + manager (ServerManager): The server connection manager + _scheduler (AsyncIOScheduler): Event scheduler + """ + self.client: Client = client + self.manager: ServerManager = manager + + self.icons: dict[str, ndarray] | None = None + self.small_icons: dict[str, ndarray] | None = None + + @slash_command() + async def stats(self: StatsExt, _ctx: SlashContext) -> None: + """Get game and player stats.""" + + @stats.subcommand() + @slash_option( + "uuid", + "Target players uuid", + OptionType.STRING, + required=True, + min_length=36, + max_length=36, + ) + @slash_option("forfeits", "Include forfeit stats", OptionType.BOOLEAN) + @slash_option("hide_uuid", "If the players's uuid should be hidden", OptionType.BOOLEAN) + async def player( + self: StatsExt, + ctx: SlashContext, + uuid: str, + *, + forfeits: bool = True, + hide_uuid: bool = False, + ) -> None: + """Get a player's stats from their uuid.""" + server = self.manager.get_server(ctx.guild_id) + + stats = await server.get_player_stats(uuid) + if stats is None: + await ctx.send("Couldn't find a player with that uuid", ephemeral=True) + return + + if hide_uuid: + await ctx.send("This message handily obscures your uuid!", ephemeral=True) + + games = stats["gamesPlayed"] - ( + 0 if forfeits else stats["forfeitWins"] + stats["forfeitLosses"] + ) + wins = stats["wins"] + stats["forfeitWins"] if forfeits else 0 + losses = stats["losses"] + stats["forfeitLosses"] if forfeits else 0 + ties = stats["ties"] + + win_rate = wins / games + tie_rate = ties / games + loss_rate = losses / games + + color: tuple[int, int, int] + if win_rate > tie_rate and win_rate > loss_rate: + color = WIN + elif tie_rate > loss_rate: + color = TIE + else: + color = LOSS + + bar = Image.new("RGBA", (200, 30), LOSS) + drawer = ImageDraw.Draw(bar) + drawer.rectangle((0, 0, floor(bar.width * win_rate), bar.height), WIN) + if tie_rate != 0: + drawer.rectangle( + ( + floor(bar.width * win_rate), + 0, + floor(bar.width * (win_rate + tie_rate)), + bar.height, + ), + TIE, + ) + + embed = ( + Embed( + "Player stats", + f"{games} game{"" if games == 1 else "s"} played.", + rgb_to_int(color), + timestamp=datetime.now(tz=timezone.utc), + ) + .set_footer("Bot by Tyrannicodin") + .add_field("Win rate", f"{win_rate:.2%}") + .add_field("Loss rate", f"{loss_rate:.2%}") + .set_image("attachment://bar.png") + ) + with BytesIO() as im_binary: + bar.save(im_binary, "PNG") + im_binary.seek(0) + await ctx.send(embed=embed, file=File(im_binary, "bar.png")) + + @stats.group("type") + async def types(self: StatsExt, _ctx: SlashContext) -> None: + """Type stat commands.""" + + @stats.subcommand(group_name="type") + async def winrate(self: StatsExt, ctx: SlashContext) -> None: + """Get win rate by type stats.""" + server = self.manager.get_server(ctx.guild_id) + + try: + result = await self.generate_type_stat( + server, + "Win rate", + "winrate", + "the average win rate of all decks with at least 1 item card of that type.", + ) + except StatsFailureError as e: + await ctx.send(e.args[0], ephemeral=True) + return + file_bytes, image, embed = result + + await ctx.send(file=image, embed=embed) + file_bytes.close() + + @stats.subcommand(group_name="type") + async def usage(self: StatsExt, ctx: SlashContext) -> None: + """Get usage by type stats.""" + server = self.manager.get_server(ctx.guild_id) + + try: + result = await self.generate_type_stat( + server, + "Usage", + "frequency", + "the average win rate of all decks with at least 1 item card of that type.", + ) + except StatsFailureError as e: + await ctx.send(e.args[0], ephemeral=True) + return + file_bytes, image, embed = result + + await ctx.send(file=image, embed=embed) + file_bytes.close() + + async def generate_type_stat( + self: StatsExt, server: Server, name: str, key: str, description: str + ) -> tuple[BytesIO, File, Embed]: + """Generate a bar chart of either win rate or usage by type.""" + stats: list[dict] = (await server.get_type_distribution_stats())["types"] + if self.icons is None or self.small_icons is None: + self.icons = {} + self.small_icons = {} + icons = await server.get_type_icons() + if not icons: + err = "Couldn't find type images" + raise StatsFailureError(err) + for hermit_type, pil_icon in icons.items(): + with BytesIO() as image_bytes: + pil_icon.resize((25, 25), Image.Resampling.BILINEAR).save(image_bytes, "png") + image_bytes.seek(0) + img = plt.imread(image_bytes) + self.icons[hermit_type] = img + with BytesIO() as image_bytes: + pil_icon.resize((12, 12), Image.Resampling.BILINEAR).save(image_bytes, "png") + image_bytes.seek(0) + img = plt.imread(image_bytes) + self.small_icons[hermit_type] = img + + if stats is None or self.icons is None or self.small_icons is None: + err = "Couldn't find stats or type images" + raise StatsFailureError(err) + + stats.sort(key=lambda stat: stat[key], reverse=True) + plt.figure() + xs = list(range(len(stats))) + ys = [float(stat[key] * 100) for stat in stats] + colors = [reduce_rgb(get_type_color(stat["type"])) for stat in stats] + + plt.bar(xs, ys, color=colors) + + gc = plt.gca() + + for i, types in enumerate(stat["type"] for stat in stats): + y_offset = 0 + for hermit_type in types: + ab = AnnotationBbox( + OffsetImage(self.small_icons[hermit_type]), + (i, 0), + xybox=(0, -8 + y_offset), + frameon=False, + xycoords="data", + boxcoords="offset points", + pad=0, + ) + y_offset -= 15 + gc.add_artist(ab) + + if gc.axes is not None: + gc.axes.get_xaxis().set_ticks([]) + gc.set_ylabel(f"{name} (%)") + plt.grid(visible=True, axis="y") + + embed = ( + Embed( + f"{name} by type", + f"{name} is {description}", + rgb_to_int(get_type_color(stats[0]["type"])), + timestamp=datetime.now(tz=timezone.utc), + ) + .set_footer("Bot by Tyrannicodin") + .set_image("attachment://graph.png") + ) + + figure_bytes = BytesIO() + plt.savefig(figure_bytes, format="PNG") + plt.close() + figure_bytes.seek(0) + return figure_bytes, File(figure_bytes, "graph.png"), embed + + @stats.subcommand() + async def games(self: StatsExt, ctx: SlashContext) -> None: + """Get game count and average game length.""" + server = self.manager.get_server(ctx.guild_id) + + game_stats = await server.get_game_stats() + if game_stats is None: + await ctx.send("Couldn't find game statistics", ephemeral=True) + return + count, length = game_stats + singular = count == 1 + embed = ( + Embed("Game history") + .add_field(f"Game{"" if singular else "s"} played", str(count)) + .add_field("Average length", length) + ) + await ctx.send(embed=embed) + + +def setup( + client: Client, + manager: ServerManager, + scheduler: AsyncIOScheduler, +) -> Extension: + """Create the extension. + + Args: + ---- + client (Client): The discord bot client + manager (ServerManager): The server connection manager + scheduler (AsyncIOScheduler): Event scheduler + """ + return StatsExt(client, manager, scheduler) diff --git a/bot/util/datagen.py b/bot/util/datagen.py index fb07329..410158f 100644 --- a/bot/util/datagen.py +++ b/bot/util/datagen.py @@ -32,6 +32,16 @@ } +def rgb_to_int(rgb: tuple[int, int, int]) -> int: + """Convert an rgb tuple to an integer. + + Args: + ---- + rgb (tuple): RGB color to convert + """ + return (rgb[0] << 16) + (rgb[1] << 8) + rgb[2] + + class Card: """Basic image generator for a card.""" diff --git a/bot/util/server.py b/bot/util/server.py index fed189b..ff260f2 100644 --- a/bot/util/server.py +++ b/bot/util/server.py @@ -1,272 +1,360 @@ -"""Handles interactions and linking discord and hc-tcg servers.""" - -from __future__ import annotations - -from datetime import datetime as dt -from datetime import timezone -from json import JSONDecodeError, loads -from time import time -from typing import Any - -from aiohttp import ClientSession -from interactions import Client, Embed, Member, Snowflake - -from bot.util.datagen import DataGenerator - - -class GamePlayer: - """A representation of a player in a game.""" - - def __init__(self: GamePlayer, data: dict[str, Any]) -> None: - """Represent a player in a game. - - Args: - ---- - data (dict): Player information - """ - self.id: str = data["playerId"] - self.name: str = data["censoredPlayerName"] - self.minecraft_name: str = data["minecraftName"] - self.lives: int = data["lives"] - - self.deck: list[str] = data["deck"] - - -class Game: - """Store data about a game.""" - - def __init__(self: Game, data: dict[str, Any]) -> None: - """Store data about a game. - - Args: - ---- - data (dict): The game data dict - """ - self.players: list[GamePlayer] = [GamePlayer(player) for player in data["players"]] - self.player_names = [player.name for player in self.players] - self.id = data["id"] - self.spectator_code: str | None = data["spectatorCode"] - self.created: dt = dt.fromtimestamp(data["createdTime"] / 1000, tz=timezone.utc) - self.spectators = data["viewers"] - len(self.players) - - print(data["state"]) - - -class QueueGame: - """Information about a private queued game.""" - - def __init__(self: QueueGame, data: dict[str, Any]) -> None: - """Information about a private queued game. - - Args: - ---- - data (dict): The game data dict - """ - self.joinCode: str = data["gameCode"] - self.spectatorCode: str = data["spectatorCode"] - self.secret: str = data["apiSecret"] - self.timeout: str = data["timeOutAt"] / 1000 - - def create_embed(self: QueueGame, *, spectators: bool = False) -> Embed: - """Create an embed with information about the game.""" - e = ( - Embed( - "Game", - f"Expires ", - timestamp=dt.now(tz=timezone.utc), - ) - .add_field("Join code", self.joinCode, inline=True) - .set_footer("Bot by Tyrannicodin16") - ) - if spectators: - e.add_field("Spectate code", self.spectatorCode, inline=True) - return e - - -class Server: - """An interface between a discord and hc-tcg server.""" - - http_session: ClientSession - data_generator: DataGenerator - - def __init__( - self: Server, - server_id: str, - server_url: str, - guild_id: str, - admins: list[str] | None = None, - tracked_forums: dict[str, list[str]] | None = None, - ) -> None: - """Create a Server object. - - Args: - ---- - server_id (str): Unique name for the server - server_url (str): The url of the hc-tcg server - server_key (str): The api key to send to the server - guild_id (str): The id of the discord server - guild_key (str): The api key sent from the server - admins (list[str]): List of users and/or roles that can use privileged - features, if blank allows all users to use privileged features - tracked_forums (list[str]): Dictionary with channel ids and tags to ignore - update_channel (str): The channel to get server updates from - """ - if admins is None: - admins = [] - if tracked_forums is None: - tracked_forums = {} - - self.server_id: str = server_id - self.last_game_count: int = 0 - self.last_game_count_time: int = 0 - self.last_queue_length: int = 0 - self.last_queue_length_time: int = 0 - - self.server_url: str = server_url - self.guild_id: str = guild_id - self.admin_roles: list[str] = admins - self.tracked_forums: dict[str, list[str]] = tracked_forums - - def create_session(self: Server) -> None: - """Create http session and data generator.""" - self.http_session = ClientSession(self.server_url + "/api/") - self.data_generator = DataGenerator(self.http_session) - - def authorize_user(self: Server, member: Member) -> bool: - """Check if a user is allowed to use privileged commands.""" - if self.admin_roles is []: - return True - admin_user = str(member.id) in self.admin_roles - admin_role = any(str(role.id) in self.admin_roles for role in member.roles) - - return admin_user or admin_role - - async def get_deck(self: Server, code: str) -> dict | None: - """Get information about a deck from the server. - - Args: - ---- - code (str): The export code of the deck to retrieve - """ - try: - async with self.http_session.get(f"deck/{code}") as response: - result = loads((await response.content.read()).decode()) - if not response.ok: - return None - except (TimeoutError, JSONDecodeError): - return None - return result - - async def create_game(self: Server) -> QueueGame | None: - """Create a server game.""" - try: - async with self.http_session.get("games/create") as response: - data: dict[str, str | int] = loads((await response.content.read()).decode()) - if not response.ok: - return None - return QueueGame(data) - except ( - ConnectionError, - JSONDecodeError, - KeyError, - ): - return None - - async def cancel_game(self: Server, game: QueueGame) -> bool: - """Cancel a queued game.""" - try: - async with self.http_session.delete( - "games/cancel", json={"code": game.secret} - ) as response: - loads((await response.content.read()).decode()) - return response.status == 200 - except ( - ConnectionError, - JSONDecodeError, - KeyError, - ): - return False - - async def get_game_count(self: Server) -> int: - """Get the number of games.""" - try: - if self.last_game_count_time > time() - 60: - return self.last_game_count - - async with self.http_session.get("games/count") as response: - data: dict[str, int] = loads((await response.content.read()).decode()) - if not response.ok: - return 0 - self.last_game_count = data["games"] - self.last_game_count_time = round(time()) - return self.last_game_count - except ( - ConnectionError, - JSONDecodeError, - KeyError, - ): - return 0 - - async def get_queue_length(self: Server) -> int: - """Get the number of games.""" - try: - if self.last_queue_length_time > time() - 60: - return self.last_queue_length - - async with self.http_session.get("games/queue/length") as response: - data: dict[str, int] = loads((await response.content.read()).decode()) - if not response.ok: - return 0 - self.last_queue_length = data["queueLength"] - self.last_queue_length_time = round(time()) - return self.last_queue_length - except ( - ConnectionError, - JSONDecodeError, - KeyError, - ): - return 0 - - -class ServerManager: - """Manage multiple servers and their functionality.""" - - def __init__(self: ServerManager, client: Client, servers: list[Server]) -> None: - """Manage multiple servers and their functionality. - - Args: - ---- - client (Client): The bot client - servers (list[Server]): A list of server objects to manage - bot_server (Application): The web server hc-tcg servers send requests to - scheduler (AsyncIOScheduler): Sheduler for repeating tasks - universe (dict): Dictionary that converts card ids to Card objects - """ - self._discord_links = {server.guild_id: server for server in servers} - - self.client = client - self.servers = servers - - def get_server(self: ServerManager, guild_id: Snowflake | None) -> Server: - """Get a server by its discord guild id. - - Args: - ---- - guild_id (str): The guild id of the discord server - """ - return ( - self._discord_links[str(guild_id)] - if guild_id in self._discord_links.keys() - else self.servers[0] - ) - - async def close_all_sessions(self: ServerManager) -> None: - """Close all server ClientSessions.""" - for server in self.servers: - await server.http_session.close() - - async def reload_all_generators(self: ServerManager) -> None: - """Close all server DataGenerators.""" - for server in self.servers: - server.create_session() - await server.data_generator.reload_all() +"""Handles interactions and linking discord and hc-tcg servers.""" + +from __future__ import annotations + +from datetime import datetime as dt +from datetime import timezone +from json import JSONDecodeError, loads +from time import time +from typing import Any + +from aiohttp import ClientSession, ContentTypeError +from interactions import Client, Embed, Member, Snowflake +from PIL import Image + +from bot.util.datagen import DataGenerator + + +class GamePlayer: + """A representation of a player in a game.""" + + def __init__(self: GamePlayer, data: dict[str, Any]) -> None: + """Represent a player in a game. + + Args: + ---- + data (dict): Player information + """ + self.id: str = data["playerId"] + self.name: str = data["censoredPlayerName"] + self.minecraft_name: str = data["minecraftName"] + self.lives: int = data["lives"] + + self.deck: list[str] = data["deck"] + + +class Game: + """Store data about a game.""" + + def __init__(self: Game, data: dict[str, Any]) -> None: + """Store data about a game. + + Args: + ---- + data (dict): The game data dict + """ + self.players: list[GamePlayer] = [GamePlayer(player) for player in data["players"]] + self.player_names = [player.name for player in self.players] + self.id = data["id"] + self.spectator_code: str | None = data["spectatorCode"] + self.created: dt = dt.fromtimestamp(data["createdTime"] / 1000, tz=timezone.utc) + self.spectators = data["viewers"] - len(self.players) + + print(data["state"]) + + +class QueueGame: + """Information about a private queued game.""" + + def __init__(self: QueueGame, data: dict[str, Any]) -> None: + """Information about a private queued game. + + Args: + ---- + data (dict): The game data dict + """ + self.joinCode: str = data["gameCode"] + self.spectatorCode: str = data["spectatorCode"] + self.secret: str = data["apiSecret"] + self.timeout: str = data["timeOutAt"] / 1000 + + def create_embed(self: QueueGame, *, spectators: bool = False) -> Embed: + """Create an embed with information about the game.""" + e = ( + Embed( + "Game", + f"Expires ", + timestamp=dt.now(tz=timezone.utc), + ) + .add_field("Join code", self.joinCode, inline=True) + .set_footer("Bot by Tyrannicodin16") + ) + if spectators: + e.add_field("Spectate code", self.spectatorCode, inline=True) + return e + + +class Server: + """An interface between a discord and hc-tcg server.""" + + http_session: ClientSession + data_generator: DataGenerator + + def __init__( + self: Server, + server_id: str, + server_url: str, + guild_id: str, + admins: list[str] | None = None, + tracked_forums: dict[str, list[str]] | None = None, + ) -> None: + """Create a Server object. + + Args: + ---- + server_id (str): Unique name for the server + server_url (str): The url of the hc-tcg server + server_key (str): The api key to send to the server + guild_id (str): The id of the discord server + guild_key (str): The api key sent from the server + admins (list[str]): List of users and/or roles that can use privileged + features, if blank allows all users to use privileged features + tracked_forums (list[str]): Dictionary with channel ids and tags to ignore + update_channel (str): The channel to get server updates from + """ + if admins is None: + admins = [] + if tracked_forums is None: + tracked_forums = {} + + self.server_id: str = server_id + + self.last_game_count: int = 0 + self.last_game_count_time: int = 0 + self.last_queue_length: int = 0 + self.last_queue_length_time: int = 0 + + self.type_data: dict[str, Image.Image] | None = None + + self.server_url: str = server_url + self.guild_id: str = guild_id + self.admin_roles: list[str] = admins + self.tracked_forums: dict[str, list[str]] = tracked_forums + + def create_session(self: Server) -> None: + """Create http session and data generator.""" + self.http_session = ClientSession(self.server_url + "/api/") + self.data_generator = DataGenerator(self.http_session) + + def authorize_user(self: Server, member: Member) -> bool: + """Check if a user is allowed to use privileged commands.""" + if self.admin_roles is []: + return True + admin_user = str(member.id) in self.admin_roles + admin_role = any(str(role.id) in self.admin_roles for role in member.roles) + + return admin_user or admin_role + + async def get_deck(self: Server, code: str) -> dict | None: + """Get information about a deck from the server. + + Args: + ---- + code (str): The export code of the deck to retrieve + """ + try: + async with self.http_session.get(f"deck/{code}") as response: + result = loads((await response.content.read()).decode()) + if not response.ok: + return None + except (TimeoutError, JSONDecodeError): + return None + return result + + async def create_game(self: Server) -> QueueGame | None: + """Create a server game.""" + try: + async with self.http_session.get("games/create") as response: + data: dict[str, str | int] = loads((await response.content.read()).decode()) + if not response.ok: + return None + return QueueGame(data) + except ( + ConnectionError, + JSONDecodeError, + KeyError, + ): + return None + + async def cancel_game(self: Server, game: QueueGame) -> bool: + """Cancel a queued game.""" + try: + async with self.http_session.delete( + "games/cancel", json={"code": game.secret} + ) as response: + loads((await response.content.read()).decode()) + return response.status == 200 + except ( + ConnectionError, + JSONDecodeError, + KeyError, + ): + return False + + async def get_game_count(self: Server) -> int: + """Get the number of games.""" + try: + if self.last_game_count_time > time() - 60: + return self.last_game_count + + async with self.http_session.get("games/count") as response: + data: dict[str, int] = loads((await response.content.read()).decode()) + if not response.ok: + return 0 + self.last_game_count = data["games"] + self.last_game_count_time = round(time()) + return self.last_game_count + except ( + ConnectionError, + JSONDecodeError, + KeyError, + ): + return 0 + + async def get_queue_length(self: Server) -> int: + """Get the number of games.""" + try: + if self.last_queue_length_time > time() - 60: + return self.last_queue_length + + async with self.http_session.get("games/queue/length") as response: + data: dict[str, int] = loads((await response.content.read()).decode()) + if not response.ok: + return 0 + self.last_queue_length = data["queueLength"] + self.last_queue_length_time = round(time()) + return self.last_queue_length + except ( + ConnectionError, + JSONDecodeError, + KeyError, + ): + return 0 + + async def get_player_stats(self: Server, uuid: str) -> dict[str, int] | None: + """Get a player's win stats from the server. + + Args: + ---- + uuid (str): The player's uuid + """ + try: + async with self.http_session.get("stats", params={"uuid": uuid}) as response: + if not response.ok: + return None + return await response.json() + except ( + ConnectionError, + JSONDecodeError, + ContentTypeError, + KeyError, + ): + return None + + async def get_type_distribution_stats(self: Server) -> list[dict[str, float | str]] | None: + """Get a player's win stats from the server. + + Args: + ---- + uuid (str): The player's uuid + """ + try: + async with self.http_session.get("stats/type-distribution") as response: + if not response.ok: + return None + return await response.json() + except ( + ConnectionError, + JSONDecodeError, + ContentTypeError, + KeyError, + ): + return None + + async def get_type_icons(self: Server) -> dict[str, Image.Image] | None: + """Get a dictionary of type icons.""" + if self.type_data is not None: + return self.type_data + try: + async with self.http_session.get("types") as response: + if not response.ok: + return None + data: list[dict[str, str]] = await response.json() + except ( + ConnectionError, + JSONDecodeError, + ContentTypeError, + KeyError, + ): + return None + self.type_data = { + hermit_type["type"]: await self.data_generator.get_image(hermit_type["icon"]) + for hermit_type in data + } + return self.type_data + + async def get_game_stats(self: Server) -> tuple[int, str] | None: + """Get number of games and average length.""" + try: + async with self.http_session.get("stats/games") as response: + if not response.ok: + return None + data = await response.json() + return ( + data["allTimeGames"], + ( + str(data["gameLength"]["averageLength"]["minutes"]) + + f" minutes, {data["gameLength"]["averageLength"]["seconds"]} seconds" + ), + ) + except ( + ConnectionError, + JSONDecodeError, + ContentTypeError, + KeyError, + ): + return None + + +class ServerManager: + """Manage multiple servers and their functionality.""" + + def __init__(self: ServerManager, client: Client, servers: list[Server]) -> None: + """Manage multiple servers and their functionality. + + Args: + ---- + client (Client): The bot client + servers (list[Server]): A list of server objects to manage + bot_server (Application): The web server hc-tcg servers send requests to + scheduler (AsyncIOScheduler): Sheduler for repeating tasks + universe (dict): Dictionary that converts card ids to Card objects + """ + self._discord_links = {server.guild_id: server for server in servers} + + self.client = client + self.servers = servers + + def get_server(self: ServerManager, guild_id: Snowflake | None) -> Server: + """Get a server by its discord guild id. + + Args: + ---- + guild_id (str): The guild id of the discord server + """ + return ( + self._discord_links[str(guild_id)] + if guild_id in self._discord_links.keys() + else self.servers[0] + ) + + async def close_all_sessions(self: ServerManager) -> None: + """Close all server ClientSessions.""" + for server in self.servers: + await server.http_session.close() + + async def reload_all_generators(self: ServerManager) -> None: + """Close all server DataGenerators.""" + for server in self.servers: + server.create_session() + await server.data_generator.reload_all()