diff --git a/src/controller/__init__.py b/src/controller/__init__.py index 61f45ee..7c07eb6 100644 --- a/src/controller/__init__.py +++ b/src/controller/__init__.py @@ -1,8 +1,5 @@ """controller handles the translation between netcode messages and model functions""" -from controller.interfaces import ClientControllerInterface, GameControllerInterface -from controller.object_controller import ObjectController -from controller.card import Card from controller.client import ClientController from controller.game import GameController diff --git a/src/controller/card.py b/src/controller/card.py deleted file mode 100644 index 571c722..0000000 --- a/src/controller/card.py +++ /dev/null @@ -1,5 +0,0 @@ -from controller import ObjectController - - -class Card(ObjectController): - pass diff --git a/src/controller/client.py b/src/controller/client.py index d73eaaa..6a48897 100644 --- a/src/controller/client.py +++ b/src/controller/client.py @@ -1,12 +1,16 @@ -from typing import Callable, Any +"""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 controller import GameControllerInterface, ClientControllerInterface from netcode.models import Message, LoginMessage -from model import Player, Team +from model import Player + +if TYPE_CHECKING: + from controller import GameController def vibe_check(t): @@ -26,10 +30,12 @@ def vibe_check(t): return decorator -class ClientController(ClientControllerInterface): +class ClientController: + """Handles communication between one WebSocket tunnel and the game""" + websocket: WebSocketServerProtocol - game_controller: GameControllerInterface + game_controller: "GameController" player: Player @@ -37,7 +43,7 @@ class ClientController(ClientControllerInterface): def __init__( self, - game_controller: GameControllerInterface, + game_controller: "GameController", websocket: WebSocketServerProtocol, ): self.websocket = websocket @@ -45,7 +51,8 @@ class ClientController(ClientControllerInterface): self.game_controller = game_controller - async def listen(self): + async def listen(self) -> None: + """Hooks the client to the correct listen callbacks""" await self.send( Message( type="context", @@ -59,10 +66,12 @@ class ClientController(ClientControllerInterface): async for message in self.websocket: await self.on_message(message) - async def send(self, 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): + 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( @@ -71,55 +80,83 @@ class ClientController(ClientControllerInterface): 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): - if self.player.team != self.game_controller.game.turn: + 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": - if message.data_type == "buy": - ## TODO : pydantic type check for message.data - await self.player.buy_cards(message.data) - elif message.data_type == "play_cards": - await self.player.play_cards(message.data) - elif message.data_type == "discard": - await self.player.discard_cards(message.data) - elif message.data_type == "draw": - await self.player.draw(message.data) - elif message.data_type == "end_turn": + # TODO : pydantic type check for message.data + match message.data_type: + case "play_card": + return await self.player.play_card(message.data) - await self.player.end_turn() + 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( + message.data.effect_id, message.data.target + ) + + case "buy": + return await self.player.turn.buy_card(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): - + 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: - await self.send_error("username must be longer than 3 characters") - return + return await self.send_error( + "username must be longer than 3 characters" + ) if self.game_controller.game.get_player(username) is not None: - await self.send_error("username already exists") - return + 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: - await self.send_error("player doesn't exists") - return + return await self.send_error("player doesn't exists") + player = player2 else: - await self.send_error("invalid message type for context") - return + return await self.send_error("invalid message type for context") self.player = player self.player.team.listeners.append(self.send_announce_dict) @@ -141,18 +178,18 @@ class ClientController(ClientControllerInterface): ) ) - # 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): - + 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() + ) diff --git a/src/controller/game.py b/src/controller/game.py index f7b2f22..b4342c6 100644 --- a/src/controller/game.py +++ b/src/controller/game.py @@ -6,18 +6,16 @@ from netcode.models import Message from model import Game, Player from controller import ( ClientController, - ClientControllerInterface, - GameControllerInterface, ) -class GameController(GameControllerInterface): +class GameController: server: WSServer game: Game - clients: List[ClientControllerInterface] + clients: List[ClientController] - def __init__(self, game: Game, server: WSServer): + def __init__(self, game: Game, server: WSServer) -> None: self.server: WSServer = server self.game: Game = game self.server.on_new_connection_callback = self.on_new_connection @@ -74,6 +72,13 @@ class GameController(GameControllerInterface): ) async def announce_game(self, websocket: WebSocketServerProtocol): + for effect in self.game.effects: + await self.announce_dict( + "create", + effect.serialize(), + websocket, + ) + for card in self.game.cards: await self.announce_dict( "create", @@ -89,6 +94,11 @@ class GameController(GameControllerInterface): ) for player in self.game.players: + await self.announce_dict( + "create", + player.turn.serialize(), + websocket, + ) await self.announce_dict( "create", player.serialize_guest(), diff --git a/src/controller/interfaces.py b/src/controller/interfaces.py deleted file mode 100644 index 9eb4685..0000000 --- a/src/controller/interfaces.py +++ /dev/null @@ -1,95 +0,0 @@ -from abc import ABC, abstractmethod -from typing import Callable, Any, List -from model import Game, Player -from netcode import WSServer -from netcode.models import Message, LoginMessage -from netcode.ws_server import WebSocketServerProtocol - - -class ClientControllerInterface(ABC): - - @abstractmethod - async def listen(self): - """Initializes the client controller""" - - @abstractmethod - async def send(self, message: Message): - pass - - @abstractmethod - async def send_error(self, error: str): - pass - - @abstractmethod - async def send_announce_dict( - self, - notification_type: str, - data: dict, - player_filter: Callable[[Player], bool] = lambda x: True, - ): - pass - - @abstractmethod - async def game_context(self, message: Message): - """Handles messages when the client is logged in""" - - @abstractmethod - async def login_context(self, message: LoginMessage): - """Handles messages when the client isn't logged in""" - - @abstractmethod - async def pregame_context(self, message: Message): - """Waits for players""" - - -class GameControllerInterface(ABC): - server: WSServer - game: Game - - clients: List[ClientControllerInterface] - - @abstractmethod - def __init__(self, game: Game, server: WSServer): - pass - - @abstractmethod - async def on_new_connection(self, websocket, callback: Callable): - pass - - @abstractmethod - async def send(self, message: Message, websocket: WebSocketServerProtocol): - pass - - @abstractmethod - async def broadcast(self, message: Message): - pass - - @abstractmethod - async def broadcast_announce_dict( - self, - notification_type: str, - data: dict, - player_filter: Callable[[Player], Any], - ): - pass - - @abstractmethod - async def announce_dict( - self, - notification_type: str, - data: dict, - websocket: WebSocketServerProtocol, - ): - pass - - @abstractmethod - async def announce_game(self, websocket: WebSocketServerProtocol): - pass - - @abstractmethod - async def guest_handler(self, websocket: WebSocketServerProtocol): - pass - - @abstractmethod - async def start_game(self): - pass diff --git a/src/controller/object_controller.py b/src/controller/object_controller.py deleted file mode 100644 index 2b57bd5..0000000 --- a/src/controller/object_controller.py +++ /dev/null @@ -1,2 +0,0 @@ -class ObjectController: - pass diff --git a/src/model/__init__.py b/src/model/__init__.py index 782d8b1..48c3e0f 100644 --- a/src/model/__init__.py +++ b/src/model/__init__.py @@ -8,8 +8,8 @@ from model.card_validator import ( CardParser, ) from model.card_loader import load_cards -from model.card import Card -from model.card_board import BoardCard +from model.card import Card, Effect, EffectType, EffectTimes from model.team import Team +from model.turn import Turn from model.player import Player from model.game import Game diff --git a/src/model/card.py b/src/model/card.py index c1f886b..c2ae0aa 100644 --- a/src/model/card.py +++ b/src/model/card.py @@ -1,40 +1,86 @@ -from typing import List +from typing import List, Optional from model import Serializable, CardRole, CardType, CardFaction -from model.card_validator import Effect +from model.card_validator import CardEffects, EffectType, EffectTimes + + +class Effect(Serializable): + """Represents an effect of a card (gold, damage, heal, draw a card...)""" + + object_type = "effect" + effect: CardEffects + amount: int + sub_effects: List["Effect"] + mutex: List["Effects"] + effect_type: Optional[EffectType] = None + times: Optional[EffectTimes] = None + + def __init__( + self, + effect: CardEffects, + amount: int, + sub_effects: Optional[List["Effect"]] = None, + effect_type: Optional[EffectType] = None, + times: Optional[EffectTimes] = None, + ): + super().__init__() + self.effect = effect + self.amount = amount + if sub_effects is None: + sub_effects = [] + self.sub_effects = sub_effects + self.effect_type = effect_type + self.mutex: List["Effect"] = [] + self.times = times + + def serialize(self) -> dict: + return super().serialize() | { + "sub_effects": [effect.uuid for effect in self.sub_effects], + "mutex": [effect.uuid for effect in self.mutex], + } class Card(Serializable): """Represents any playable card""" + id: int object_type = "card" sprite: str name: str - cost: int + cost: int | None role: CardRole card_type: CardType + card_role: CardRole faction: CardFaction effects: List[Effect] + defense: Optional[int] + guard: Optional[bool] def __init__( self, + card_id: int, sprite: str, name: str, cost: int | None, card_type: CardType, + card_role: CardRole, faction: CardFaction, effects: List[Effect], + defense: Optional[int], + guard: Optional[bool], ): super().__init__() + self.card_id = card_id self.sprite = sprite self.name = name self.cost = cost self.card_type = card_type + self.card_role = card_role self.faction = faction self.effects = effects + self.defense = defense + self.guard = guard - -class CardChampion(Card): - """Represents any champion card""" - - defense: int - guard: bool + def serialize(self) -> dict: + return super().serialize() | { + "effects": [effect.uuid for effect in self.effects], + } diff --git a/src/model/card_board.py b/src/model/card_board.py deleted file mode 100644 index 24f3637..0000000 --- a/src/model/card_board.py +++ /dev/null @@ -1,15 +0,0 @@ -from typing import List -from model import Card, Serializable - - -class BoardCard(Serializable): - """Represents a card on a player's board during a game turn""" - - card: Card - - def __init__(self, card: Card): - super().__init__() - self.card = card - - def reset(self): - pass diff --git a/src/model/card_loader.py b/src/model/card_loader.py index af1fc75..476df48 100644 --- a/src/model/card_loader.py +++ b/src/model/card_loader.py @@ -3,15 +3,16 @@ from pathlib import Path from pydantic import BaseModel, ValidationError from typing import List -from model.card_validator import CardParser +from model.card_validator import CardParser, EffectParser +from model.card import Effect, Card class CardContainerFile(BaseModel): cards: List[CardParser] -def load_cards(): - cards = [] +def load_cards() -> List[CardParser]: + cards: List[CardParser] = [] for p in Path("./src/cards").glob("**/*.json"): print("loading " + p.name) with open(p, encoding="UTF-8") as f_in: @@ -24,3 +25,38 @@ def load_cards(): except ValidationError as e: print("Incorrect or missing JSON data : " + p.name + " : " + str(e)) return cards + + +def create_card(card: CardParser, effects: List[Effect]) -> Card: + return Card( + card_id=card.id, + card_type=card.card_type, + cost=card.cost, + effects=effects, + faction=card.faction, + name=card.name, + sprite="spritetest", + guard=card.guard, + defense=card.defense, + card_role=card.role, + ) + + +def create_card_effects(card: CardParser) -> List[Effect]: + combined: List[Effect] = [] + for effect in card.effects: + combined.append(create_effect(effect)) + return combined + + +def create_effect(effect: EffectParser) -> Effect: + sub_effects: List[Effect] = [] + for s in effect.sub_effects: + sub_effects.append(create_effect(s)) + return Effect( + effect=effect.effect, + amount=effect.amount, + effect_type=effect.effect_type, + sub_effects=sub_effects, + times=effect.times, + ) diff --git a/src/model/card_validator.py b/src/model/card_validator.py index 7667d89..85bdac5 100644 --- a/src/model/card_validator.py +++ b/src/model/card_validator.py @@ -45,7 +45,7 @@ class CardEffects(str, Enum): RESTACK_DISCARDED_CHAMPION = "restack_discarded_champion" RESTACK_DISCARDED_CARD = "restack_discarded_card" STACK_NEXT_ACTION_BOUGHT = "stack_next_action_bought" - STACK_NEXT_CARD_BOUGHT = "stack_next_card_bough" + STACK_NEXT_CARD_BOUGHT = "stack_next_card_bought" PLAY_NEXT_CARD_BOUGHT = "play_next_card_bought" @@ -62,15 +62,17 @@ class EffectType(str, Enum): FACTION_COMBO = "faction_combo" -class Effect(BaseModel): +class EffectParser(BaseModel): effect: CardEffects amount: int effect_type: Optional[EffectType] = None - sub_effects: Optional[List["Effect"]] = None + sub_effects: List["EffectParser"] = [] + times: Optional[EffectTimes] = None mutex: Optional[int] = None class CardParser(BaseModel): + id: int role: CardRole card_type: CardType faction: CardFaction @@ -79,5 +81,4 @@ class CardParser(BaseModel): cost: Optional[int] = None defense: Optional[int] = None guard: Optional[bool] = None - init_amount: int - effects: List[Effect] + effects: List[EffectParser] diff --git a/src/model/game.py b/src/model/game.py index a9a541e..385151b 100644 --- a/src/model/game.py +++ b/src/model/game.py @@ -1,15 +1,11 @@ +"""Instance of the game, with players and a market""" + import asyncio from typing import List import random -from model import ( - Card, - CardRole, - CardParser, - Player, - Team, - Serializable, - load_cards, -) +from model import Card, CardRole, CardParser, Player, Team, Serializable, Effect + +from model.card_loader import load_cards, create_card_effects, create_card class Game(Serializable): @@ -19,6 +15,7 @@ class Game(Serializable): loaded_cards: List[CardParser] = load_cards() cards: List[Card] + effects: List[Effect] market_stack: List[Card] market: List[Card] gem_stack: List[Card] @@ -29,63 +26,47 @@ class Game(Serializable): default_health = 50 - turn: Team + current_turn: Team - def __init__(self): + def __init__(self) -> None: super().__init__() self.market_stack: List[Card] = [] - self.market: List[Card] = [None for _ in range(5)] + self.market: List[Card | None] = [None for _ in range(5)] self.gem_stack: List[Card] = [] self.players: List[Player] = [] self.cards: List[Card] = [] + self.effects: List[Effect] = [] self.teams: List[Team] = [] self.started = False - self.turn = None + self.current_turn = None async def init_game(self): await self.init_cards() await self.distribute_market_cards() + async def instanciate_card(self, card_parser: CardParser) -> Card: + effects = create_card_effects(card_parser) + for e in effects: + self.effects.append(e) + await self.notify_controller("create", e.serialize()) + card = create_card(card_parser, effects) + self.cards.append(card) + await self.notify_controller("create", card.serialize()) + return card + async def init_cards(self): """Creates market cards and fire gem cards from previously loaded cards""" market_only = [c for c in self.loaded_cards if c.role == CardRole.MARKET] for c in market_only: for _ in range(c.init_amount): - self.market_stack.append( - await self.create_card( - Card( - sprite="spritetest", - name=c.name, - cost=c.cost, - card_type=c.card_type, - faction=c.faction, - effects=c.effects, - ) - ) - ) + self.market_stack.append(await self.instanciate_card(c)) random.shuffle(self.market_stack) fire_gems = [c for c in self.loaded_cards if c.role == CardRole.FIRE_GEM] for c in fire_gems: for _ in range(c.init_amount): - self.gem_stack.append( - await self.create_card( - Card( - sprite="spritetest", - name=c.name, - cost=c.cost, - card_type=c.card_type, - faction=c.faction, - effects=c.effects, - ) - ) - ) - - async def create_card(self, card: Card) -> Card: - self.cards.append(card) - await self.notify_controller("create", card.serialize_guest()) - return card + self.gem_stack.append(await self.instanciate_card(c)) async def create_base_deck(self, base_deck_name: str = "base") -> List[Card]: """Creates the starting base deck for a player. @@ -98,18 +79,7 @@ class Game(Serializable): base_deck = [] for c in base_deck_loaded: for _ in range(c.init_amount): - base_deck.append( - await self.create_card( - Card( - sprite="spritetest", - name=c.name, - cost=c.cost, - card_type=c.card_type, - faction=c.faction, - effects=c.effects, - ) - ) - ) + base_deck.append(await self.instanciate_card(c)) random.shuffle(base_deck) return base_deck @@ -121,6 +91,7 @@ class Game(Serializable): self.players.append(player) player.stack_pile = await self.create_base_deck("base") + await self.notify_controller("create", player.turn.serialize()) await self.notify_controller("create", player.serialize_guest()) await self.notify_controller("update", self.serialize_guest()) return player @@ -158,21 +129,24 @@ class Game(Serializable): "teams": [team.uuid for team in self.teams], "started": self.started, "default_health": self.default_health, - "turn": None if self.turn is None else self.turn.uuid, + "current_turn": ( + None if self.current_turn is None else self.current_turn.uuid + ), + "effects": [effect.uuid for effect in self.effects], } return {k: data[k] for k in data.keys() - ["loaded_cards"]} async def start_game(self): self.started = True - self.turn = random.choice(self.teams) + self.current_turn = random.choice(self.teams) await self.notify_controller("update", self.serialize_guest()) draws = [] - for p in [p for p in self.players if p.team == self.turn]: + for p in [p for p in self.players if p.team == self.current_turn]: draws.append(p.draw(3)) - for p in [p for p in self.players if p.team != self.turn]: + for p in [p for p in self.players if p.team != self.current_turn]: draws.append(p.draw(5)) await asyncio.gather(*draws) @@ -191,6 +165,14 @@ class Game(Serializable): ) async def end_turn(self): - team_index = self.teams.index(self.turn) - self.turn = self.teams[(team_index + 1) % len(self.teams)] + team_index = self.teams.index(self.current_turn) + self.current_turn = self.teams[(team_index + 1) % len(self.teams)] + for p in self.players: + p.start_turn() await self.notify_controller("update", self.serialize_guest()) + + def search_card_in_pile(self, card_id: str, pile: List[Card]): + return next( + (m for m in [f for f in pile if f is not None] if str(m.uuid) == card_id), + None, + ) diff --git a/src/model/player.py b/src/model/player.py index 86d8807..e7a728d 100644 --- a/src/model/player.py +++ b/src/model/player.py @@ -1,8 +1,13 @@ +"""Game player, controller by a Client""" + import asyncio import random -from typing import List +from typing import List, Optional, TYPE_CHECKING -from model import Card, Team, Serializable, BoardCard, CardType +from model import Card, Team, Serializable, Turn, CardType, CardRole + +if TYPE_CHECKING: + from model import Game class Player(Serializable): @@ -16,9 +21,10 @@ class Player(Serializable): team: Team hand: List[Card] - board: List[BoardCard] + board: List[Card] - game: Serializable + game: "Game" + turn: Turn _ready: bool = False @@ -34,15 +40,16 @@ class Player(Serializable): def ready(self): del self._ready - def __init__(self, game: Serializable, username: str, team: Team): + def __init__(self, game: "Game", username: str, team: Team) -> None: super().__init__() self.game = game self.stack_pile: List[Card] = [] self.discard_pile: List[Card] = [] self.hand: List[Card] = [] - self.board: List[BoardCard] = [] + self.board: List[Card] = [] self.team = team self.username = username + self.turn = Turn(self) def serialize_guest(self) -> dict: """Get data available to the public""" @@ -51,7 +58,8 @@ class Player(Serializable): "discard_pile": [card.uuid for card in self.discard_pile], "team": self.team.uuid, "hand": [None for card in self.hand], - "board": [card_board.card.uuid for card_board in self.board], + "turn": self.turn.uuid, + "board": [card.uuid for card in self.board], } return {k: data[k] for k in data.keys() - ["game"]} @@ -74,21 +82,29 @@ class Player(Serializable): self.stack_pile, self.discard_pile = self.discard_pile, self.stack_pile random.shuffle(self.stack_pile) - async def draw(self, amount: int): - """Draw a card from the stack. Handles stack rebuild from discard automatically""" + async def start_turn(self): + self.turn.start_turn() - while amount > 0: - if len(self.stack_pile) == 0: - if len(self.discard_pile) == 0: - print("No more cards to draw !") - break - await self._rebuild_stack() + async def end_turn(self): + temp_board = [*self.board] + for c in self.board: + for c in self.board: + if c.card_type not in [ + CardType.HERO, + CardType.HERO_ABILITY, + CardType.CHAMPION, + ]: + temp_board.remove(c) + self.discard_pile.append(c) - if len(self.stack_pile) > 0: - self.hand.append(self.stack_pile.pop()) - amount = amount - 1 + self.board = temp_board - await self.notify_update() + for c in self.hand: + self.discard_pile.append(c) + self.hand = [] + + await self.draw(5) + await self.game.end_turn() async def notify_update(self): """Correctly updates player data for team and other clients. @@ -104,95 +120,161 @@ class Player(Serializable): await asyncio.gather(*notifies) - async def play_cards(self, cards_ids: List[str]): - for s in cards_ids: - card = next( - (m for m in self.hand if str(m.uuid) == s), - None, - ) + def source_card(self, card_id: str, source: List[Card]) -> Card: + card = self.game.search_card_in_pile(card_id, source) + if card is None: + return None + source.remove(card) + return card + + async def play_card(self, card_id: str): + card = self.source_card(card_id, self.hand) + if card is None: + return + self.board.append(card) + await self.notify_update() + + async def discard_card(self, card_id: str): + card = self.source_card(card_id, self.hand) + if card is not None: + self.discard_pile.append(card) + else: + board_card = self.source_card(card_id, self.board) + if board_card is not None: + self.discard_pile.append(board_card) + else: + print("Card not found in board or hand") + return + await self.notify_update() + + async def draw(self, amount: int) -> bool: + """Draw a card from the stack. Handles stack rebuild from discard automatically""" + + while amount > 0: + if len(self.stack_pile) == 0: + if len(self.discard_pile) == 0: + print("No more cards to draw !") + break + await self._rebuild_stack() + + if len(self.stack_pile) > 0: + self.hand.append(self.stack_pile.pop()) + amount = amount - 1 + + await self.notify_update() + return True + + async def buy_card(self, card: Card) -> bool: + + if card in self.game.gem_stack: + self.game.gem_stack.remove(card) + elif card in self.game.market: + index = self.game.market.index(card) + self.game.market[index] = None + await self.game.distribute_market_cards() + else: + print("Card not found in market or gem_stack") + return False + + self.discard_pile.append(card) + await self.game.notify_controller("update", self.game.serialize_guest()) + await self.notify_update() + return True + + async def sacrifice_card(self, card_id: str) -> bool: + board_card = self.source_card(card_id, self.board) + if board_card is None: + card = self.source_card(card_id, self.discard_pile) + if card is None: + print("Card not found in board or hand or discard") + return False + + if card.card_role == CardRole.FIRE_GEM: + self.game.gem_stack.append(card) + await self.notify_update() + return True + + async def draw_from_discard(self, card_id: str): + + card = self.source_card(card_id, self.discard_pile) + if card is not None: + self.hand.append(card) + else: + print("Card not found in discard pile") + return + + await self.notify_update() + + async def put_on_stack_from_discard(self, card: Card) -> bool: + if card not in self.discard_pile: + print("Card not found in discard pile") + return False + + self.discard_pile.remove(card) + self.stack_pile.append(card) + await self.notify_update() + return True + + async def buy_in_hand(self, card: Card) -> bool: + if card in self.game.gem_stack: + self.game.gem_stack.remove(card) + elif card in self.game.market: + index = self.game.market.index(card) + self.game.market[index] = None + await self.game.distribute_market_cards() + else: + print("Card not found in market or gem_stack") + return False + + self.hand.append(card) + await self.game.notify_controller("update", self.game.serialize_guest()) + await self.notify_update() + return True + + async def buy_on_stack(self, card: Card) -> bool: + if card in self.game.gem_stack: + self.game.gem_stack.remove(card) + elif card in self.game.market: + index = self.game.market.index(card) + self.game.market[index] = None + await self.game.distribute_market_cards() + else: + print("Card not found in market or gem_stack") + return False + + self.stack_pile.append(card) + await self.game.notify_controller("update", self.game.serialize_guest()) + await self.notify_update() + return True + + async def stun(self, card_id: str) -> bool: + # TODO : Handle Predator Rule + for p in self.game.players: + if p.team == self.team: + continue + card = self.game.search_card_in_pile(card.id, p.board) if card is None: - print("Card not found in player hand") continue - self.board.append(BoardCard(card)) - self.hand.remove(card) - await self.notify_update() - - async def discard_cards(self, cards_ids: List[str]): - for s in cards_ids: - card = next( - (m for m in self.hand if str(m.uuid) == s), - None, - ) - if card is not None: - self.discard_pile.append(card) - self.hand.remove(card) else: - board_card = next( - (m for m in self.board if str(m.card.uuid) == s), - None, - ) - if board_card is not None: - self.discard_pile.append(board_card.card) - self.board.remove(board_card) - else: - print("Card not found in board or hand") - continue - await self.notify_update() + card = self.source_card(card_id, p.board) + p.discard_pile.append(card) + await p.notify_update() + return True - async def end_turn(self): - temp_board = [*self.board] - for c in self.board: - if c.card.card_type in [ - CardType.HERO, - CardType.HERO_ABILITY, - CardType.CHAMPION, - ]: - c.reset() - else: - temp_board.remove(c) - self.discard_pile.append(c.card) + print("Cannot stun : card not found") + return False - self.board = temp_board - - for c in self.hand: - self.discard_pile.append(c) - self.hand = [] - - await self.draw(5) - await self.game.end_turn() - - async def buy_cards(self, card_ids: List[str]): - for card_id in card_ids: - card = next( - ( - m - for m in [f for f in self.game.market if f is not None] - if str(m.uuid) == card_id - ), - None, - ) - if card is not None: - index = self.game.market.index(card) - self.game.market[index] = None - self.discard_pile.append(card) - await self.game.distribute_market_cards() - else: - card = next( - ( - m - for m in [f for f in self.game.gem_stack if f is not None] - if str(m.uuid) == card_id - ), - None, - ) - if card is not None: - self.game.gem_stack.remove(card) - self.discard_pile.append(card) - await self.game.notify_controller( - "update", self.game.serialize_guest() - ) - else: - print("Card not found in market or gem_stack") - continue - - await self.notify_update() + async def make_enemy_discard(self, player_id: str) -> bool: + p = next( + (p for p in self.game.players if str(p.uuid) == player_id), + None, + ) + if p is None: + print("player not found") + return False + if p.turn.discard_markers >= len(p.hand): + print("player has already too many discard markers!") + return False + p.turn.discard_markers += 1 + await self.game.notify_controller("update", p.turn.serialize()) + return True diff --git a/src/model/serializable.py b/src/model/serializable.py index e375b34..b1c6b1a 100644 --- a/src/model/serializable.py +++ b/src/model/serializable.py @@ -3,22 +3,22 @@ from uuid import uuid4, UUID class Serializable: - object_type = "generic" + object_type = "serializable" uuid: UUID listeners: List[Callable[[str, dict, Callable], Awaitable[Any]]] - def __init__(self): + def __init__(self) -> None: self.uuid = uuid4() self.object_type = self.__class__.object_type self.listeners = [] - def serialize(self): + def serialize(self) -> dict: """Returns a dict to be saved on the database""" return {k: self.__dict__[k] for k in self.__dict__.keys() - ["listeners"]} - def serialize_guest(self): + def serialize_guest(self) -> dict: """Returns a dict with some elements masked for the players""" return {k: self.__dict__[k] for k in self.__dict__.keys() - ["listeners"]} diff --git a/src/model/team.py b/src/model/team.py index 24ce4ed..072ee2c 100644 --- a/src/model/team.py +++ b/src/model/team.py @@ -1,3 +1,5 @@ +"""Represents a team (Health and Turn) of players""" + from model import Serializable @@ -9,7 +11,7 @@ class Team(Serializable): color: str name: str - def __init__(self, health=80, color="red", name="unnamed Team"): + def __init__(self, health=80, color="red", name="unnamed Team") -> None: super().__init__() self.health = health self.color = color diff --git a/src/model/turn.py b/src/model/turn.py new file mode 100644 index 0000000..538aaf4 --- /dev/null +++ b/src/model/turn.py @@ -0,0 +1,275 @@ +"""Stores which effects have been locked or used in a turn +Checks the actions of the user and prevent the user to do something illegal +""" + +from typing import TYPE_CHECKING, List, Dict, cast, Optional, Callable +from uuid import UUID +from model import ( + Serializable, + CardType, + Card, + Effect, + CardEffects, + EffectType, + EffectTimes, +) + + +if TYPE_CHECKING: + from model import Player + + +class Turn(Serializable): + """Stores which effects have been locked or used in a turn + Checks the actions of the user and prevent the user to do something illegal""" + + object_type = "turn" + + player: "Player" + cards_locked: List[Card] + effects_availables: List[Effect] + champions_health: Dict[UUID, int] + discard_markers: int + damage_reserve: int + heal_reserve: int + gold_reserve: int + + def __init__(self, player: "Player") -> None: + super().__init__() + self.player = player + self.damage_reserve = 0 + self.heal_reserve = 0 + self.gold_reserve = 0 + self.champions_health: Dict[UUID, int] = {} + self.effects_availables: List[Effect] = [] + self.cards_locked: List[Card] = [] + self.discard_markers = 0 + + def end_turn(self) -> None: + """Resets card champion health and card effects""" + if len(self.player.hand) > 0: + return print("You must play all your cards before ending a turn!") + + self.damage_reserve = 0 + self.heal_reserve = 0 + self.gold_reserve = 0 + self.effects_availables = [] + self.cards_locked = [] + self.discard_markers = 0 + self.player.end_turn() + + def lock_card(self, card_id: str): + if self.discard_markers > 0: + return print("You cannot lock a card while having discard markers") + card = self.player.game.search_card_in_pile(card_id, self.player.board) + if card is None: + return print("Couldn't find card in player board") + if card in self.cards_locked: + return print("Card already locked") + self.cards_locked.append(card) + for effect in card.effects: + for _ in range(effect.amount): + match effect.times: + case EffectTimes.PER_CARD_OF_SAME_FACTION: + for c in self.player.board: + if c.faction == card.faction: + self.effects_availables.append(effect) + case EffectTimes.PER_CHAMPION: + for c in self.player.board: + if c.card_role == CardType.CHAMPION: + self.effects_availables.append(effect) + case EffectTimes.PER_OTHER_CHAMPION: + for c in self.player.board: + if c.card_role == CardType.CHAMPION and c != card: + self.effects_availables.append(effect) + case EffectTimes.PER_OTHER_GUARD: + for c in self.player.board: + if ( + c.card_role == CardType.CHAMPION + and c.guard is True + and c != card + ): + self.effects_availables.append(effect) + case None: + self.effects_availables.append(effect) + case _: + raise NotImplementedError( + "EffectTimes " + effect.times + " not implemented" + ) + + def discard_card(self, card_id: str): + if self.discard_markers <= 0: + return print("You don't have any discard markers!") + card = self.player.game.search_card_in_pile(card_id, self.cards_locked) + if card is not None: + return print("Cannot discard: card is locked") + return self.player.discard_card(card_id) + + def sacrifice_card(self, card_id: str): + card = self.player.game.search_card_in_pile(card_id, self.cards_locked) + if card is not None: + return print("Cannot sacrifice : card is locked") + return self.player.discard_card(card_id) + + async def use_effect(self, effect_id: str, target: Optional[str]): + if self.discard_markers > 0: + return print("You cannot use an effect while having discard markers") + effect = next( + (e for e in self.effects_availables if str(e.uuid) == effect_id), + None, + ) + if effect is None: + return print("Effect not available") + + card = next( + (m for m in self.player.board if effect in m), + None, + ) + if card is None: + # TODO: Could break for subEffects + # Maybe store a reference to card in effect to fix this + return print( + "Something went wrong while finding the card parent of an effect" + ) + match effect.effect_type: + case EffectType.SUICIDE: + success = False + success = self.player.sacrifice_card(card.uuid) + if not success: + return print("something went wrong while trying to suicide a card") + for e in card.effects: + if e != effect: + self.effects_availables.remove(e) + for se in e.sub_effects: + if se != effect: + self.effects_availables.remove(se) + case EffectType.FACTION_COMBO: + if ( + len([c for c in self.player.board if c.faction == card.faction]) + <= 1 + ): + return print("effect cannot run : faction combo not reached") + case None: + pass + case _: + raise NotImplementedError( + "EffectType " + effect.effect_type + " not implemented" + ) + + success = False + + match effect.effect_type: + case CardEffects.GOLD: + self.gold_reserve += effect.amount + success = False + case CardEffects.DAMAGE: + self.damage_reserve += effect.amount + success = False + case CardEffects.HEAL: + self.heal_reserve += effect.amount + success = False + case CardEffects.PERPARE: + success = await self.prepare(target) + case CardEffects.STUN: + success = await self.player.stun(target) + case CardEffects.SACRIFICE: + success = await self.sacrifice_card(target) + case CardEffects.DRAW: + success = await self.player.draw(1) + case CardEffects.DRAW_AND_DISCARD: + self.discard_markers += 1 + success = await self.player.draw(1) + case CardEffects.DISCARD: + success = await self.player.make_enemy_discard(target) + case CardEffects.RESTACK_DISCARDED_CHAMPION: + success = await self.restack_discarded_champion(target) + case CardEffects.RESTACK_DISCARDED_CARD: + success = await self.player.put_on_stack_from_discard(target) + case CardEffects.STACK_NEXT_ACTION_BOUGHT: + success = await self.buy_card(target, self.player.buy_on_stack) + case CardEffects.STACK_NEXT_CARD_BOUGHT: + success = await self.buy_card(target, self.stack_next_card_bought_check) + case CardEffects.PLAY_NEXT_CARD_BOUGHT: + success = await self.buy_card(target, self.player.buy_in_hand) + case _: + return + + if not success: + return print("something went wrong while executing effect") + + self.effects_availables.remove(effect) + for e in effect.sub_effects: + self.effects_availables.append(e) + self.player.game.notify_controller("update", self.serialize()) + + def prepare(self, card_id: str) -> bool: + card = next( + (e for e in self.cards_locked if str(e.uuid) == card_id), + None, + ) + if card is None: + print("Card is not locked") + return False + for e in card.effects: + if e.effect_type != EffectType.CHAMPION_ACTION: + continue + if e in self.effects_availables: + continue + self.effects_availables.append(e) + return True + + def start_turn(self) -> None: + for c in self.player.board: + if c.card_type == CardType.CHAMPION: + self.champions_health[c.uuid] = c.defense + + def serialize(self): + data = super().serialize() | { + "cards_locked": [card.uuid for card in self.cards_locked], + "effects_availables": [effect.uuid for effect in self.effects_availables], + } + + return {k: data[k] for k in data.keys() - ["player"]} + + async def restack_discarded_champion(self, card_id) -> bool: + card = self.player.game.search_card_in_pile(card_id, self.player.discard_pile) + if card is None: + print("card not found in discard") + return False + if card.card_type != CardType.CHAMPION: + print("card is not a champion") + return False + else: + return await self.player.put_on_stack_from_discard(self, card) + + async def stack_next_card_bought_check(self, card: Card) -> bool: + + if self.gold_reserve < card.cost: + print("Not enough gold to buy card!") + return False + + return self.player.buy_on_stack(card) + + async def buy_card( + self, card_id: str, buy_function: Optional[Callable[[Card], bool]] = None + ): + if buy_function is None: + buy_function = self.player.buy_card + card = self.player.game.search_card_in_pile(card_id, self.player.game.market) + if card is None: + card = self.player.game.search_card_in_pile( + card_id, self.player.game.gem_stack + ) + if card is None: + print("Card not found in market or gem_stack") + return False + + if self.gold_reserve < card.cost: + print("Not enough gold to buy card!") + return False + + if buy_function(card): + self.gold_reserve -= card.cost + return True + + return False diff --git a/src/netcode/ws_server.py b/src/netcode/ws_server.py index 2d84e92..9c61f7a 100644 --- a/src/netcode/ws_server.py +++ b/src/netcode/ws_server.py @@ -14,7 +14,7 @@ class WSServer: port: int - def __init__(self): + def __init__(self) -> None: self.connections = set() self.message_callbacks = []