"""Handles communication between one WebSocket tunnel and the game""" import os from typing import Callable, Any, TYPE_CHECKING, Awaitable from pydantic import ValidationError from websockets.server import WebSocketServerProtocol from websockets.exceptions import ConnectionClosedOK from netcode.models import Message, LoginMessage, MessageType from model import Player from .auth import discord_process_code if TYPE_CHECKING: from controller import GameController from model import Game def vibe_check(t): """Enforces the selected type for the message""" def decorator(func): async def new_f(self, message): try: message = t.model_validate_json(message) return await func(self, message) except ValidationError as e: print(e) await self.send_error(e.errors()) return new_f return decorator class ClientController: """Handles communication between one WebSocket tunnel and the game""" websocket: WebSocketServerProtocol game_controller: "GameController" player: Player track_connection: Callable on_message = Callable[[Message], Awaitable[None]] def __init__( self, game_controller: "GameController", websocket: WebSocketServerProtocol, ): self.websocket = websocket self.game_controller = game_controller self.game_controller.clients.append(self) self.on_message = self.on_message_stub async def on_message_stub(self, message: Message): return async def close(self): """Remove self from Game broadcasting list""" self.game_controller.clients.remove(self) async def send(self, message: Message) -> None: """Send a Message object to the connected client""" try: await self.websocket.send(message.model_dump_json()) except ConnectionClosedOK: print("Connection Closed") async def send_error(self, error: str) -> None: """Send an error string to the connected client""" await self.send(Message(type=MessageType.ERROR, data_type="error", data=error)) async def send_announce_dict( self, notification_type: MessageType, data: dict, player_filter: Callable[[Player], Any] = lambda x: True, ): """Send a, Object dict to the connected client player_filter is a filter which is called using the connected player """ if hasattr(self, "player") is False or player_filter(self.player): await self.send( Message(type=notification_type, data_type="object", data=data) ) async def join_game( self, game: "Game", username: str, discord_id: str, discord_avatar: str ) -> None: player = next( (p for p in game.players if p.discord_id == discord_id), None, ) if player is None: if self.game_controller.game.started: await self.send( Message( type=MessageType.SET, data_type="current_game", data=game.uuid, ) ) await self.send( Message( type=MessageType.CONTEXT, data_type="game_context_id", data="spectator", ) ) return await self.send_error( "you cannot interact with a game that has already started" ) player = await self.game_controller.game.create_player(username) player.discord_id = discord_id player.image = ( "https://cdn.discordapp.com/avatars/" + discord_id + "/" + discord_avatar + ".webp" ) self.player = player self.player.team.listeners.append(self.send_announce_dict) if self.game_controller.game.started: self.on_message = self.game_context await self.send( Message( type=MessageType.CONTEXT, data_type="game_context_id", data="game" ) ) for p in self.game_controller.game.players: if p.team == player.team: await self.send( Message( type=MessageType.UPDATE, data_type="object", data=player.serialize_team(), ) ) else: self.on_message = self.pregame_context await self.send( Message( type=MessageType.CONTEXT, data_type="game_context_id", data="pregame", ) ) print("user logged in : " + username) await self.send( Message( type=MessageType.SET, data_type="self", data=self.player.uuid, ) ) await self.send( Message( type=MessageType.SET, data_type="current_game", data=self.player.game.uuid, ) ) @vibe_check(Message) async def pregame_context(self, message: Message) -> None: """Handles all messages related to when the current client is logged in but the game isn't yet started""" if message.type == "ready": self.player.ready = bool(message.data) await self.player.notify_update() if all(p.ready for p in self.game_controller.game.players): if ("DEV_MODE" in os.environ) or len( self.game_controller.game.players ) > 1: await self.game_controller.start_game() @vibe_check(Message) async def game_context(self, message: Message) -> None: """Handles all messages related to when the current client is logged in and connected to a running game""" # if message.type == "temp_action": # if message.data_type == "set_health": # self.player.team.health = message.data # return await self.game_controller.game.notify_controller( # "update", self.player.team.serialize_guest() # ) if self.game_controller.game.ended is True: return await self.send_error("game is already finished!") if self.player.team != self.game_controller.game.current_turn: return await self.send_error("it is not your turn to play!") if message.type == "action": # TODO : pydantic type check for message.data match message.data_type: case "play_card": if not isinstance(message.data, str): return return await self.player.play_card(message.data) case "discard": if not isinstance(message.data, str): return return await self.player.turn.discard_card(message.data) case "lock_card": if not isinstance(message.data, str): return return await self.player.turn.lock_card(message.data) case "use_effect": return await self.player.turn.use_effect_with_check( message.data.get("effect_id"), message.data.get("target") ) case "buy": if not isinstance(message.data, str): return return await self.player.turn.buy_card(message.data) case "heal": if not isinstance(message.data, int): return return await self.player.turn.heal(message.data) case "damage_team": if not isinstance(message.data, str): return return await self.player.turn.damage_team(message.data) case "damage_champion": if not isinstance(message.data, str): return return await self.player.turn.damage_champion(message.data) case "end_turn": return await self.player.turn.end_turn() return await self.send_error( "Couldn't process action with data_type : " + message.data_type )