"""Main application entry point""" import asyncio import argparse import sys from pathlib import Path # Add src to path sys.path.insert(0, str(Path(__file__).parent)) from core.telnet_client import TelnetClient from core.ansi_parser import ANSIParser from core.game_state import GameState from core.parsers import ParserFactory, OutputBuffer, EventType from config.settings import load_config from utils.logger import setup_logger from utils.human_behavior import HumanBehavior from modules.decision import DecisionEngine from modules.decision.combat_ai import CombatConfig from modules.mapper import WorldMapper from modules.character import CharacterCreator, CreatorConfig from modules.quest import QuestManager from modules.progression import ProgressionOptimizer from modules.llm import LLMBrain, LLMConfig, SocialAI logger = None async def main_loop(config): """Main game loop""" global logger # Initialize components telnet = TelnetClient(config.mud.host, config.mud.port) ansi_parser = ANSIParser() game_state = GameState() human_behavior = HumanBehavior( config.behavior.reaction_time_min_ms, config.behavior.reaction_time_max_ms, config.behavior.pause_probability, ) # Create MUD-specific parser via factory mud_parser = ParserFactory.create(config.mud.mud_type) output_buffer = OutputBuffer(mud_parser) # Initialize Decision Engine (Phase 4) combat_cfg = CombatConfig() decision_engine = DecisionEngine( combat_config=combat_cfg, exploration_strategy=config.ai.exploration_strategy, ) decision_engine.goal_manager.set_default_goals() # Initialize WorldMapper (Phase 3 — hybrid SQLite/Memvid) world_mapper = WorldMapper(db_path=config.mapper.active_storage.db_path) # Initialize Character Creator (Phase 2) creator_cfg = CreatorConfig( preferred_race=config.character.preferred_race, preferred_class=config.character.preferred_class, ) character_creator = CharacterCreator(creator_cfg) # Initialize Quest Manager & Progression Optimizer quest_manager = QuestManager() progression = ProgressionOptimizer() # Initialize LLM Brain (hybrid AI) llm_config = LLMConfig( base_url=config.ai.llm.base_url, model=config.ai.llm.model, temperature=config.ai.llm.temperature, max_tokens=config.ai.llm.max_tokens, timeout_seconds=config.ai.llm.timeout_seconds, max_retries=config.ai.llm.max_retries, system_prompt=config.ai.llm.system_prompt, ) llm_brain = LLMBrain(llm_config) social_ai = SocialAI( client=llm_brain.client, character_name=config.character.name, character_class=config.character.preferred_class, ) if config.ai.llm.enabled: await llm_brain.initialize() logger.info( "Parser: %s (mud_type=%s)", type(mud_parser).__name__, config.mud.mud_type, ) logger.info( "Decision engine ready | strategy=%s | goals=%d", config.ai.exploration_strategy, decision_engine.goal_manager.active_goal_count, ) logger.info( "WorldMapper ready | rooms=%d | exits=%d", world_mapper.total_rooms, world_mapper.total_exits, ) logger.info( "LLM Brain: %s | model=%s", "online" if llm_brain.is_available else "offline (rule-based fallback)", config.ai.llm.model, ) # Connect to MUD logger.info("Connecting to %s:%s", config.mud.host, config.mud.port) if not await telnet.connect(): logger.error("Failed to connect to MUD server") return # Login state machine login_state = "username" # username -> password -> logged_in # ------------------------------------------------------------------ # Receive callback — called for every chunk from the server # ------------------------------------------------------------------ async def on_receive(data: str): nonlocal login_state # Strip ANSI codes clean_text = ansi_parser.strip_ansi(data) logger.debug("Received: %s", clean_text.strip()[:200]) # ----- Handle login sequence ----- if login_state == "username": if "login" in clean_text.lower() or "name" in clean_text.lower(): logger.info("Sending username: %s", config.mud.username) await human_behavior.delay() await telnet.send(config.mud.username) login_state = "password" return elif login_state == "password": if "password" in clean_text.lower(): logger.info("Sending password") await human_behavior.delay() await telnet.send(config.mud.password) login_state = "logged_in" return # Handle reconnect / take over if "take over" in clean_text.lower() or "copy" in clean_text.lower(): logger.info("Taking over existing session") await human_behavior.delay() await telnet.send("y") login_state = "logged_in" return # ----- Game logic (only after login) ----- if login_state != "logged_in": return # Feed text into the output buffer; get back complete blocks blocks = output_buffer.feed(clean_text) for block in blocks: # Parse the block into structured events events = mud_parser.parse_block(block) for event in events: if event.event_type == EventType.PROMPT and event.prompt: game_state.update_from_prompt(event.prompt) logger.info( "Status: HP=%d/%d MP=%d/%d SP=%d/%d", event.prompt.hp or 0, event.prompt.max_hp or 0, event.prompt.mp or 0, event.prompt.max_mp or 0, event.prompt.sp or 0, event.prompt.max_sp or 0, ) elif event.event_type == EventType.ROOM and event.room: game_state.update_room(event.room) # Update automapper world_mapper.on_room_enter( title=event.room.title or "", description=event.room.description or "", exits=event.room.exits, npcs=event.room.npcs, items=event.room.items, ) logger.info( "Room: '%s' | Exits: %s | NPCs: %s | Items: %s | mapped=%d", event.room.title, event.room.exits, event.room.npcs, event.room.items, world_mapper.total_rooms, ) elif event.event_type == EventType.COMBAT and event.combat: game_state.update_combat(event.combat) logger.info( "Combat: %s -> %s (%s, dmg=%s)", event.combat.attacker, event.combat.target, event.combat.action, event.combat.damage, ) # --- Social AI: detect player messages in the raw block --- if llm_brain.is_available: social_event = social_ai.detect_social_event(block) if social_event: logger.info( "Social: %s from %s: %s", social_event["type"], social_event["speaker"], social_event["message"][:100], ) reply_cmd = await social_ai.respond(social_event) if reply_cmd: await human_behavior.delay() await telnet.send(reply_cmd) logger.info("Social reply: %s", reply_cmd) telnet.set_receive_callback(on_receive) # Start receive loop receive_task = asyncio.create_task(telnet.receive_loop()) try: # Main game loop logger.info("Starting main game loop") # Keep connection alive last_keep_alive = asyncio.get_running_loop().time() last_decision = asyncio.get_running_loop().time() # Decision interval: how often (in seconds) the engine ticks decision_interval_min = config.ai.decision_delay_ms[0] / 1000.0 decision_interval_max = ( config.ai.decision_delay_ms[1] / 1000.0 if len(config.ai.decision_delay_ms) > 1 else decision_interval_min ) import random as _rng next_decision_delay = _rng.uniform(decision_interval_min, decision_interval_max) while telnet.connected: current_time = asyncio.get_running_loop().time() # Send keep-alive if logged in if login_state == "logged_in": if current_time - last_keep_alive >= config.mud.keep_alive_interval: logger.debug( "Sending keep-alive: '%s'", config.mud.keep_alive_command ) await telnet.send(config.mud.keep_alive_command) last_keep_alive = current_time # ----- Decision Engine Tick ----- if current_time - last_decision >= next_decision_delay: # Let the engine evaluate the world and queue actions decision_engine.tick(game_state) # Execute the next action from the queue action = decision_engine.next_action() if action: logger.info( "Executing: '%s' [%s] (%s)", action.command, action.priority.name, action.description, ) # Notify mapper of movement commands if action.source == "explore" and action.priority.name == "MOVE": world_mapper.on_move(action.command) await human_behavior.delay() await telnet.send(action.command) last_decision = current_time next_decision_delay = _rng.uniform( decision_interval_min, decision_interval_max ) await asyncio.sleep(0.25) except KeyboardInterrupt: logger.info("Shutting down...") finally: # Flush any remaining buffered text remaining = output_buffer.flush() if remaining: events = mud_parser.parse_block(remaining) for event in events: if event.event_type == EventType.ROOM and event.room: game_state.update_room(event.room) # Close mapper and save state logger.info( "Session stats: %d rooms mapped, %d exits", world_mapper.total_rooms, world_mapper.total_exits, ) world_mapper.close() await telnet.disconnect() receive_task.cancel() def main(): """Entry point""" global logger # Parse arguments parser = argparse.ArgumentParser(description="MUD AI Player") parser.add_argument( "--config", default="config.yaml", help="Path to configuration file", ) args = parser.parse_args() # Load configuration try: config = load_config(args.config) except FileNotFoundError: print(f"Error: Configuration file '{args.config}' not found") print("Please create a config.yaml file (see examples/config.example.yaml)") sys.exit(1) except Exception as e: print(f"Error loading configuration: {e}") sys.exit(1) # Setup logging logger = setup_logger( level=config.logging.level, log_file=config.logging.file, console=config.logging.console, max_file_size_mb=config.logging.max_file_size_mb, backup_count=config.logging.backup_count, ) logger.info("=" * 60) logger.info("MUD AI Player Starting") logger.info("=" * 60) # Run main loop try: asyncio.run(main_loop(config)) except Exception as e: logger.exception("Fatal error: %s", e) sys.exit(1) if __name__ == "__main__": main()