Update app.py

app.py

@@ -1,26 +1,55 @@
 import os
 import sys
-import gradio as gr
-import subprocess
 import json
-import torch
 from pathlib import Path
+import tempfile
+import shutil
 
-# Set environment variables for HF Spaces
-os.environ["GRADIO_SERVER_NAME"] = "0.0.0.0"
-os.environ["GRADIO_SERVER_PORT"] = "7860"
+# Safe imports with fallbacks
+try:
+    import gradio as gr
+except ImportError:
+    print("Installing gradio...")
+    os.system(f"{sys.executable} -m pip install gradio")
+    import gradio as gr
+
+try:
+    import subprocess
+except ImportError:
+    subprocess = None
+
+try:
+    import torch
+    HAS_TORCH = True
+except ImportError:
+    HAS_TORCH = False
+    print("PyTorch not available, using CPU mode")
 
-# Pre-download models cache
-os.environ["HF_HUB_CACHE"] = "/tmp/hf_cache"
-os.environ["HUGGINGFACE_HUB_CACHE"] = "/tmp/hf_cache"
+# Ensure temp directories exist
+TEMP_DIR = tempfile.gettempdir()
+HF_CACHE_DIR = os.path.join(TEMP_DIR, "hf_cache")
+os.makedirs(HF_CACHE_DIR, exist_ok=True)
 
-# Fix potential Hunyuan Video Avatar issues
+# Set environment variables safely
+os.environ["GRADIO_SERVER_NAME"] = "0.0.0.0"
+os.environ["GRADIO_SERVER_PORT"] = "7860"
+os.environ["HF_HUB_CACHE"] = HF_CACHE_DIR
+os.environ["HUGGINGFACE_HUB_CACHE"] = HF_CACHE_DIR
 os.environ["TOKENIZERS_PARALLELISM"] = "false"
 os.environ["PYTORCH_CUDA_ALLOC_CONF"] = "max_split_size_mb:512"
 
+def safe_import(module_name, package_name=None):
+    """Safely import a module with fallback"""
+    if package_name is None:
+        package_name = module_name
+    try:
+        return __import__(module_name.replace("-", "_"))
+    except ImportError:
+        print(f"Module {module_name} not found, functionality limited")
+        return None
+
 def setup_environment():
-    """Setup environment for HF Spaces with WanGP v6.3"""
-    # Install additional dependencies if needed
+    """Setup environment with error handling"""
     dependencies = [
         "sageattention==1.0.6",
         "insightface",
@@ -38,55 +67,72 @@ def setup_environment():
     
     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)
+            module_name = dep.split("==")[0].split(">=")[0].split("[")[0]
+            safe_import(module_name)
+        except Exception as e:
+            print(f"Could not process {dep}: {e}")
+            if subprocess:
+                try:
+                    subprocess.run(
+                        [sys.executable, "-m", "pip", "install", dep], 
+                        check=False, 
+                        capture_output=True,
+                        timeout=30
+                    )
+                except Exception as install_error:
+                    print(f"Failed to install {dep}: {install_error}")
 
 def download_essential_models():
-    """Pre-download essential models for faster startup"""
+    """Pre-download models with full error handling"""
     try:
         from huggingface_hub import snapshot_download
         
-        print("Downloading Hunyuan Video Avatar models...")
+        print("Attempting to download 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
-        )
+        try:
+            snapshot_download(
+                repo_id="tencent/HunyuanVideo-Avatar", 
+                cache_dir=HF_CACHE_DIR,
+                allow_patterns=["*.safetensors", "*.json", "*.txt", "*.bin"],
+                ignore_patterns=["*.mp4", "*.avi", "*.mov"],
+                resume_download=True,
+                max_workers=2
+            )
+        except Exception as e:
+            print(f"Could not download HunyuanVideo-Avatar: {e}")
         
