app.py

import os
import sys
import gradio as gr
import subprocess
import json
import torch
from pathlib import Path

# Set environment variables for HF Spaces
os.environ["GRADIO_SERVER_NAME"] = "0.0.0.0"
os.environ["GRADIO_SERVER_PORT"] = "7860"

# Pre-download models cache
os.environ["HF_HUB_CACHE"] = "/tmp/hf_cache"
os.environ["HUGGINGFACE_HUB_CACHE"] = "/tmp/hf_cache"

# Fix potential Hunyuan Video Avatar issues
os.environ["TOKENIZERS_PARALLELISM"] = "false"
os.environ["PYTORCH_CUDA_ALLOC_CONF"] = "max_split_size_mb:512"

def setup_environment():
    """Setup environment for HF Spaces with WanGP v6.3"""
    # Install additional dependencies if needed
    dependencies = [
        "sageattention==1.0.6",
        "insightface",
        "facexlib", 
        "diffusers>=0.30.0",
        "transformers>=4.44.0",
        "accelerate>=0.34.0",
        "xformers",
        "opencv-python",
        "imageio[ffmpeg]",
        "moviepy",
        "librosa",
        "soundfile"
    ]
    
    for dep in dependencies:
        try:
            module_name = dep.split("==")[0].split(">=")[0]
            __import__(module_name.replace("-", "_"))
        except ImportError:
            print(f"Installing {dep}...")
            subprocess.run([sys.executable, "-m", "pip", "install", dep], 
                         check=True, capture_output=True)

def download_essential_models():
    """Pre-download essential models for faster startup"""
    try:
        from huggingface_hub import snapshot_download
        
        print("Downloading Hunyuan Video Avatar models...")
        
        # Download Hunyuan Video Avatar base models
        snapshot_download(
            repo_id="tencent/HunyuanVideo-Avatar", 
            cache_dir="/tmp/hf_cache",
            allow_patterns=["*.safetensors", "*.json", "*.txt", "*.bin"],
            ignore_patterns=["*.mp4", "*.avi", "*.mov"]  # Skip demo videos
        )
        
        # Download base Hunyuan Video model
        snapshot_download(
            repo_id="tencent/HunyuanVideo",
            cache_dir="/tmp/hf_cache", 
            allow_patterns=["*.safetensors", "*.json", "*.txt"],
            ignore_patterns=["*.mp4", "*.avi"]
        )
        
        print("✅ Models downloaded successfully!")
        
    except Exception as e:
        print(f"Model download warning: {e}")
        print("Models will be downloaded on-demand during first use.")

def create_hf_config():
    """Create optimized config for HF Spaces deployment"""
    config = {
        "model_settings": {
            "profile": 3,  # Optimized for A10G Large
            "quantize_transformer": True,
            "attention_mode": "sage",
            "compile": False,  # Disable for stability on HF
            "teacache": "2.0"
        },
        "avatar_settings": {
            "max_frames": 120,  # ~5 seconds at 24fps
            "resolution": "512x512",  # Balanced quality/performance
            "emotion_control": True,
            "multi_character": True
        },
        "memory_optimization": {
            "enable_vae_tiling": True,
            "enable_cpu_offload": True,
            "max_batch_size": 1,
            "gradient_checkpointing": True
        },
        "audio_processing": {
            "sample_rate": 16000,
            "max_duration": 15,  # seconds
            "supported_formats": ["wav", "mp3", "m4a"]
        }
    }
    
    config_path = "/tmp/hf_config.json"
    with open(config_path, "w") as f:
        json.dump(config, f, indent=2)
    
    return config

class WanGPInterface:
    """WanGP Interface for HF Spaces"""
    
    def __init__(self, config):
        self.config = config
        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        self.models_loaded = False
        
    def load_models(self):
        """Load models on demand"""
        if self.models_loaded:
            return
            
        try:
            print("Loading Hunyuan Video Avatar models...")
            # Model loading logic would go here
            # This is a placeholder for the actual model loading
            self.models_loaded = True
            print("✅ Models loaded successfully!")
        except Exception as e:
            print(f"❌ Error loading models: {e}")
            raise e
    
    def generate_avatar_video(self, audio_file, avatar_image, prompt="", emotion="neutral"):
        try:
            self.load_models()
    
            # Gọi model thật để sinh video từ ảnh và audio
            output_video_path = "/tmp/generated_avatar.mp4"
            
            # TODO: thay bằng đoạn code gọi mô hình HunyuanAvatar thực sự
            # Giả sử video đã được lưu vào `output_video_path`
    
            return output_video_path, "✅ Video generated successfully!"
        
        except Exception as e:
            return None, f"❌ Error: {str(e)}"

    
    def generate_video(self, prompt, duration=5, resolution="512x512"):
        """Generate video from text prompt"""
        try:
            self.load_models()
            
            # Placeholder for video generation logic
            return f"Generated video for prompt: {prompt}"
            
        except Exception as e:
            return f"Error: {str(e)}"

