8ecce0f5ad
Add ability to force specific play outcomes instead of random dice rolls, enabling targeted testing of specific game scenarios. Changes: - play_resolver.resolve_play(): Add forced_outcome parameter, bypass dice rolls when provided, create dummy AbRoll with placeholder values - game_engine.resolve_play(): Accept and pass through forced_outcome param - terminal_client/commands.py: Pass forced_outcome to game engine Testing: - Verified TRIPLE, HOMERUN, and STRIKEOUT outcomes work correctly - Dummy AbRoll properly constructed with all required fields - Game state updates correctly with forced outcomes Example usage in REPL: resolve_with triple resolve_with homerun Fixes terminal client testing workflow to allow controlled scenarios. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
504 lines
18 KiB
Python
504 lines
18 KiB
Python
"""
|
|
Shared command implementations for terminal client.
|
|
|
|
This module contains the core logic for game commands that can be
|
|
used by both the REPL (repl.py) and CLI (main.py) interfaces.
|
|
|
|
Author: Claude
|
|
Date: 2025-10-27
|
|
"""
|
|
import asyncio
|
|
import logging
|
|
from uuid import UUID, uuid4
|
|
from typing import Optional, List, Tuple
|
|
|
|
from app.core.game_engine import game_engine
|
|
from app.core.state_manager import state_manager
|
|
from app.models.game_models import DefensiveDecision, OffensiveDecision
|
|
from app.config import PlayOutcome
|
|
from app.database.operations import DatabaseOperations
|
|
from terminal_client import display
|
|
from terminal_client.config import Config
|
|
|
|
logger = logging.getLogger(f'{__name__}.commands')
|
|
|
|
|
|
class GameCommands:
|
|
"""Shared command implementations for game operations."""
|
|
|
|
def __init__(self):
|
|
self.db_ops = DatabaseOperations()
|
|
|
|
async def create_new_game(
|
|
self,
|
|
league: str = 'sba',
|
|
home_team: int = 1,
|
|
away_team: int = 2,
|
|
set_current: bool = True
|
|
) -> Tuple[UUID, bool]:
|
|
"""
|
|
Create a new game with lineups and start it.
|
|
|
|
Args:
|
|
league: 'sba' or 'pd'
|
|
home_team: Home team ID
|
|
away_team: Away team ID
|
|
set_current: Whether to set as current game
|
|
|
|
Returns:
|
|
Tuple of (game_id, success)
|
|
"""
|
|
gid = uuid4()
|
|
|
|
try:
|
|
# Step 1: Create game in memory and database
|
|
display.print_info("Step 1: Creating game...")
|
|
|
|
state = await state_manager.create_game(
|
|
game_id=gid,
|
|
league_id=league,
|
|
home_team_id=home_team,
|
|
away_team_id=away_team
|
|
)
|
|
|
|
await self.db_ops.create_game(
|
|
game_id=gid,
|
|
league_id=league,
|
|
home_team_id=home_team,
|
|
away_team_id=away_team,
|
|
game_mode="friendly",
|
|
visibility="public"
|
|
)
|
|
|
|
display.print_success(f"Game created: {gid}")
|
|
|
|
if set_current:
|
|
Config.set_current_game(gid)
|
|
display.print_info(f"Current game set to: {gid}")
|
|
|
|
# Step 2: Setup lineups
|
|
display.print_info("Step 2: Creating test lineups...")
|
|
await self._create_test_lineups(gid, league, home_team, away_team)
|
|
|
|
# Step 3: Start the game
|
|
display.print_info("Step 3: Starting game...")
|
|
state = await game_engine.start_game(gid)
|
|
|
|
display.print_success(f"Game started - Inning {state.inning} {state.half}")
|
|
display.display_game_state(state)
|
|
|
|
return gid, True
|
|
|
|
except Exception as e:
|
|
display.print_error(f"Failed to create new game: {e}")
|
|
logger.exception("New game error")
|
|
return gid, False
|
|
|
|
async def _create_test_lineups(
|
|
self,
|
|
game_id: UUID,
|
|
league: str,
|
|
home_team: int,
|
|
away_team: int
|
|
) -> None:
|
|
"""Create test lineups for both teams."""
|
|
positions = ['P', 'C', '1B', '2B', '3B', 'SS', 'LF', 'CF', 'RF']
|
|
|
|
for team_id in [home_team, away_team]:
|
|
team_name = "Home" if team_id == home_team else "Away"
|
|
|
|
for i, position in enumerate(positions, start=1):
|
|
if league == 'sba':
|
|
player_id = (team_id * 100) + i
|
|
await self.db_ops.add_sba_lineup_player(
|
|
game_id=game_id,
|
|
team_id=team_id,
|
|
player_id=player_id,
|
|
position=position,
|
|
batting_order=i,
|
|
is_starter=True
|
|
)
|
|
else:
|
|
card_id = (team_id * 100) + i
|
|
await self.db_ops.add_pd_lineup_card(
|
|
game_id=game_id,
|
|
team_id=team_id,
|
|
card_id=card_id,
|
|
position=position,
|
|
batting_order=i,
|
|
is_starter=True
|
|
)
|
|
|
|
display.console.print(f" ✓ {team_name} team lineup created (9 players)")
|
|
|
|
async def submit_defensive_decision(
|
|
self,
|
|
game_id: UUID,
|
|
alignment: str = 'normal',
|
|
infield: str = 'normal',
|
|
outfield: str = 'normal',
|
|
hold_runners: Optional[List[int]] = None
|
|
) -> bool:
|
|
"""
|
|
Submit defensive decision.
|
|
|
|
Returns:
|
|
True if successful, False otherwise
|
|
"""
|
|
try:
|
|
decision = DefensiveDecision(
|
|
alignment=alignment,
|
|
infield_depth=infield,
|
|
outfield_depth=outfield,
|
|
hold_runners=hold_runners or []
|
|
)
|
|
|
|
state = await game_engine.submit_defensive_decision(game_id, decision)
|
|
display.print_success("Defensive decision submitted")
|
|
display.display_decision("defensive", decision)
|
|
display.display_game_state(state)
|
|
return True
|
|
|
|
except Exception as e:
|
|
display.print_error(f"Failed to submit defensive decision: {e}")
|
|
logger.exception("Defensive decision error")
|
|
return False
|
|
|
|
async def submit_offensive_decision(
|
|
self,
|
|
game_id: UUID,
|
|
approach: str = 'normal',
|
|
steal_attempts: Optional[List[int]] = None,
|
|
hit_and_run: bool = False,
|
|
bunt_attempt: bool = False
|
|
) -> bool:
|
|
"""
|
|
Submit offensive decision.
|
|
|
|
Returns:
|
|
True if successful, False otherwise
|
|
"""
|
|
try:
|
|
decision = OffensiveDecision(
|
|
approach=approach,
|
|
steal_attempts=steal_attempts or [],
|
|
hit_and_run=hit_and_run,
|
|
bunt_attempt=bunt_attempt
|
|
)
|
|
|
|
state = await game_engine.submit_offensive_decision(game_id, decision)
|
|
display.print_success("Offensive decision submitted")
|
|
display.display_decision("offensive", decision)
|
|
display.display_game_state(state)
|
|
return True
|
|
|
|
except Exception as e:
|
|
display.print_error(f"Failed to submit offensive decision: {e}")
|
|
logger.exception("Offensive decision error")
|
|
return False
|
|
|
|
async def resolve_play(self, game_id: UUID, forced_outcome: Optional[PlayOutcome] = None) -> bool:
|
|
"""
|
|
Resolve the current play.
|
|
|
|
Args:
|
|
game_id: Game to resolve
|
|
forced_outcome: If provided, use this outcome instead of rolling dice
|
|
|
|
Returns:
|
|
True if successful, False otherwise
|
|
"""
|
|
try:
|
|
if forced_outcome:
|
|
display.print_info(f"🎯 Forcing outcome: {forced_outcome.value}")
|
|
|
|
result = await game_engine.resolve_play(game_id, forced_outcome)
|
|
state = await game_engine.get_game_state(game_id)
|
|
|
|
if state:
|
|
display.display_play_result(result, state)
|
|
display.display_game_state(state)
|
|
return True
|
|
else:
|
|
display.print_error(f"Game {game_id} not found after resolution")
|
|
return False
|
|
|
|
except Exception as e:
|
|
display.print_error(f"Failed to resolve play: {e}")
|
|
logger.exception("Resolve play error")
|
|
return False
|
|
|
|
def list_outcomes(self) -> None:
|
|
"""
|
|
Display all available PlayOutcome values for manual selection.
|
|
"""
|
|
from rich.table import Table
|
|
from rich.console import Console
|
|
|
|
console = Console()
|
|
|
|
# Create categorized table
|
|
table = Table(title="Available Play Outcomes", show_header=True, header_style="bold cyan")
|
|
table.add_column("Category", style="yellow", width=20)
|
|
table.add_column("Outcome", style="green", width=25)
|
|
table.add_column("Description", style="white", width=50)
|
|
|
|
# Outs
|
|
table.add_row("Outs", "strikeout", "Batter strikes out")
|
|
table.add_row("", "groundball_a", "Groundball - double play if possible")
|
|
table.add_row("", "groundball_b", "Groundball - standard")
|
|
table.add_row("", "groundball_c", "Groundball - weak contact")
|
|
table.add_row("", "flyout_a", "Flyout variant A")
|
|
table.add_row("", "flyout_b", "Flyout variant B (medium depth)")
|
|
table.add_row("", "flyout_c", "Flyout variant C (deep)")
|
|
table.add_row("", "lineout", "Line drive out")
|
|
table.add_row("", "popout", "Pop fly out")
|
|
|
|
# Hits
|
|
table.add_row("Hits", "single_1", "Single - standard advancement")
|
|
table.add_row("", "single_2", "Single - enhanced advancement")
|
|
table.add_row("", "single_uncapped", "Single (uncapped) - decision tree")
|
|
table.add_row("", "double_2", "Double to 2nd base")
|
|
table.add_row("", "double_3", "Double to 3rd base")
|
|
table.add_row("", "double_uncapped", "Double (uncapped) - decision tree")
|
|
table.add_row("", "triple", "Triple")
|
|
table.add_row("", "homerun", "Home run")
|
|
|
|
# Walks/HBP
|
|
table.add_row("Walks/HBP", "walk", "Base on balls")
|
|
table.add_row("", "hbp", "Hit by pitch")
|
|
table.add_row("", "intentional_walk", "Intentional walk")
|
|
|
|
# Errors
|
|
table.add_row("Errors", "error", "Defensive error")
|
|
|
|
# Interrupts
|
|
table.add_row("Interrupts", "wild_pitch", "Wild pitch (pa=0)")
|
|
table.add_row("", "passed_ball", "Passed ball (pa=0)")
|
|
table.add_row("", "stolen_base", "Stolen base (pa=0)")
|
|
table.add_row("", "caught_stealing", "Caught stealing (pa=0)")
|
|
table.add_row("", "balk", "Balk (pa=0)")
|
|
table.add_row("", "pick_off", "Pick off (pa=0)")
|
|
|
|
# Ballpark Power
|
|
table.add_row("Ballpark", "bp_homerun", "Ballpark home run")
|
|
table.add_row("", "bp_single", "Ballpark single")
|
|
table.add_row("", "bp_flyout", "Ballpark flyout")
|
|
table.add_row("", "bp_lineout", "Ballpark lineout")
|
|
|
|
console.print(table)
|
|
console.print("\n[cyan]Usage:[/cyan] [green]resolve_with[/green] [yellow]<outcome>[/yellow]")
|
|
console.print("[dim]Example: resolve_with single_1[/dim]")
|
|
|
|
async def quick_play_rounds(
|
|
self,
|
|
game_id: UUID,
|
|
count: int = 1
|
|
) -> int:
|
|
"""
|
|
Execute multiple plays with default decisions.
|
|
|
|
Returns:
|
|
Number of plays successfully executed
|
|
"""
|
|
plays_completed = 0
|
|
|
|
for i in range(count):
|
|
try:
|
|
state = await game_engine.get_game_state(game_id)
|
|
if not state or state.status != "active":
|
|
display.print_warning(f"Game ended at play {i + 1}")
|
|
break
|
|
|
|
display.print_info(f"Play {i + 1}/{count}")
|
|
|
|
# Submit default decisions
|
|
await game_engine.submit_defensive_decision(game_id, DefensiveDecision())
|
|
await game_engine.submit_offensive_decision(game_id, OffensiveDecision())
|
|
|
|
# Resolve
|
|
result = await game_engine.resolve_play(game_id)
|
|
state = await game_engine.get_game_state(game_id)
|
|
|
|
if state:
|
|
display.print_success(f"{result.description}")
|
|
display.console.print(
|
|
f"[cyan]Score: Away {state.away_score} - {state.home_score} Home, "
|
|
f"Inning {state.inning} {state.half}, {state.outs} outs[/cyan]"
|
|
)
|
|
plays_completed += 1
|
|
|
|
await asyncio.sleep(0.3) # Brief pause for readability
|
|
|
|
except Exception as e:
|
|
display.print_error(f"Error on play {i + 1}: {e}")
|
|
logger.exception("Quick play error")
|
|
break
|
|
|
|
# Show final state
|
|
state = await game_engine.get_game_state(game_id)
|
|
if state:
|
|
display.print_info("Final state:")
|
|
display.display_game_state(state)
|
|
|
|
return plays_completed
|
|
|
|
async def show_game_status(self, game_id: UUID) -> bool:
|
|
"""
|
|
Display current game state.
|
|
|
|
Returns:
|
|
True if successful, False otherwise
|
|
"""
|
|
try:
|
|
state = await game_engine.get_game_state(game_id)
|
|
if state:
|
|
display.display_game_state(state)
|
|
return True
|
|
else:
|
|
display.print_error(f"Game {game_id} not found")
|
|
return False
|
|
except Exception as e:
|
|
display.print_error(f"Failed to get game status: {e}")
|
|
return False
|
|
|
|
async def show_box_score(self, game_id: UUID) -> bool:
|
|
"""
|
|
Display box score.
|
|
|
|
Returns:
|
|
True if successful, False otherwise
|
|
"""
|
|
try:
|
|
state = await game_engine.get_game_state(game_id)
|
|
if state:
|
|
display.display_box_score(state)
|
|
return True
|
|
else:
|
|
display.print_error(f"Game {game_id} not found")
|
|
return False
|
|
except Exception as e:
|
|
display.print_error(f"Failed to get box score: {e}")
|
|
return False
|
|
|
|
def validate_manual_outcome(self, outcome: str, location: Optional[str] = None) -> bool:
|
|
"""
|
|
Validate a manual outcome submission.
|
|
|
|
Args:
|
|
outcome: PlayOutcome enum value (e.g., 'groundball_c')
|
|
location: Optional hit location (e.g., 'SS')
|
|
|
|
Returns:
|
|
True if valid, False otherwise
|
|
"""
|
|
from app.models.game_models import ManualOutcomeSubmission
|
|
from pydantic import ValidationError
|
|
|
|
try:
|
|
# Try to create ManualOutcomeSubmission
|
|
submission = ManualOutcomeSubmission(
|
|
outcome=outcome,
|
|
hit_location=location
|
|
)
|
|
|
|
# Show success
|
|
display.print_success(f"✅ Valid manual outcome submission")
|
|
display.console.print(f" [cyan]Outcome:[/cyan] [green]{submission.outcome}[/green]")
|
|
if submission.hit_location:
|
|
display.console.print(f" [cyan]Location:[/cyan] [green]{submission.hit_location}[/green]")
|
|
else:
|
|
display.console.print(f" [cyan]Location:[/cyan] [dim]None (not required for this outcome)[/dim]")
|
|
|
|
# Check if location is required
|
|
outcome_enum = PlayOutcome(outcome)
|
|
if outcome_enum.requires_hit_location():
|
|
if not location:
|
|
display.print_warning("⚠️ Note: This outcome typically requires a hit location")
|
|
display.console.print(" [dim]Groundballs and flyouts need location for runner advancement[/dim]")
|
|
|
|
return True
|
|
|
|
except ValidationError as e:
|
|
display.print_error("❌ Invalid manual outcome submission")
|
|
for error in e.errors():
|
|
field = error['loc'][0] if error['loc'] else 'unknown'
|
|
message = error['msg']
|
|
display.console.print(f" [red]•[/red] [yellow]{field}:[/yellow] {message}")
|
|
return False
|
|
except Exception as e:
|
|
display.print_error(f"Validation error: {e}")
|
|
return False
|
|
|
|
def test_hit_location(self, outcome: str, handedness: str = 'R', count: int = 10) -> None:
|
|
"""
|
|
Test hit location calculation for a given outcome and handedness.
|
|
|
|
Args:
|
|
outcome: PlayOutcome enum value (e.g., 'groundball_c')
|
|
handedness: Batter handedness ('L' or 'R')
|
|
count: Number of samples to generate
|
|
|
|
Displays:
|
|
Distribution of hit locations
|
|
"""
|
|
from app.config.result_charts import calculate_hit_location, PlayOutcome
|
|
from collections import Counter
|
|
from rich.table import Table
|
|
|
|
try:
|
|
outcome_enum = PlayOutcome(outcome)
|
|
except ValueError:
|
|
display.print_error(f"Invalid outcome: {outcome}")
|
|
display.console.print(" [dim]Use 'list_outcomes' to see valid outcomes[/dim]")
|
|
return
|
|
|
|
# Generate samples
|
|
locations = []
|
|
for _ in range(count):
|
|
location = calculate_hit_location(outcome_enum, handedness)
|
|
if location:
|
|
locations.append(location)
|
|
|
|
if not locations:
|
|
display.print_info(f"Outcome '{outcome}' does not require hit location")
|
|
display.console.print(" [dim]Location only tracked for groundballs and flyouts[/dim]")
|
|
return
|
|
|
|
# Count distribution
|
|
counter = Counter(locations)
|
|
total = len(locations)
|
|
|
|
# Display results
|
|
display.print_success(f"Hit Location Distribution for {outcome} ({handedness}HB)")
|
|
|
|
table = Table(show_header=True, header_style="bold cyan")
|
|
table.add_column("Location", style="yellow", width=15)
|
|
table.add_column("Count", style="green", width=10, justify="right")
|
|
table.add_column("Percentage", style="cyan", width=15, justify="right")
|
|
table.add_column("Visual", style="white", width=30)
|
|
|
|
for location in sorted(counter.keys()):
|
|
count_val = counter[location]
|
|
pct = (count_val / total) * 100
|
|
bar = "█" * int(pct / 3) # Scale bar to fit
|
|
table.add_row(
|
|
location,
|
|
str(count_val),
|
|
f"{pct:.1f}%",
|
|
bar
|
|
)
|
|
|
|
display.console.print(table)
|
|
|
|
# Show pull rates info
|
|
display.console.print(f"\n[dim]Pull rates: 45% pull, 35% center, 20% opposite[/dim]")
|
|
if handedness == 'R':
|
|
display.console.print(f"[dim]RHB pulls left (3B, SS, LF)[/dim]")
|
|
else:
|
|
display.console.print(f"[dim]LHB pulls right (1B, 2B, RF)[/dim]")
|
|
|
|
|
|
# Singleton instance
|
|
game_commands = GameCommands()
|