554178dc6e
Added stadium_owner_id field to GameState to track who played the stadium: - stadium_owner_id: str | None tracks the player who played the current stadium - When a stadium is replaced, old stadium discards to OWNER's pile (not current player) - Added stadium_owner_id to VisibleGameState for client visibility - Updated existing test and added 2 new tests for stadium ownership This fixes the bug where replacing an opponent's stadium would discard to the current player's pile instead of the opponent's. 797 tests passing.
616 lines
23 KiB
Python
616 lines
23 KiB
Python
"""Game state models for the Mantimon TCG game engine.
|
|
|
|
This module defines the complete game state hierarchy:
|
|
- Zone: A collection of cards (deck, hand, bench, etc.) with operations
|
|
- PlayerState: All zones and state for a single player
|
|
- GameState: Complete game state including both players, rules, and turn tracking
|
|
|
|
The GameState is designed to be self-contained - it includes the card registry
|
|
so that the game can be serialized/deserialized without external dependencies.
|
|
This supports both live multiplayer and offline/standalone play.
|
|
|
|
Usage:
|
|
# Create a new game
|
|
game = GameState(
|
|
game_id="match-123",
|
|
rules=RulesConfig(),
|
|
card_registry={card.id: card for card in cards},
|
|
players={
|
|
"player1": PlayerState(player_id="player1"),
|
|
"player2": PlayerState(player_id="player2"),
|
|
},
|
|
)
|
|
|
|
# Access player zones
|
|
player = game.players["player1"]
|
|
card = player.hand.draw()
|
|
player.deck.shuffle(rng)
|
|
"""
|
|
|
|
from typing import Any
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.core.config import RulesConfig
|
|
from app.core.models.card import CardDefinition, CardInstance
|
|
from app.core.models.enums import GameEndReason, TurnPhase
|
|
from app.core.rng import RandomProvider
|
|
|
|
|
|
class ForcedAction(BaseModel):
|
|
"""Represents an action a player must take before the game can proceed.
|
|
|
|
When a forced action is set, only the specified player can act, and only
|
|
with the specified action type. This is used for situations like:
|
|
- Selecting a new active Pokemon after a knockout
|
|
- Selecting prize cards to take
|
|
- Discarding cards when required by an effect
|
|
|
|
Attributes:
|
|
player_id: The player who must take the action.
|
|
action_type: The type of action required (e.g., "select_active", "select_prize").
|
|
reason: Human-readable explanation of why this action is required.
|
|
params: Additional parameters for the action (e.g., {"count": 2} for "discard 2 cards").
|
|
"""
|
|
|
|
player_id: str
|
|
action_type: str
|
|
reason: str
|
|
params: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class Zone(BaseModel):
|
|
"""A collection of cards representing a game zone.
|
|
|
|
Zones are the fundamental building blocks of game state. Each zone
|
|
(deck, hand, bench, etc.) is a Zone instance with appropriate operations.
|
|
|
|
The zone stores CardInstance objects by their instance_id. The actual
|
|
CardDefinition data is looked up from the GameState.card_registry.
|
|
|
|
Attributes:
|
|
cards: List of CardInstance objects in this zone.
|
|
zone_type: Identifier for this zone type (for debugging/logging).
|
|
"""
|
|
|
|
cards: list[CardInstance] = Field(default_factory=list)
|
|
zone_type: str = "generic"
|
|
|
|
def __len__(self) -> int:
|
|
"""Return the number of cards in this zone."""
|
|
return len(self.cards)
|
|
|
|
def __bool__(self) -> bool:
|
|
"""Return True if the zone has any cards."""
|
|
return len(self.cards) > 0
|
|
|
|
def __contains__(self, instance_id: str) -> bool:
|
|
"""Check if a card with the given instance_id is in this zone."""
|
|
return any(card.instance_id == instance_id for card in self.cards)
|
|
|
|
def is_empty(self) -> bool:
|
|
"""Check if the zone has no cards."""
|
|
return len(self.cards) == 0
|
|
|
|
def add(self, card: CardInstance) -> None:
|
|
"""Add a card to this zone (appends to the end)."""
|
|
self.cards.append(card)
|
|
|
|
def add_to_top(self, card: CardInstance) -> None:
|
|
"""Add a card to the top of this zone (index 0, like top of deck)."""
|
|
self.cards.insert(0, card)
|
|
|
|
def add_to_bottom(self, card: CardInstance) -> None:
|
|
"""Add a card to the bottom of this zone (end of list)."""
|
|
self.cards.append(card)
|
|
|
|
def remove(self, instance_id: str) -> CardInstance | None:
|
|
"""Remove and return a card by instance_id, or None if not found."""
|
|
for i, card in enumerate(self.cards):
|
|
if card.instance_id == instance_id:
|
|
return self.cards.pop(i)
|
|
return None
|
|
|
|
def get(self, instance_id: str) -> CardInstance | None:
|
|
"""Get a card by instance_id without removing it."""
|
|
for card in self.cards:
|
|
if card.instance_id == instance_id:
|
|
return card
|
|
return None
|
|
|
|
def draw(self) -> CardInstance | None:
|
|
"""Remove and return the top card (index 0), or None if empty.
|
|
|
|
This is the standard "draw from deck" operation.
|
|
"""
|
|
if self.cards:
|
|
return self.cards.pop(0)
|
|
return None
|
|
|
|
def draw_bottom(self) -> CardInstance | None:
|
|
"""Remove and return the bottom card, or None if empty."""
|
|
if self.cards:
|
|
return self.cards.pop()
|
|
return None
|
|
|
|
def peek(self, count: int = 1) -> list[CardInstance]:
|
|
"""Look at the top N cards without removing them.
|
|
|
|
Args:
|
|
count: Number of cards to peek at.
|
|
|
|
Returns:
|
|
List of cards from the top (may be fewer if zone has less).
|
|
"""
|
|
return self.cards[:count]
|
|
|
|
def peek_bottom(self, count: int = 1) -> list[CardInstance]:
|
|
"""Look at the bottom N cards without removing them."""
|
|
if count >= len(self.cards):
|
|
return list(self.cards)
|
|
return self.cards[-count:]
|
|
|
|
def shuffle(self, rng: RandomProvider) -> None:
|
|
"""Shuffle the cards in this zone using the provided RNG.
|
|
|
|
Args:
|
|
rng: RandomProvider instance for deterministic shuffling.
|
|
"""
|
|
rng.shuffle(self.cards)
|
|
|
|
def clear(self) -> list[CardInstance]:
|
|
"""Remove and return all cards from this zone."""
|
|
cards = self.cards.copy()
|
|
self.cards.clear()
|
|
return cards
|
|
|
|
def get_all(self) -> list[CardInstance]:
|
|
"""Return a copy of all cards in this zone."""
|
|
return list(self.cards)
|
|
|
|
def find_by_definition(self, definition_id: str) -> list[CardInstance]:
|
|
"""Find all cards with a specific definition_id."""
|
|
return [card for card in self.cards if card.definition_id == definition_id]
|
|
|
|
def count(self) -> int:
|
|
"""Return the number of cards (alias for __len__)."""
|
|
return len(self.cards)
|
|
|
|
|
|
class PlayerState(BaseModel):
|
|
"""Complete state for a single player in a game.
|
|
|
|
Contains all zones (deck, hand, active, bench, etc.) and per-player
|
|
state like score, turn counters, and action tracking.
|
|
|
|
Attributes:
|
|
player_id: Unique identifier for this player.
|
|
deck: The player's draw deck.
|
|
hand: Cards in the player's hand.
|
|
active: The active Pokemon zone (0 or 1 card).
|
|
bench: Benched Pokemon (up to max_bench_size from rules).
|
|
discard: Discard pile.
|
|
prizes: Prize cards (hidden until taken, in point-based mode this is unused).
|
|
energy_deck: Separate energy deck (Pokemon Pocket style).
|
|
score: Points scored (knockouts).
|
|
energy_attachments_this_turn: Number of energy cards attached this turn.
|
|
supporters_played_this_turn: Number of Supporter cards played this turn.
|
|
stadiums_played_this_turn: Number of Stadium cards played this turn.
|
|
items_played_this_turn: Number of Item cards played this turn.
|
|
retreats_this_turn: Number of retreats performed this turn.
|
|
gx_attack_used: Whether this player has used their GX attack (once per game).
|
|
vstar_power_used: Whether this player has used their VSTAR power (once per game).
|
|
"""
|
|
|
|
player_id: str
|
|
|
|
# Zones
|
|
deck: Zone = Field(default_factory=lambda: Zone(zone_type="deck"))
|
|
hand: Zone = Field(default_factory=lambda: Zone(zone_type="hand"))
|
|
active: Zone = Field(default_factory=lambda: Zone(zone_type="active"))
|
|
bench: Zone = Field(default_factory=lambda: Zone(zone_type="bench"))
|
|
discard: Zone = Field(default_factory=lambda: Zone(zone_type="discard"))
|
|
prizes: Zone = Field(default_factory=lambda: Zone(zone_type="prizes"))
|
|
energy_deck: Zone = Field(default_factory=lambda: Zone(zone_type="energy_deck"))
|
|
energy_zone: Zone = Field(default_factory=lambda: Zone(zone_type="energy_zone"))
|
|
|
|
# Score tracking (point-based system)
|
|
score: int = 0
|
|
|
|
# Per-turn action counters (reset at turn start)
|
|
# These are integers to support configurable rules (e.g., 2 energy per turn)
|
|
energy_attachments_this_turn: int = 0
|
|
supporters_played_this_turn: int = 0
|
|
stadiums_played_this_turn: int = 0
|
|
items_played_this_turn: int = 0
|
|
retreats_this_turn: int = 0
|
|
|
|
# Per-game flags
|
|
gx_attack_used: bool = False
|
|
vstar_power_used: bool = False
|
|
|
|
def get_active_pokemon(self) -> CardInstance | None:
|
|
"""Get the active Pokemon, or None if no active."""
|
|
if self.active.cards:
|
|
return self.active.cards[0]
|
|
return None
|
|
|
|
def has_active_pokemon(self) -> bool:
|
|
"""Check if this player has an active Pokemon."""
|
|
return len(self.active) > 0
|
|
|
|
def has_benched_pokemon(self) -> bool:
|
|
"""Check if this player has any benched Pokemon."""
|
|
return len(self.bench) > 0
|
|
|
|
def has_pokemon_in_play(self) -> bool:
|
|
"""Check if this player has any Pokemon in play (active or bench)."""
|
|
return self.has_active_pokemon() or self.has_benched_pokemon()
|
|
|
|
def get_all_pokemon_in_play(self) -> list[CardInstance]:
|
|
"""Get all Pokemon in play (active + bench)."""
|
|
pokemon = []
|
|
if self.active.cards:
|
|
pokemon.extend(self.active.cards)
|
|
pokemon.extend(self.bench.cards)
|
|
return pokemon
|
|
|
|
def can_attach_energy(self, rules: RulesConfig) -> bool:
|
|
"""Check if this player can attach energy based on rules and turn state.
|
|
|
|
Compares the number of energy attachments made this turn against the
|
|
rules.energy.attachments_per_turn limit.
|
|
|
|
Args:
|
|
rules: The RulesConfig governing this game.
|
|
|
|
Returns:
|
|
True if more energy can be attached this turn.
|
|
"""
|
|
return self.energy_attachments_this_turn < rules.energy.attachments_per_turn
|
|
|
|
def can_play_supporter(self, rules: RulesConfig) -> bool:
|
|
"""Check if this player can play a Supporter card.
|
|
|
|
Compares the number of Supporters played this turn against the
|
|
rules.trainer.supporters_per_turn limit.
|
|
|
|
Args:
|
|
rules: The RulesConfig governing this game.
|
|
|
|
Returns:
|
|
True if more Supporters can be played this turn.
|
|
"""
|
|
return self.supporters_played_this_turn < rules.trainer.supporters_per_turn
|
|
|
|
def can_play_stadium(self, rules: RulesConfig) -> bool:
|
|
"""Check if this player can play a Stadium card.
|
|
|
|
Compares the number of Stadiums played this turn against the
|
|
rules.trainer.stadiums_per_turn limit.
|
|
|
|
Args:
|
|
rules: The RulesConfig governing this game.
|
|
|
|
Returns:
|
|
True if more Stadiums can be played this turn.
|
|
"""
|
|
return self.stadiums_played_this_turn < rules.trainer.stadiums_per_turn
|
|
|
|
def can_play_item(self, rules: RulesConfig) -> bool:
|
|
"""Check if this player can play an Item card.
|
|
|
|
Compares the number of Items played this turn against the
|
|
rules.trainer.items_per_turn limit. If items_per_turn is None,
|
|
unlimited Items can be played.
|
|
|
|
Args:
|
|
rules: The RulesConfig governing this game.
|
|
|
|
Returns:
|
|
True if more Items can be played this turn.
|
|
"""
|
|
if rules.trainer.items_per_turn is None:
|
|
return True # Unlimited items
|
|
return self.items_played_this_turn < rules.trainer.items_per_turn
|
|
|
|
def can_retreat(self, rules: RulesConfig) -> bool:
|
|
"""Check if this player can retreat based on rules and turn state.
|
|
|
|
Compares the number of retreats performed this turn against the
|
|
rules.retreat.retreats_per_turn limit.
|
|
|
|
Args:
|
|
rules: The RulesConfig governing this game.
|
|
|
|
Returns:
|
|
True if more retreats are allowed this turn.
|
|
"""
|
|
return self.retreats_this_turn < rules.retreat.retreats_per_turn
|
|
|
|
def bench_space_available(self, rules: RulesConfig) -> int:
|
|
"""Return how many more Pokemon can be placed on the bench."""
|
|
return max(0, rules.bench.max_size - len(self.bench))
|
|
|
|
def can_bench_pokemon(self, rules: RulesConfig) -> bool:
|
|
"""Check if there's room on the bench for another Pokemon."""
|
|
return self.bench_space_available(rules) > 0
|
|
|
|
def reset_turn_state(self) -> None:
|
|
"""Reset all per-turn counters. Called at the start of each turn.
|
|
|
|
Resets all action counters (energy attachments, trainer plays, retreats)
|
|
to zero, and resets ability usage on all Pokemon in play.
|
|
"""
|
|
self.energy_attachments_this_turn = 0
|
|
self.supporters_played_this_turn = 0
|
|
self.stadiums_played_this_turn = 0
|
|
self.items_played_this_turn = 0
|
|
self.retreats_this_turn = 0
|
|
|
|
# Also reset ability usage on all Pokemon in play
|
|
for pokemon in self.get_all_pokemon_in_play():
|
|
pokemon.reset_turn_state()
|
|
|
|
|
|
class GameState(BaseModel):
|
|
"""Complete state of a game in progress.
|
|
|
|
This is the top-level model containing everything needed to represent
|
|
and resume a game. It's designed to be self-contained - the card_registry
|
|
holds all CardDefinitions used in this game, so the state can be
|
|
serialized and loaded without external dependencies.
|
|
|
|
Attributes:
|
|
game_id: Unique identifier for this game.
|
|
rules: The RulesConfig governing this game.
|
|
card_registry: Mapping of card definition IDs to CardDefinition objects.
|
|
players: Mapping of player IDs to PlayerState objects.
|
|
current_player_id: ID of the player whose turn it is.
|
|
turn_number: Current turn number (starts at 1).
|
|
phase: Current phase of the turn.
|
|
winner_id: ID of the winning player, if game has ended.
|
|
end_reason: Why the game ended, if it has ended.
|
|
stadium_in_play: The current Stadium card in play, if any.
|
|
stadium_owner_id: Player ID of who played the current stadium (for discard).
|
|
turn_order: List of player IDs in turn order.
|
|
first_turn_completed: Whether the very first turn of the game is done.
|
|
forced_actions: Queue of ForcedAction items that must be completed before game proceeds.
|
|
Actions are processed in FIFO order (first added = first to resolve).
|
|
action_log: Log of actions taken (for replays/debugging).
|
|
"""
|
|
|
|
game_id: str
|
|
rules: RulesConfig = Field(default_factory=RulesConfig)
|
|
card_registry: dict[str, CardDefinition] = Field(default_factory=dict)
|
|
players: dict[str, PlayerState] = Field(default_factory=dict)
|
|
|
|
# Turn tracking
|
|
current_player_id: str = ""
|
|
turn_number: int = 0
|
|
phase: TurnPhase = TurnPhase.SETUP
|
|
|
|
# Game end state
|
|
winner_id: str | None = None
|
|
end_reason: GameEndReason | None = None
|
|
|
|
# Shared game state
|
|
stadium_in_play: CardInstance | None = None
|
|
stadium_owner_id: str | None = None
|
|
|
|
# Turn order (for 2+ player support)
|
|
turn_order: list[str] = Field(default_factory=list)
|
|
|
|
# First turn tracking
|
|
first_turn_completed: bool = False
|
|
|
|
# Forced actions queue (e.g., select new active after KO, select prizes)
|
|
# Actions are processed in FIFO order - first added is first to resolve
|
|
forced_actions: list[ForcedAction] = Field(default_factory=list)
|
|
|
|
# Optional action log for replays
|
|
action_log: list[dict[str, Any]] = Field(default_factory=list)
|
|
|
|
def get_current_player(self) -> PlayerState:
|
|
"""Get the PlayerState for the current player.
|
|
|
|
Raises:
|
|
KeyError: If current_player_id is not in players dict.
|
|
"""
|
|
return self.players[self.current_player_id]
|
|
|
|
def get_opponent_id(self, player_id: str) -> str:
|
|
"""Get the opponent's player ID (assumes 2-player game).
|
|
|
|
Args:
|
|
player_id: The player whose opponent we want.
|
|
|
|
Returns:
|
|
The opponent's player ID.
|
|
|
|
Raises:
|
|
ValueError: If player_id is not in the game or game has != 2 players.
|
|
"""
|
|
if len(self.players) != 2:
|
|
raise ValueError("get_opponent_id only works for 2-player games")
|
|
if player_id not in self.players:
|
|
raise ValueError(f"Player {player_id} not found in game")
|
|
for pid in self.players:
|
|
if pid != player_id:
|
|
return pid
|
|
# This should be unreachable with 2 players where one is player_id
|
|
raise ValueError(f"Could not find opponent for {player_id}")
|
|
|
|
def get_opponent(self, player_id: str) -> PlayerState:
|
|
"""Get the PlayerState for a player's opponent (assumes 2-player game)."""
|
|
return self.players[self.get_opponent_id(player_id)]
|
|
|
|
def get_card_definition(self, definition_id: str) -> CardDefinition | None:
|
|
"""Look up a CardDefinition from the registry."""
|
|
return self.card_registry.get(definition_id)
|
|
|
|
def is_player_turn(self, player_id: str) -> bool:
|
|
"""Check if it's the specified player's turn."""
|
|
return self.current_player_id == player_id
|
|
|
|
def is_first_turn(self) -> bool:
|
|
"""Check if this is the very first turn of the game.
|
|
|
|
The first turn is turn 1, before first_turn_completed is set True.
|
|
"""
|
|
return self.turn_number == 1 and not self.first_turn_completed
|
|
|
|
def is_game_over(self) -> bool:
|
|
"""Check if the game has ended."""
|
|
return self.winner_id is not None
|
|
|
|
# =========================================================================
|
|
# Forced Action Queue Management
|
|
# =========================================================================
|
|
|
|
def has_forced_action(self) -> bool:
|
|
"""Check if there are any pending forced actions."""
|
|
return len(self.forced_actions) > 0
|
|
|
|
def get_current_forced_action(self) -> ForcedAction | None:
|
|
"""Get the current (first) forced action without removing it.
|
|
|
|
Returns:
|
|
The first forced action in the queue, or None if queue is empty.
|
|
"""
|
|
if self.forced_actions:
|
|
return self.forced_actions[0]
|
|
return None
|
|
|
|
def add_forced_action(self, action: ForcedAction) -> None:
|
|
"""Add a forced action to the queue.
|
|
|
|
Actions are processed in FIFO order, so this adds to the end.
|
|
|
|
Args:
|
|
action: The ForcedAction to add.
|
|
"""
|
|
self.forced_actions.append(action)
|
|
|
|
def pop_forced_action(self) -> ForcedAction | None:
|
|
"""Remove and return the current (first) forced action.
|
|
|
|
Returns:
|
|
The first forced action that was removed, or None if queue was empty.
|
|
"""
|
|
if self.forced_actions:
|
|
return self.forced_actions.pop(0)
|
|
return None
|
|
|
|
def clear_forced_actions(self) -> None:
|
|
"""Clear all pending forced actions."""
|
|
self.forced_actions.clear()
|
|
|
|
def get_player_count(self) -> int:
|
|
"""Return the number of players in this game."""
|
|
return len(self.players)
|
|
|
|
def advance_turn(self) -> None:
|
|
"""Advance to the next player's turn.
|
|
|
|
This handles:
|
|
- Setting first_turn_completed after turn 1
|
|
- Incrementing turn_number
|
|
- Switching current_player_id to the next player
|
|
- Resetting the new player's turn flags
|
|
- Setting phase to DRAW
|
|
"""
|
|
# Mark first turn as completed if we're ending turn 1
|
|
if self.turn_number == 1:
|
|
self.first_turn_completed = True
|
|
|
|
# Find next player in turn order
|
|
if self.turn_order:
|
|
current_index = self.turn_order.index(self.current_player_id)
|
|
next_index = (current_index + 1) % len(self.turn_order)
|
|
self.current_player_id = self.turn_order[next_index]
|
|
|
|
# Only increment turn number when we wrap back to first player
|
|
if next_index == 0:
|
|
self.turn_number += 1
|
|
else:
|
|
# Fallback for games without explicit turn order
|
|
self.turn_number += 1
|
|
|
|
# Reset the new player's turn state
|
|
self.get_current_player().reset_turn_state()
|
|
|
|
# Start at draw phase
|
|
self.phase = TurnPhase.DRAW
|
|
|
|
def set_winner(self, player_id: str, reason: GameEndReason) -> None:
|
|
"""Set the winner and end the game.
|
|
|
|
Args:
|
|
player_id: The winning player's ID.
|
|
reason: Why the game ended.
|
|
"""
|
|
self.winner_id = player_id
|
|
self.end_reason = reason
|
|
|
|
def log_action(self, action: dict[str, Any]) -> None:
|
|
"""Add an action to the action log for replay support."""
|
|
self.action_log.append(action)
|
|
|
|
def get_all_cards_in_play(self) -> list[CardInstance]:
|
|
"""Get all cards currently in play (all players' active and bench)."""
|
|
cards = []
|
|
for player in self.players.values():
|
|
cards.extend(player.get_all_pokemon_in_play())
|
|
return cards
|
|
|
|
def find_card_instance(self, instance_id: str) -> tuple[CardInstance | None, str | None]:
|
|
"""Find a CardInstance anywhere in the game.
|
|
|
|
Searches all zones of all players for the card, including cards attached
|
|
to Pokemon (energy, tools) and cards underneath evolved Pokemon.
|
|
|
|
Args:
|
|
instance_id: The instance_id to search for.
|
|
|
|
Returns:
|
|
Tuple of (CardInstance, zone_type) if found, (None, None) if not.
|
|
Zone types include standard zones plus:
|
|
- "attached_energy": Energy attached to a Pokemon in play
|
|
- "attached_tools": Tools attached to a Pokemon in play
|
|
- "cards_underneath": Cards in an evolution stack
|
|
"""
|
|
for player in self.players.values():
|
|
# Search standard zones
|
|
for zone_name in [
|
|
"deck",
|
|
"hand",
|
|
"active",
|
|
"bench",
|
|
"discard",
|
|
"prizes",
|
|
"energy_deck",
|
|
"energy_zone",
|
|
]:
|
|
zone: Zone = getattr(player, zone_name)
|
|
card = zone.get(instance_id)
|
|
if card:
|
|
return card, zone_name
|
|
|
|
# Search cards attached to Pokemon in play (active and bench)
|
|
for pokemon in player.get_all_pokemon_in_play():
|
|
# Check attached energy
|
|
for energy in pokemon.attached_energy:
|
|
if energy.instance_id == instance_id:
|
|
return energy, "attached_energy"
|
|
# Check attached tools
|
|
for tool in pokemon.attached_tools:
|
|
if tool.instance_id == instance_id:
|
|
return tool, "attached_tools"
|
|
# Check evolution stack (cards underneath)
|
|
for card in pokemon.cards_underneath:
|
|
if card.instance_id == instance_id:
|
|
return card, "cards_underneath"
|
|
|
|
return None, None
|