def create_gradio_interface(wangp_interface):
    """Create Gradio interface for WanGP"""
    
    with gr.Blocks(title="WanGP v6.3 - Hunyuan Video Avatar", theme=gr.themes.Soft()) as demo:
        gr.HTML("""
        <div style="text-align: center; margin-bottom: 20px;">
            <h1>🎭 WanGP v6.3 - Hunyuan Video Avatar</h1>
            <p>Advanced AI Video Generation with Audio-Driven Human Animation</p>
        </div>
        """)
        
        with gr.Tabs():
            # Avatar Generation Tab
            with gr.TabItem("🎭 Avatar Generation"):
                with gr.Row():
                    with gr.Column():
                        audio_input = gr.Audio(
                            label="Audio Input",
                            type="filepath",
                            format="wav"
                        )
                        avatar_image = gr.Image(
                            label="Avatar Image",
                            type="filepath"
                        )
                        emotion_control = gr.Dropdown(
                            choices=["neutral", "happy", "sad", "angry", "surprised"],
                            value="neutral",
                            label="Emotion Control"
                        )
                        avatar_prompt = gr.Textbox(
                            label="Additional Prompt (Optional)",
                            placeholder="Describe additional details..."
                        )
                        generate_avatar_btn = gr.Button("Generate Avatar Video", variant="primary")
                    
                    with gr.Column():
                        avatar_output = gr.Video(label="Generated Avatar Video")
                        avatar_status = gr.Textbox(label="Status", interactive=False)
            
            # Text-to-Video Tab
            with gr.TabItem("📹 Text to Video"):
                with gr.Row():
                    with gr.Column():
                        video_prompt = gr.Textbox(
                            label="Video Prompt",
                            placeholder="Describe the video you want to generate...",
                            lines=3
                        )
                        duration_slider = gr.Slider(
                            minimum=2,
                            maximum=10,
                            value=5,
                            step=1,
                            label="Duration (seconds)"
                        )
                        resolution_dropdown = gr.Dropdown(
                            choices=["512x512", "768x768", "1024x1024"],
                            value="512x512",
                            label="Resolution"
                        )
                        generate_video_btn = gr.Button("Generate Video", variant="primary")
                    
                    with gr.Column():
                        video_output = gr.Video(label="Generated Video")
                        video_status = gr.Textbox(label="Status", interactive=False)
        
        # Event handlers
        generate_avatar_btn.click(
            fn=wangp_interface.generate_avatar_video,
            inputs=[audio_input, avatar_image, avatar_prompt, emotion_control],
            outputs=[avatar_output, avatar_status]  # trả ra cả video và status
        )
        
        generate_video_btn.click(
            fn=wangp_interface.generate_video,
            inputs=[video_prompt, duration_slider, resolution_dropdown],
            outputs=[video_status]
        )
        
        gr.HTML("""
        <div style="text-align: center; margin-top: 20px; color: #666;">
            <p>Powered by Hunyuan Video Avatar & WanGP v6.3</p>
        </div>
        """)
    
    return demo

if __name__ == "__main__":
    print("🚀 Starting WanGP v6.3 with Hunyuan Video Avatar...")
    
    # Setup environment
    setup_environment()
    
    # Create configuration
    config = create_hf_config()
    
    # Download models in background
    try:
        download_essential_models()
    except Exception as e:
        print(f"Model download failed: {e}")
    
    # Initialize WanGP interface
    wangp_interface = WanGPInterface(config)
    
    # Create and launch Gradio interface
    demo = create_gradio_interface(wangp_interface)
    
    print("✅ Setup complete! Launching application...")
    
    demo.launch(
        server_name="0.0.0.0",
        server_port=7860,
        share=False,  # HF Spaces handles sharing
        debug=False,
        show_error=True
    )