""" Interactive REPL for terminal client. Provides an interactive shell that keeps game state in memory across commands. Uses Python's cmd module for readline support and command completion. Author: Claude Date: 2025-10-26 """ import asyncio import logging import cmd from uuid import UUID, uuid4 from typing import Optional 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.database.operations import DatabaseOperations from terminal_client import display from terminal_client.config import Config from terminal_client.commands import game_commands from terminal_client.completions import GameREPLCompletions from terminal_client.help_text import show_help, HelpFormatter from terminal_client.arg_parser import ( parse_new_game_args, parse_defensive_args, parse_offensive_args, parse_quick_play_args, parse_use_game_args, ArgumentParseError ) logger = logging.getLogger(f'{__name__}.repl') class GameREPL(GameREPLCompletions, cmd.Cmd): """Interactive REPL for game engine testing.""" intro = """ ╔══════════════════════════════════════════════════════════════════════════════╗ ║ Paper Dynasty Game Engine - Terminal Client ║ ║ Interactive Mode ║ ╚══════════════════════════════════════════════════════════════════════════════╝ Type 'help' to see all available commands. Type 'help ' for detailed information about a specific command. Use TAB for auto-completion of commands and options. Quick start: new_game Create and start a new game status Show current game state defensive Submit defensive decision offensive Submit offensive decision resolve Resolve the play quick_play 10 Auto-play 10 plays Press Ctrl+D or type 'quit' to exit. """ prompt = '⚾ > ' def __init__(self): # Initialize both parent classes cmd.Cmd.__init__(self) GameREPLCompletions.__init__(self) self.current_game_id: Optional[UUID] = None self.db_ops = DatabaseOperations() # Create persistent event loop for entire REPL session self.loop = asyncio.new_event_loop() asyncio.set_event_loop(self.loop) # Try to load current game from config saved_game = Config.get_current_game() if saved_game: self.current_game_id = saved_game display.print_info(f"Loaded saved game: {saved_game}") def _ensure_game(self) -> UUID: """Ensure current game is set.""" if self.current_game_id is None: display.print_error("No current game. Use 'new_game' first.") raise ValueError("No current game") return self.current_game_id async def _ensure_game_loaded(self, game_id: UUID) -> None: """ Ensure game is loaded in state_manager. If game exists in database but not in memory, recover it using state_manager.recover_game() which replays plays to rebuild state, then prepare the next play to populate snapshot fields. """ # Check if already in memory state = state_manager.get_state(game_id) if state is not None: return # Already loaded # Try to recover from database try: display.print_info(f"Loading game {game_id} from database...") recovered_state = await state_manager.recover_game(game_id) if recovered_state and recovered_state.status == "active": # Call _prepare_next_play to populate snapshot fields # (batter_id, pitcher_id, catcher_id, on_base_code) await game_engine._prepare_next_play(recovered_state) logger.debug(f"Prepared snapshot for recovered game {game_id}") display.print_success("Game loaded successfully") except Exception as e: display.print_error(f"Failed to load game: {e}") logger.error(f"Game recovery failed for {game_id}: {e}", exc_info=True) raise ValueError(f"Game {game_id} not found") def _run_async(self, coro): """ Helper to run async functions using persistent event loop. This keeps database connections alive across commands. """ return self.loop.run_until_complete(coro) # ==================== Game Management Commands ==================== def do_new_game(self, arg): """ Create a new game with lineups and start it. Usage: new_game [--league sba|pd] [--home-team N] [--away-team N] Examples: new_game new_game --league pd new_game --home-team 5 --away-team 3 """ async def _new_game(): try: # Parse arguments with robust parser args = parse_new_game_args(arg) # Use shared command gid, success = await game_commands.create_new_game( league=args['league'], home_team=args['home_team'], away_team=args['away_team'], set_current=True ) if success: self.current_game_id = gid except ArgumentParseError as e: display.print_error(f"Invalid arguments: {e}") except Exception as e: display.print_error(f"Failed to create game: {e}") logger.exception("New game error") self._run_async(_new_game()) def do_defensive(self, arg): """ Submit defensive decision. Usage: defensive [--alignment TYPE] [--infield DEPTH] [--outfield DEPTH] [--hold BASES] Options: --alignment normal, shifted_left, shifted_right, extreme_shift --infield in, normal, back, double_play --outfield in, normal, back --hold Comma-separated bases (e.g., 1,3) Examples: defensive defensive --alignment shifted_left defensive --infield double_play --hold 1,3 """ async def _defensive(): try: gid = self._ensure_game() await self._ensure_game_loaded(gid) # Parse arguments with robust parser args = parse_defensive_args(arg) # Submit decision await game_commands.submit_defensive_decision( game_id=gid, alignment=args['alignment'], infield=args['infield'], outfield=args['outfield'], hold_runners=args['hold'] ) except ArgumentParseError as e: display.print_error(f"Invalid arguments: {e}") except ValueError: pass # Already printed error except Exception as e: display.print_error(f"Failed: {e}") logger.exception("Defensive error") self._run_async(_defensive()) def do_offensive(self, arg): """ Submit offensive decision. Usage: offensive [--approach TYPE] [--steal BASES] [--hit-run] [--bunt] Options: --approach normal, contact, power, patient --steal Comma-separated bases (e.g., 2,3) --hit-run Enable hit-and-run (flag) --bunt Attempt bunt (flag) Examples: offensive offensive --approach power offensive --steal 2 --hit-run """ async def _offensive(): try: gid = self._ensure_game() await self._ensure_game_loaded(gid) # Parse arguments with robust parser args = parse_offensive_args(arg) # Submit decision await game_commands.submit_offensive_decision( game_id=gid, approach=args['approach'], steal_attempts=args['steal'], hit_and_run=args['hit_run'], bunt_attempt=args['bunt'] ) except ArgumentParseError as e: display.print_error(f"Invalid arguments: {e}") except ValueError: pass except Exception as e: display.print_error(f"Failed: {e}") logger.exception("Offensive error") self._run_async(_offensive()) def do_resolve(self, arg): """ Resolve the current play. Usage: resolve Both defensive and offensive decisions must be submitted first. """ async def _resolve(): try: gid = self._ensure_game() await self._ensure_game_loaded(gid) # Use shared command await game_commands.resolve_play(gid) except ValueError: pass except Exception as e: display.print_error(f"Failed: {e}") logger.exception("Resolve error") self._run_async(_resolve()) def do_list_outcomes(self, arg): """ List all available PlayOutcome values for manual outcome testing. Usage: list_outcomes Displays a categorized table of all play outcomes that can be used with the 'resolve_with' command for testing specific scenarios. """ # This is synchronous, no need for async game_commands.list_outcomes() def do_resolve_with(self, arg): """ Resolve the current play with a specific outcome (for testing). Usage: resolve_with resolve_with x-check [[+]] Arguments: outcome PlayOutcome value (e.g., single_1, homerun, strikeout) position For x-check: P, C, 1B, 2B, 3B, SS, LF, CF, RF result For x-check: G1, G2, G3, F1, F2, F3, SI1, SI2, DO2, DO3, TR3, FO, PO error For x-check: NO, E1, E2, E3, RP (default: NO if not specified) This command allows you to force a specific outcome instead of rolling dice, useful for testing runner advancement, specific game states, and edge cases. Use 'list_outcomes' to see all available outcome values. Examples: resolve_with single_1 resolve_with homerun resolve_with groundball_a resolve_with double_uncapped resolve_with x-check SS # Test X-Check to shortstop (random result) resolve_with x-check LF DO2 # Force double to LF with no error resolve_with x-check 2B G2+E1 # Force groundout to 2B with E1 error resolve_with x-check SS SI2+E2 # Force single to SS with E2 error """ async def _resolve_with(): try: gid = self._ensure_game() await self._ensure_game_loaded(gid) # Parse arguments args = arg.strip().lower().split() if not args: display.print_error("Missing outcome argument") display.print_info("Usage: resolve_with ") display.print_info(" resolve_with x-check [[+]]") display.print_info("Use 'list_outcomes' to see available values") return # Check for x-check with position outcome_str = args[0] xcheck_position = None xcheck_result = None xcheck_error = None if outcome_str in ['x-check', 'xcheck', 'x_check']: if len(args) < 2: display.print_error("Missing position for x-check") display.print_info("Usage: resolve_with x-check [[+]]") display.print_info("Valid positions: P, C, 1B, 2B, 3B, SS, LF, CF, RF") return outcome_str = 'x_check' xcheck_position = args[1].upper() # Validate position valid_positions = ['P', 'C', '1B', '2B', '3B', 'SS', 'LF', 'CF', 'RF'] if xcheck_position not in valid_positions: display.print_error(f"Invalid position: {xcheck_position}") display.print_info(f"Valid positions: {', '.join(valid_positions)}") return # Parse optional result+error (e.g., "DO2+E1" or "G2") if len(args) >= 3: result_spec = args[2].upper() # Split on '+' to separate result and error if '+' in result_spec: parts = result_spec.split('+') xcheck_result = parts[0] xcheck_error = parts[1] if len(parts) > 1 else 'NO' else: xcheck_result = result_spec xcheck_error = 'NO' # Default to no error # Validate result code valid_results = ['G1', 'G2', 'G3', 'F1', 'F2', 'F3', 'SI1', 'SI2', 'DO2', 'DO3', 'TR3', 'FO', 'PO'] if xcheck_result not in valid_results: display.print_error(f"Invalid X-Check result: {xcheck_result}") display.print_info(f"Valid results: {', '.join(valid_results)}") return # Validate error code valid_errors = ['NO', 'E1', 'E2', 'E3', 'RP'] if xcheck_error not in valid_errors: display.print_error(f"Invalid error code: {xcheck_error}") display.print_info(f"Valid errors: {', '.join(valid_errors)}") return # Try to convert string to PlayOutcome enum from app.config import PlayOutcome try: outcome = PlayOutcome(outcome_str) except ValueError: display.print_error(f"Invalid outcome: {outcome_str}") display.print_info("Use 'list_outcomes' to see available values") return # Use shared command with forced outcome await game_commands.resolve_play( gid, forced_outcome=outcome, xcheck_position=xcheck_position, xcheck_result=xcheck_result, xcheck_error=xcheck_error ) except ValueError: pass except Exception as e: display.print_error(f"Failed: {e}") logger.exception("Resolve with outcome error") self._run_async(_resolve_with()) def do_status(self, arg): """ Display current game state. Usage: status """ async def _status(): try: gid = self._ensure_game() await self._ensure_game_loaded(gid) # Use shared command await game_commands.show_game_status(gid) except ValueError: pass except Exception as e: display.print_error(f"Failed: {e}") self._run_async(_status()) def do_rollback(self, arg): """ Roll back the last N plays. Usage: rollback Arguments: num_plays Number of plays to roll back (must be > 0) Deletes the specified number of plays from the database and reconstructs the game state by replaying the remaining plays. Also removes any substitutions that occurred during the rolled-back plays. Use this for correcting mistakes or recovering from corrupted plays. Examples: rollback 1 # Undo the last play rollback 3 # Undo the last 3 plays rollback 5 # Undo the last 5 plays """ async def _rollback(): try: gid = self._ensure_game() await self._ensure_game_loaded(gid) # Parse argument if not arg: display.print_error("Usage: rollback ") display.console.print("Example: [cyan]rollback 3[/cyan]") return try: num_plays = int(arg.strip()) except ValueError: display.print_error(f"Invalid number: {arg}") return # Use shared command await game_commands.rollback_plays(gid, num_plays) except ValueError: pass except Exception as e: display.print_error(f"Failed: {e}") logger.exception("Rollback error") self._run_async(_rollback()) def do_quick_play(self, arg): """ Auto-play multiple plays with default decisions. Usage: quick_play [COUNT] Examples: quick_play # Play 1 play quick_play 10 # Play 10 plays quick_play 27 # Play ~3 innings """ async def _quick_play(): try: gid = self._ensure_game() await self._ensure_game_loaded(gid) # Parse arguments with robust parser args = parse_quick_play_args(arg) # Execute quick play plays_completed = await game_commands.quick_play_rounds( game_id=gid, count=args['count'] ) display.print_success(f"Completed {plays_completed} plays") except ArgumentParseError as e: display.print_error(f"Invalid arguments: {e}") except ValueError: pass except Exception as e: display.print_error(f"Failed: {e}") logger.exception("Quick play error") self._run_async(_quick_play()) def do_box_score(self, arg): """ Display box score. Usage: box-score """ async def _box_score(): try: gid = self._ensure_game() await self._ensure_game_loaded(gid) # Use shared command await game_commands.show_box_score(gid) except ValueError: pass except Exception as e: display.print_error(f"Failed: {e}") self._run_async(_box_score()) def do_roll_dice(self, arg): """ Roll dice for manual outcome mode. Server rolls dice and displays results. Players then read their physical cards and submit the outcome using manual_outcome command. Usage: roll_dice Example: roll_dice # Read physical card based on dice manual_outcome groundball_c SS """ game_id = self.current_game_id if not game_id: display.print_error("No game selected. Create one with 'new_game'") return self._run_async(game_commands.roll_manual_dice(game_id)) def do_manual_outcome(self, arg): """ Submit manual outcome from physical card (manual mode). After rolling dice with 'roll_dice', read your physical card and submit the outcome you see. Usage: manual_outcome [location] Arguments: outcome PlayOutcome enum value (e.g., groundball_c, single_1) location Hit location (e.g., SS, 1B, LF) - required for groundballs/flyouts Examples: roll_dice # First, roll the dice manual_outcome strikeout # Submit outcome from card manual_outcome groundball_c SS manual_outcome flyout_b LF manual_outcome walk # Location not needed for walks Note: Must call 'roll_dice' first before submitting outcome. """ game_id = self.current_game_id if not game_id: display.print_error("No game selected. Create one with 'new_game'") return parts = arg.split() if not parts: display.print_error("Usage: manual_outcome [location]") display.console.print("[dim]Example: manual_outcome groundball_c SS[/dim]") display.console.print("[dim]Must call 'roll_dice' first[/dim]") return outcome = parts[0] location = parts[1] if len(parts) > 1 else None self._run_async(game_commands.submit_manual_outcome(game_id, outcome, location)) def do_test_location(self, arg): """ Test hit location distribution for an outcome. Usage: test_location [handedness] [count] Arguments: outcome PlayOutcome enum value (e.g., groundball_c) handedness 'L' or 'R' (default: R) count Number of samples (default: 100) Examples: test_location groundball_c test_location groundball_c L test_location flyout_b R 200 Shows distribution of hit locations based on pull rates. """ parts = arg.split() if not parts: display.print_error("Usage: test_location [handedness] [count]") display.console.print("[dim]Example: test_location groundball_c R 100[/dim]") return outcome = parts[0] handedness = parts[1] if len(parts) > 1 else 'R' count = int(parts[2]) if len(parts) > 2 else 100 # Validate handedness if handedness not in ['L', 'R']: display.print_error("Handedness must be 'L' or 'R'") return game_commands.test_hit_location(outcome, handedness, count) def do_list_games(self, arg): """ List all games in state manager. Usage: list-games """ games = state_manager.list_games() if not games: display.print_warning("No active games in memory") return display.print_info(f"Active games: {len(games)}") for game_id in games: marker = "* " if game_id == self.current_game_id else " " display.console.print(f"{marker}{game_id}") def do_use_game(self, arg): """ Switch to a different game. Usage: use_game Example: use_game a1b2c3d4-e5f6-7890-abcd-ef1234567890 """ try: # Parse arguments with robust parser args = parse_use_game_args(arg) gid = UUID(args['game_id']) self.current_game_id = gid Config.set_current_game(gid) display.print_success(f"Switched to game: {gid}") except ArgumentParseError as e: display.print_error(f"Invalid arguments: {e}") except ValueError: display.print_error(f"Invalid UUID: {arg}") def do_config(self, arg): """ Show configuration. Usage: config """ config_path = Config.get_config_path() display.print_info(f"Config file: {config_path}") if self.current_game_id: display.console.print(f"\n[green]Current game:[/green] {self.current_game_id}") else: display.console.print("\n[yellow]No current game set[/yellow]") # ==================== Enhanced Help System ==================== def do_help(self, arg): """ Show help for commands. Usage: help List all commands help Show detailed help for a command """ if arg: # Show detailed help for specific command show_help(arg) else: # Show command list HelpFormatter.show_command_list() def help_new_game(self): """Show detailed help for new_game command.""" show_help('new_game') def help_defensive(self): """Show detailed help for defensive command.""" show_help('defensive') def help_offensive(self): """Show detailed help for offensive command.""" show_help('offensive') def help_resolve(self): """Show detailed help for resolve command.""" show_help('resolve') def help_quick_play(self): """Show detailed help for quick_play command.""" show_help('quick_play') def help_status(self): """Show detailed help for status command.""" show_help('status') def help_box_score(self): """Show detailed help for box_score command.""" show_help('box_score') def help_list_games(self): """Show detailed help for list_games command.""" show_help('list_games') def help_use_game(self): """Show detailed help for use_game command.""" show_help('use_game') def help_config(self): """Show detailed help for config command.""" show_help('config') def help_clear(self): """Show detailed help for clear command.""" show_help('clear') # ==================== Interrupt Play Commands ==================== def do_force_wild_pitch(self, arg): """ Force a wild pitch interrupt play. Usage: force_wild_pitch Wild pitch advances all runners one base. This is an interrupt play (pa=0) and does not count as a plate appearance. Example: force_wild_pitch # Runner on 2nd advances to 3rd """ async def _force_wild_pitch(): try: gid = self._ensure_game() await self._ensure_game_loaded(gid) await game_commands.force_wild_pitch(gid) except ValueError: pass # Already printed error self._run_async(_force_wild_pitch()) def do_force_passed_ball(self, arg): """ Force a passed ball interrupt play. Usage: force_passed_ball Passed ball advances all runners one base. This is an interrupt play (pa=0) and does not count as a plate appearance. Example: force_passed_ball # Runner on 1st advances to 2nd """ async def _force_passed_ball(): try: gid = self._ensure_game() await self._ensure_game_loaded(gid) await game_commands.force_passed_ball(gid) except ValueError: pass # Already printed error self._run_async(_force_passed_ball()) # ==================== Jump Roll Testing Commands ==================== def do_roll_jump(self, arg): """ Roll jump dice for stolen base testing. Usage: roll_jump [league] Jump roll components: - 1d20 check roll (1=pickoff, 2=balk, 3+=normal) - 2d6 for normal jump (if check >= 3) - 1d20 resolution (if check == 1 or 2) Arguments: league 'sba' or 'pd' (default: sba) Examples: roll_jump # Roll for SBA league roll_jump pd # Roll for PD league The jump roll determines steal attempt outcomes: - Pickoff check (5%): Pitcher attempts to pick off runner - Balk check (5%): Pitcher may commit balk - Normal jump (90%): Use 2d6 total for steal success check """ parts = arg.split() league = parts[0] if parts else 'sba' if league not in ['sba', 'pd']: display.print_error("League must be 'sba' or 'pd'") return game_commands.roll_jump(league, self.current_game_id) def do_test_jump(self, arg): """ Test jump roll distribution. Usage: test_jump [count] [league] Rolls N jump rolls and displays distribution statistics including: - Pickoff check frequency (expected: 5%) - Balk check frequency (expected: 5%) - Normal jump frequency (expected: 90%) - Jump total distribution (2d6, expected avg: 7.0) Arguments: count Number of rolls (default: 10) league 'sba' or 'pd' (default: sba) Examples: test_jump # 10 rolls for SBA test_jump 100 # 100 rolls for SBA test_jump 50 pd # 50 rolls for PD """ parts = arg.split() count = int(parts[0]) if parts else 10 league = parts[1] if len(parts) > 1 else 'sba' if league not in ['sba', 'pd']: display.print_error("League must be 'sba' or 'pd'") return game_commands.test_jump(count, league) # ==================== Fielding Roll Testing Commands ==================== def do_roll_fielding(self, arg): """ Roll fielding check dice for testing. Usage: roll_fielding [league] Fielding roll components: - 1d20: Range check - 3d6: Error total (3-18) - 1d100: Rare play check Arguments: position P, C, 1B, 2B, 3B, SS, LF, CF, RF (required) league 'sba' or 'pd' (default: sba) Examples: roll_fielding SS # Roll for shortstop (SBA) roll_fielding P pd # Roll for pitcher (PD) roll_fielding CF # Roll for center field Rare plays: - SBA: d100 = 1 (1% chance) - PD: error_total = 5 (~2.78% chance) """ parts = arg.split() if not parts: display.print_error("Usage: roll_fielding [league]") display.console.print("[dim]Valid positions: P, C, 1B, 2B, 3B, SS, LF, CF, RF[/dim]") return position = parts[0] league = parts[1] if len(parts) > 1 else 'sba' if league not in ['sba', 'pd']: display.print_error("League must be 'sba' or 'pd'") return game_commands.roll_fielding(position, league, self.current_game_id) def do_test_fielding(self, arg): """ Test fielding roll distribution for a position. Usage: test_fielding [count] [league] Rolls N fielding rolls and displays distribution statistics including: - Rare play frequency - Average range roll (d20, expected: 10.5) - Average error total (3d6, expected: 10.5) - Error total distribution (3-18) Arguments: position P, C, 1B, 2B, 3B, SS, LF, CF, RF (required) count Number of rolls (default: 10) league 'sba' or 'pd' (default: sba) Examples: test_fielding SS # 10 rolls for shortstop (SBA) test_fielding 2B 100 # 100 rolls for second base test_fielding CF 50 pd # 50 rolls for center field (PD) """ parts = arg.split() if not parts: display.print_error("Usage: test_fielding [count] [league]") display.console.print("[dim]Valid positions: P, C, 1B, 2B, 3B, SS, LF, CF, RF[/dim]") return position = parts[0] count = int(parts[1]) if len(parts) > 1 else 10 league = parts[2] if len(parts) > 2 else 'sba' if league not in ['sba', 'pd']: display.print_error("League must be 'sba' or 'pd'") return game_commands.test_fielding(position, count, league) # ==================== REPL Control Commands ==================== def do_clear(self, arg): """Clear the screen.""" display.console.clear() def do_quit(self, arg): """Exit the REPL.""" display.print_info("Goodbye!") # Clean up event loop self.loop.close() return True def do_exit(self, arg): """Exit the REPL.""" return self.do_quit(arg) def do_EOF(self, arg): """Handle Ctrl+D.""" print() # New line return self.do_quit(arg) def emptyline(self): """Do nothing on empty line.""" pass def default(self, line): """Handle unknown commands.""" display.print_error(f"Unknown command: {line}") display.print_info("Type 'help' for available commands") def start_repl(): """Start the interactive REPL.""" repl = GameREPL() repl.cmdloop() if __name__ == "__main__": start_repl()