Delete app.py

app.py

@@ -1,365 +0,0 @@
-import os
-import subprocess
-import sys
-import json
-import struct
-import logging
-import random
-import tempfile
-import hashlib
-import shutil
-import gc
-from pathlib import Path
-
-# Disable torch.compile / dynamo before any torch import
-os.environ["TORCH_COMPILE_DISABLE"] = "1"
-os.environ["TORCHDYNAMO_DISABLE"] = "1"
-
-# Clone LTX-2 repo and install packages
-LTX_REPO_URL = "https://github.com/Lightricks/LTX-2.git"
-LTX_REPO_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "LTX-2")
-LTX_COMMIT = "ae855f8538843825f9015a419cf4ba5edaf5eec2"
-
-if not os.path.exists(LTX_REPO_DIR):
-    subprocess.run(["git", "clone", LTX_REPO_URL, LTX_REPO_DIR], check=True)
-    subprocess.run(["git", "checkout", LTX_COMMIT], cwd=LTX_REPO_DIR, check=True)
-
-subprocess.run(
-    [sys.executable, "-m", "pip", "install", "--force-reinstall", "--no-deps", "-e",
-     os.path.join(LTX_REPO_DIR, "packages", "ltx-core"),
-     "-e", os.path.join(LTX_REPO_DIR, "packages", "ltx-pipelines")],
-    check=True,
-)
-
-sys.path.insert(0, os.path.join(LTX_REPO_DIR, "packages", "ltx-pipelines", "src"))
-sys.path.insert(0, os.path.join(LTX_REPO_DIR, "packages", "ltx-core", "src"))
-
-import spaces
-import torch
-import gradio as gr
-import numpy as np
-from huggingface_hub import hf_hub_download, snapshot_download
-from ltx_core.loader.primitives import StateDict, LoraPathStrengthAndSDOps
-from ltx_core.loader.sft_loader import SafetensorsStateDictLoader
-from ltx_core.loader.sd_ops import LTXV_LORA_COMFY_RENAMING_MAP
-from ltx_core.components.diffusion_steps import EulerDiffusionStep
-from ltx_core.components.noisers import GaussianNoiser
-from ltx_core.model.audio_vae import encode_audio as vae_encode_audio
-from ltx_core.model.upsampler import upsample_video
-from ltx_core.model.video_vae import TilingConfig, get_video_chunks_number, decode_video as vae_decode_video
-from ltx_core.quantization import QuantizationPolicy
-from ltx_core.types import Audio, AudioLatentShape, VideoPixelShape
-from ltx_pipelines.distilled import DistilledPipeline
-from ltx_pipelines.utils import euler_denoising_loop
-from ltx_pipelines.utils.args import ImageConditioningInput
-from ltx_pipelines.utils.constants import DISTILLED_SIGMA_VALUES, STAGE_2_DISTILLED_SIGMA_VALUES
-from ltx_pipelines.utils.helpers import (
-    cleanup_memory, combined_image_conditionings, denoise_video_only,
-    encode_prompts, simple_denoising_func,
-)
-from ltx_pipelines.utils.media_io import decode_audio_from_file, encode_video
-
-# --- ZERO-GPU PATCHES ---
-_SAFETENSORS_DTYPE_MAP = {
-    "F64": torch.float64, "F32": torch.float32, "F16": torch.float16, "BF16": torch.bfloat16,
-    "F8_E5M2": torch.float8_e5m2, "F8_E4M3": torch.float8_e4m3fn, "I64": torch.int64,
-    "I32": torch.int32, "I16": torch.int16, "I8": torch.int8, "U8": torch.uint8, "BOOL": torch.bool,
-}
-
-def _patched_load(self, path, sd_ops, device=None):
-    sd = {}
-    size, dtype = 0, set()
-    device = torch.device("cpu") 
-    model_paths = path if isinstance(path, list) else [path]
-    for shard_path in model_paths:
-        with open(shard_path, "rb") as f:
-            header_len = struct.unpack("<Q", f.read(8))[0]
-            header = json.loads(f.read(header_len).decode("utf-8"))
-            data_base = 8 + header_len
-            for name, meta in header.items():
-                if name == "__metadata__": continue
-                expected_name = name if sd_ops is None else sd_ops.apply_to_key(name)
-                if expected_name is None: continue
-                start, end = meta["data_offsets"]
-                f.seek(data_base + start)
-                buf = f.read(end - start)
-                t = torch.frombuffer(bytearray(buf), dtype=_SAFETENSORS_DTYPE_MAP[meta["dtype"]]).reshape(meta["shape"])
-                t = t.to(device=device, non_blocking=True, copy=False)
-                kvs = ((expected_name, t),) if sd_ops is None else sd_ops.apply_to_key_value(expected_name, t)
-                for key, v in kvs:
-                    size += v.nbytes
-                    dtype.add(v.dtype)
-                    sd[key] = v
-    return StateDict(sd=sd, device=device, size=size, dtype=dtype)
-
-SafetensorsStateDictLoader.load = _patched_load
-
-_original_tensor_to = torch.Tensor.to
-def _spaces_safe_to(self, *args, **kwargs):
-    def _is_cuda_target(x):
-        return x == "cuda" or (isinstance(x, torch.device) and x.type == "cuda") or (isinstance(x, str) and x.startswith("cuda"))
-    if args and _is_cuda_target(args[0]):
-        return _original_tensor_to(self, "cuda", *args[1:], **{k: v for k, v in kwargs.items() if k not in ("non_blocking", "copy")})
-    if kwargs.get("device") is not None and _is_cuda_target(kwargs["device"]):
-        return _original_tensor_to(self, *args, **{k: v for k, v in kwargs.items() if k not in ("non_blocking", "copy")})
-    return _original_tensor_to(self, *args, **kwargs)
-
-torch.Tensor.to = _spaces_safe_to
-
-class LTX23DistilledA2VPipeline(DistilledPipeline):
-    def __call__(self, prompt, seed, height, width, num_frames, frame_rate, images, audio_path=None, tiling_config=None, enhance_prompt=False):
-        if audio_path is None:
-            return super().__call__(prompt=prompt, seed=seed, height=height, width=width, num_frames=num_frames, frame_rate=frame_rate, images=images, tiling_config=tiling_config, enhance_prompt=enhance_prompt)
-        generator = torch.Generator(device=self.device).manual_seed(seed)
-        noiser = GaussianNoiser(generator=generator)
-        stepper = EulerDiffusionStep()
-        dtype = torch.bfloat16
-        (ctx_p,) = encode_prompts([prompt], self.model_ledger, enhance_first_prompt=enhance_prompt, enhance_prompt_image=images[0].path if len(images) > 0 else None)
-        video_context, audio_context = ctx_p.video_encoding, ctx_p.audio_encoding
-        video_duration = num_frames / frame_rate
-        decoded_audio = decode_audio_from_file(audio_path, self.device, 0.0, video_duration)
-        if decoded_audio is None: raise ValueError("Audio extraction failed")
-        encoded_audio_latent = vae_encode_audio(decoded_audio, self.model_ledger.audio_encoder())
-        audio_shape = AudioLatentShape.from_duration(batch=1, duration=video_duration, channels=8, mel_bins=16)
-        expected_frames = audio_shape.frames
-        actual_frames = encoded_audio_latent.shape[2]
-        if actual_frames > expected_frames: encoded_audio_latent = encoded_audio_latent[:, :, :expected_frames, :]
-        elif actual_frames < expected_frames:
-            pad = torch.zeros(encoded_audio_latent.shape[0], encoded_audio_latent.shape[1], expected_frames - actual_frames, encoded_audio_latent.shape[3], device=encoded_audio_latent.device, dtype=encoded_audio_latent.dtype)
-            encoded_audio_latent = torch.cat([encoded_audio_latent, pad], dim=2)
-        video_encoder = self.model_ledger.video_encoder()
-        transformer = self.model_ledger.transformer()
-        stage_1_sigmas = torch.tensor(DISTILLED_SIGMA_VALUES, device=self.device)
-        def denoising_loop(sigmas, video_state, audio_state, stepper):
-            return euler_denoising_loop(sigmas=sigmas, video_state=video_state, audio_state=audio_state, stepper=stepper, denoise_fn=simple_denoising_func(video_context=video_context, audio_context=audio_context, transformer=transformer))
-        stage_1_output_shape = VideoPixelShape(batch=1, frames=num_frames, width=width // 2, height=height // 2, fps=frame_rate)
-        stage_1_conditionings = combined_image_conditionings(images=images, height=stage_1_output_shape.height, width=stage_1_output_shape.width, video_encoder=video_encoder, dtype=dtype, device=self.device)
-        video_state = denoise_video_only(output_shape=stage_1_output_shape, conditionings=stage_1_conditionings, noiser=noiser, sigmas=stage_1_sigmas, stepper=stepper, denoising_loop_fn=denoising_loop, components=self.pipeline_components, dtype=dtype, device=self.device, initial_audio_latent=encoded_audio_latent)
-        torch.cuda.synchronize()
-        cleanup_memory()
-        upscaled_video_latent = upsample_video(latent=video_state.latent[:1], video_encoder=video_encoder, upsampler=self.model_ledger.spatial_upsampler())
-        stage_2_sigmas = torch.tensor(STAGE_2_DISTILLED_SIGMA_VALUES, device=self.device)
-        stage_2_output_shape = VideoPixelShape(batch=1, frames=num_frames, width=width, height=height, fps=frame_rate)
-        stage_2_conditionings = combined_image_conditionings(images=images, height=stage_2_output_shape.height, width=stage_2_output_shape.width, video_encoder=video_encoder, dtype=dtype, device=self.device)
-        video_state = denoise_video_only(output_shape=stage_2_output_shape, conditionings=stage_2_conditionings, noiser=noiser, sigmas=stage_2_sigmas, stepper=stepper, denoising_loop_fn=denoising_loop, components=self.pipeline_components, dtype=dtype, device=self.device, noise_scale=stage_2_sigmas[0], initial_video_latent=upscaled_video_latent, initial_audio_latent=encoded_audio_latent)
-        torch.cuda.synchronize()
-        del transformer, video_encoder
-        cleanup_memory()
-        decoded_video = vae_decode_video(video_state.latent, self.model_ledger.video_decoder(), tiling_config, generator)
-        original_audio = Audio(waveform=decoded_audio.waveform.squeeze(0), sampling_rate=decoded_audio.sampling_rate)
-        return decoded_video, original_audio
-
-# --- MODEL LOADING ---
-LTX_MODEL_REPO = "Lightricks/LTX-2.3"
-GEMMA_REPO ="Lightricks/gemma-3-12b-it-qat-q4_0-unquantized"
-weights_dir = Path("weights")
-weights_dir.mkdir(exist_ok=True)
-checkpoint_path = hf_hub_download(repo_id="ibyteohdear/Lightricks-LTX-2.3-DISTILLED-10-Eros", filename="LTX2.3_DISTILLED_BAKED_LTX_SULPHUR_STYLE_IS_10Eros_v14_r768.safetensors", local_dir=str(weights_dir))
-spatial_upsampler_path = hf_hub_download(repo_id=LTX_MODEL_REPO, filename="ltx-2.3-spatial-upscaler-x2-1.1.safetensors")
-gemma_root = snapshot_download(repo_id=GEMMA_REPO)
-
-LORA_REPO = "dagloop5/LoRA"
-LORA_FILES = {
-    "singularity": ("TenStrip/LTX2.3_DMD_Lora", "LTX2.3_DMD_reshaped_r256.safetensors"),
-    "teneros": (LORA_REPO, "LTX2.3-Furry-2D-NSFW-Multi-Purpose-Lora+Cum.safetensors"),
-    "sulphur": (LORA_REPO, "ltx23E28093SlowMotion26.Pkrs.safetensors"),
-    "pose": (LORA_REPO, "LTX2_3_NSFW_furry_concat_v2.safetensors"),
-    "general": (LORA_REPO, "LTX2.3_reasoning_Sulphur-2_I2V_V4.safetensors"),
-    "motion": (LORA_REPO, "Sulphur_LTX 2.3_better _NSFW_motion.safetensors"),
-    "dreamlay": ("lynaNSFW/DR34ML4Y_AIO_NSFW_LTX23", "DR34ML4Y_LTXXX_V2.safetensors"),
-    "mself": (LORA_REPO, "LTX2.3_2d_NSFW_motion_enhancer.safetensors"),
-    "dramatic": ("Muapi/valiantcat-ltx-2.3-transition-lora", "valiantcat-ltx-2.3-transition-lora.safetensors"),
-    "fluid": (LORA_REPO, "Cr3ampi3_animation_sulphur-2_i2v_v1.0.safetensors"),
-    "liquid": (LORA_REPO, "liquid_wet_dr1pp_ltx2_v1.0_scaled.safetensors"),
-    "demopose": (LORA_REPO, "clapping-cheeks-audio-v001-alpha.safetensors"),
-    "voice": (LORA_REPO, "hentai_voice_ltx23_v2.comfy.safetensors"),
-    "realism": (LORA_REPO, "FurryenhancerLTX2.3V4.094fused.safetensors"),
-    "transition": (LORA_REPO, "LTX-2_takerpov_lora_v1.2.safetensors"),
-    "physics": (LORA_REPO, "LTX2.3_Physics_V2_000002000.safetensors"),
-    "reasoning": ("LiconStudio/Ltx2.3-VBVR-lora-I2V", "Ltx2.3-Licon-VBVR-I2V-390K-R32.safetensors"),
-    "twostep": (LORA_REPO, "LTX2.3_Multi_step_video_reasoning_V0.1.safetensors"),
-    "mcfurry": (LORA_REPO, "mvmt_lora_v2_600.safetensors"),
-    "dm": (LORA_REPO, "Doggy_mission_sulphur-2_v0.5.safetensors"),
-    "praxis": (LORA_REPO, "Penile_Praxis_V4.safetensors"),
-    "threed": (LORA_REPO, "ltx2-3d-animations-12500-steps-k3nk.safetensors"),
-    "concept": (LORA_REPO, "ltx23_nsfw_helper_multi_concept_lora_v2.safetensors"),
-    "bulge": (LORA_REPO, "stomach_bulge_10eros_sulphur_v1.safetensors"),
-}
-LORA_PATHS = {k: hf_hub_download(repo_id=v[0], filename=v[1]) for k, v in LORA_FILES.items()}
-
-pipeline = LTX23DistilledA2VPipeline(
-    distilled_checkpoint_path=checkpoint_path,
-    spatial_upsampler_path=spatial_upsampler_path,
-    gemma_root=gemma_root,
-    loras=[],
-    quantization=QuantizationPolicy.fp8_cast(),
-)
-
-ledger = pipeline.model_ledger
-_transformer = ledger.transformer()
-_video_encoder = ledger.video_encoder()
-_video_decoder = ledger.video_decoder()
-_audio_encoder = ledger.audio_encoder()
-_audio_decoder = ledger.audio_decoder()
-_vocoder = ledger.vocoder()
-_spatial_upsampler = ledger.spatial_upsampler()
-_text_encoder = ledger.text_encoder()
-_embeddings_processor = ledger.gemma_embeddings_processor()
-
-ledger.transformer = lambda: _transformer
-ledger.video_encoder = lambda: _video_encoder
-ledger.video_decoder = lambda: _video_decoder
-ledger.audio_encoder = lambda: _audio_encoder
-ledger.audio_decoder = lambda: _audio_decoder
-ledger.vocoder = lambda: _vocoder
-ledger.spatial_upsampler = lambda: _spatial_upsampler
-ledger.text_encoder = lambda: _text_encoder
-ledger.gemma_embeddings_processor = lambda: _embeddings_processor
-
-# --- IMPROVED CPU LORA PREPARATION ---
-current_lora_key = None
-PENDING_LORA_STATE = None
-
-def prepare_lora_cache(
-    *strengths,
-    progress=gr.Progress(track_tqdm=True),
-):
-    global current_lora_key, PENDING_LORA_STATE
-    
-    # Build key based on strength values
-    strength_vals = [round(float(x), 2) for x in strengths]
-    key_str = "|".join(map(str, strength_vals))
-    key = hashlib.sha256(key_str.encode()).hexdigest()
-    
-    if current_lora_key == key:
-        return "LoRA state already cached."
-
-    progress(0.1, desc="Filtering active LoRAs")
-    active_entries = []
-    for (name, path), strength in zip(LORA_PATHS.items(), strengths):
-        if float(strength) != 0.0:
-            active_entries.append(LoraPathStrengthAndSDOps(path, float(strength), LTXV_LORA_COMFY_RENAMING_MAP))
-
-    if not active_entries:
-        PENDING_LORA_STATE = None
-        current_lora_key = key
-        return "No LoRAs selected."
-
-    try:
-        progress(0.3, desc="Applying LoRAs to base weights (CPU)")
-        # CRITICAL: We apply to the existing _transformer on CPU instead of creating a new ledger/model
-        # We save the original state to restore it later
-        original_state = {k: v.clone() for k, v in _transformer.state_dict().items() if "lora" not in k}
-        
-        _transformer.load_loras(active_entries)
-        
-        progress(0.7, desc="Extracting fused state")
-        # Only store the delta or the fused state to avoid keeping the whole model in memory twice
-        PENDING_LORA_STATE = {k: v.detach().cpu().contiguous() for k, v in _transformer.state_dict().items()}
-        
-        # Restore base weights so the pipeline remains "clean" for the next prepare call
-        _transformer.load_state_dict(original_state, strict=False)
-        
-        current_lora_key = key
-        return "Built LoRA state successfully."
-    except Exception as e:
-        return f"Error: {e}"
-    finally:
-        gc.collect()
-
-def apply_lora_state():
-    global PENDING_LORA_STATE
-    if PENDING_LORA_STATE is None: return
-    with torch.no_grad():
-        _transformer.load_state_dict(PENDING_LORA_STATE, strict=False)
-
-def get_gpu_duration(*args, gpu_duration=75.0, **kwargs):
-    return int(gpu_duration)
-
-@spaces.GPU(size="xlarge", duration=get_gpu_duration)
-@torch.inference_mode()
-def generate_video(first_image, last_image, input_audio, prompt, duration, gpu_duration, enhance_prompt, seed, randomize_seed, height, width, *strengths, progress=gr.Progress(track_tqdm=True)):
-    try:
-        apply_lora_state()
-        
-        current_seed = random.randint(0, np.iinfo(np.int32).max) if randomize_seed else int(seed)
-        frame_rate = 24.0
-        num_frames = int(duration * frame_rate) + 1
-        num_frames = ((num_frames - 1 + 7) // 8) * 8 + 1
-        images = []
-        output_dir = Path("outputs")
-        output_dir.mkdir(exist_ok=True)
-        if first_image is not None:
-            temp_first_path = output_dir / f"temp_first_{current_seed}.jpg"
-            if hasattr(first_image, "save"): first_image.save(temp_first_path)
-            else: temp_first_path = Path(first_image)
-            images.append(ImageConditioningInput(path=str(temp_first_path), frame_idx=0, strength=1.0))
-        if last_image is not None:
-            temp_last_path = output_dir / f"temp_last_{current_seed}.jpg"
-            if hasattr(last_image, "save"): last_image.save(temp_last_path)
-            else: temp_last_path = Path(last_image)
-            images.append(ImageConditioningInput(path=str(temp_last_path), frame_idx=num_frames - 1, strength=1.0))
-        
-        tiling_config = TilingConfig.default()
-        video_chunks_number = get_video_chunks_number(num_frames, tiling_config)
-        
-        video, audio = pipeline(prompt=prompt, seed=current_seed, height=int(height), width=int(width), num_frames=num_frames, frame_rate=frame_rate, images=images, audio_path=input_audio, tiling_config=tiling_config, enhance_prompt=enhance_prompt)
-        
-        output_path = tempfile.mktemp(suffix=".mp4")
-        encode_video(video=video, fps=frame_rate, audio=audio, output_path=output_path, video_chunks_number=video_chunks_number)
-        
-        return str(output_path), current_seed
-    except Exception as e:
-        return None, current_seed
-
-RESOLUTIONS = {
-    "low":  {"16:9": (768, 512), "9:16": (512, 768), "1:1": (768, 768), "4:3": (768, 576), "3:4": (576, 768), "21:9": (768, 384)},
-    "high": {"16:9": (1536, 1024), "9:16": (1024, 1536), "1:1": (1024, 1024), "4:3": (1536, 1152), "3:4": (1152, 1536), "21:9": (1536, 768)},
-}
-
-def update_res(first, last, high_res):
-    ref = first if first is not None else last
-    if ref is None: return gr.update(value=1536), gr.update(value=1024)
-    w, h = (ref.size if hasattr(ref, "size") else ref.shape[:2][::-1])
-    ratio = w / h
-    aspect = min({"16:9": 16/9, "9:16": 9/16, "1:1": 1.0}, key=lambda k: abs(ratio - (16/9 if k=="16:9" else 9/16 if k=="9:16" else 1.0)))
-    tier = "high" if high_res else "low"
-    res = RESOLUTIONS[tier][aspect]
-    return gr.update(value=res[0]), gr.update(value=res[1])
-
-with gr.Blocks(title="LTX-2.3 CPU Optimized") as demo:
-    gr.Markdown("# LTX-2.3 CPU-Optimized LoRA")
-    with gr.Row():
-        with gr.Column():
-            with gr.Row():
-                first_image = gr.Image(label="First Frame", type="pil")
-                last_image = gr.Image(label="Last Frame", type="pil")
-            input_audio = gr.Audio(label="Audio Input", type="filepath")
-            prompt = gr.Textbox(label="Prompt", value="Make this image come alive", lines=3)
-            duration = gr.Slider(label="Duration (s)", minimum=1.0, maximum=30.0, value=10.0, step=0.1)
-            generate_btn = gr.Button("Generate Video", variant="primary", size="lg")
-            with gr.Accordion("Advanced Settings", open=False):
-                seed = gr.Slider(label="Seed", minimum=0, maximum=np.iinfo(np.int32).max, value=10, step=1)
-                randomize_seed = gr.Checkbox(label="Randomize Seed", value=True)
-                with gr.Row():
-                    width = gr.Number(label="Width", value=1536, precision=0)
-                    height = gr.Number(label="Height", value=1024, precision=0)
-                with gr.Row():
-                    enhance_prompt = gr.Checkbox(label="Enhance Prompt", value=False)
-                    high_res = gr.Checkbox(label="High Resolution", value=True)
-                gr.Markdown("### LoRA Adapter Strengths")
-                lora_sliders = [gr.Slider(label=f"{k} strength", minimum=0.0, maximum=2.0, value=0.0, step=0.01) for k in LORA_FILES.keys()]
-                prepare_btn = gr.Button("Prepare LoRA Cache", variant="secondary")
-                lora_status = gr.Textbox(label="Status", interactive=False)
-        with gr.Column():
-            output_video = gr.Video(label="Generated Video")
-            gpu_duration = gr.Slider(label="ZeroGPU Duration (s)", minimum=30.0, maximum=240.0, value=75.0, step=1.0)
-
-    first_image.change(fn=update_res, inputs=[first_image, last_image, high_res], outputs=[width, height])
-    last_image.change(fn=update_res, inputs=[first_image, last_image, high_res], outputs=[width, height])
-    high_res.change(fn=update_res, inputs=[first_image, last_image, high_res], outputs=[width, height])
-    prepare_btn.click(fn=prepare_lora_cache, inputs=lora_sliders, outputs=lora_status)
-    generate_btn.click(fn=generate_video, inputs=[first_image, last_image, input_audio, prompt, duration, gpu_duration, enhance_prompt, seed, randomize_seed, height, width] + lora_sliders, outputs=[output_video, seed])
-
-if __name__ == "__main__":
-    demo.launch(theme=gr.themes.Citrus())