strat-gameplay-webapp/backend/app/core/play_resolver.py
Cal Corum a696473d0a CLAUDE: Integrate flyball advancement with RunnerAdvancement system
Major Phase 2 refactoring to consolidate runner advancement logic:

**Flyball System Enhancement**:
- Add FLYOUT_BQ variant (medium-shallow depth)
- 4 flyball types with clear semantics: A (deep), B (medium), BQ (medium-shallow), C (shallow)
- Updated helper methods to include FLYOUT_BQ

**RunnerAdvancement Integration**:
- Extend runner_advancement.py to handle both groundballs AND flyballs
- advance_runners() routes to _advance_runners_groundball() or _advance_runners_flyball()
- Comprehensive flyball logic with proper DECIDE mechanics per flyball type
- No-op movements recorded for state recovery consistency

**PlayResolver Refactoring**:
- Consolidate all 4 flyball outcomes to delegate to RunnerAdvancement (DRY)
- Eliminate duplicate flyball resolution code
- Rename helpers for clarity: _advance_on_single_1/_advance_on_single_2 (was _advance_on_single)
- Fix single/double advancement logic for different hit types

**State Recovery Fix**:
- Fix state_manager.py game recovery to build LineupPlayerState objects properly
- Use get_lineup_player() helper to construct from lineup data
- Correctly track runners in on_first/on_second/on_third fields (matches Phase 2 model)

**Database Support**:
- Add runner tracking fields to play data for accurate recovery
- Include batter_id, on_first_id, on_second_id, on_third_id, and *_final fields

**Type Safety Improvements**:
- Fix lineup_id access throughout runner_advancement.py (was accessing on_first directly, now on_first.lineup_id)
- Make current_batter_lineup_id non-optional (always set by _prepare_next_play)
- Add type: ignore for known SQLAlchemy false positives

**Documentation**:
- Update CLAUDE.md with comprehensive flyball documentation
- Add flyball types table, usage examples, and test coverage notes
- Document differences between groundball and flyball mechanics

**Testing**:
- Add test_flyball_advancement.py with 21 flyball tests
- Coverage: all 4 types, DECIDE scenarios, no-op movements, edge cases

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

Co-Authored-By: Claude <noreply@anthropic.com>
2025-10-31 17:04:23 -05:00

559 lines
20 KiB
Python

