Archived
153 lines
4.6 KiB
Python
153 lines
4.6 KiB
Python
from typing import Callable, Any
|
|
from pydantic import ValidationError
|
|
|
|
from websockets.server import WebSocketServerProtocol
|
|
from controller import GameControllerInterface, ClientControllerInterface
|
|
|
|
from netcode.models import Message, LoginMessage
|
|
|
|
from model import Player, Team
|
|
|
|
|
|
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(ClientControllerInterface):
|
|
websocket: WebSocketServerProtocol
|
|
|
|
game_controller: GameControllerInterface
|
|
|
|
player: Player
|
|
|
|
track_connection: Callable
|
|
|
|
def __init__(
|
|
self,
|
|
game_controller: GameControllerInterface,
|
|
websocket: WebSocketServerProtocol,
|
|
):
|
|
self.websocket = websocket
|
|
self.on_message = self.login_context
|
|
|
|
self.game_controller = game_controller
|
|
|
|
async def listen(self):
|
|
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):
|
|
await self.websocket.send(message.model_dump_json())
|
|
|
|
async def send_error(self, error: str):
|
|
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,
|
|
):
|
|
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):
|
|
if self.player.team != self.game_controller.game.turn:
|
|
return await self.send_error("it is not your turn to play!")
|
|
if message.type == "action":
|
|
if message.data_type == "buy":
|
|
for c in message.data:
|
|
await self.player.buy_card(c)
|
|
elif message.data_type == "play_cards":
|
|
await self.player.play_cards(message.data)
|
|
|
|
@vibe_check(LoginMessage)
|
|
async def login_context(self, message: LoginMessage):
|
|
|
|
username = message.data
|
|
if message.type == "register":
|
|
|
|
if len(username) < 3:
|
|
await self.send_error("username must be longer than 3 characters")
|
|
return
|
|
|
|
if self.game_controller.game.get_player(username) is not None:
|
|
await self.send_error("username already exists")
|
|
return
|
|
|
|
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:
|
|
await self.send_error("player doesn't exists")
|
|
return
|
|
player = player2
|
|
|
|
else:
|
|
await self.send_error("invalid message type for context")
|
|
return
|
|
|
|
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",
|
|
)
|
|
)
|
|
|
|
# for player in self.game_controller.game.players:
|
|
# if player.team != self.player.team:
|
|
# continue
|
|
|
|
# await self.send(
|
|
# Message(type="update", data_type="object", data=player.serialize_team())
|
|
# )
|
|
|
|
@vibe_check(Message)
|
|
async def pregame_context(self, message: Message):
|
|
|
|
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()
|
|
self.on_message = self.game_context
|