-        # 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"]
-        )
+        try:
+            snapshot_download(
+                repo_id="tencent/HunyuanVideo",
+                cache_dir=HF_CACHE_DIR, 
+                allow_patterns=["*.safetensors", "*.json", "*.txt"],
+                ignore_patterns=["*.mp4", "*.avi"],
+                resume_download=True,
+                max_workers=2
+            )
+        except Exception as e:
+            print(f"Could not download HunyuanVideo: {e}")
         
-        print("✅ Models downloaded successfully!")
+        print("Model download attempt completed")
         
+    except ImportError:
+        print("huggingface_hub not available, skipping model download")
     except Exception as e:
-        print(f"Model download warning: {e}")
-        print("Models will be downloaded on-demand during first use.")
+        print(f"Model download error: {e}")
 
 def create_hf_config():
-    """Create optimized config for HF Spaces deployment"""
+    """Create config with error handling"""
     config = {
         "model_settings": {
-            "profile": 3,  # Optimized for A10G Large
+            "profile": 3,
             "quantize_transformer": True,
             "attention_mode": "sage",
-            "compile": False,  # Disable for stability on HF
+            "compile": False,
             "teacache": "2.0"
         },
         "avatar_settings": {
-            "max_frames": 120,  # ~5 seconds at 24fps
-            "resolution": "512x512",  # Balanced quality/performance
+            "max_frames": 120,
+            "resolution": "512x512",
             "emotion_control": True,
             "multi_character": True
         },
@@ -98,182 +144,333 @@ def create_hf_config():
         },
         "audio_processing": {
             "sample_rate": 16000,
-            "max_duration": 15,  # seconds
+            "max_duration": 15,
             "supported_formats": ["wav", "mp3", "m4a"]
         }
     }
     
-    config_path = "/tmp/hf_config.json"
-    with open(config_path, "w") as f:
-        json.dump(config, f, indent=2)
+    config_path = os.path.join(TEMP_DIR, "hf_config.json")
+    try:
+        with open(config_path, "w", encoding='utf-8') as f:
+            json.dump(config, f, indent=2)
+    except Exception as e:
+        print(f"Could not save config: {e}")
+        config_path = None
     
     return config
 
