back-python/src/controller/client.py

"""Handles communication between one WebSocket tunnel and the game"""

from typing import Callable, Any, TYPE_CHECKING
from pydantic import ValidationError

from websockets.server import WebSocketServerProtocol

from netcode.models import Message, LoginMessage

from model import Player

if TYPE_CHECKING:
    from controller import GameController


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

    def __init__(
        self,
        game_controller: "GameController",
        websocket: WebSocketServerProtocol,
    ):
        self.websocket = websocket
        self.on_message = self.login_context

        self.game_controller = game_controller

    async def listen(self) -> None:
        """Hooks the client to the correct listen callbacks"""
        await self.send(
            Message(
                type="context",
                data_type="game_context_id",
                data="login",
            )
        )

        await self.game_controller.announce_game(self.websocket)

        async for message in self.websocket:
            await self.on_message(message)

    async def send(self, message: Message) -> None:
        """Send a Message object to the connected client"""
        await self.websocket.send(message.model_dump_json())

    async def send_error(self, error: str) -> None:
        """Send an error string to the connected client"""
        await self.send(Message(type="error", data_type="error", data=error))

    async def send_announce_dict(
        self,
        notification_type: str,
        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)
            )

    @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.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":
                    return await self.player.play_card(message.data)

                case "discard":
                    return await self.player.turn.discard_card(message.data)

                case "lock_card":
                    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":
                    return await self.player.turn.buy_card(message.data)

                case "heal":
                    return await self.player.turn.heal(message.data)

                case "damage":
                    return await self.player.turn.damage(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
            )

    @vibe_check(LoginMessage)
    async def login_context(self, message: LoginMessage) -> None:
        """Handles all messages related to when the current client isn't yet logged in"""
        username = message.data
        if self.game_controller.game.started:
            return await self.send_error(
                "you cannot interact with a game that has already started"
            )
        if message.type == "register":
            if len(username) < 3:
                return await self.send_error(
                    "username must be longer than 3 characters"
                )

            if self.game_controller.game.get_player(username) is not None:
                return await self.send_error("username already exists")

            player = await self.game_controller.game.create_player(username)
        elif message.type == "login":
            player2 = self.game_controller.game.get_player(username)

            if player2 is None:
                return await self.send_error("player doesn't exists")

            player = player2

        else:
            return await self.send_error("invalid message type for context")

        self.player = player
        self.player.team.listeners.append(self.send_announce_dict)
        self.on_message = self.pregame_context
        print("user logged in : " + username)

        await self.send(
            Message(
                type="set",
                data_type="self",
                data=self.player.uuid,
            )
        )
        await self.send(
            Message(
                type="context",
                data_type="game_context_id",
                data="pregame",
            )
        )

    @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)
            if all(p.ready for p in self.game_controller.game.players):
                await self.game_controller.start_game()

        elif message.type == "temp_action":
            if message.data_type == "set_health":
                self.player.team.health = message.data
                await self.game_controller.game.notify_controller(
                    "update", self.player.team.serialize_guest()
                )