Archived
264 lines
9.1 KiB
Python
264 lines
9.1 KiB
Python
"""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 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
|
|
|
|
|
|
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=MessageType.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"""
|
|
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)
|
|
)
|
|
|
|
@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 message.type == MessageType.REGISTER:
|
|
if self.game_controller.game.started:
|
|
return await self.send_error(
|
|
"you cannot interact with a game that has already started"
|
|
)
|
|
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 == MessageType.LOGIN:
|
|
player2 = self.game_controller.game.get_player(username)
|
|
|
|
if player2 is None:
|
|
return await self.send_error("player doesn't exists")
|
|
|
|
player = player2
|
|
elif message.type == MessageType.DISCORD_LOGIN:
|
|
user = discord_process_code(message.data)
|
|
if user is None:
|
|
await self.send_error("discord login not successful")
|
|
return
|
|
username = user["global_name"]
|
|
player = next(
|
|
(
|
|
p
|
|
for p in self.game_controller.game.players
|
|
if p.discord_id == user["id"]
|
|
),
|
|
None,
|
|
)
|
|
if player is None:
|
|
if self.game_controller.game.started:
|
|
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 = user["id"]
|
|
player.image = (
|
|
"https://cdn.discordapp.com/avatars/"
|
|
+ user["id"]
|
|
+ "/"
|
|
+ user["avatar"]
|
|
+ ".webp"
|
|
)
|
|
else:
|
|
return await self.send_error("invalid message type for context")
|
|
|
|
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,
|
|
)
|
|
)
|
|
|
|
@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()
|
|
|
|
@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":
|
|
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
|
|
)
|