+def create_dummy_video(output_path, duration=5, fps=24, width=512, height=512):
+    """Create a dummy video file for testing"""
+    try:
+        import numpy as np
+        import cv2
+        
+        fourcc = cv2.VideoWriter_fourcc(*'mp4v')
+        out = cv2.VideoWriter(output_path, fourcc, fps, (width, height))
+        
+        for i in range(duration * fps):
+            # Create gradient frame
+            frame = np.ones((height, width, 3), dtype=np.uint8) * 50
+            text = f"Frame {i+1}"
+            cv2.putText(frame, text, (width//4, height//2), 
+                       cv2.FONT_HERSHEY_SIMPLEX, 1, (255, 255, 255), 2)
+            out.write(frame)
+        
+        out.release()
+        return True
+    except Exception as e:
+        print(f"Could not create video with OpenCV: {e}")
+        # Create empty file as fallback
+        try:
+            with open(output_path, 'wb') as f:
+                f.write(b'dummy video content')
+            return True
+        except:
+            return False
+
 class WanGPInterface:
-    """WanGP Interface for HF Spaces"""
+    """WanGP Interface with full error handling"""
     
     def __init__(self, config):
-        self.config = config
-        self.device = "cuda" if torch.cuda.is_available() else "cpu"
+        self.config = config or {}
+        self.device = "cpu"
+        if HAS_TORCH:
+            try:
+                self.device = "cuda" if torch.cuda.is_available() else "cpu"
+            except:
+                self.device = "cpu"
         self.models_loaded = False
         
     def load_models(self):
-        """Load models on demand"""
+        """Load models with error handling"""
         if self.models_loaded:
-            return
+            return True
             
         try:
-            print("Loading Hunyuan Video Avatar models...")
-            # Model loading logic would go here
-            # This is a placeholder for the actual model loading
+            print("Loading Hunyuan Video Avatar models (placeholder)...")
+            # Placeholder for actual model loading
+            import time
+            time.sleep(0.5)  # Simulate loading
             self.models_loaded = True
-            print("✅ Models loaded successfully!")
+            print("✅ Models loaded successfully (simulated)!")
+            return True
         except Exception as e:
-            print(f"❌ Error loading models: {e}")
-            raise e
+            print(f"Error in model loading: {e}")
+            self.models_loaded = False
+            return False
     
     def generate_avatar_video(self, audio_file, avatar_image, prompt="", emotion="neutral"):
+        """Generate avatar video with comprehensive error handling"""
         try:
-            self.load_models()
-    
-            # Gọi model thật để sinh video từ ảnh và audio
-            output_video_path = "/tmp/generated_avatar.mp4"
+            # Validate inputs
+            if audio_file is None:
+                return None, "❌ Error: No audio file provided"
+            
+            if avatar_image is None:
+                return None, "❌ Error: No avatar image provided"
+            
+            # Ensure model is loaded
+            if not self.load_models():
+                print("Models not loaded, using dummy generation")
+            
+            # Create output path
+            output_filename = f"avatar_{os.getpid()}_{id(self)}.mp4"
+            output_path = os.path.join(TEMP_DIR, output_filename)
+            
+            # Create dummy video
+            if create_dummy_video(output_path, duration=5):
+                if os.path.exists(output_path):
+                    return output_path, "✅ Video generated successfully (demo mode)!"
+                else:
+                    return None, "❌ Error: Failed to create output file"
+            else:
+                return None, "❌ Error: Video generation failed"
             
-            # 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)}"
-
+            error_msg = f"❌ Error in avatar generation: {str(e)}"
+            print(error_msg)
+            return None, error_msg
     
     def generate_video(self, prompt, duration=5, resolution="512x512"):
-        """Generate video from text prompt"""
+        """Generate video from text with error handling"""
         try:
-            self.load_models()
+            if not prompt:
+                return "❌ Error: No prompt provided"
+            
+            # Parse resolution
+            try:
+                width, height = map(int, resolution.split('x'))
+            except:
+                width, height = 512, 512
             
-            # Placeholder for video generation logic
-            return f"Generated video for prompt: {prompt}"
+            # Ensure model is loaded
+            if not self.load_models():
+                print("Models not loaded, using dummy generation")
             
+            # Create output path
+            output_filename = f"video_{os.getpid()}_{id(self)}.mp4"
+            output_path = os.path.join(TEMP_DIR, output_filename)
+            
+            # Create dummy video
+            if create_dummy_video(output_path, duration=int(duration), width=width, height=height):
+                if os.path.exists(output_path):
+                    return output_path, f"✅ Generated video for prompt: {prompt[:50]}..."
+                else:
+                    return None, "❌ Error: Failed to create output file"
+            else:
+                return None, "❌ Error: Video generation failed"
+                
         except Exception as e:
-            return f"Error: {str(e)}"
+            error_msg = f"❌ Error in video generation: {str(e)}"
+            print(error_msg)
+            return None, error_msg
 
 def create_gradio_interface(wangp_interface):
-    """Create Gradio interface for WanGP"""
+    """Create Gradio interface with error handling"""
     
-    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)
+    try:
+        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>
+                <p style="color: orange;">⚠️ Running in Demo Mode - Using placeholder outputs</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"
+                            )
+                            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...",
+                                value=""
+                            )
+                            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, value="Ready")
+                
+                # 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,
+                                value=""
+                            )
+                            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, value="Ready")
+            
+            # Event handlers with error handling
+            def safe_avatar_generation(*args):
+                try:
+                    return wangp_interface.generate_avatar_video(*args)
+                except Exception as e:
+                    return None, f"❌ Unexpected error: {str(e)}"
+            
+            def safe_video_generation(*args):
+                try:
+                    result = wangp_interface.generate_video(*args)
+                    if isinstance(result, tuple):
+                        return result
+                    else:
+                        return None, result
+                except Exception as e:
+                    return None, f"❌ Unexpected error: {str(e)}"
             
