2b8fac405f
Major refactor to properly track attached cards and evolution history: Model Changes (app/core/models/card.py): - Change attached_energy from list[str] to list[CardInstance] - Change attached_tools from list[str] to list[CardInstance] - Add cards_underneath field for evolution stack tracking - Update attach_energy/detach_energy to work with CardInstance - Add attach_tool/detach_tool methods - Add get_all_attached_cards helper Engine Changes (app/core/engine.py): - _execute_attach_energy: Pass full CardInstance to attach_energy - _execute_evolve: Build evolution stack, transfer attachments, clear status - _execute_retreat: Detached energy goes to discard pile - Fix: Evolution now clears status conditions (Pokemon TCG standard) Game State (app/core/models/game_state.py): - find_card_instance now searches attached_energy, attached_tools, cards_underneath Turn Manager (app/core/turn_manager.py): - process_knockout: Discard all attached energy, tools, and evolution stack Effects (app/core/effects/handlers.py): - discard_energy: Find owner's discard pile and move detached energy there - NEW devolve effect: Remove evolution stages with configurable destination - Fix: Use EffectType.SPECIAL instead of non-existent EffectType.ZONE Rules Validator (app/core/rules_validator.py): - Update energy type checking to iterate CardInstance objects Tests: - Update existing tests for new CardInstance-based energy attachment - NEW test_evolution_stack.py with 28 comprehensive tests covering: - Evolution stack building (Basic -> Stage 1 -> Stage 2) - Energy/tool transfer and damage carryover on evolution - Devolve effect (single/multi stage, hand/discard destination, KO check) - Knockout processing with all attachments going to discard - find_card_instance for attached cards and evolution stack All 765 tests pass.
593 lines
22 KiB
Python
593 lines
22 KiB
Python
"""Turn and phase management for the Mantimon TCG game engine.
|
|
|
|
This module implements the turn/phase state machine, handling:
|
|
- Phase transitions within a turn (DRAW -> MAIN -> ATTACK -> END)
|
|
- Turn start/end processing
|
|
- Between-turn effects (poison/burn damage, status recovery)
|
|
- Automatic energy from energy deck (Pokemon Pocket style)
|
|
|
|
The TurnManager is stateless - it operates on a GameState and returns the
|
|
modified state. This allows for deterministic replays and testing.
|
|
|
|
Turn Structure:
|
|
1. SETUP: Initial game setup (handled separately by game initialization)
|
|
2. DRAW: Draw a card from deck (auto-advances to MAIN)
|
|
3. MAIN: Play cards, attach energy, evolve, use abilities, retreat
|
|
4. ATTACK: Declare and resolve attack (optional, can skip to END)
|
|
5. END: Apply end-of-turn effects, check knockouts, advance turn
|
|
|
|
Valid Phase Transitions:
|
|
- SETUP -> DRAW (game start)
|
|
- DRAW -> MAIN (automatic after draw)
|
|
- MAIN -> ATTACK (player choice)
|
|
- MAIN -> END (player skips attack)
|
|
- ATTACK -> END (automatic after attack resolution)
|
|
- END -> (next player's DRAW)
|
|
|
|
Usage:
|
|
from app.core.turn_manager import TurnManager
|
|
from app.core.rng import SeededRandom
|
|
|
|
manager = TurnManager()
|
|
rng = SeededRandom(seed=42)
|
|
|
|
# Start a new turn
|
|
result = manager.start_turn(game, rng)
|
|
if result.win_result:
|
|
# Player can't draw - opponent wins
|
|
game = apply_win_result(game, result.win_result)
|
|
|
|
# Advance through phases
|
|
manager.advance_to_main(game) # After draw phase actions
|
|
manager.advance_to_attack(game) # Ready to attack
|
|
manager.end_turn(game, rng) # Process end of turn
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import TYPE_CHECKING
|
|
|
|
from pydantic import BaseModel
|
|
|
|
from app.core.models.enums import GameEndReason, StatusCondition, TurnPhase
|
|
from app.core.win_conditions import WinResult, check_cannot_draw
|
|
|
|
if TYPE_CHECKING:
|
|
from app.core.models.game_state import GameState
|
|
from app.core.rng import RandomProvider
|
|
|
|
|
|
class TurnStartResult(BaseModel):
|
|
"""Result of starting a turn.
|
|
|
|
Attributes:
|
|
success: Whether the turn started successfully.
|
|
drew_card: Whether a card was drawn (False if deck empty or first turn no-draw).
|
|
energy_flipped: Whether energy was flipped from energy deck.
|
|
win_result: If set, the game ended due to deck-out.
|
|
message: Description of what happened.
|
|
"""
|
|
|
|
success: bool
|
|
drew_card: bool = False
|
|
energy_flipped: bool = False
|
|
win_result: WinResult | None = None
|
|
message: str = ""
|
|
|
|
|
|
class TurnEndResult(BaseModel):
|
|
"""Result of ending a turn.
|
|
|
|
Attributes:
|
|
success: Whether the turn ended successfully.
|
|
between_turn_damage: Dict of instance_id -> damage applied by status.
|
|
status_removed: Dict of instance_id -> list of status conditions removed.
|
|
knockouts: List of instance_ids of Pokemon knocked out by status damage.
|
|
win_result: If set, the game ended due to knockouts or other condition.
|
|
message: Description of what happened.
|
|
"""
|
|
|
|
success: bool
|
|
between_turn_damage: dict[str, int] = {}
|
|
status_removed: dict[str, list[StatusCondition]] = {}
|
|
knockouts: list[str] = []
|
|
win_result: WinResult | None = None
|
|
message: str = ""
|
|
|
|
|
|
class PhaseTransitionError(Exception):
|
|
"""Raised when an invalid phase transition is attempted."""
|
|
|
|
pass
|
|
|
|
|
|
# Valid phase transitions as a graph
|
|
VALID_TRANSITIONS: dict[TurnPhase, list[TurnPhase]] = {
|
|
TurnPhase.SETUP: [TurnPhase.DRAW],
|
|
TurnPhase.DRAW: [TurnPhase.MAIN],
|
|
TurnPhase.MAIN: [TurnPhase.ATTACK, TurnPhase.END],
|
|
TurnPhase.ATTACK: [TurnPhase.END],
|
|
TurnPhase.END: [TurnPhase.DRAW], # Next player's turn starts at DRAW
|
|
}
|
|
|
|
|
|
class TurnManager:
|
|
"""Manages turn and phase transitions for a game.
|
|
|
|
The TurnManager is stateless and operates on GameState objects.
|
|
All methods that modify state do so in-place on the provided GameState.
|
|
|
|
The manager handles:
|
|
- Phase advancement within a turn
|
|
- Turn start processing (draw, energy flip, counter resets)
|
|
- Turn end processing (status damage, recovery checks, turn advancement)
|
|
- Between-turn effects (poison, burn, sleep/paralysis recovery)
|
|
|
|
Example:
|
|
manager = TurnManager()
|
|
rng = SeededRandom(seed=42)
|
|
|
|
# Start player's turn
|
|
result = manager.start_turn(game, rng)
|
|
|
|
# Player takes actions during MAIN phase...
|
|
|
|
# End turn and process effects
|
|
result = manager.end_turn(game, rng)
|
|
"""
|
|
|
|
def can_transition(self, from_phase: TurnPhase, to_phase: TurnPhase) -> bool:
|
|
"""Check if a phase transition is valid.
|
|
|
|
Args:
|
|
from_phase: Current phase.
|
|
to_phase: Target phase.
|
|
|
|
Returns:
|
|
True if the transition is allowed.
|
|
"""
|
|
valid_targets = VALID_TRANSITIONS.get(from_phase, [])
|
|
return to_phase in valid_targets
|
|
|
|
def get_valid_transitions(self, phase: TurnPhase) -> list[TurnPhase]:
|
|
"""Get list of valid phases to transition to from the current phase.
|
|
|
|
Args:
|
|
phase: Current phase.
|
|
|
|
Returns:
|
|
List of phases that can be transitioned to.
|
|
"""
|
|
return VALID_TRANSITIONS.get(phase, [])
|
|
|
|
def _set_phase(self, game: GameState, new_phase: TurnPhase) -> None:
|
|
"""Set the game phase, validating the transition.
|
|
|
|
Args:
|
|
game: GameState to modify.
|
|
new_phase: Phase to transition to.
|
|
|
|
Raises:
|
|
PhaseTransitionError: If the transition is invalid.
|
|
"""
|
|
if not self.can_transition(game.phase, new_phase):
|
|
raise PhaseTransitionError(
|
|
f"Invalid phase transition: {game.phase.value} -> {new_phase.value}. "
|
|
f"Valid transitions: {[p.value for p in VALID_TRANSITIONS.get(game.phase, [])]}"
|
|
)
|
|
game.phase = new_phase
|
|
|
|
def start_turn(self, game: GameState, rng: RandomProvider) -> TurnStartResult:
|
|
"""Start the current player's turn.
|
|
|
|
This method:
|
|
1. Resets the player's per-turn counters
|
|
2. Checks for deck-out (cannot draw) win condition
|
|
3. Draws a card (if rules allow)
|
|
4. Flips energy from energy deck (if enabled)
|
|
5. Advances phase to MAIN
|
|
|
|
Call this at the beginning of each player's turn.
|
|
|
|
Args:
|
|
game: GameState to modify.
|
|
rng: RandomProvider for any random effects.
|
|
|
|
Returns:
|
|
TurnStartResult with details of turn start.
|
|
"""
|
|
player = game.get_current_player()
|
|
rules = game.rules
|
|
messages = []
|
|
|
|
# Reset per-turn state
|
|
player.reset_turn_state()
|
|
|
|
# Set phase to DRAW
|
|
game.phase = TurnPhase.DRAW
|
|
|
|
# Check deck-out before drawing (if enabled)
|
|
if rules.win_conditions.cannot_draw:
|
|
win_result = check_cannot_draw(game)
|
|
if win_result is not None:
|
|
return TurnStartResult(
|
|
success=False,
|
|
win_result=win_result,
|
|
message=f"Player {player.player_id} cannot draw - deck is empty",
|
|
)
|
|
|
|
# Determine if we should draw
|
|
is_first_turn = game.is_first_turn()
|
|
should_draw = rules.first_turn.can_draw or not is_first_turn
|
|
|
|
drew_card = False
|
|
if should_draw and len(player.deck) > 0:
|
|
card = player.deck.draw()
|
|
if card:
|
|
player.hand.add(card)
|
|
drew_card = True
|
|
messages.append("Drew a card")
|
|
elif not should_draw:
|
|
messages.append("First turn - no draw")
|
|
|
|
# Flip energy from energy deck (Pokemon Pocket style)
|
|
energy_flipped = False
|
|
if rules.energy.auto_flip_from_deck and len(player.energy_deck) > 0:
|
|
# First turn energy attach restriction check
|
|
can_attach_first_turn = rules.first_turn.can_attach_energy or not is_first_turn
|
|
if can_attach_first_turn:
|
|
energy = player.energy_deck.draw()
|
|
if energy:
|
|
player.energy_zone.add(energy)
|
|
energy_flipped = True
|
|
messages.append("Flipped energy to zone")
|
|
|
|
# Advance to MAIN phase
|
|
game.phase = TurnPhase.MAIN
|
|
|
|
return TurnStartResult(
|
|
success=True,
|
|
drew_card=drew_card,
|
|
energy_flipped=energy_flipped,
|
|
message="; ".join(messages) if messages else "Turn started",
|
|
)
|
|
|
|
def advance_to_main(self, game: GameState) -> None:
|
|
"""Advance from DRAW phase to MAIN phase.
|
|
|
|
This is typically called after start_turn() if you need explicit control,
|
|
but start_turn() already advances to MAIN automatically.
|
|
|
|
Args:
|
|
game: GameState to modify.
|
|
|
|
Raises:
|
|
PhaseTransitionError: If not currently in DRAW phase.
|
|
"""
|
|
self._set_phase(game, TurnPhase.MAIN)
|
|
|
|
def advance_to_attack(self, game: GameState) -> None:
|
|
"""Advance from MAIN phase to ATTACK phase.
|
|
|
|
Call this when the player declares an attack.
|
|
|
|
Args:
|
|
game: GameState to modify.
|
|
|
|
Raises:
|
|
PhaseTransitionError: If not currently in MAIN phase.
|
|
"""
|
|
self._set_phase(game, TurnPhase.ATTACK)
|
|
|
|
def advance_to_end(self, game: GameState) -> None:
|
|
"""Advance to END phase.
|
|
|
|
Can be called from MAIN (skipping attack) or ATTACK (after attack resolution).
|
|
|
|
Args:
|
|
game: GameState to modify.
|
|
|
|
Raises:
|
|
PhaseTransitionError: If transition is not valid from current phase.
|
|
"""
|
|
self._set_phase(game, TurnPhase.END)
|
|
|
|
def skip_attack(self, game: GameState) -> None:
|
|
"""Skip the attack phase and go directly to END.
|
|
|
|
This is an alias for advance_to_end() when called from MAIN phase.
|
|
|
|
Args:
|
|
game: GameState to modify.
|
|
|
|
Raises:
|
|
PhaseTransitionError: If not currently in MAIN phase.
|
|
"""
|
|
if game.phase != TurnPhase.MAIN:
|
|
raise PhaseTransitionError(
|
|
f"Can only skip attack from MAIN phase, currently in {game.phase.value}"
|
|
)
|
|
self._set_phase(game, TurnPhase.END)
|
|
|
|
def end_turn(self, game: GameState, rng: RandomProvider) -> TurnEndResult:
|
|
"""End the current player's turn and apply between-turn effects.
|
|
|
|
This method:
|
|
1. Applies poison/burn damage to the current player's active Pokemon
|
|
2. Checks for status recovery (burn flip, sleep flip)
|
|
3. Removes paralysis (wears off after one turn)
|
|
4. Processes knockouts from status damage (moves Pokemon to discard, awards points)
|
|
5. Checks win conditions (after knockout processing)
|
|
6. Advances to the next player's turn
|
|
|
|
Call this when the current player's turn should end.
|
|
|
|
Args:
|
|
game: GameState to modify.
|
|
rng: RandomProvider for coin flips.
|
|
|
|
Returns:
|
|
TurnEndResult with details of what happened.
|
|
"""
|
|
# Ensure we're in END phase
|
|
if game.phase != TurnPhase.END:
|
|
self.advance_to_end(game)
|
|
|
|
player = game.get_current_player()
|
|
rules = game.rules
|
|
|
|
between_turn_damage: dict[str, int] = {}
|
|
status_removed: dict[str, list[StatusCondition]] = {}
|
|
knockouts: list[str] = []
|
|
messages: list[str] = []
|
|
|
|
# Process status effects on active Pokemon
|
|
active = player.get_active_pokemon()
|
|
if active:
|
|
# Apply poison damage
|
|
if StatusCondition.POISONED in active.status_conditions:
|
|
damage = rules.status.poison_damage
|
|
active.damage += damage
|
|
between_turn_damage[active.instance_id] = (
|
|
between_turn_damage.get(active.instance_id, 0) + damage
|
|
)
|
|
messages.append(f"Poison dealt {damage} damage")
|
|
|
|
# Apply burn damage and check for recovery
|
|
if StatusCondition.BURNED in active.status_conditions:
|
|
damage = rules.status.burn_damage
|
|
active.damage += damage
|
|
between_turn_damage[active.instance_id] = (
|
|
between_turn_damage.get(active.instance_id, 0) + damage
|
|
)
|
|
messages.append(f"Burn dealt {damage} damage")
|
|
|
|
# Flip to potentially remove burn
|
|
if rules.status.burn_flip_to_remove:
|
|
if rng.coin_flip():
|
|
active.remove_status(StatusCondition.BURNED)
|
|
if active.instance_id not in status_removed:
|
|
status_removed[active.instance_id] = []
|
|
status_removed[active.instance_id].append(StatusCondition.BURNED)
|
|
messages.append("Burn removed (heads)")
|
|
else:
|
|
messages.append("Burn remains (tails)")
|
|
|
|
# Check for sleep recovery
|
|
if (
|
|
StatusCondition.ASLEEP in active.status_conditions
|
|
and rules.status.sleep_flip_to_wake
|
|
):
|
|
if rng.coin_flip():
|
|
active.remove_status(StatusCondition.ASLEEP)
|
|
if active.instance_id not in status_removed:
|
|
status_removed[active.instance_id] = []
|
|
status_removed[active.instance_id].append(StatusCondition.ASLEEP)
|
|
messages.append("Woke up from sleep (heads)")
|
|
else:
|
|
messages.append("Still asleep (tails)")
|
|
|
|
# Paralysis wears off at end of turn
|
|
if StatusCondition.PARALYZED in active.status_conditions:
|
|
active.remove_status(StatusCondition.PARALYZED)
|
|
if active.instance_id not in status_removed:
|
|
status_removed[active.instance_id] = []
|
|
status_removed[active.instance_id].append(StatusCondition.PARALYZED)
|
|
messages.append("Paralysis wore off")
|
|
|
|
# Check for knockout from status damage
|
|
card_def = game.get_card_definition(active.definition_id)
|
|
if card_def and card_def.hp and active.is_knocked_out(card_def.hp):
|
|
knockouts.append(active.instance_id)
|
|
messages.append(f"{card_def.name} knocked out by status damage!")
|
|
|
|
# Process knockouts BEFORE checking win conditions
|
|
# The opponent (who will be the next player) scores points for status KOs
|
|
win_result = None
|
|
opponent_id = game.get_opponent_id(player.player_id)
|
|
|
|
for knocked_out_id in knockouts:
|
|
ko_result = self.process_knockout(game, knocked_out_id, opponent_id)
|
|
if ko_result:
|
|
win_result = ko_result
|
|
break # Game ended
|
|
|
|
# Advance to next player's turn
|
|
game.advance_turn()
|
|
|
|
return TurnEndResult(
|
|
success=True,
|
|
between_turn_damage=between_turn_damage,
|
|
status_removed=status_removed,
|
|
knockouts=knockouts,
|
|
win_result=win_result,
|
|
message="; ".join(messages) if messages else "Turn ended",
|
|
)
|
|
|
|
def process_knockout(
|
|
self, game: GameState, knocked_out_id: str, opponent_id: str
|
|
) -> WinResult | None:
|
|
"""Process a Pokemon knockout and check for win conditions.
|
|
|
|
This method:
|
|
1. Moves the knocked out Pokemon to discard
|
|
2. Awards points to the opponent (based on variant)
|
|
3. Checks for win by points
|
|
4. Sets up forced action if player needs new active
|
|
|
|
Note: This should be called by the game engine after damage resolution.
|
|
|
|
Args:
|
|
game: GameState to modify.
|
|
knocked_out_id: Instance ID of the knocked out Pokemon.
|
|
opponent_id: Player ID of the opponent (who scores points).
|
|
|
|
Returns:
|
|
WinResult if the knockout ends the game, None otherwise.
|
|
"""
|
|
# Find the knockout Pokemon and its owner
|
|
for player_id, player in game.players.items():
|
|
card = player.active.get(knocked_out_id)
|
|
if card:
|
|
# Found in active - remove from zone
|
|
player.active.remove(knocked_out_id)
|
|
|
|
# TODO: Future hook point - pre_knockout_discard event
|
|
# This would allow effects to redirect energy/tools elsewhere
|
|
|
|
# Discard all attached energy
|
|
for energy in card.attached_energy:
|
|
player.discard.add(energy)
|
|
card.attached_energy = []
|
|
|
|
# Discard all attached tools
|
|
for tool in card.attached_tools:
|
|
player.discard.add(tool)
|
|
card.attached_tools = []
|
|
|
|
# Discard entire evolution stack (cards underneath)
|
|
for underneath in card.cards_underneath:
|
|
player.discard.add(underneath)
|
|
card.cards_underneath = []
|
|
|
|
# Finally discard the Pokemon itself
|
|
player.discard.add(card)
|
|
|
|
# Award points to opponent
|
|
opponent = game.players[opponent_id]
|
|
card_def = game.get_card_definition(card.definition_id)
|
|
if card_def and card_def.variant:
|
|
points = game.rules.prizes.points_for_knockout(card_def.variant)
|
|
opponent.score += points
|
|
|
|
# Check for win by points
|
|
if opponent.score >= game.rules.prizes.count:
|
|
loser_id = game.get_opponent_id(opponent_id)
|
|
return WinResult(
|
|
winner_id=opponent_id,
|
|
loser_id=loser_id,
|
|
end_reason=GameEndReason.PRIZES_TAKEN,
|
|
reason=f"Player {opponent_id} scored {opponent.score} points",
|
|
)
|
|
|
|
# Check if owner has no Pokemon left
|
|
if not player.has_pokemon_in_play():
|
|
return WinResult(
|
|
winner_id=opponent_id,
|
|
loser_id=player_id,
|
|
end_reason=GameEndReason.NO_POKEMON,
|
|
reason=f"Player {player_id} has no Pokemon in play",
|
|
)
|
|
|
|
# Set up forced action to select new active
|
|
if player.has_benched_pokemon() and not player.has_active_pokemon():
|
|
from app.core.models.game_state import ForcedAction
|
|
|
|
game.forced_action = ForcedAction(
|
|
player_id=player_id,
|
|
action_type="select_active",
|
|
reason="Your active Pokemon was knocked out. Select a new active Pokemon.",
|
|
)
|
|
|
|
return None
|
|
|
|
# Check bench too (for bench damage knockouts)
|
|
card = player.bench.get(knocked_out_id)
|
|
if card:
|
|
# Remove from bench
|
|
player.bench.remove(knocked_out_id)
|
|
|
|
# TODO: Future hook point - pre_knockout_discard event
|
|
|
|
# Discard all attached energy
|
|
for energy in card.attached_energy:
|
|
player.discard.add(energy)
|
|
card.attached_energy = []
|
|
|
|
# Discard all attached tools
|
|
for tool in card.attached_tools:
|
|
player.discard.add(tool)
|
|
card.attached_tools = []
|
|
|
|
# Discard entire evolution stack
|
|
for underneath in card.cards_underneath:
|
|
player.discard.add(underneath)
|
|
card.cards_underneath = []
|
|
|
|
# Discard the Pokemon
|
|
player.discard.add(card)
|
|
|
|
# Award points
|
|
opponent = game.players[opponent_id]
|
|
card_def = game.get_card_definition(card.definition_id)
|
|
if card_def and card_def.variant:
|
|
points = game.rules.prizes.points_for_knockout(card_def.variant)
|
|
opponent.score += points
|
|
|
|
# Check for win by points
|
|
if opponent.score >= game.rules.prizes.count:
|
|
loser_id = game.get_opponent_id(opponent_id)
|
|
return WinResult(
|
|
winner_id=opponent_id,
|
|
loser_id=loser_id,
|
|
end_reason=GameEndReason.PRIZES_TAKEN,
|
|
reason=f"Player {opponent_id} scored {opponent.score} points",
|
|
)
|
|
|
|
# Check if owner has no Pokemon left
|
|
if not player.has_pokemon_in_play():
|
|
return WinResult(
|
|
winner_id=opponent_id,
|
|
loser_id=player_id,
|
|
end_reason=GameEndReason.NO_POKEMON,
|
|
reason=f"Player {player_id} has no Pokemon in play",
|
|
)
|
|
|
|
return None
|
|
|
|
return None
|
|
|
|
def check_turn_limit(self, game: GameState) -> WinResult | None:
|
|
"""Check if the turn limit has been reached.
|
|
|
|
Should be called at the start of each turn (before start_turn).
|
|
If the turn limit is reached, returns a WinResult based on score.
|
|
|
|
Args:
|
|
game: GameState to check.
|
|
|
|
Returns:
|
|
WinResult if turn limit reached, None otherwise.
|
|
"""
|
|
rules = game.rules
|
|
if not rules.win_conditions.turn_limit_enabled:
|
|
return None
|
|
|
|
if game.turn_number > rules.win_conditions.turn_limit:
|
|
# Turn limit reached - determine winner by score
|
|
from app.core.win_conditions import check_turn_limit
|
|
|
|
return check_turn_limit(game)
|
|
|
|
return None
|