""" Decoupled game engine for BaoLife. Provides a synchronous, testable game loop that can run without WebSocket dependencies. This module extracts the core game logic from initLifeSim() for testing and CLI use. """ import random from datetime import date from typing import Optional from storage import get_storage, IGameStorage from output import get_output, IGameOutput from services import get_conversation_service, IConversationService from config import config class GameEngine: """ Decoupled game engine that can run with or without WebSocket. This class provides a synchronous interface to the game loop, making it testable and usable from CLI. """ def __init__(self, storage: Optional[IGameStorage] = None, output: Optional[IGameOutput] = None, conversation_service: Optional[IConversationService] = None): """ Initialize the game engine. Args: storage: Storage backend (defaults to factory-created) output: Output backend (defaults to factory-created) conversation_service: Conversation service (defaults to factory-created) """ self.storage = storage or get_storage() self.output = output or get_output() self.conversation_service = conversation_service or get_conversation_service() async def run_game_tick(self, player, force_update: bool = False): """ Run one game tick for a player. This is the core game loop logic extracted from initLifeSim(). Args: player: The player object to update force_update: Force update regardless of gameSpeed Returns: Updated player object """ from functions import ( handleDeath, updateAge, get_season, handleFinances, handleMoods, handleEducation, handleJob, handleRelationships, handleHabitChanges, set_avatar, getPeakEnergy, updateDeathChance, getIntradayActivity, parseLocations, updateBio, messageFunction ) from stats.stats_manager import checkEvents, checkDilemmas, checkDayEvents, parseOneTimeEvents from intradayActivity import get_dailyPlan player.ticks += 1 # Check if we should run this tick if player.ticks % player.gameSpeed != 0 and not force_update: return player # Check if game is active if player.controller != 'active' and not force_update: return player # Check if player is creating character if player.status == "creating" and not force_update: return player # Handle death if player.c.status == "dead": handleDeath(player) self.storage.save_game(player) return player # Increment time player.minuteOfHour += 1 player.time = str(player.hourOfDay) + ":" + str(player.minuteOfHour) if player.minuteOfHour == 60: player.minuteOfHour = 0 # Hourly ticks if player.minuteOfHour == 0: updateObject = { 'date': player.date, 'hourOfDay': player.hourOfDay, 'minuteOfHour': player.minuteOfHour, 'weekDayText': player.weekDayText, 'energy': player.c.energy, 'calcEnergy': player.c.calcEnergy, 'money': player.c.money, 'diamonds': player.c.diamonds, 'prestige': player.c.prestige, 'stress': player.c.stress, 'happiness': player.c.happiness, 'occupation': player.c.occupation, 'location': player.c.location, 'schedules': player.c.schedules, 'intraDayMessage': player.c.intraDayMessage, 'dailyPlan': player.c.dailyPlan, 'gameSpeed': player.gameSpeed, } player.message = False player.hourOfDay += 1 # Update age result = updateAge(player) if hasattr(result, 'type') and result.type == 'messageEvent': await self.output.send_event_message(result) # Daily ticks (at hour 24) if player.hourOfDay == 24: player.hourOfDay = 0 # Update day counter if player.dayOfYear == 365: player.dayOfYear = 1 else: player.dayOfYear += 1 if player.dayOfWeek == 7: player.dayOfWeek = 1 else: player.dayOfWeek += 1 # Update date player.date = date.fromordinal( date(2022, 1, 1).toordinal() + player.dayOfYear - 1 ).strftime('%m-%d') player.monthOfYear = int(player.date.split('-')[0]) player.season = get_season(player.monthOfYear) player.date = str(player.date) player.time = str(player.hourOfDay) + ":00" player.weekDayText = ["Sunday", "Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday"][player.dayOfWeek - 1] # School days calculation player.daysUntilSchoolEnds = (152 - player.dayOfYear) if player.dayOfYear > 244: player.daysUntilSchoolEnds = (365 - 244) + player.daysUntilSchoolEnds player.daysSinceSchoolStarted = (player.dayOfYear - 244) if player.dayOfYear < 244: player.daysSinceSchoolStarted = (121 + player.dayOfYear) if player.dayOfYear > 152 and player.dayOfYear < 244: player.summerVacation = True else: player.summerVacation = False # Weekly ticks (Monday at hour 0) if player.dayOfWeek == 1 and player.hourOfDay == 0: for i in range(0, len(player.r)): handleFinances(player.r[i]) handleMoods(player, player.r[i]) handleEducation(player.r[i]) handleJob(player, player.r[i]) handleRelationships(player, player.r[i]) player.r[i].image = set_avatar(player.r[i]) handleFinances(player.c) handleMoods(player, player.c) handleRelationships(player, player.c) handleJob(player, player.c) handleEducation(player.c) handleHabitChanges(player, player.c) player.c.image = set_avatar(player.c) self.storage.save_game(player) await self.output.send_user_info(player) # Daily events check (at hour 0) if player.hourOfDay == 0: player.dayEvent = False getPeakEnergy(player.c) player.c.energy += 1 if player.c.energy < 100 else 0 # First day message if player.c.ageDays == 1 and player.c.firstname: player.message = (player.c.firstname.capitalize() + " is starting their life, full of opportunities.") await self.output.send_user_info(player) # Birthday if player.c.ageDays > 0 and player.c.ageDays % 365 == 0: player.c.ageYears += 1 player.c.deathChance = updateDeathChance(player.c) player.message = (player.c.firstname.capitalize() + " is " + str(player.c.ageYears) + " years old.") self.storage.save_game(player) await self.output.send_user_info(player) # Death check if player.c.deathChance * 100 > (random.random() * 100) or player.c.ageYears > 120: player.c.status = "dead" player.message = (player.c.firstname.capitalize() + " has died at the age of " + str(player.c.ageYears) + " years.") await self.output.send_user_info(player) # Check day events result = checkDayEvents(player, 'check') if result: if result.type == 'messageEvent': player.events.add(result.id) await self.output.send_event_message(result) elif result.type == 'questionEvent': await self.output.send_event_message(result) # Generate daily plans player.c = get_dailyPlan(player, player.c) if player.gameSpeed > 10: updateObject['dailyPlan'] = player.c.dailyPlan # Daily plans for relationships for i in range(0, len(player.r)): if player.r[i].status == "alive" and player.r[i].familiarity > 0: player.r[i].familiarity = player.r[i].familiarity - 3 player.r[i] = get_dailyPlan(player, player.r[i]) # Intraday activities player.c = getIntradayActivity(player, player.c) for i in range(0, len(player.r)): player.r[i] = getIntradayActivity(player, player.r[i]) player = parseLocations(player) parseOneTimeEvents(player) player = updateBio(player) # Message queue if player.messageQueue and len(player.messageQueue) > 0: player.message = player.messageQueue.pop(0) player.messageLog.append(player.message) await self.output.send_event_message( messageFunction('queue', player.message, player, True) ) player.message = "" # Update client if player.updateClient: await self.output.send_user_info(player) player.updateClient = False # Check regular events if player.gameSpeed < config.SPEED_QUESTION_PAUSE: result = checkEvents(player, 'check') if result: if hasattr(result, 'type'): if result.type == 'messageEvent': player.events.add(result.id) await self.output.send_event_message(result) elif result.type == 'questionEvent': await self.output.send_event_message(result) # Check dilemmas result = checkDilemmas(player) if result: await self.output.send_event_message(result) # Send update object if updateObject and updateObject != {}: await self.output.send_dict(updateObject) elif player.gameSpeed > 10: await self.output.send_dict({'type': 'u', 'minuteOfHour': player.minuteOfHour}) return player def run_game_tick_sync(self, player, force_update: bool = False): """ Synchronous wrapper for run_game_tick. For testing and CLI use where async is not needed. Args: player: The player object to update force_update: Force update regardless of gameSpeed Returns: Updated player object """ import asyncio try: # Check if there's a running loop asyncio.get_running_loop() # If we're here, there's already a loop running # This is typically in a test context with pytest-asyncio raise RuntimeError("Cannot use run_game_tick_sync from async context. Use await run_game_tick() instead.") except RuntimeError as e: if "no running event loop" in str(e).lower(): # No running loop, create a new one return asyncio.run(self.run_game_tick(player, force_update)) else: # Already in async context raise