● Terminal Client Improvement Plan - Part 5: Player Name Caching and Display Overview Enhance the display system to show player names, positions, and stats instead of just lineup IDs. This requires integration with the player model system from Week 6 and adds a caching layer for performance. Note: This enhancement requires Week 6 player models to be implemented first. The code provided here creates the infrastructure that will be activated once player models are available. Files to Create 1. Create backend/terminal_client/player_cache.py """ Player data caching for terminal client. Caches player/card data from league APIs to avoid repeated lookups and provide fast name/stat display in the REPL. Author: Claude Date: 2025-10-27 """ import logging from typing import Dict, Optional, Any from uuid import UUID from dataclasses import dataclass from datetime import datetime, timedelta logger = logging.getLogger(f'{__name__}.player_cache') @dataclass class CachedPlayer: """Cached player data.""" # Identification card_id: Optional[int] = None player_id: Optional[int] = None # Basic info name: str = "Unknown" position: str = "?" team_id: int = 0 # Display info image_url: Optional[str] = None handedness: Optional[str] = None # League-specific league: str = "unknown" # Cache metadata cached_at: datetime = None def get_display_name(self) -> str: """Get formatted display name.""" if self.handedness: hand_symbol = {"R": "⟩", "L": "⟨", "S": "⟨⟩"}.get(self.handedness, "") return f"{self.name} {hand_symbol} ({self.position})" return f"{self.name} ({self.position})" def get_short_display(self) -> str: """Get short display format.""" return f"{self.name} {self.position}" class PlayerCache: """ Cache for player data to avoid repeated API lookups. Maintains in-memory cache with TTL and per-game player rosters. """ def __init__(self, ttl_seconds: int = 3600): """ Initialize player cache. Args: ttl_seconds: Time to live for cached entries (default: 1 hour) """ self.ttl_seconds = ttl_seconds # Cache by card_id (PD league) self._card_cache: Dict[int, CachedPlayer] = {} # Cache by player_id (SBA league) self._player_cache: Dict[int, CachedPlayer] = {} # Cache by lineup_id for quick lookups self._lineup_cache: Dict[int, CachedPlayer] = {} # Track which game uses which players self._game_rosters: Dict[UUID, Dict[int, CachedPlayer]] = {} logger.info(f"PlayerCache initialized with TTL={ttl_seconds}s") def get_by_lineup_id(self, lineup_id: int) -> Optional[CachedPlayer]: """ Get player by lineup ID (fastest lookup for display). Args: lineup_id: Lineup entry ID Returns: CachedPlayer or None if not cached """ player = self._lineup_cache.get(lineup_id) if player and self._is_expired(player): logger.debug(f"Lineup cache expired for {lineup_id}") del self._lineup_cache[lineup_id] return None return player def get_by_card_id(self, card_id: int) -> Optional[CachedPlayer]: """ Get player by card ID (PD league). Args: card_id: Card ID Returns: CachedPlayer or None if not cached """ player = self._card_cache.get(card_id) if player and self._is_expired(player): logger.debug(f"Card cache expired for {card_id}") del self._card_cache[card_id] return None return player def get_by_player_id(self, player_id: int) -> Optional[CachedPlayer]: """ Get player by player ID (SBA league). Args: player_id: Player ID Returns: CachedPlayer or None if not cached """ player = self._player_cache.get(player_id) if player and self._is_expired(player): logger.debug(f"Player cache expired for {player_id}") del self._player_cache[player_id] return None return player def add_player( self, player_data: Dict[str, Any], lineup_id: Optional[int] = None ) -> CachedPlayer: """ Add player to cache. Args: player_data: Player data from database/API lineup_id: Optional lineup ID for quick lookups Returns: CachedPlayer instance """ now = datetime.utcnow() player = CachedPlayer( card_id=player_data.get('card_id'), player_id=player_data.get('player_id'), name=player_data.get('name', 'Unknown'), position=player_data.get('position', '?'), team_id=player_data.get('team_id', 0), image_url=player_data.get('image_url'), handedness=player_data.get('handedness'), league=player_data.get('league', 'unknown'), cached_at=now ) # Cache by appropriate ID if player.card_id: self._card_cache[player.card_id] = player logger.debug(f"Cached card {player.card_id}: {player.name}") if player.player_id: self._player_cache[player.player_id] = player logger.debug(f"Cached player {player.player_id}: {player.name}") # Cache by lineup ID if provided if lineup_id: self._lineup_cache[lineup_id] = player logger.debug(f"Cached lineup {lineup_id}: {player.name}") return player async def load_game_roster(self, game_id: UUID, league: str) -> int: """ Load all players for a game into cache. This should be called when switching to a game to pre-populate the cache with all players that will be referenced. Args: game_id: Game UUID league: League ID ('sba' or 'pd') Returns: Number of players loaded """ # Import here to avoid circular dependency from app.database.operations import DatabaseOperations db_ops = DatabaseOperations() try: # Get both team lineups home_lineup = await db_ops.get_active_lineup(game_id, None) # Will need team_id away_lineup = await db_ops.get_active_lineup(game_id, None) all_players = home_lineup + away_lineup game_roster = {} for lineup_entry in all_players: # Extract player data based on league player_data = { 'card_id': getattr(lineup_entry, 'card_id', None), 'player_id': getattr(lineup_entry, 'player_id', None), 'name': getattr(lineup_entry, 'name', 'Unknown'), 'position': lineup_entry.position, 'team_id': lineup_entry.team_id, 'league': league } # Add to cache player = self.add_player(player_data, lineup_id=lineup_entry.id) game_roster[lineup_entry.id] = player # Store game roster mapping self._game_rosters[game_id] = game_roster logger.info(f"Loaded {len(all_players)} players for game {game_id}") return len(all_players) except Exception as e: logger.error(f"Failed to load game roster: {e}", exc_info=True) return 0 def get_game_roster(self, game_id: UUID) -> Dict[int, CachedPlayer]: """ Get cached roster for a game. Args: game_id: Game UUID Returns: Dictionary mapping lineup_id to CachedPlayer """ return self._game_rosters.get(game_id, {}) def clear_game(self, game_id: UUID) -> None: """ Clear cached data for a specific game. Args: game_id: Game UUID """ if game_id in self._game_rosters: del self._game_rosters[game_id] logger.info(f"Cleared cache for game {game_id}") def clear_expired(self) -> int: """ Remove expired entries from cache. Returns: Number of entries removed """ count = 0 # Clear card cache expired_cards = [ cid for cid, player in self._card_cache.items() if self._is_expired(player) ] for cid in expired_cards: del self._card_cache[cid] count += 1 # Clear player cache expired_players = [ pid for pid, player in self._player_cache.items() if self._is_expired(player) ] for pid in expired_players: del self._player_cache[pid] count += 1 # Clear lineup cache expired_lineups = [ lid for lid, player in self._lineup_cache.items() if self._is_expired(player) ] for lid in expired_lineups: del self._lineup_cache[lid] count += 1 if count > 0: logger.info(f"Cleared {count} expired cache entries") return count def get_stats(self) -> Dict[str, int]: """ Get cache statistics. Returns: Dictionary with cache stats """ return { 'card_cache_size': len(self._card_cache), 'player_cache_size': len(self._player_cache), 'lineup_cache_size': len(self._lineup_cache), 'games_cached': len(self._game_rosters), 'ttl_seconds': self.ttl_seconds } def _is_expired(self, player: CachedPlayer) -> bool: """Check if cached player is expired.""" if player.cached_at is None: return True age = (datetime.utcnow() - player.cached_at).total_seconds() return age > self.ttl_seconds # Global cache instance player_cache = PlayerCache() 2. Create backend/terminal_client/enhanced_display.py """ Enhanced display functions with player names. Extends the base display module with player name lookups and richer formatting. Author: Claude Date: 2025-10-27 """ import logging from typing import Optional from rich.text import Text from app.models.game_models import GameState from terminal_client import display from terminal_client.player_cache import player_cache, CachedPlayer logger = logging.getLogger(f'{__name__}.enhanced_display') def get_player_display(lineup_id: Optional[int]) -> str: """ Get display string for a player by lineup ID. Args: lineup_id: Lineup entry ID Returns: Formatted player string (name if available, ID otherwise) """ if lineup_id is None: return "None" # Try to get from cache player = player_cache.get_by_lineup_id(lineup_id) if player: return player.get_short_display() # Fall back to lineup ID return f"Lineup #{lineup_id}" def display_game_state_enhanced(state: GameState) -> None: """ Display game state with player names (enhanced version). This is a drop-in replacement for display.display_game_state() that adds player names when available. Args: state: Current game state """ from rich.panel import Panel from rich import box # Status color based on game status status_color = { "pending": "yellow", "active": "green", "paused": "yellow", "completed": "blue" } # Build state display state_text = Text() state_text.append(f"Game ID: {state.game_id}\n", style="bold") state_text.append(f"League: {state.league_id.upper()}\n") state_text.append(f"Status: ", style="bold") state_text.append(f"{state.status}\n", style=status_color.get(state.status, "white")) state_text.append("\n") # Score state_text.append("Score: ", style="bold cyan") state_text.append(f"Away {state.away_score} - {state.home_score} Home\n", style="cyan") # Inning state_text.append("Inning: ", style="bold magenta") state_text.append(f"{state.inning} {state.half.capitalize()}\n", style="magenta") # Outs state_text.append("Outs: ", style="bold yellow") state_text.append(f"{state.outs}\n", style="yellow") # Runners (enhanced with names) runners = state.get_all_runners() if runners: state_text.append("\nRunners: ", style="bold green") runner_displays = [] for base, runner in runners: player_name = get_player_display(runner.lineup_id) runner_displays.append(f"{base}B: {player_name}") state_text.append(f"{', '.join(runner_displays)}\n", style="green") else: state_text.append("\nBases: ", style="bold") state_text.append("Empty\n", style="dim") # Current players (enhanced with names) if state.current_batter_lineup_id: batter_name = get_player_display(state.current_batter_lineup_id) state_text.append(f"\nBatter: {batter_name}\n") if state.current_pitcher_lineup_id: pitcher_name = get_player_display(state.current_pitcher_lineup_id) state_text.append(f"Pitcher: {pitcher_name}\n") # Pending decision if state.pending_decision: state_text.append(f"\nPending: ", style="bold red") state_text.append(f"{state.pending_decision} decision\n", style="red") # Last play result if state.last_play_result: state_text.append(f"\nLast Play: ", style="bold") state_text.append(f"{state.last_play_result}\n", style="italic") # Display panel panel = Panel( state_text, title=f"[bold]Game State[/bold]", border_style="blue", box=box.ROUNDED ) display.console.print(panel) def show_cache_stats() -> None: """Display player cache statistics.""" stats = player_cache.get_stats() display.console.print("\n[bold cyan]Player Cache Statistics:[/bold cyan]") display.console.print(f" Card cache: {stats['card_cache_size']} entries") display.console.print(f" Player cache: {stats['player_cache_size']} entries") display.console.print(f" Lineup cache: {stats['lineup_cache_size']} entries") display.console.print(f" Games cached: {stats['games_cached']}") display.console.print(f" TTL: {stats['ttl_seconds']} seconds\n") Files to Update 3. Update backend/terminal_client/repl.py # Add imports at top from terminal_client.player_cache import player_cache from terminal_client.enhanced_display import ( display_game_state_enhanced, show_cache_stats ) class GameREPL(GameREPLCompletions, cmd.Cmd): # ... existing code ... 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. Also pre-load player cache for better display. """ # Check if already in memory state = state_manager.get_state(game_id) if state is not None: # Game loaded, check if cache needs refresh roster = player_cache.get_game_roster(game_id) if not roster: # Load player names into cache try: count = await player_cache.load_game_roster(game_id, state.league_id) if count > 0: logger.debug(f"Loaded {count} players into cache") except Exception as e: logger.warning(f"Could not load player cache: {e}") return # 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": await game_engine._prepare_next_play(recovered_state) logger.debug(f"Prepared snapshot for recovered game {game_id}") # Load player cache try: count = await player_cache.load_game_roster( game_id, recovered_state.league_id ) if count > 0: display.print_success(f"Loaded {count} players into display cache") except Exception as e: logger.warning(f"Could not load player cache: {e}") 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") # Add new command for cache management def do_cache(self, arg): """ Manage player cache. Usage: cache Show cache statistics cache clear Clear expired entries cache reload Reload current game roster """ async def _cache(): args = arg.split() if not args or args[0] == 'stats': # Show stats show_cache_stats() elif args[0] == 'clear': # Clear expired count = player_cache.clear_expired() display.print_success(f"Cleared {count} expired entries") elif args[0] == 'reload': # Reload current game try: gid = self._ensure_game() state = state_manager.get_state(gid) if state: count = await player_cache.load_game_roster(gid, state.league_id) display.print_success(f"Loaded {count} players") else: display.print_error("No game state found") except ValueError: pass else: display.print_error(f"Unknown cache command: {args[0]}") display.print_info("Usage: cache [stats|clear|reload]") self._run_async(_cache()) # Update status command to use enhanced display def do_status(self, arg): """Display current game state with player names.""" async def _status(): try: gid = self._ensure_game() await self._ensure_game_loaded(gid) state = await game_engine.get_game_state(gid) if state: # Use enhanced display if cache is populated roster = player_cache.get_game_roster(gid) if roster: display_game_state_enhanced(state) else: display.display_game_state(state) else: display.print_error("Game state not found") except ValueError: pass except Exception as e: display.print_error(f"Failed: {e}") self._run_async(_status()) Testing Plan 4. Create backend/tests/unit/terminal_client/test_player_cache.py """ Unit tests for player cache. """ import pytest from datetime import datetime, timedelta from uuid import uuid4 from terminal_client.player_cache import PlayerCache, CachedPlayer class TestCachedPlayer: """Tests for CachedPlayer dataclass.""" def test_get_display_name_with_handedness(self): """Test display name includes handedness symbol.""" player = CachedPlayer( name="Mike Trout", position="CF", handedness="R" ) assert "⟩" in player.get_display_name() assert "Mike Trout" in player.get_display_name() assert "CF" in player.get_display_name() def test_get_display_name_without_handedness(self): """Test display name without handedness.""" player = CachedPlayer( name="Mike Trout", position="CF" ) result = player.get_display_name() assert result == "Mike Trout (CF)" def test_get_short_display(self): """Test short display format.""" player = CachedPlayer( name="Mike Trout", position="CF" ) assert player.get_short_display() == "Mike Trout CF" class TestPlayerCache: """Tests for PlayerCache.""" @pytest.fixture def cache(self): """Create fresh cache for each test.""" return PlayerCache(ttl_seconds=3600) def test_add_and_get_by_card_id(self, cache): """Test adding and retrieving by card ID.""" player_data = { 'card_id': 123, 'name': 'Test Player', 'position': 'SS', 'team_id': 1, 'league': 'pd' } player = cache.add_player(player_data) assert player.name == 'Test Player' assert player.position == 'SS' # Retrieve by card ID retrieved = cache.get_by_card_id(123) assert retrieved is not None assert retrieved.name == 'Test Player' def test_add_and_get_by_player_id(self, cache): """Test adding and retrieving by player ID.""" player_data = { 'player_id': 456, 'name': 'Test Player', 'position': 'CF', 'team_id': 2, 'league': 'sba' } player = cache.add_player(player_data) # Retrieve by player ID retrieved = cache.get_by_player_id(456) assert retrieved is not None assert retrieved.name == 'Test Player' def test_add_with_lineup_id(self, cache): """Test adding with lineup ID for quick lookups.""" player_data = { 'card_id': 123, 'name': 'Test Player', 'position': 'SS', 'team_id': 1, 'league': 'pd' } player = cache.add_player(player_data, lineup_id=10) # Retrieve by lineup ID retrieved = cache.get_by_lineup_id(10) assert retrieved is not None assert retrieved.name == 'Test Player' def test_expired_entry_removed(self, cache): """Test that expired entries are not returned.""" cache_short_ttl = PlayerCache(ttl_seconds=0) player_data = { 'card_id': 123, 'name': 'Test Player', 'position': 'SS', 'team_id': 1, 'league': 'pd' } player = cache_short_ttl.add_player(player_data) player.cached_at = datetime.utcnow() - timedelta(seconds=10) # Should return None for expired entry retrieved = cache_short_ttl.get_by_card_id(123) assert retrieved is None def test_clear_expired(self, cache): """Test clearing expired entries.""" # Add fresh entry cache.add_player({ 'card_id': 1, 'name': 'Fresh', 'position': 'P', 'team_id': 1, 'league': 'pd' }) # Add expired entry old_player = cache.add_player({ 'card_id': 2, 'name': 'Old', 'position': 'C', 'team_id': 1, 'league': 'pd' }) old_player.cached_at = datetime.utcnow() - timedelta(seconds=10000) # Clear expired count = cache.clear_expired() assert count > 0 assert cache.get_by_card_id(1) is not None # Fresh still there assert cache.get_by_card_id(2) is None # Old removed def test_get_stats(self, cache): """Test getting cache statistics.""" cache.add_player({'card_id': 1, 'name': 'P1', 'position': 'P', 'team_id': 1, 'league': 'pd'}) cache.add_player({'player_id': 2, 'name': 'P2', 'position': 'C', 'team_id': 1, 'league': 'sba'}) stats = cache.get_stats() assert stats['card_cache_size'] == 1 assert stats['player_cache_size'] == 1 assert stats['ttl_seconds'] == 3600 def test_clear_game(self, cache): """Test clearing game-specific cache.""" game_id = uuid4() # Simulate game roster cache._game_rosters[game_id] = { 1: CachedPlayer(name="Player 1", position="P"), 2: CachedPlayer(name="Player 2", position="C") } assert len(cache.get_game_roster(game_id)) == 2 cache.clear_game(game_id) assert len(cache.get_game_roster(game_id)) == 0 Configuration and Activation 5. Create backend/terminal_client/config.py update # Add to existing config.py class ClientConfig: """Configuration for terminal client features.""" # Feature flags ENABLE_PLAYER_NAMES = False # Set to True when Week 6 models are ready ENABLE_PLAYER_CACHE = False # Set to True when Week 6 models are ready # Cache settings PLAYER_CACHE_TTL = 3600 # 1 hour AUTO_LOAD_ROSTER = True # Auto-load roster when switching games @classmethod def is_enhanced_display_enabled(cls) -> bool: """Check if enhanced display with player names is enabled.""" return cls.ENABLE_PLAYER_NAMES and cls.ENABLE_PLAYER_CACHE Activation Instructions (For Future) When Week 6 player models are implemented: 1. Update feature flags: # In terminal_client/config.py ENABLE_PLAYER_NAMES = True ENABLE_PLAYER_CACHE = True 2. Implement player data fetching: - Add methods to fetch player data from league APIs - Integrate with backend/app/data/api_client.py 3. Update database operations: - Ensure lineup queries return player names - Add joins to get player/card details 4. Test with real data: - Verify cache performance - Check TTL behavior - Validate display formatting Example Output (When Activated) Before (Current): Batter: Lineup #1 Pitcher: Lineup #10 Runners: 1B(#2), 3B(#5) After (With Player Names): Batter: Mike Trout CF Pitcher: Clayton Kershaw P Runners: 1B: Aaron Judge RF, 3B: Mookie Betts RF Benefits 1. Better UX: See player names instead of cryptic IDs 2. Performance: Caching prevents repeated database/API lookups 3. Flexibility: Works with both PD and SBA leagues 4. Future-ready: Infrastructure ready for Week 6 integration 5. Optional: Can be disabled if not needed