Delete FlowFacade_aoti.py

FlowFacade_aoti.py

@@ -1,185 +0,0 @@
-import os
-import torch
-import numpy as np
-from PIL import Image
-from typing import Tuple, Optional
-from VideoEngine_aoti import VideoEngineAOTI
-from TextProcessor import TextProcessor
-
-try:
-    import spaces
-    HAS_SPACES = True
-except ImportError:
-    HAS_SPACES = False
-    class spaces:
-        @staticmethod
-        def GPU(duration=120):
-            def decorator(func):
-                return func
-            return decorator
-
-
-class FlowFacadeAOTI:
-    """
-    Unified interface coordinating all subsystems with AOTI optimization.
-    This is the AOTI-enabled version of FlowFacade.
-    """
-
-    def __init__(self):
-        self.is_spaces = os.environ.get('SPACE_ID') is not None
-        self.video_engine = VideoEngineAOTI()
-        self.text_processor = TextProcessor(resource_manager=None)
-        print("✓ DeltaFlow (AOTI) initialized")
-
-    def _calculate_gpu_duration(self, image: Image.Image, duration_seconds: float,
-                                num_inference_steps: int, enable_prompt_expansion: bool, **kwargs) -> int:
-        """Calculate dynamic GPU duration for ZeroGPU with AOTI optimization."""
-        BASE_FRAMES_HEIGHT_WIDTH = 81 * 832 * 624
-        BASE_STEP_DURATION = 8
-
-        if self.video_engine.use_aoti:
-            BASE_STEP_DURATION = int(BASE_STEP_DURATION * 0.6)
-
-        resized_image = self.video_engine.resize_image(image)
-        width, height = resized_image.width, resized_image.height
-        frames = self.video_engine.get_num_frames(duration_seconds)
-
-        factor = frames * width * height / BASE_FRAMES_HEIGHT_WIDTH
-        step_duration = BASE_STEP_DURATION * factor ** 1.5
-        total_duration = int(num_inference_steps) * step_duration
-
-        if not self.video_engine.is_loaded:
-            total_duration += 150
-
-        if enable_prompt_expansion:
-            total_duration += 40
-
-        return max(int(total_duration), 300)
-
-    @spaces.GPU(duration=_calculate_gpu_duration)
-    def generate_video_from_image(self, image: Image.Image, user_instruction: str,
-                                  duration_seconds: float = 3.0, num_inference_steps: int = 4,
-                                  guidance_scale: float = 1.0, guidance_scale_2: float = 1.0,
-                                  seed: int = 42, randomize_seed: bool = False,
-                                  enable_prompt_expansion: bool = False,
-                                  progress=None) -> Tuple[str, str, int]:
-        """
-        Generate video from image with AOTI-optimized pipeline.
-
-        Returns:
-            tuple: (video_path, final_prompt, seed_used)
-        """
-        if image is None:
-            raise ValueError("No image provided")
-        if not user_instruction or user_instruction.strip() == "":
-            raise ValueError("Please provide a motion instruction")
-
-        try:
-            if randomize_seed:
-                seed = np.random.randint(0, 2147483647)
-
-            if enable_prompt_expansion:
-                if progress:
-                    progress(0.1, desc="AI expanding your prompt...")
-                final_prompt = self.text_processor.process(user_instruction, auto_unload=True)
-            else:
-                final_prompt = user_instruction
-
-            if progress:
-                progress(0.2, desc="Preparing GPU memory...")
-
-            if not self.video_engine.is_loaded:
-                import gc
-                gc.collect()
-                if torch.cuda.is_available():
-                    torch.cuda.empty_cache()
-                    torch.cuda.ipc_collect()
-
-                if progress:
-                    progress(0.25, desc="Loading video generation model (with AOTI)...")
-                self.video_engine.load_model()
-
-            gc.collect()
-            if torch.cuda.is_available():
-                torch.cuda.empty_cache()
-
-            if progress:
-                aoti_status = "AOTI enabled" if self.video_engine.use_aoti else "FP8 only"
-                progress(0.3, desc=f"Generating video ({aoti_status})...")
-
-            video_path = self.video_engine.generate_video(
-                image=image, prompt=final_prompt, duration_seconds=duration_seconds,
-                num_inference_steps=num_inference_steps, guidance_scale=guidance_scale,
-                guidance_scale_2=guidance_scale_2, seed=seed
-            )
-
-            if progress:
-                progress(1.0, desc="Complete!")
-
-            return video_path, final_prompt, seed
-
-        except Exception as e:
-            import traceback
-            print(f"\n✗ Generation error: {type(e).__name__}: {str(e)}")
-            if os.environ.get('DEBUG'):
-                print(traceback.format_exc())
-            raise RuntimeError(f"Generation failed: {type(e).__name__}: {str(e)}")
-
-    def cleanup(self) -> None:
-        """Cleanup resources."""
-        try:
-            if hasattr(self.text_processor, 'is_loaded') and self.text_processor.is_loaded:
-                self.text_processor.unload_model()
-            torch.cuda.empty_cache()
-        except Exception as e:
-            if os.environ.get('DEBUG'):
-                print(f"⚠ Cleanup warning: {str(e)}")
-
-    def get_system_info(self) -> dict:
-        """Get system information including AOTI status."""
-        quantization_type = "None"
-        if torch.cuda.is_available():
-            cuda_cap = torch.cuda.get_device_capability()
-            fp8_supported = cuda_cap[0] > 8 or (cuda_cap[0] == 8 and cuda_cap[1] >= 9)
-            quantization_type = "FP8" if fp8_supported else "INT8"
-
-        optimizations = [
-            "Lightning LoRA (4-8 steps)",
-            f"{quantization_type} Quantization"
-        ]
-
-        if self.video_engine.use_aoti:
-            optimizations.append("AOTI Compilation (1.5-1.8x)")
-
-        return {
-            "device": self.video_engine.device,
-            "video_model": VideoEngineAOTI.MODEL_ID,
-            "text_model": TextProcessor.MODEL_ID,
-            "lightning_lora": "Enabled",
-            "quantization": quantization_type,
-            "aoti": "Enabled" if self.video_engine.use_aoti else "Disabled",
-            "optimizations": optimizations
-        }
-
-    def validate_image(self, image: Image.Image) -> bool:
-        """Validate image dimensions."""
-        if image is None:
-            return False
-
-        min_dim, max_dim = 256, 4096
-
-        if image.width < min_dim or image.height < min_dim:
-            print(f"⚠ Image too small: {image.width}x{image.height}")
-            return False
-
-        if image.width > max_dim or image.height > max_dim:
-            print(f"⚠ Image too large: {image.width}x{image.height}")
-            return False
-
-        return True
-
-    def __del__(self):
-        try:
-            self.cleanup()
-        except:
-            pass