import gradio as gr import spaces import os import json import torch import gc from datetime import datetime from pathlib import Path # Disable torch dynamo globally to avoid ConstantVariable errors torch._dynamo.config.suppress_errors = True # Initialize directories DATA_DIR = Path("/data") if os.path.exists("/data") else Path("./data") DATA_DIR.mkdir(exist_ok=True) (DATA_DIR / "users").mkdir(exist_ok=True) (DATA_DIR / "monsters").mkdir(exist_ok=True) (DATA_DIR / "models").mkdir(exist_ok=True) (DATA_DIR / "cache").mkdir(exist_ok=True) # Ensure Gradio cache directory exists import tempfile gradio_cache_dir = Path("/tmp/gradio") gradio_cache_dir.mkdir(parents=True, exist_ok=True) # Set environment variable for Gradio cache os.environ.setdefault("GRADIO_TEMP_DIR", str(gradio_cache_dir)) # Import modules (to be created) from core.ai_pipeline import MonsterGenerationPipeline from core.game_mechanics import GameMechanics from core.state_manager import StateManager from core.auth_manager import AuthManager from ui.themes import get_cyberpunk_theme, CYBERPUNK_CSS from ui.interfaces import create_voice_interface, create_visual_interface # Initialize with GPU optimization @spaces.GPU(duration=300) def initialize_systems(): """Initialize all core systems with GPU""" pipeline = MonsterGenerationPipeline() return pipeline # Initialize core systems (defer GPU initialization) pipeline = None def get_pipeline(): """Get or initialize the pipeline with GPU support""" global pipeline if pipeline is None: try: pipeline = initialize_systems() except Exception as e: print(f"GPU initialization failed, falling back to CPU: {e}") pipeline = MonsterGenerationPipeline(device="cpu") return pipeline game_mechanics = GameMechanics() state_manager = StateManager(DATA_DIR) auth_manager = AuthManager() # Main generation function @spaces.GPU(duration=180) def generate_monster(oauth_profile: gr.OAuthProfile | None, audio_input=None, text_input=None, reference_images=None, training_focus="balanced", care_level="normal"): """Generate a new monster with AI pipeline""" if oauth_profile is None: return { "message": "🔒 Please log in to create monsters!", "image": None, "model_3d": None, "stats": None, "dialogue": None } user_id = oauth_profile.username try: # Generate monster using AI pipeline current_pipeline = get_pipeline() result = current_pipeline.generate_monster( audio_input=audio_input, text_input=text_input, reference_images=reference_images, user_id=user_id ) # Create game monster from AI result monster = game_mechanics.create_monster(result, { "training_focus": training_focus, "care_level": care_level }, user_id) # Save to persistent storage state_manager.save_monster(user_id, monster) # Prepare response response_dict = { "message": f"✨ {monster.name} has been created!", "image": result.get('image'), "model_3d": result.get('model_3d'), "stats": monster.get_stats_display(), "dialogue": result.get('dialogue', "🤖💚1️⃣0️⃣0️⃣") } return ( response_dict["message"], response_dict["image"], response_dict["model_3d"], response_dict["stats"], response_dict["dialogue"] ) except Exception as e: print(f"Error generating monster: {str(e)}") # Use fallback generation current_pipeline = get_pipeline() fallback_result = current_pipeline.fallback_generation(text_input or "friendly digital creature") fallback_dict = { "message": "⚡ Created using quick generation mode", "image": fallback_result.get('image'), "model_3d": None, "stats": fallback_result.get('stats'), "dialogue": "🤖❓9️⃣9️⃣" } return ( fallback_dict["message"], fallback_dict["image"], fallback_dict["model_3d"], fallback_dict["stats"], fallback_dict["dialogue"] ) # Training function def train_monster(oauth_profile: gr.OAuthProfile | None, training_type, intensity): """Train the active monster""" if oauth_profile is None: return "🔒 Please log in to train monsters!", None, None user_id = oauth_profile.username current_monster = state_manager.get_current_monster(user_id) if not current_monster: return "❌ No active monster to train!", None, None # Apply training result = game_mechanics.train_monster(current_monster, training_type, intensity) if result['success']: state_manager.update_monster(user_id, current_monster) return ( result['message'], current_monster.get_stats_display(), result.get('evolution_check') ) else: return result['message'], None, None # Care functions def feed_monster(oauth_profile: gr.OAuthProfile | None, food_type): """Feed the active monster""" if oauth_profile is None: return "🔒 Please log in to care for monsters!" user_id = oauth_profile.username current_monster = state_manager.get_current_monster(user_id) if not current_monster: return "❌ No active monster to feed!" result = game_mechanics.feed_monster(current_monster, food_type) state_manager.update_monster(user_id, current_monster) return result['message'] # Build the Gradio interface with gr.Blocks( theme=get_cyberpunk_theme(), css=CYBERPUNK_CSS, title="DigiPal - Digital Monster Companion" ) as demo: # Header with cyberpunk styling gr.HTML("""
Your AI-Powered Digital Monster Companion