""" Pydantic models for in-memory game state management. These models represent the active game state cached in memory for fast gameplay. They are separate from the SQLAlchemy database models (db_models.py) to optimize for different use cases: - game_models.py: Fast in-memory operations, type-safe validation, WebSocket serialization - db_models.py: Database persistence, relationships, audit trail Author: Claude Date: 2025-10-22 """ import logging from dataclasses import dataclass from typing import TYPE_CHECKING, Any, Optional from uuid import UUID from pydantic import BaseModel, ConfigDict, Field, field_validator from app.config.result_charts import PlayOutcome if TYPE_CHECKING: from app.models.player_models import PositionRating logger = logging.getLogger(f"{__name__}") # ============================================================================ # LINEUP STATE # ============================================================================ class LineupPlayerState(BaseModel): """ Represents a player in the game lineup. Contains both lineup-specific data (position, batting_order) and player data (name, image) loaded at game start for efficient access. Phase 3E-Main: Now includes position_rating for X-Check resolution. """ lineup_id: int card_id: int position: str batting_order: int | None = None is_active: bool = True is_starter: bool = True # Player data (loaded at game start from SBA/PD API) player_name: str | None = None player_image: str | None = None # Card image player_headshot: str | None = None # Headshot for UI circles # Phase 3E-Main: Position rating (loaded at game start for PD league) position_rating: Optional["PositionRating"] = None @field_validator("position") @classmethod def validate_position(cls, v: str) -> str: """Ensure position is valid""" valid_positions = ["P", "C", "1B", "2B", "3B", "SS", "LF", "CF", "RF", "DH"] if v not in valid_positions: raise ValueError(f"Position must be one of {valid_positions}") return v @field_validator("batting_order") @classmethod def validate_batting_order(cls, v: int | None) -> int | None: """Ensure batting order is 1-9 if provided""" if v is not None and (v < 1 or v > 9): raise ValueError("batting_order must be between 1 and 9") return v class TeamLineupState(BaseModel): """ Represents a team's active lineup in the game. Provides helper methods for common lineup queries. """ team_id: int players: list[LineupPlayerState] = Field(default_factory=list) def get_batting_order(self) -> list[LineupPlayerState]: """ Get players in batting order (1-9). Returns: List of players sorted by batting_order """ return sorted( [p for p in self.players if p.batting_order is not None], key=lambda x: x.batting_order or 0, # Type narrowing: filtered None above ) def get_pitcher(self) -> LineupPlayerState | None: """ Get the active pitcher for this team. Returns: Active pitcher or None if not found """ pitchers = [p for p in self.players if p.position == "P" and p.is_active] return pitchers[0] if pitchers else None def get_player_by_lineup_id(self, lineup_id: int) -> LineupPlayerState | None: """ Get player by lineup ID. Args: lineup_id: The lineup entry ID Returns: Player or None if not found """ for player in self.players: if player.lineup_id == lineup_id: return player return None def get_batter(self, batting_order_idx: int) -> LineupPlayerState | None: """ Get batter by batting order index (0-8). Args: batting_order_idx: Index in batting order (0 = leadoff, 8 = 9th batter) Returns: Player at that position in the order, or None """ order = self.get_batting_order() if 0 <= batting_order_idx < len(order): return order[batting_order_idx] return None def get_player_by_card_id(self, card_id: int) -> LineupPlayerState | None: """ Get player by card ID. Args: card_id: The card/player ID Returns: Player or None if not found """ for player in self.players: if player.card_id == card_id: return player return None # ============================================================================ # DECISION STATE # ============================================================================ class DefensiveDecision(BaseModel): """ Defensive team strategic decisions for a play. These decisions affect play outcomes (e.g., infield depth affects double play chances). """ infield_depth: str = "normal" # infield_in, normal, corners_in outfield_depth: str = "normal" # normal, shallow hold_runners: list[int] = Field(default_factory=list) # [1, 3] = hold 1st and 3rd @field_validator("infield_depth") @classmethod def validate_infield_depth(cls, v: str) -> str: """Validate infield depth""" valid = ["infield_in", "normal", "corners_in"] if v not in valid: raise ValueError(f"infield_depth must be one of {valid}") return v @field_validator("outfield_depth") @classmethod def validate_outfield_depth(cls, v: str) -> str: """Validate outfield depth""" valid = ["normal", "shallow"] if v not in valid: raise ValueError(f"outfield_depth must be one of {valid}") return v class OffensiveDecision(BaseModel): """ Offensive team strategic decisions for a play. These decisions affect baserunner actions. Session 2 Update (2025-01-14): Replaced approach field with action field. Valid actions: swing_away, steal, check_jump, hit_and_run, sac_bunt, squeeze_bunt When action="steal", steal_attempts must specify which bases to steal. """ # Specific action choice action: str = "swing_away" # Base stealing - only used when action="steal" steal_attempts: list[int] = Field( default_factory=list ) # [2] = steal second, [2, 3] = double steal @field_validator("action") @classmethod def validate_action(cls, v: str) -> str: """Validate action is one of the allowed values""" valid_actions = [ "swing_away", "steal", "check_jump", "hit_and_run", "sac_bunt", "squeeze_bunt", ] if v not in valid_actions: raise ValueError(f"action must be one of {valid_actions}") return v @field_validator("steal_attempts") @classmethod def validate_steal_attempts(cls, v: list[int]) -> list[int]: """Validate steal attempt bases""" for base in v: if base not in [2, 3, 4]: # 2nd, 3rd, home raise ValueError("steal_attempts must contain only bases 2, 3, or 4") return v class ManualOutcomeSubmission(BaseModel): """ Model for human players submitting play outcomes in manual mode. In manual mode (SBA + PD manual), players roll dice, read physical cards, and submit the outcome they see. The system then validates and processes. Usage: - Players submit via WebSocket after reading their physical card - Outcome is required (what the card shows) - Hit location is optional (only needed for certain outcomes) - System validates location is provided when required Example: ManualOutcomeSubmission( outcome=PlayOutcome.GROUNDBALL_C, hit_location='SS' ) """ outcome: PlayOutcome # PlayOutcome enum from result_charts hit_location: str | None = ( None # '1B', '2B', 'SS', '3B', 'LF', 'CF', 'RF', 'P', 'C' ) @field_validator("hit_location") @classmethod def validate_hit_location(cls, v: str | None) -> str | None: """Validate hit location is a valid position.""" if v is None: return v valid_locations = ["1B", "2B", "SS", "3B", "LF", "CF", "RF", "P", "C"] if v not in valid_locations: raise ValueError(f"hit_location must be one of {valid_locations}") return v # ============================================================================ # X-CHECK RESULT # ============================================================================ @dataclass class XCheckResult: """ Intermediate state for X-Check play resolution. Tracks all dice rolls, table lookups, and conversions from initial x-check through final outcome determination. Resolution Flow: 1. Roll 1d20 + 3d6 2. Look up base_result from defense table[d20][defender_range] 3. Apply SPD test if needed (base_result = 'SPD') 4. Apply G2#/G3# → SI2 conversion if conditions met 5. Look up error_result from error chart[error_rating][3d6] 6. Determine final_outcome (may be ERROR if out+error) Attributes: position: Position being checked (SS, LF, 3B, etc.) d20_roll: Defense range table row selector (1-20) d6_roll: Error chart lookup value (3-18, sum of 3d6) defender_range: Defender's range rating (1-5, adjusted for playing in) defender_error_rating: Defender's error rating (0-25) base_result: Initial result from defense table (G1, F2, SI2, SPD, etc.) converted_result: Result after SPD/G2#/G3# conversions (may equal base_result) error_result: Error type from error chart (NO, E1, E2, E3, RP) final_outcome: Final PlayOutcome after all conversions defender_id: Player ID of defender hit_type: Combined result string for Play.hit_type (e.g. 'single_2_plus_error_1') """ position: str d20_roll: int d6_roll: int defender_range: int defender_error_rating: int defender_id: int base_result: str converted_result: str error_result: str # 'NO', 'E1', 'E2', 'E3', 'RP' final_outcome: PlayOutcome hit_type: str # Optional: SPD test details if applicable spd_test_roll: int | None = None spd_test_target: int | None = None spd_test_passed: bool | None = None def to_dict(self) -> dict: """Convert to dict for WebSocket transmission.""" return { "position": self.position, "d20_roll": self.d20_roll, "d6_roll": self.d6_roll, "defender_range": self.defender_range, "defender_error_rating": self.defender_error_rating, "defender_id": self.defender_id, "base_result": self.base_result, "converted_result": self.converted_result, "error_result": self.error_result, "final_outcome": self.final_outcome.value, "hit_type": self.hit_type, "spd_test": { "roll": self.spd_test_roll, "target": self.spd_test_target, "passed": self.spd_test_passed, } if self.spd_test_roll else None, } # ============================================================================ # GAME STATE # ============================================================================ class GameState(BaseModel): """ Complete in-memory game state. This is the core state model representing an active game. It is optimized for fast in-memory operations during gameplay. Attributes: game_id: Unique game identifier league_id: League identifier ('sba' or 'pd') home_team_id: Home team ID (from league API) away_team_id: Away team ID (from league API) home_team_is_ai: Whether home team is AI-controlled away_team_is_ai: Whether away team is AI-controlled status: Game status (pending, active, paused, completed) inning: Current inning (1-9+) half: Current half ('top' or 'bottom') outs: Current outs (0-2) home_score: Home team score away_score: Away team score runners: List of runners currently on base away_team_batter_idx: Away team batting order position (0-8) home_team_batter_idx: Home team batting order position (0-8) current_batter: Snapshot - LineupPlayerState for current batter (required) current_pitcher: Snapshot - LineupPlayerState for current pitcher (optional) current_catcher: Snapshot - LineupPlayerState for current catcher (optional) current_on_base_code: Snapshot - bit field of occupied bases (1=1st, 2=2nd, 4=3rd) pending_decision: Type of decision awaiting ('defensive', 'offensive', 'result_selection') decisions_this_play: Accumulated decisions for current play play_count: Total plays so far last_play_result: Description of last play outcome """ game_id: UUID league_id: str # Teams home_team_id: int away_team_id: int home_team_is_ai: bool = False away_team_is_ai: bool = False # Creator (for demo/testing - creator can control home team) creator_discord_id: str | None = None # Resolution mode auto_mode: bool = ( False # True = auto-generate outcomes (PD only), False = manual submissions ) # Game rules (configurable per game) regulation_innings: int = Field(default=9, ge=1) # Standard 9, doubleheader 7, etc. outs_per_inning: int = Field(default=3, ge=1) # Standard 3 # Game state status: str = "pending" # pending, active, paused, completed inning: int = Field(default=1, ge=1) half: str = "top" # top or bottom outs: int = Field(default=0, ge=0, le=2) # Score home_score: int = Field(default=0, ge=0) away_score: int = Field(default=0, ge=0) # Runners (direct references matching DB structure) on_first: LineupPlayerState | None = None on_second: LineupPlayerState | None = None on_third: LineupPlayerState | None = None # Batting order tracking (per team) - indexes into batting order (0-8) away_team_batter_idx: int = Field(default=0, ge=0, le=8) home_team_batter_idx: int = Field(default=0, ge=0, le=8) # Current play snapshot (set by _prepare_next_play) # These capture the state BEFORE each play for accurate record-keeping # Changed to full objects for consistency with on_first/on_second/on_third current_batter: LineupPlayerState current_pitcher: LineupPlayerState | None = None current_catcher: LineupPlayerState | None = None current_on_base_code: int = Field( default=0, ge=0 ) # Bit field: 1=1st, 2=2nd, 4=3rd, 7=loaded # Decision tracking pending_decision: str | None = None # 'defensive', 'offensive', 'result_selection' decisions_this_play: dict[str, Any] = Field(default_factory=dict) # Phase 3: Enhanced decision workflow pending_defensive_decision: DefensiveDecision | None = None pending_offensive_decision: OffensiveDecision | None = None decision_phase: str = ( "idle" # idle, awaiting_defensive, awaiting_offensive, resolving, completed ) decision_deadline: str | None = None # ISO8601 timestamp for timeout # Manual mode support (Week 7 Task 7) pending_manual_roll: Any | None = ( None # AbRoll stored when dice rolled in manual mode ) # Play tracking play_count: int = Field(default=0, ge=0) last_play_result: str | None = None @field_validator("league_id") @classmethod def validate_league_id(cls, v: str) -> str: """Ensure league_id is valid""" valid_leagues = ["sba", "pd"] if v not in valid_leagues: raise ValueError(f"league_id must be one of {valid_leagues}") return v @field_validator("status") @classmethod def validate_status(cls, v: str) -> str: """Ensure status is valid""" valid_statuses = ["pending", "active", "paused", "completed"] if v not in valid_statuses: raise ValueError(f"status must be one of {valid_statuses}") return v @field_validator("half") @classmethod def validate_half(cls, v: str) -> str: """Ensure half is valid""" if v not in ["top", "bottom"]: raise ValueError("half must be 'top' or 'bottom'") return v @field_validator("pending_decision") @classmethod def validate_pending_decision(cls, v: str | None) -> str | None: """Ensure pending_decision is valid""" if v is not None: valid = ["defensive", "offensive", "result_selection", "substitution"] if v not in valid: raise ValueError(f"pending_decision must be one of {valid}") return v @field_validator("decision_phase") @classmethod def validate_decision_phase(cls, v: str) -> str: """Ensure decision_phase is valid""" valid = [ "idle", "awaiting_defensive", "awaiting_offensive", "resolving", "completed", ] if v not in valid: raise ValueError(f"decision_phase must be one of {valid}") return v # Helper methods def get_batting_team_id(self) -> int: """Get the ID of the team currently batting""" return self.away_team_id if self.half == "top" else self.home_team_id def get_fielding_team_id(self) -> int: """Get the ID of the team currently fielding""" return self.home_team_id if self.half == "top" else self.away_team_id def is_batting_team_ai(self) -> bool: """Check if the currently batting team is AI-controlled""" return self.away_team_is_ai if self.half == "top" else self.home_team_is_ai def is_fielding_team_ai(self) -> bool: """Check if the currently fielding team is AI-controlled""" return self.home_team_is_ai if self.half == "top" else self.away_team_is_ai def get_defender_for_position( self, position: str, state_manager: Any, # 'StateManager' - avoid circular import ) -> LineupPlayerState | None: """ Get the defender playing at specified position. Uses StateManager's lineup cache to find the defensive player. Args: position: Position code (P, C, 1B, 2B, 3B, SS, LF, CF, RF) state_manager: StateManager instance for lineup access Returns: LineupPlayerState if found, None otherwise Phase 3E-Main: Integrated with X-Check resolution """ # Get fielding team's lineup from cache fielding_team_id = self.get_fielding_team_id() lineup = state_manager.get_lineup(self.game_id, fielding_team_id) if not lineup: logger.warning(f"No lineup found for fielding team {fielding_team_id}") return None # Find active player at the specified position for player in lineup.players: if player.position == position and player.is_active: return player logger.warning( f"No active player found at position {position} for team {fielding_team_id}" ) return None def is_runner_on_first(self) -> bool: """Check if there's a runner on first base""" return self.on_first is not None def is_runner_on_second(self) -> bool: """Check if there's a runner on second base""" return self.on_second is not None def is_runner_on_third(self) -> bool: """Check if there's a runner on third base""" return self.on_third is not None def get_runner_at_base(self, base: int) -> LineupPlayerState | None: """Get runner at specified base (1, 2, or 3)""" if base == 1: return self.on_first if base == 2: return self.on_second if base == 3: return self.on_third return None def bases_occupied(self) -> list[int]: """Get list of occupied bases""" bases = [] if self.on_first: bases.append(1) if self.on_second: bases.append(2) if self.on_third: bases.append(3) return bases def get_all_runners(self) -> list[tuple[int, LineupPlayerState]]: """Returns list of (base, player) tuples for occupied bases""" runners = [] if self.on_first: runners.append((1, self.on_first)) if self.on_second: runners.append((2, self.on_second)) if self.on_third: runners.append((3, self.on_third)) return runners def clear_bases(self) -> None: """Clear all runners from bases""" self.on_first = None self.on_second = None self.on_third = None def add_runner(self, player: LineupPlayerState, base: int) -> None: """ Add a runner to a base Args: player: LineupPlayerState to place on base base: Base number (1, 2, or 3) """ if base not in [1, 2, 3]: raise ValueError("base must be 1, 2, or 3") if base == 1: self.on_first = player elif base == 2: self.on_second = player elif base == 3: self.on_third = player def advance_runner(self, from_base: int, to_base: int) -> None: """ Advance a runner from one base to another. Args: from_base: Starting base (1, 2, or 3) to_base: Ending base (2, 3, or 4 for home) """ runner = self.get_runner_at_base(from_base) if runner: # Remove from current base if from_base == 1: self.on_first = None elif from_base == 2: self.on_second = None elif from_base == 3: self.on_third = None if to_base == 4: # Runner scored if self.half == "top": self.away_score += 1 else: self.home_score += 1 else: # Runner advanced to another base if to_base == 1: self.on_first = runner elif to_base == 2: self.on_second = runner elif to_base == 3: self.on_third = runner def increment_outs(self) -> bool: """ Increment outs by 1. Returns: True if half-inning is over (3 outs), False otherwise """ self.outs += 1 if self.outs >= 3: return True return False def end_half_inning(self) -> None: """End the current half-inning and prepare for next""" self.outs = 0 self.clear_bases() if self.half == "top": # Switch to bottom of same inning self.half = "bottom" else: # End inning, go to top of next inning self.half = "top" self.inning += 1 def is_game_over(self) -> bool: """ Check if game is over. Game ends after regulation innings if home team is ahead or tied, or immediately if home team takes lead in bottom of final inning or later. Uses self.regulation_innings (default 9) for game length. """ reg = self.regulation_innings if self.inning < reg: return False if self.inning == reg and self.half == "bottom": # Bottom of final regulation inning - game ends if home team ahead if self.home_score > self.away_score: return True if self.inning > reg and self.half == "bottom": # Extra innings, bottom half - walk-off possible if self.home_score > self.away_score: return True if ( self.inning >= reg and self.half == "top" and self.outs >= self.outs_per_inning ): # Top of final inning or later just ended if self.home_score != self.away_score: return True return False model_config = ConfigDict( json_schema_extra={ "example": { "game_id": "550e8400-e29b-41d4-a716-446655440000", "league_id": "sba", "home_team_id": 1, "away_team_id": 2, "home_team_is_ai": False, "away_team_is_ai": True, "status": "active", "inning": 3, "half": "top", "outs": 1, "home_score": 2, "away_score": 1, "on_first": None, "on_second": { "lineup_id": 5, "card_id": 123, "position": "CF", "batting_order": 2, }, "on_third": None, "away_team_batter_idx": 3, "home_team_batter_idx": 5, "current_batter": { "lineup_id": 8, "card_id": 125, "position": "RF", "batting_order": 4, }, "current_pitcher": { "lineup_id": 10, "card_id": 130, "position": "P", "batting_order": 9, }, "current_catcher": { "lineup_id": 11, "card_id": 131, "position": "C", "batting_order": 2, }, "current_on_base_code": 2, "pending_decision": None, "decisions_this_play": {}, "play_count": 15, "last_play_result": "Single to left field", } } ) # ============================================================================ # MODEL REBUILD (for forward references) # ============================================================================ # Import PositionRating at runtime for model rebuild from app.models.player_models import PositionRating # noqa: E402 # Rebuild models that use forward references LineupPlayerState.model_rebuild() GameState.model_rebuild() # ============================================================================ # EXPORTS # ============================================================================ __all__ = [ "LineupPlayerState", "TeamLineupState", "DefensiveDecision", "OffensiveDecision", "GameState", ]