heronfief/back-python
Archived
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

only legal actions can be taken

Browse Source
This commit is contained in:
legonzaur
2024-04-24 23:55:39 +02:00
parent e156d3a951
commit 8b397484cb
17 changed files with 710 additions and 359 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/controller/__init__.py
-3
View File
@@ -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
src/controller/card.py
-5
View File
@@ -1,5 +0,0 @@
from controller import ObjectController
class Card(ObjectController):
pass
src/controller/client.py
+80 -43
View File
@@ -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()
)
src/controller/game.py
+15 -5
View File
@@ -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(),
src/controller/interfaces.py
-95
View File
@@ -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
src/controller/object_controller.py
-2
View File
@@ -1,2 +0,0 @@
class ObjectController:
pass
src/model/__init__.py
+2 -2
View File
@@ -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
src/model/card.py
+55 -9
View File
@@ -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],
}
src/model/card_board.py
-15
View File
@@ -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
src/model/card_loader.py
+39 -3
View File
@@ -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,
)
src/model/card_validator.py
+6 -5
View File
@@ -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]
src/model/game.py
+42 -60
View File
@@ -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,
)
src/model/player.py
+188 -106
View File
@@ -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
src/model/serializable.py
+4 -4
View File
@@ -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"]}
src/model/team.py
+3 -1
View File
@@ -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
src/model/turn.py
+275
View File
@@ -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
src/netcode/ws_server.py
+1 -1
View File
@@ -14,7 +14,7 @@ class WSServer:
port: int
def __init__(self):
def __init__(self) -> None:
self.connections = set()
self.message_callbacks = []