"""
Play Resolver - Resolves play outcomes based on dice rolls.
Architecture: Outcome-first design where manual resolution is primary.
- resolve_outcome(): Core resolution logic (works for both manual and auto)
- resolve_manual_play(): Wrapper for manual submissions (most games)
- resolve_auto_play(): Wrapper for PD auto mode (rare)
Author: Claude
Date: 2025-10-24
Updated: 2025-10-31 - Week 7 Task 6: Integrated RunnerAdvancement and outcome-first architecture
"""
import logging
from dataclasses import dataclass
from typing import Optional, List, TYPE_CHECKING
import pendulum
from app.core.dice import dice_system
from app.core.roll_types import AbRoll, RollType
from app.core.runner_advancement import RunnerAdvancement
from app.models.game_models import GameState, DefensiveDecision, OffensiveDecision, ManualOutcomeSubmission
from app.config import PlayOutcome, get_league_config
from app.config.result_charts import calculate_hit_location, PdAutoResultChart, ManualResultChart
if TYPE_CHECKING:
from app.models.player_models import PdPlayer
logger = logging.getLogger(f'{__name__}.PlayResolver')
@dataclass
class PlayResult:
"""Result of a resolved play"""
outcome: PlayOutcome
outs_recorded: int
runs_scored: int
batter_result: Optional[int] # None = out, 1-4 = base reached
runners_advanced: List[tuple[int, int]] # [(from_base, to_base), ...]
description: str
ab_roll: AbRoll # Full at-bat roll for audit trail
hit_location: Optional[str] = None # '1B', '2B', '3B', 'SS', 'LF', 'CF', 'RF', 'P', 'C'
# Statistics
is_hit: bool = False
is_out: bool = False
is_walk: bool = False
class PlayResolver:
"""
Resolves play outcomes based on dice rolls and game state.
Architecture: Outcome-first design
- Manual mode (primary): Players submit outcomes after reading physical cards
- Auto mode (rare): System generates outcomes from digitized ratings (PD only)
Args:
league_id: 'sba' or 'pd'
auto_mode: If True, use result charts to auto-generate outcomes
Only supported for leagues with digitized card data
Raises:
ValueError: If auto_mode requested for league that doesn't support it
"""
def __init__(self, league_id: str, auto_mode: bool = False):
self.league_id = league_id
self.auto_mode = auto_mode
self.runner_advancement = RunnerAdvancement()
# Get league config for validation
league_config = get_league_config(league_id)
# Validate auto mode support
if auto_mode and not league_config.supports_auto_mode():
raise ValueError(
f"Auto mode not supported for {league_id} league. "
f"This league does not have digitized card data."
)
# Initialize result chart for auto mode only
if auto_mode:
self.result_chart = PdAutoResultChart()
logger.info(f"PlayResolver initialized in AUTO mode for {league_id}")
else:
self.result_chart = None
logger.info(f"PlayResolver initialized in MANUAL mode for {league_id}")
# ========================================
# PUBLIC METHODS - Primary API
# ========================================
def resolve_manual_play(
self,
submission: ManualOutcomeSubmission,
state: GameState,
defensive_decision: DefensiveDecision,
offensive_decision: OffensiveDecision,
ab_roll: AbRoll
) -> PlayResult:
"""
Resolve a manually submitted play (SBA + PD manual mode).
This is the PRIMARY method for most games. Players read physical cards
and submit the outcome they see via WebSocket.
Args:
submission: Player's submitted outcome + optional hit location
state: Current game state
defensive_decision: Defensive team's choices
offensive_decision: Offensive team's choices
ab_roll: Server-rolled dice for audit trail
Returns:
PlayResult with complete outcome
"""
logger.info(f"Resolving manual play - {submission.outcome} at {submission.hit_location}")
# Convert string to PlayOutcome enum
outcome = PlayOutcome(submission.outcome)
# Delegate to core resolution
return self.resolve_outcome(
outcome=outcome,
hit_location=submission.hit_location,
state=state,
defensive_decision=defensive_decision,
offensive_decision=offensive_decision,
ab_roll=ab_roll
)
def resolve_auto_play(
self,
state: GameState,
batter: 'PdPlayer',
pitcher: 'PdPlayer',
defensive_decision: DefensiveDecision,
offensive_decision: OffensiveDecision
) -> PlayResult:
"""
Resolve an auto-generated play (PD auto mode only).
This is RARE - only used for PD games with auto mode enabled.
System generates outcome from digitized player ratings.
Args:
state: Current game state
batter: Batting player (PdPlayer with ratings)
pitcher: Pitching player (PdPlayer with ratings)
defensive_decision: Defensive team's choices
offensive_decision: Offensive team's choices
Returns:
PlayResult with complete outcome
Raises:
ValueError: If called when not in auto mode
"""
if not self.auto_mode:
raise ValueError("resolve_auto_play() can only be called in auto mode")
logger.info(f"Resolving auto play - {batter.name} vs {pitcher.name}")
# Roll dice
ab_roll = dice_system.roll_ab(league_id=state.league_id, game_id=state.game_id)
# Generate outcome from ratings
outcome, hit_location = self.result_chart.get_outcome( #type: ignore
roll=ab_roll,
state=state,
batter=batter,
pitcher=pitcher
)
# Delegate to core resolution
return self.resolve_outcome(
outcome=outcome,
hit_location=hit_location,
state=state,
defensive_decision=defensive_decision,
offensive_decision=offensive_decision,
ab_roll=ab_roll
)
def resolve_outcome(
self,
outcome: PlayOutcome,
hit_location: Optional[str],
state: GameState,
defensive_decision: DefensiveDecision,
offensive_decision: OffensiveDecision,
ab_roll: AbRoll
) -> PlayResult:
"""
CORE resolution method - all play resolution logic lives here.
This method handles all outcome types and delegates to RunnerAdvancement
for groundball outcomes. Works for both manual and auto modes.
Args:
outcome: The play outcome (from card or auto-generated)
hit_location: Where ball was hit ('1B', '2B', '3B', 'SS', 'LF', 'CF', 'RF', 'P', 'C') or None
state: Current game state
defensive_decision: Defensive team's positioning/strategy
offensive_decision: Offensive team's strategy
ab_roll: Dice roll for audit trail
Returns:
PlayResult with complete outcome, runner movements, and statistics
"""
logger.info(f"Resolving {outcome.value} - Inning {state.inning} {state.half}, {state.outs} outs")
# ==================== Strikeout ====================
if outcome == PlayOutcome.STRIKEOUT:
return PlayResult(
outcome=outcome,
outs_recorded=1,
runs_scored=0,
batter_result=None,
runners_advanced=[],
description="Strikeout looking",
ab_roll=ab_roll,
hit_location=None,
is_out=True
)
# ==================== Groundballs ====================
elif outcome in [PlayOutcome.GROUNDBALL_A, PlayOutcome.GROUNDBALL_B, PlayOutcome.GROUNDBALL_C]:
# Delegate to RunnerAdvancement for all groundball outcomes
advancement_result = self.runner_advancement.advance_runners(
outcome=outcome,
hit_location=hit_location or 'SS', # Default to SS if location not specified
state=state,
defensive_decision=defensive_decision
)
# Convert RunnerMovement list to tuple format for PlayResult
runners_advanced = [
(movement.from_base, movement.to_base)
for movement in advancement_result.movements
if not movement.is_out and movement.from_base > 0 # Exclude batter, include only runners
]
# Extract batter result from movements
batter_movement = next(
(m for m in advancement_result.movements if m.from_base == 0),
None
)
batter_result = batter_movement.to_base if batter_movement and not batter_movement.is_out else None
return PlayResult(
outcome=outcome,
outs_recorded=advancement_result.outs_recorded,
runs_scored=advancement_result.runs_scored,
batter_result=batter_result,
runners_advanced=runners_advanced,
description=advancement_result.description,
ab_roll=ab_roll,
hit_location=hit_location,
is_out=(advancement_result.outs_recorded > 0)
)
# ==================== Flyouts ====================
elif outcome in [PlayOutcome.FLYOUT_A, PlayOutcome.FLYOUT_B, PlayOutcome.FLYOUT_BQ, PlayOutcome.FLYOUT_C]:
# Delegate to RunnerAdvancement for all flyball outcomes
advancement_result = self.runner_advancement.advance_runners(
outcome=outcome,
hit_location=hit_location or 'CF', # Default to CF if location not specified
state=state,
defensive_decision=defensive_decision
)
# Convert RunnerMovement list to tuple format for PlayResult
runners_advanced = [
(movement.from_base, movement.to_base)
for movement in advancement_result.movements
if not movement.is_out and movement.from_base > 0 # Exclude batter, include only runners
]
# Extract batter result from movements (always out for flyouts)
batter_movement = next(
(m for m in advancement_result.movements if m.from_base == 0),
None
)
batter_result = batter_movement.to_base if batter_movement and not batter_movement.is_out else None
return PlayResult(
outcome=outcome,
outs_recorded=advancement_result.outs_recorded,
runs_scored=advancement_result.runs_scored,
batter_result=batter_result,
runners_advanced=runners_advanced,
description=advancement_result.description,
ab_roll=ab_roll,
hit_location=hit_location,
is_out=(advancement_result.outs_recorded > 0)
)
# ==================== Lineout ====================
elif outcome == PlayOutcome.LINEOUT:
return PlayResult(
outcome=outcome,
outs_recorded=1,
runs_scored=0,
batter_result=None,
runners_advanced=[],
description="Lineout",
ab_roll=ab_roll,
is_out=True
)
elif outcome == PlayOutcome.WALK:
# Walk - batter to first, runners advance if forced
runners_advanced = self._advance_on_walk(state)
runs_scored = sum(1 for (from_base, to_base) in runners_advanced if to_base == 4)
return PlayResult(
outcome=outcome,
outs_recorded=0,
runs_scored=runs_scored,
batter_result=1,
runners_advanced=runners_advanced,
description="Walk",
ab_roll=ab_roll,
is_walk=True
)
# ==================== Singles ====================
elif outcome == PlayOutcome.SINGLE_1:
# Single with standard advancement
runners_advanced = self._advance_on_single_1(state)
runs_scored = sum(1 for (from_base, to_base) in runners_advanced if to_base == 4)
return PlayResult(
outcome=outcome,
outs_recorded=0,
runs_scored=runs_scored,
batter_result=1,
runners_advanced=runners_advanced,
description="Single to left field",
ab_roll=ab_roll,
is_hit=True
)
elif outcome == PlayOutcome.SINGLE_2:
# Single with enhanced advancement (more aggressive)
runners_advanced = self._advance_on_single_2(state)
runs_scored = sum(1 for (from_base, to_base) in runners_advanced if to_base == 4)
return PlayResult(
outcome=outcome,
outs_recorded=0,
runs_scored=runs_scored,
batter_result=1,
runners_advanced=runners_advanced,
description="Single to right field",
ab_roll=ab_roll,
is_hit=True
)
elif outcome == PlayOutcome.SINGLE_UNCAPPED:
# TODO Phase 3: Implement uncapped hit decision tree
# For now, treat as SINGLE_1
runners_advanced = self._advance_on_single_1(state)
runs_scored = sum(1 for (from_base, to_base) in runners_advanced if to_base == 4)
return PlayResult(
outcome=outcome,
outs_recorded=0,
runs_scored=runs_scored,
batter_result=1,
runners_advanced=runners_advanced,
description="Single to center (uncapped)",
ab_roll=ab_roll,
is_hit=True
)
# ==================== Doubles ====================
elif outcome == PlayOutcome.DOUBLE_2:
# Double to 2nd base
runners_advanced = self._advance_on_double_2(state)
runs_scored = sum(1 for (from_base, to_base) in runners_advanced if to_base == 4)
return PlayResult(
outcome=outcome,
outs_recorded=0,
runs_scored=runs_scored,
batter_result=2,
runners_advanced=runners_advanced,
description="Double to left-center",
ab_roll=ab_roll,
is_hit=True
)
elif outcome == PlayOutcome.DOUBLE_3:
# Double with extra advancement (batter to 3rd)
runners_advanced = self._advance_on_double_3(state)
runs_scored = sum(1 for (from_base, to_base) in runners_advanced if to_base == 4)
return PlayResult(
outcome=outcome,
outs_recorded=0,
runs_scored=runs_scored,
batter_result=3, # Batter goes to 3rd
runners_advanced=runners_advanced,
description="Double to right-center gap (batter to 3rd)",
ab_roll=ab_roll,
is_hit=True
)
elif outcome == PlayOutcome.DOUBLE_UNCAPPED:
# TODO Phase 3: Implement uncapped hit decision tree
# For now, treat as DOUBLE_2
runners_advanced = self._advance_on_double_2(state)
runs_scored = sum(1 for (from_base, to_base) in runners_advanced if to_base == 4)
return PlayResult(
outcome=outcome,
outs_recorded=0,
runs_scored=runs_scored,
batter_result=2,
runners_advanced=runners_advanced,
description="Double (uncapped)",
ab_roll=ab_roll,
is_hit=True
)
elif outcome == PlayOutcome.TRIPLE:
# All runners score
runners_advanced = [(base, 4) for base, _ in state.get_all_runners()]
runs_scored = len(runners_advanced)
return PlayResult(
outcome=outcome,
outs_recorded=0,
runs_scored=runs_scored,
batter_result=3,
runners_advanced=runners_advanced,
description="Triple to right-center gap",
ab_roll=ab_roll,
is_hit=True
)
elif outcome == PlayOutcome.HOMERUN:
# Everyone scores
runners_advanced = [(base, 4) for base, _ in state.get_all_runners()]
runs_scored = len(runners_advanced) + 1
return PlayResult(
outcome=outcome,
outs_recorded=0,
runs_scored=runs_scored,
batter_result=4,
runners_advanced=runners_advanced,
description="Home run to left field",
ab_roll=ab_roll,
is_hit=True
)
elif outcome == PlayOutcome.WILD_PITCH:
# Runners advance one base
runners_advanced = [(base, base + 1) for base, _ in state.get_all_runners()]
runs_scored = sum(1 for (from_base, to_base) in runners_advanced if to_base == 4)
return PlayResult(
outcome=outcome,
outs_recorded=0,
runs_scored=runs_scored,
batter_result=None, # Batter stays at plate
runners_advanced=runners_advanced,
description="Wild pitch - runners advance",
ab_roll=ab_roll
)
elif outcome == PlayOutcome.PASSED_BALL:
# Runners advance one base
runners_advanced = [(base, base + 1) for base, _ in state.get_all_runners()]
runs_scored = sum(1 for (from_base, to_base) in runners_advanced if to_base == 4)
return PlayResult(
outcome=outcome,
outs_recorded=0,
runs_scored=runs_scored,
batter_result=None, # Batter stays at plate
runners_advanced=runners_advanced,
description="Passed ball - runners advance",
ab_roll=ab_roll
)
else:
raise ValueError(f"Unhandled outcome: {outcome}")
def _advance_on_walk(self, state: GameState) -> List[tuple[int, int]]:
"""Calculate runner advancement on walk"""
advances = []
# Only forced runners advance
if state.on_first:
# First occupied - check second
if state.on_second:
# Bases loaded scenario
if state.on_third:
# Bases loaded - force runner home
advances.append((3, 4))
advances.append((2, 3))
advances.append((1, 2))
return advances
def _advance_on_single_1(self, state: GameState) -> List[tuple[int, int]]:
"""Calculate runner advancement on single (simplified)"""
advances = []
if state.on_third:
# Runner on third scores
advances.append((3, 4))
if state.on_second:
advances.append((2, 3))
if state.on_first:
advances.append((1, 2))
return advances
def _advance_on_single_2(self, state: GameState) -> List[tuple[int, int]]:
"""Calculate runner advancement on single (simplified)"""
advances = []
if state.on_third:
# Runner on third scores
advances.append((3, 4))
if state.on_second:
# Runner on second scores
advances.append((2, 4))
if state.on_first:
# Runner on first to third
advances.append((1, 3))
return advances
def _advance_on_double_2(self, state: GameState) -> List[tuple[int, int]]:
"""Calculate runner advancement on double"""
advances = []
# All runners score on double (simplified)
for base, _ in state.get_all_runners():
advances.append((base, 4))
return advances
def _advance_on_double_3(self, state: GameState) -> List[tuple[int, int]]:
"""Calculate runner advancement on double"""
advances = []
# All runners score on double (simplified)
for base, _ in state.get_all_runners():
advances.append((base, 4))
return advances