strat-gameplay-webapp/backend/app/models/game_models.py
Cal Corum a1f42a93b8 CLAUDE: Implement Phase 3A - X-Check data models and enums
Add foundational data structures for X-Check play resolution system:

Models Added:
- PositionRating: Defensive ratings (range 1-5, error 0-88) for X-Check resolution
- XCheckResult: Dataclass tracking complete X-Check resolution flow with dice rolls,
  conversions (SPD test, G2#/G3#→SI2), error results, and final outcomes
- BasePlayer.active_position_rating: Optional field for current defensive position

Enums Extended:
- PlayOutcome.X_CHECK: New outcome type requiring special resolution
- PlayOutcome.is_x_check(): Helper method for type checking

Documentation Enhanced:
- Play.check_pos: Documented as X-Check position identifier
- Play.hit_type: Documented with examples (single_2_plus_error_1, etc.)

Utilities Added:
- app/core/cache.py: Redis cache key helpers for player positions and game state

Implementation Planning:
- Complete 6-phase implementation plan (3A-3F) documented in .claude/implementation/
- Phase 3A complete with all acceptance criteria met
- Zero breaking changes, all existing tests passing

Next: Phase 3B will add defense tables, error charts, and advancement logic

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
2025-11-01 15:32:09 -05:00

655 lines
22 KiB
Python

"""
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 Optional, Dict, List, Any
from uuid import UUID
from pydantic import BaseModel, Field, field_validator, ConfigDict
from app.config.result_charts import PlayOutcome
logger = logging.getLogger(f'{__name__}')
# ============================================================================
# LINEUP STATE
# ============================================================================
class LineupPlayerState(BaseModel):
"""
Represents a player in the game lineup.
This is a lightweight reference to a player - the full player data
(ratings, attributes, etc.) will be cached separately in Week 6.
"""
lineup_id: int
card_id: int
position: str
batting_order: Optional[int] = None
is_active: bool = True
@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: Optional[int]) -> Optional[int]:
"""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) -> Optional[LineupPlayerState]:
"""
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) -> Optional[LineupPlayerState]:
"""
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) -> Optional[LineupPlayerState]:
"""
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
# ============================================================================
# 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).
"""
alignment: str = "normal" # normal, shifted_left, shifted_right, extreme_shift
infield_depth: str = "normal" # infield_in, normal, corners_in
outfield_depth: str = "normal" # in, normal
hold_runners: List[int] = Field(default_factory=list) # [1, 3] = hold 1st and 3rd
@field_validator('alignment')
@classmethod
def validate_alignment(cls, v: str) -> str:
"""Validate alignment"""
valid = ['normal', 'shifted_left', 'shifted_right', 'extreme_shift']
if v not in valid:
raise ValueError(f"alignment must be one of {valid}")
return v
@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 = ['in', 'normal']
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 batter approach and baserunner actions.
"""
approach: str = "normal" # normal, contact, power, patient
steal_attempts: List[int] = Field(default_factory=list) # [2] = steal second, [2, 3] = double steal
hit_and_run: bool = False
bunt_attempt: bool = False
@field_validator('approach')
@classmethod
def validate_approach(cls, v: str) -> str:
"""Validate batting approach"""
valid = ['normal', 'contact', 'power', 'patient']
if v not in valid:
raise ValueError(f"approach must be one of {valid}")
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(f"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: Optional[str] = None # '1B', '2B', 'SS', '3B', 'LF', 'CF', 'RF', 'P', 'C'
@field_validator('hit_location')
@classmethod
def validate_hit_location(cls, v: Optional[str]) -> Optional[str]:
"""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: Optional[int] = None
spd_test_target: Optional[int] = None
spd_test_passed: Optional[bool] = 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_lineup_id: Snapshot - batter for current play
current_pitcher_lineup_id: Snapshot - pitcher for current play
current_catcher_lineup_id: Snapshot - catcher for current play
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
# Resolution mode
auto_mode: bool = False # True = auto-generate outcomes (PD only), False = manual submissions
# 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: Optional[LineupPlayerState] = None
on_second: Optional[LineupPlayerState] = None
on_third: Optional[LineupPlayerState] = 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
current_batter_lineup_id: int
current_pitcher_lineup_id: Optional[int] = None
current_catcher_lineup_id: Optional[int] = 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: Optional[str] = None # 'defensive', 'offensive', 'result_selection'
decisions_this_play: Dict[str, Any] = Field(default_factory=dict)
# Phase 3: Enhanced decision workflow
pending_defensive_decision: Optional[DefensiveDecision] = None
pending_offensive_decision: Optional[OffensiveDecision] = None
decision_phase: str = "idle" # idle, awaiting_defensive, awaiting_offensive, resolving, completed
decision_deadline: Optional[str] = None # ISO8601 timestamp for timeout
# Manual mode support (Week 7 Task 7)
pending_manual_roll: Optional[Any] = None # AbRoll stored when dice rolled in manual mode
# Play tracking
play_count: int = Field(default=0, ge=0)
last_play_result: Optional[str] = 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: Optional[str]) -> Optional[str]:
"""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 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) -> Optional[LineupPlayerState]:
"""Get runner at specified base (1, 2, or 3)"""
if base == 1:
return self.on_first
elif base == 2:
return self.on_second
elif 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 9 innings if home team is ahead or tied,
or immediately if home team takes lead in bottom of 9th or later.
"""
if self.inning < 9:
return False
if self.inning == 9 and self.half == "bottom":
# Bottom of 9th - game ends if home team ahead
if self.home_score > self.away_score:
return True
if self.inning > 9 and self.half == "bottom":
# Extra innings, bottom half - walk-off possible
if self.home_score > self.away_score:
return True
if self.inning >= 9 and self.half == "top" and self.outs >= 3:
# Top of 9th 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,
"current_pitcher_lineup_id": 10,
"current_catcher_lineup_id": 11,
"current_on_base_code": 2,
"pending_decision": None,
"decisions_this_play": {},
"play_count": 15,
"last_play_result": "Single to left field"
}
}
)
# ============================================================================
# EXPORTS
# ============================================================================
__all__ = [
'LineupPlayerState',
'TeamLineupState',
'DefensiveDecision',
'OffensiveDecision',
'GameState',
]