Update app.py from anycoder

app.py

@@ -1,568 +1 @@
-import gradio as gr
-import torch
-import gc
-import os
-import sys
-import time
-import tempfile
-import traceback
-from pathlib import Path
-from typing import Optional, Generator
-from PIL import Image
-import logging
-
-logging.basicConfig(level=logging.INFO, format='%(asctime)s | %(levelname)s | %(message)s')
-logger = logging.getLogger(__name__)
-
-# Environment - Electric Jungle Setup
-if Path("/data").exists():
-    os.environ["HF_HOME"] = "/data/.huggingface"
-    os.environ["PYTORCH_CUDA_ALLOC_CONF"] = "expandable_segments:True"
-
-HF_TOKEN = os.getenv("HF_TOKEN")
-LORA_REPO = "Playtime-AI/Wan2.2-Loras"
-LORA_CACHE = Path(os.environ.get("HF_HOME", ".")) / "loras"
-LORA_CACHE.mkdir(parents=True, exist_ok=True)
-
-# ZeroGPU Detection
-try:
-    import spaces
-    ZEROGPU = True
-except ImportError:
-    ZEROGPU = False
-
-# Global State
-lora_cache: dict = {}
-selected_lora: Optional[str] = None
-MODEL_ID = "Wan-AI/Wan2.2-TI2V-5B-Diffusers"
-
-# === ELECTRIC PROGRESS TRACKER ===
-class ElectricProgress:
-    """Vibrant progress tracking with visual flair"""
-    STAGES = [
-        ("init", "🔥 Igniting Core", 5),
-        ("download_check", "⚡ Scanning Cache", 5),
-        ("download", "🌩️ Downloading Model", 40),
-        ("load_pipeline", "🚀 Pipeline Activation", 20),
-        ("load_lora", "🎨 Style Fusion", 5),
-        ("prepare", "✨ Frame Prep", 5),
-        ("generate", "🎬 Creative Explosion", 15),
-        ("export", "🎥 Video Manifest", 5),
-    ]
-
-    def __init__(self):
-        self.current_stage = 0
-        self.stage_progress = 0
-        self.start_time = time.time()
-        self.stage_start_time = time.time()
-        self.messages = []
-
-    def _get_stage_weight(self, stage_name: str) -> int:
-        for name, _, weight in self.STAGES:
-            if name == stage_name: return weight
-        return 10
-
-    def _get_cumulative_progress(self) -> int:
-        total = 0
-        for i, (name, _, weight) in enumerate(self.STAGES):
-            if i < self.current_stage:
-                total += weight
-            elif i == self.current_stage:
-                total += int(weight * self.stage_progress / 100)
-                break
-        return min(total, 100)
-
-    def start_stage(self, stage_name: str, message: str = ""):
-        for i, (name, label, _) in enumerate(self.STAGES):
-            if name == stage_name:
-                self.current_stage = i
-                self.stage_progress = 0
-                self.stage_start_time = time.time()
-                self.messages.append(f"⚡ {label}" + (f": {message}" if message else ""))
-                break
-
-    def update_stage(self, progress: int, detail: str = ""):
-        self.stage_progress = min(progress, 100)
-        if detail and self.messages:
-            stage_label = self.STAGES[self.current_stage][1]
-            self.messages[-1] = f"⚡ {stage_label}: {detail}"
-
-    def complete_stage(self, message: str = ""):
-        self.stage_progress = 100
-        elapsed = time.time() - self.stage_start_time
-        stage_label = self.STAGES[self.current_stage][1]
-        self.messages[-1] = f"✅ {stage_label} ({elapsed:.1f}s)" + (f" - {message}" if message else "")
-
-    def error(self, message: str):
-        self.messages.append(f"💥 {message}")
-
-    def format_display(self) -> str:
-        overall = self._get_cumulative_progress()
-        elapsed = time.time() - self.start_time
-        
-        if overall > 5:
-            estimated_total = elapsed / (overall / 100)
-            remaining = max(0, estimated_total - elapsed)
-            time_str = f"⏱️ ~{int(remaining)}s remaining"
-        else:
-            time_str = "⏳ calculating..."
-
-        bar_width = 40
-        filled = int(bar_width * overall / 100)
-        bar = "🟩" * filled + "⬜" * (bar_width - filled)
-        
-        lines = [
-            f"# 🎨 Progress: {overall}%",
-            f"```{bar}``` {time_str}",
-            "",
-            "**⚡ Creative Pipeline:**",
-        ]
-        lines.extend(self.messages[-8:])
-        return "\n".join(lines)
-
-# === VALIDATION ===
-def validate_image(image) -> tuple[bool, str, Optional[Image.Image]]:
-    if image is None:
-        return False, "🎨 **Upload your creative spark!**\n\nDrop an image to ignite the jungle.", None
-
-    try:
-        if isinstance(image, str):
-            if not os.path.exists(image):
-                return False, "🔍 **Image not found**\n\nTry uploading again.", None
-            img = Image.open(image)
-        elif isinstance(image, Image.Image):
-            img = image
-        else:
-            return False, f"❌ **Invalid image type**: {type(image).__name__}", None
-
-        if img.mode not in ('RGB', 'RGBA', 'L'):
-            img = img.convert('RGB')
-
-        width, height = img.size
-        if width < 64 or height < 64:
-            return False, f"📏 **Image too small**: {width}x{height}\n\nMinimum: 64x64", None
-        if width > 4096 or height > 4096:
-            return False, f"📏 **Image too large**: {width}x{height}\n\nMax: 4096x4096", None
-
-        return True, "✅ **Canvas ready!**", img.convert('RGB')
-    except Exception as e:
-        return False, f"💥 **Image processing failed**: {str(e)}", None
-
-def validate_prompt(prompt: str) -> tuple[bool, str]:
-    if not prompt or not prompt.strip():
-        return False, "✍️ **Describe the motion!**\n\n'exhale gently', 'hair flows in wind', 'subtle smile'"
-    if len(prompt) > 1000:
-        return False, f"📜 **Prompt too long**: {len(prompt)} chars\n\nMax 1000 chars"
-    return True, "✅ **Vision captured!**"
-
-# === CORE CREATION ENGINE ===
-def generate_with_progress(image, prompt, frames, height, width, guidance, steps, lora_scale, seed) -> Generator:
-    progress = ElectricProgress()
-    
-    try:
-        # Stage 1: Ignition
-        progress.start_stage("init", "Validating creative inputs")
-        yield None, progress.format_display()
-
-        valid, msg, processed_image = validate_image(image)
-        if not valid:
-            progress.error(msg)
-            yield None, progress.format_display()
-            return
-
-        valid, msg = validate_prompt(prompt)
-        if not valid:
-            progress.error(msg)
-            yield None, progress.format_display()
-            return
-
-        progress.complete_stage("Vision locked!")
-        yield None, progress.format_display()
-
-        # Stage 2: Model Scan
-        progress.start_stage("download_check", "Scanning neural cache")
-        yield None, progress.format_display()
-
-        try:
-            from huggingface_hub import try_to_load_from_cache
-            cached = try_to_load_from_cache(MODEL_ID, "model_index.json")
-            model_cached = cached is not None
-            progress.complete_stage("Neural pathways active" if model_cached else "Fetching cosmic data")
-        except:
-            model_cached = False
-            progress.complete_stage("Cache scan complete")
-        yield None, progress.format_display()
-
-        # Stage 3: Model Download (if needed)
-        if not model_cached:
-            progress.start_stage("download", "Channeling 20GB neural network")
-            yield None, progress.format_display()
-            progress.update_stage(25, "Syncing cosmic intelligence...")
-            yield None, progress.format_display()
-            time.sleep(0.8)
-            progress.update_stage(75, "Neural weights downloading...")
-            yield None, progress.format_display()
-            time.sleep(0.8)
-        else:
-            progress.start_stage("download", "Using cached neural pathways")
-            progress.complete_stage("Instant activation")
-        yield None, progress.format_display()
-
-        # Stage 4: Pipeline Activation
-        progress.start_stage("load_pipeline", "Awakening AI pipeline")
-        yield None, progress.format_display()
-
-        try:
-            from diffusers import WanImageToVideoPipeline
-            progress.update_stage(25, "Initializing creative core...")
-            yield None, progress.format_display()
-
-            pipe = WanImageToVideoPipeline.from_pretrained(
-                MODEL_ID,
-                torch_dtype=torch.bfloat16,
-                use_safetensors=True,
-                token=HF_TOKEN,
-            )
-
-            progress.update_stage(60, "Electrifying GPU pathways...")
-            yield None, progress.format_display()
-
-            if torch.cuda.is_available():
-                pipe.enable_model_cpu_offload()
-                if hasattr(pipe, 'vae'):
-                    pipe.vae.enable_slicing()
-                    pipe.vae.enable_tiling()
-                progress.update_stage(90, "GPU symphony engaged")
-            else:
-                pipe = pipe.to("cpu")
-                progress.update_stage(90, "CPU mode activated")
-            
-            progress.complete_stage()
-            yield None, progress.format_display()
-
-        except Exception as e:
-            progress.error(f"Pipeline failed: {str(e)[:100]}")
-            yield None, progress.format_display()
-            return
-
-        # Stage 5: Style Fusion (LoRA)
-        progress.start_stage("load_lora")
-        yield None, progress.format_display()
-
-        if selected_lora and selected_lora in lora_cache:
-            try:
-                from huggingface_hub import hf_hub_download
-                progress.update_stage(30, f"Fusing {selected_lora} style...")
-                yield None, progress.format_display()
-
-                filename = lora_cache[selected_lora]
-                local_path = LORA_CACHE / Path(filename).name
-
-                if not local_path.exists():
-                    progress.update_stage(60, "Downloading style weights...")
-                    yield None, progress.format_display()
-                    hf_hub_download(
-                        repo_id=LORA_REPO,
-                        filename=filename,
-                        local_dir=str(LORA_CACHE),
-                        token=HF_TOKEN
-                    )
-
-                progress.update_stage(90, "Style infusion complete")
-                pipe.load_lora_weights(str(local_path.parent), weight_name=local_path.name)
-                progress.complete_stage(f"{selected_lora} activated")
-            except Exception as e:
-                progress.error(f"LoRA failed: {str(e)[:50]}")
-        else:
-            progress.complete_stage("Pure generation mode")
-        yield None, progress.format_display()
-
-        # Stage 6: Creative Prep
-        progress.start_stage("prepare", "Framing the vision")
-        yield None, progress.format_display()
-
-        w = (int(width) // 8) * 8
-        h = (int(height) // 8) * 8
-        img = processed_image.resize((w, h), Image.Resampling.LANCZOS)
-        progress.update_stage(50, f"Canvas: {w}x{h}")
-        yield None, progress.format_display()
-
-        gen = None
-        if seed >= 0:
-            device = "cuda" if torch.cuda.is_available() else "cpu"
-            gen = torch.Generator(device=device).manual_seed(int(seed))
-            progress.update_stage(80, f"Creative seed: {int(seed)}")
-        else:
-            progress.update_stage(80, "Chaos seed activated")
-        yield None, progress.format_display()
-        progress.complete_stage()
-        yield None, progress.format_display()
-
-        # Stage 7: Creative Explosion
-        progress.start_stage("generate", f"Exploding {int(frames)} frames")
-        yield None, progress.format_display()
-
-        kwargs = {
-            "prompt": prompt,
-            "image": img,
-            "height": h,
-            "width": w,
-            "num_frames": int(frames),
-            "guidance_scale": float(guidance),
-            "num_inference_steps": int(steps),
-            "generator": gen,
-        }
-
-        if selected_lora:
-            kwargs["cross_attention_kwargs"] = {"scale": float(lora_scale)}
-
-        def progress_callback(pipe, step, timestep, callback_kwargs):
-            pct = int((step / int(steps)) * 100)
-            progress.update_stage(pct, f"Frame {step}/{int(steps)}")
-            return callback_kwargs
-
-        kwargs["callback_on_step_end"] = progress_callback
-        progress.update_stage(5, "Initiating creative burst...")
-        yield None, progress.format_display()
-
-        output = pipe(**kwargs)
-        frames_out = output.frames[0]
-        progress.complete_stage(f"{len(frames_out)} frames materialized")
-        yield None, progress.format_display()
-
-        # Stage 8: Video Manifest
-        progress.start_stage("export", "Manifesting video masterpiece")
-        yield None, progress.format_display()
-
-        from diffusers.utils import export_to_video
-        fps = min(8 + (frames // 10), 16)
-        progress.update_stage(50, f"Encoding symphony @ {fps}fps")
-        yield None, progress.format_display()
-
-        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
-            export_to_video(frames_out, f.name, fps=fps)
-            video_path = f.name
-
-        progress.complete_stage()
-
-        # Cleanup & Victory
-        del pipe
-        gc.collect()
-        if torch.cuda.is_available():
-            torch.cuda.empty_cache()
-
-        total_time = time.time() - progress.start_time
-        final_status = f"""# 🎉 Masterpiece Complete!
-
-**{len(frames_out)} frames** @ **{fps}fps** = **{len(frames_out)/fps:.1f}s** of pure creativity
-
-**Canvas:** {w}x{h} | **Time:** {total_time:.1f}s | **Style:** {selected_lora or 'Pure'}
-
-**Your electric jungle creation is ready!** ⚡🎬✨
-"""
-        yield video_path, final_status
-
-    except torch.cuda.OutOfMemoryError:
-        error_msg = """# 💥 GPU Overload!
-
-**The jungle got too electric!** 
-
-**Quick fixes:**
-- Canvas: 256x384 
-- Frames: 17 
-- Refresh & retry
-
-*Heavy creativity needs space to breathe* 🌩️"""
-        yield None, error_msg
-
-    except Exception as e:
-        tb = traceback.format_exc()
-        logger.error(f"Creation failed: {e}")
-        error_msg = f"""# 💥 Creation Interrupted
-
-**Error:** {type(e).__name__}: {str(e)[:200]}
-
-**Reset the jungle:**
-1. New image
-2. Simple prompt  
-3. Lower settings
-4. Refresh page
-
-```
-{tb[-800:]}
-```"""
-        yield None, error_msg
-
-# === CREATIVE WRAPPERS ===
-if ZEROGPU:
-    @spaces.GPU(duration=300)
-    def generate_video(image, prompt, frames, height, width, guidance, steps, lora_scale, seed):
-        result = None
-        status = ""
-        for video, msg in generate_with_progress(image, prompt, frames, height, width, guidance, steps, lora_scale, seed):
-            result = video
-            status = msg
-        return result, status
-else:
-    def generate_video(image, prompt, frames, height, width, guidance, steps, lora_scale, seed):
-        result = None
-        status = ""
-        for video, msg in generate_with_progress(image, prompt, frames, height, width, guidance, steps, lora_scale, seed):
-            result = video
-            status = msg
-        return result, status
-
-# === ELECTRIC UI FUNCTIONS ===
-def get_status() -> str:
-    gpu = "⚡ ZeroGPU" if ZEROGPU else ("🔥 CUDA" if torch.cuda.is_available() else "💻 CPU")
-    lora = selected_lora or "Pure Creation"
-    return f"**{gpu} | Style: {lora}**"
-
-def refresh_loras():
-    global lora_cache
-    try:
-        from huggingface_hub import list_repo_files
-        files = list_repo_files(LORA_REPO, token=HF_TOKEN)
-        lora_cache = {Path(f).stem: f for f in files if f.endswith('.safetensors')}
-        names = sorted(lora_cache.keys())
-        return gr.Dropdown(choices=names, value=names[0] if names else None), f"✅ **{len(names)} styles** ready to ignite!"
-    except Exception as e:
-        return gr.Dropdown(choices=[], value=None), f"⚠️ **Style scan failed**: {str(e)[:80]}"
-
-def select_lora(name: str) -> str:
-    global selected_lora
-    if name and name in lora_cache:
-        selected_lora = name
-        return f"🎨 **{name}** electrified!\n\n*Style will pulse through your creation*"
-    selected_lora = None
-    return "🌿 **Pure creation mode** - raw AI power"
-
-# === GRAND ELECTRIC INTERFACE ===
-css = """
-.electric-jungle {
-    background: linear-gradient(135deg, #0f0f23 0%, #1a0033 50%, #0f0f23 100%);
-    font-family: 'Segoe UI', -apple-system, sans-serif;
-    color: #e0e0ff;
-}
-.progress-box {
-    font-family: 'Fira Code', monospace;
-    background: linear-gradient(145deg, #1a1a3e, #16213e);
-    padding: 20px;
-    border-radius: 12px;
-    border: 1px solid #00d4ff;
-    box-shadow: 0 8px 32px rgba(0, 212, 255, 0.1);
-}
-.btn-electric {
-    background: linear-gradient(45deg, #ff00aa, #00d4ff, #ffaa00);
-    border: none;
-    border-radius: 12px;
-    font-weight: bold;
-    transition: all 0.3s ease;
-}
-.btn-electric:hover {
-    transform: translateY(-2px);
-    box-shadow: 0 12px 40px rgba(255, 0, 170, 0.4);
-}
-"""
-
-with gr.Blocks(title="Electric Jungle - Wan2.2 Creator") as demo:
-    
-    gr.Markdown("""
-# ⚡ **Electric Jungle Creator**
-### *Wan2.2 LoRA Video Generator*
-
-**Transform static images into pulsing video art**  
-*First run: 3-5min download | Cached: 45-90s magic*
-
-<div style="background: linear-gradient(90deg, #ff00aa, #00d4ff); height: 4px; border-radius: 2px; margin: 20px 0;"></div>
-    """)
-
-    status = gr.Markdown(get_status())
-
-    with gr.Tabs():
-        with gr.TabItem("🎬 Create"):
-            with gr.Row():
-                with gr.Column(scale=1):
-                    img_in = gr.Image(type="pil", label="🌿 Your Creative Spark", height=320)
-                    prompt_in = gr.Textbox(
-                        label="✨ Motion Vision",
-                        lines=3,
-                        placeholder="gentle breathing, wind in hair, subtle pulsing light, cosmic dance...",
-                        max_lines=5
-                    )
-                    
-                    with gr.Accordion("⚙️ Creative Controls", open=False):
-                        gr.Markdown("*Lower = faster lightning*")
-                        with gr.Row():
-                            frames_sl = gr.Slider(17, 49, 25, step=8, label="⚡ Frames")
-                            h_sl = gr.Slider(256, 480, 384, step=32, label="📐 Height")
-                            w_sl = gr.Slider(256, 640, 512, step=32, label="📐 Width")
-                        with gr.Row():
-                            cfg_sl = gr.Slider(4, 12, 7.0, step=0.5, label="🎯 Vision Lock")
-                            steps_sl = gr.Slider(15, 30, 22, step=2, label="🔮 Quality")
-                        lora_sl = gr.Slider(0, 1.5, 0.85, step=0.05, label="🎨 Style Pulse")
-                        seed_in = gr.Number(1337, label="🌌 Seed", info="-1 = cosmic chaos")
-
-                    gen_btn = gr.Button("🚀 IGNITE CREATION", variant="primary", size="lg")
-
-                with gr.Column(scale=1):
-                    gen_status = gr.Markdown("*Drop your spark and ignite*", elem_classes=["progress-box"])
-                    vid_out = gr.Video(label="🎥 Your Electric Masterpiece", height=380)
-
-        with gr.TabItem("🎨 Styles"):
-            gr.Markdown("""
-## 🎨 **Style Fusion Chamber**
-
-*Select artistic styles that pulse through your creations*
-            """)
-            lora_dd = gr.Dropdown(choices=[], label="⚡ Available Styles")
-            with gr.Row():
-                refresh_btn = gr.Button("🔄 Scan Styles")
-                select_btn = gr.Button("✅ Fuse Style", variant="primary")
-            lora_status = gr.Markdown("🌿 Ready for pure creation...")
-
-        with gr.TabItem("🧠 Neural Codex"):
-            gr.Markdown("""
-## ⚡ **Electric Jungle Codex**
-
-### **How to Ignite**
-1. **Drop Image** - Portrait, scene, abstract art
-2. **Vision Prompt** - "gentle sway", "cosmic pulse", "hair flows wildly"  
-3. **⚙️ Tweak** - Lower settings = faster lightning
-4. **IGNITE** - Watch magic unfold
-
-### **Pro Tips**
-- **Fast**: 320x512, 17 frames, 18 steps
-- **Epic**: 512x768, 33 frames, 25 steps  
-- **Styles**: Amplify with LoRA fusion
-
-### **Tech Stack**
-| Component | Power |
-|-----------|-------|
-| Wan2.2-TI2V | 5B params |
-| ZeroGPU | H200 cluster |
-| LoRA Styles | Infinite |
-| Output | 1080p@16fps |
-
-*Creativity has no limits in the Electric Jungle* 🌩️✨
-            """)
-
-    # Event Symphony
-    refresh_btn.click(refresh_loras, outputs=[lora_dd, lora_status])
-    select_btn.click(select_lora, [lora_dd], [lora_status]).then(get_status, outputs=[status])
-    gen_btn.click(
-        generate_video,
-        [img_in, prompt_in, frames_sl, h_sl, w_sl, cfg_sl, steps_sl, lora_sl, seed_in],
-        [vid_out, gen_status]
-    )
-
-    demo.load(refresh_loras, outputs=[lora_dd, lora_status])
-
-if __name__ == "__main__":
-    demo.queue(max_size=3).launch(
-        server_name="0.0.0.0", 
-        server_port=7860, 
-        show_error=True,
-        share=False,
-        css=css
-    )
+\n{error_message}\n