""" Headless game engine for BaoLife. Provides a CLI interface to run game simulations without WebSocket, useful for testing and automation. """ import sys import os import asyncio from typing import Optional, List, Callable from datetime import date, timedelta # Set test mode environment variable os.environ['BAOLIFE_TEST_MODE'] = 'true' from game_engine import GameEngine from tests.mocks.storage_mock import InMemoryStorage from tests.mocks.output_mock import MockGameOutput from tests.mocks.services_mock import MockConversationService from output import CollectorOutput class HeadlessGame: """ Headless game engine that runs without WebSocket. Allows running game simulations from CLI or tests with full control over time progression and event handling. """ def __init__(self, player=None, player_id: Optional[str] = None, use_real_storage: bool = False): """ Initialize the headless game. Args: player: Optional existing player object player_id: Optional player ID to load from storage use_real_storage: Use real MySQL storage (default: in-memory) """ # Set up storage if use_real_storage: from storage import MySQLStorage self.storage = MySQLStorage() else: self.storage = InMemoryStorage() # Set up output collector self.output = CollectorOutput() # Set up conversation service self.conversation_service = MockConversationService() self.conversation_service.set_response( "This is a test conversation response." ) # Create game engine self.engine = GameEngine( storage=self.storage, output=self.output, conversation_service=self.conversation_service ) # Load or set player if player: self.player = player elif player_id: self.player = self.storage.load_game(player_id) if not self.player: raise ValueError(f"No game found for player ID: {player_id}") else: # Create a new test player from tests.fixtures.player_fixtures import create_newborn_player self.player = create_newborn_player() # Event loop self.loop = asyncio.new_event_loop() asyncio.set_event_loop(self.loop) def tick(self, count: int = 1, force: bool = True) -> 'HeadlessGame': """ Advance the game by N ticks. Args: count: Number of ticks to advance force: Force update regardless of gameSpeed Returns: Self for method chaining """ for _ in range(count): self.player = self.loop.run_until_complete( self.engine.run_game_tick(self.player, force_update=force) ) return self def advance_to_date(self, target_date: date, max_ticks: int = 100000) -> 'HeadlessGame': """ Fast-forward to a specific date. Args: target_date: Target date to reach max_ticks: Maximum ticks to prevent infinite loops Returns: Self for method chaining Raises: RuntimeError: If max_ticks exceeded """ from datetime import datetime ticks = 0 while ticks < max_ticks: # Parse current date current_date = datetime.strptime(self.player.date, '%m-%d').date() current_date = current_date.replace(year=target_date.year) if current_date >= target_date: break self.tick(1, force=True) ticks += 1 if ticks >= max_ticks: raise RuntimeError(f"Max ticks exceeded while advancing to {target_date}") return self def advance_to_hour(self, target_hour: int) -> 'HeadlessGame': """ Advance to a specific hour of the day. Args: target_hour: Target hour (0-23) Returns: Self for method chaining """ max_ticks = 24 * 60 # One day's worth of minutes ticks = 0 while self.player.hourOfDay != target_hour and ticks < max_ticks: self.tick(1, force=True) ticks += 1 return self def advance_to_event(self, event_name: str, max_ticks: int = 10000, check_fn: Optional[Callable] = None) -> 'HeadlessGame': """ Run until a specific event triggers. Args: event_name: Name of event to wait for max_ticks: Maximum ticks before giving up check_fn: Optional custom function to check for event Returns: Self for method chaining Raises: RuntimeError: If max_ticks exceeded without finding event """ ticks = 0 if check_fn is None: # Default: check if event is in player.events def default_check(): return event_name in self.player.events check_fn = default_check while not check_fn() and ticks < max_ticks: self.tick(1, force=True) ticks += 1 if not check_fn(): raise RuntimeError( f"Event '{event_name}' not found after {max_ticks} ticks" ) return self def advance_minutes(self, minutes: int) -> 'HeadlessGame': """ Advance by a specific number of game minutes. Args: minutes: Number of minutes to advance Returns: Self for method chaining """ self.tick(minutes, force=True) return self def advance_hours(self, hours: int) -> 'HeadlessGame': """ Advance by a specific number of game hours. Args: hours: Number of hours to advance Returns: Self for method chaining """ return self.advance_minutes(hours * 60) def advance_days(self, days: int) -> 'HeadlessGame': """ Advance by a specific number of game days. Args: days: Number of days to advance Returns: Self for method chaining """ return self.advance_minutes(days * 24 * 60) def get_events(self) -> List[dict]: """ Get all events that have occurred. Returns: List of event dictionaries """ return self.output.get_events() def get_questions(self) -> List[dict]: """ Get all questions that have been asked. Returns: List of question dictionaries """ return self.output.get_questions() def get_player_updates(self) -> List[dict]: """ Get all player updates that have been sent. Returns: List of player update dictionaries """ return self.output.get_player_updates() def answer_question(self, question_id: str, response: str) -> 'HeadlessGame': """ Answer a pending question. Args: question_id: ID of the question to answer response: The response/answer Returns: Self for method chaining """ # This would integrate with the question handling system # For now, we'll just record the answer # Implementation would depend on how questions are handled in the game # Remove from asked questions if present if question_id in self.player.askedQuestions: self.player.askedQuestions.remove(question_id) return self def get_last_event(self) -> Optional[dict]: """ Get the most recent event. Returns: Event dictionary or None """ events = self.get_events() return events[-1] if events else None def get_last_question(self) -> Optional[dict]: """ Get the most recent question. Returns: Question dictionary or None """ questions = self.get_questions() return questions[-1] if questions else None def has_event_with_text(self, text: str) -> bool: """ Check if any event contains specific text. Args: text: Text to search for Returns: True if found, False otherwise """ return self.output.collector.has_event_with_text(text) def save(self) -> 'HeadlessGame': """ Save the current game state. Returns: Self for method chaining """ self.storage.save_game(self.player) return self def reset_output(self) -> 'HeadlessGame': """ Clear all collected output. Returns: Self for method chaining """ self.output.clear() return self def get_player(self): """ Get the current player object. Returns: Player object """ return self.player def close(self): """Clean up resources.""" self.loop.close() def __repr__(self): return (f"HeadlessGame(player={self.player.c.firstname if self.player else 'None'}, " f"date={self.player.date if self.player else 'None'}, " f"hour={self.player.hourOfDay if self.player else 'None'})") def __enter__(self): """Context manager entry.""" return self def __exit__(self, exc_type, exc_val, exc_tb): """Context manager exit.""" self.close() def main(): """ Main CLI entry point for running headless simulations. Example usage: python cli_runner.py --player-id test123 --days 30 """ import argparse parser = argparse.ArgumentParser(description='Run BaoLife game simulation headlessly') parser.add_argument('--player-id', help='Player ID to load') parser.add_argument('--days', type=int, default=1, help='Days to simulate') parser.add_argument('--real-storage', action='store_true', help='Use real MySQL storage instead of in-memory') parser.add_argument('--verbose', action='store_true', help='Verbose output') args = parser.parse_args() # Create headless game with HeadlessGame(player_id=args.player_id, use_real_storage=args.real_storage) as game: print(f"Starting simulation: {game}") # Run simulation game.advance_days(args.days) print(f"Simulation complete: {game}") if args.verbose: print(f"\nEvents: {len(game.get_events())}") print(f"Questions: {len(game.get_questions())}") for event in game.get_events(): print(f" - {event.get('description', 'No description')}") # Save if using real storage if args.real_storage: game.save() print("Game saved to database") if __name__ == '__main__': main()