-            # 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)
+            generate_avatar_btn.click(
+                fn=safe_avatar_generation,
+                inputs=[audio_input, avatar_image, avatar_prompt, emotion_control],
+                outputs=[avatar_output, avatar_status]
+            )
+            
+            generate_video_btn.click(
+                fn=safe_video_generation,
+                inputs=[video_prompt, duration_slider, resolution_dropdown],
+                outputs=[video_output, video_status]
+            )
+            
+            gr.HTML("""
+            <div style="text-align: center; margin-top: 20px; color: #666;">
+                <p>Powered by Hunyuan Video Avatar & WanGP v6.3</p>
+                <p style="font-size: 12px;">Note: This is a demonstration interface with placeholder outputs</p>
+            </div>
+            """)
         
-        # 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
-        )
+        return demo
         
-        generate_video_btn.click(
-            fn=wangp_interface.generate_video,
-            inputs=[video_prompt, duration_slider, resolution_dropdown],
-            outputs=[video_status]
+    except Exception as e:
+        print(f"Error creating Gradio interface: {e}")
+        # Return minimal interface
+        demo = gr.Interface(
+            fn=lambda x: f"Error: {str(e)}",
+            inputs="text",
+            outputs="text",
+            title="WanGP v6.3 - Error State"
         )
-        
-        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
+        return demo
 
-if __name__ == "__main__":
+def main():
+    """Main function with comprehensive error handling"""
     print("🚀 Starting WanGP v6.3 with Hunyuan Video Avatar...")
     
-    # Setup environment
-    setup_environment()
+    try:
+        # Setup environment
+        setup_environment()
+    except Exception as e:
+        print(f"Environment setup warning: {e}")
     
-    # Create configuration
-    config = create_hf_config()
+    try:
+        # Create configuration
+        config = create_hf_config()
+    except Exception as e:
+        print(f"Config creation warning: {e}")
+        config = {}
     
-    # Download models in background
     try:
+        # Download models in background
         download_essential_models()
     except Exception as e:
-        print(f"Model download failed: {e}")
-    
-    # Initialize WanGP interface
-    wangp_interface = WanGPInterface(config)
+        print(f"Model download skipped: {e}")
     
-    # Create and launch Gradio interface
-    demo = create_gradio_interface(wangp_interface)
-    
-    print("✅ Setup complete! Launching application...")
+    try:
+        # Initialize WanGP interface
+        wangp_interface = WanGPInterface(config)
+    except Exception as e:
+        print(f"Interface initialization error: {e}")
+        # Create minimal interface
+        class MinimalInterface:
+            def __init__(self):
+                self.config = {}
+            def generate_avatar_video(self, *args):
+                return None, "Service temporarily unavailable"
+            def generate_video(self, *args):
+                return None, "Service temporarily unavailable"
+        wangp_interface = MinimalInterface()
     
-    demo.launch(
-        server_name="0.0.0.0",
-        server_port=7860,
-        share=False,  # HF Spaces handles sharing
-        debug=False,
-        show_error=True
-    )
+    try:
+        # Create and launch Gradio interface
+        demo = create_gradio_interface(wangp_interface)
+        
+        print("✅ Setup complete! Launching application...")
+        
+        # Launch with error handling
+        demo.launch(
+            server_name="0.0.0.0",
+            server_port=int(os.environ.get("GRADIO_SERVER_PORT", 7860)),
+            share=False,
+            debug=False,
+            show_error=True,
+            prevent_thread_lock=False
+        )
+    except Exception as e:
+        print(f"❌ Failed to launch Gradio: {e}")
+        print("Attempting fallback launch...")
+        try:
+            # Minimal fallback
+            import gradio as gr
+            gr.Interface(
+                fn=lambda x: "System Error - Please restart",
+                inputs="text", 
+                outputs="text"
+            ).launch()
+        except:
+            print("❌ Complete failure. Exiting.")
+            sys.exit(1)
+
+if __name__ == "__main__":
+    main()