app.py

import json
import logging
import os
import random
import struct
import subprocess
import sys
import tempfile
from pathlib import Path

# Disable torch.compile / dynamo before any torch import
os.environ["TORCH_COMPILE_DISABLE"] = "1"
os.environ["TORCHDYNAMO_DISABLE"] = "1"
os.environ.setdefault("PYTORCH_CUDA_ALLOC_CONF", "expandable_segments:True")

# Install xformers for memory-efficient attention
subprocess.run(
    [sys.executable, "-m", "pip", "install", "xformers==0.0.32.post2", "--no-build-isolation"],
    check=False,
)

# 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_SHA = "780984275fd47128b02bef9b5c085404276866ee"


def _ensure_ltx_repo() -> None:
    import shutil
    if os.path.exists(LTX_REPO_DIR):
        head = subprocess.run(
            ["git", "-C", LTX_REPO_DIR, "rev-parse", "HEAD"],
            capture_output=True,
            text=True,
            check=False,
        )
        if head.returncode == 0 and head.stdout.strip() == LTX_COMMIT_SHA:
            return
        shutil.rmtree(LTX_REPO_DIR, ignore_errors=True)

    print(f"Cloning {LTX_REPO_URL} @ {LTX_COMMIT_SHA[:8]}...")
    os.makedirs(LTX_REPO_DIR, exist_ok=True)
    subprocess.run(["git", "init", LTX_REPO_DIR], check=True)
    subprocess.run(["git", "remote", "add", "origin", LTX_REPO_URL], cwd=LTX_REPO_DIR, check=True)
    subprocess.run(["git", "fetch", "--depth", "1", "origin", LTX_COMMIT_SHA], cwd=LTX_REPO_DIR, check=True)
    subprocess.run(["git", "checkout", LTX_COMMIT_SHA], cwd=LTX_REPO_DIR, check=True)


_ensure_ltx_repo()

print("Installing ltx-core and ltx-pipelines from cloned repo...")
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 av
import torch

torch._dynamo.config.suppress_errors = True
torch._dynamo.config.disable = True

import gradio as gr
import numpy as np
import spaces
from huggingface_hub import hf_hub_download, snapshot_download
from PIL import Image

from ltx_core.loader import LTXV_LORA_COMFY_RENAMING_MAP, LoraPathStrengthAndSDOps
from ltx_core.loader.primitives import StateDict
from ltx_core.loader.sft_loader import SafetensorsStateDictLoader
from ltx_core.model.video_vae import TilingConfig, get_video_chunks_number
from ltx_core.quantization.fp8_cast import build_policy as build_fp8_cast_policy
from ltx_pipelines.distilled import DistilledPipeline
from ltx_pipelines.utils.args import ImageConditioningInput
from ltx_pipelines.utils.media_io import encode_video

from ltx_core.model.transformer import attention as _attn_mod

print(f"[ATTN] Before patch: memory_efficient_attention={_attn_mod.memory_efficient_attention}")
try:
    from xformers.ops import memory_efficient_attention as _mea
    _attn_mod.memory_efficient_attention = _mea
    print(f"[ATTN] After patch: memory_efficient_attention={_attn_mod.memory_efficient_attention}")
except Exception as e:
    print(f"[ATTN] xformers patch FAILED: {type(e).__name__}: {e}")

try:
    from xformers.ops.fmha import _set_use_fa3
    _set_use_fa3(False)
    print("[ATTN] xformers FA3 dispatch disabled")
except Exception as e:
    print(f"[ATTN] FA3 disable FAILED: {type(e).__name__}: {e}")

_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 = 0
    dtype = set()
    device = device or 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
print("[FUSE-PATCH] SafetensorsStateDictLoader.load replaced (chunked-read)")

logging.getLogger().setLevel(logging.INFO)

MAX_SEED = np.iinfo(np.int32).max
DEFAULT_LORA_STRENGTH = 0.6
TOKEN_ENV_NAMES = ("HF_TOKEN", "HUGGINGFACE_HUB_TOKEN", "HUGGING_FACE_HUB_TOKEN", "HUGGINGFACE_TOKEN")
HUB_MODEL_ID = os.environ.get("PINKCHERRY_HUB_REPO", "SexGod1979/PinkCherry_NSFW_LTX23")
DATA_MOUNT = os.environ.get("LTX_DATA_ROOT", "/data")
CACHE_DIR = Path(os.environ.get("LTX_CACHE_DIR", str(Path.home() / ".cache" / "pinkcherry-ltx")))
CACHE_DIR.mkdir(parents=True, exist_ok=True)
OUTPUT_DIR = Path("outputs")
OUTPUT_DIR.mkdir(exist_ok=True)
LAST_FRAME_PATH = OUTPUT_DIR / "last_frame.jpg"

CHECKPOINT_NAME = "SexGod_PinkCherry_dev_bf16_LTX23_v1.safetensors"
UPSCALER_FILENAME = "ltx-2.3-spatial-upscaler-x2-1.1.safetensors"
FALLBACK_LORA_NAME = "ltx-2.3-22b-distilled-lora-1.1_fro90_ceil72_condsafe.safetensors"

CHECKPOINT_CANDIDATES = (
    f"model/{CHECKPOINT_NAME}",
    f"v1/model/{CHECKPOINT_NAME}",
)
LORA_SEARCH_DIRS = (
    "distil_lora",
    "v1/distil_lora",
    "lora",
    "loras",
)
HUB_LORA_CANDIDATES = tuple(f"{d}/{FALLBACK_LORA_NAME}" for d in ("distil_lora", "v1/distil_lora"))

_pipeline_cache: dict[str, object] = {"key": None, "pipeline": None}
pipeline = None


def _get_hf_token() -> str | None:
    for name in TOKEN_ENV_NAMES:
        token = os.environ.get(name)
        if token and token.strip():
            return token.strip()
    return None


def _resolve_asset(candidates: tuple[str, ...], label: str) -> tuple[str, str]:
    for relpath in candidates:
        bucket_path = os.path.join(DATA_MOUNT, relpath)
        if os.path.isfile(bucket_path) and os.path.getsize(bucket_path) > 0:
            print(f"[ASSET] {label}: bucket -> {bucket_path}")
            return bucket_path, "bucket"
    token = _get_hf_token()
    last_error = None
    for relpath in candidates:
        try:
            hub_path = hf_hub_download(
                HUB_MODEL_ID,
                relpath,
                token=token,
                local_dir=str(CACHE_DIR / "hub-mirror"),
            )
            print(f"[ASSET] {label}: hub -> {hub_path}")
            return hub_path, "hub"
        except Exception as exc:
            last_error = exc
    raise FileNotFoundError(
        f"Could not resolve {label}. Checked bucket under {DATA_MOUNT} and hub repo {HUB_MODEL_ID}: {last_error}"
    )


def _ensure_supporting_assets() -> tuple[str, str]:
    upscaler_path = os.path.join(DATA_MOUNT, UPSCALER_FILENAME)
    if not (os.path.isfile(upscaler_path) and os.path.getsize(upscaler_path) > 0):
        upscaler_path = hf_hub_download(
            "Lightricks/LTX-2.3",
            UPSCALER_FILENAME,
            token=_get_hf_token(),
            local_dir=str(CACHE_DIR),
        )
        print(f"[ASSET] spatial upsampler: hub -> {upscaler_path}")
    gemma_root = os.environ.get("GEMMA_ROOT", str(CACHE_DIR / "gemma-3-12b-it"))
    gemma_path = Path(gemma_root)
    if not gemma_path.exists() or not any(gemma_path.rglob("model*.safetensors")):
        snapshot_download(
            "google/gemma-3-12b-it-qat-q4_0-unquantized",
            token=_get_hf_token(),
            local_dir=gemma_root,
        )
        print(f"[ASSET] gemma: downloaded -> {gemma_root}")
    return upscaler_path, gemma_root


def scan_lora_files() -> dict[str, str]:
    found: dict[str, str] = {}
    for subdir in LORA_SEARCH_DIRS:
        root = Path(DATA_MOUNT) / subdir
        if not root.is_dir():
            continue
        for path in sorted(root.glob("*.safetensors")):
            if path.is_file() and path.stat().st_size > 0:
                label = f"{subdir}/{path.name}"
                found[label] = str(path)
    return found


def _default_lora_label(lora_map: dict[str, str]) -> str | None:
    if not lora_map:
        return None
    for label in lora_map:
        if "distilled" in label.lower():
            return label
    return next(iter(lora_map))


def _resolve_lora_path(lora_label: str | None, lora_map: dict[str, str]) -> str | None:
    if lora_label and lora_label in lora_map:
        return lora_map[lora_label]
    if lora_map:
        return lora_map[_default_lora_label(lora_map)]
    path, _ = _resolve_asset(HUB_LORA_CANDIDATES, "distilled lora")
    return path


def _build_pipeline(lora_path: str | None, lora_strength: float) -> DistilledPipeline:
    cache_key = (lora_path, round(float(lora_strength), 4))
    cached = _pipeline_cache.get("key")
    if cached == cache_key and _pipeline_cache.get("pipeline") is not None:
        print(f"[CACHE] Hit for key {cache_key}")
        return _pipeline_cache["pipeline"]
    print(f"[PIPELINE] Cache miss or rebuild forced. Instantiating lora={lora_path} @ {lora_strength}")
    loras = []
    if lora_path:
        loras = [
            LoraPathStrengthAndSDOps(
                lora_path,
                float(lora_strength),
                LTXV_LORA_COMFY_RENAMING_MAP,
            )
        ]
    new_pipeline = DistilledPipeline(
        distilled_checkpoint_path=distilled_checkpoint_path,
        spatial_upsampler_path=spatial_upsampler_path,
        gemma_root=gemma_root,
        loras=loras,
        quantization=build_fp8_cast_policy(distilled_checkpoint_path),
    )
    _pipeline_cache["key"] = cache_key
    _pipeline_cache["pipeline"] = new_pipeline
    if torch.cuda.is_available():
        torch.cuda.empty_cache()
    return new_pipeline


distilled_checkpoint_path, checkpoint_source = _resolve_asset(CHECKPOINT_CANDIDATES, "checkpoint")
spatial_upsampler_path, gemma_root = _ensure_supporting_assets()
LORA_FILES = scan_lora_files()
DEFAULT_LORA_LABEL = _default_lora_label(LORA_FILES)

print(f"[PIPELINE] checkpoint={distilled_checkpoint_path} ({checkpoint_source})")
print(f"[PIPELINE] upsampler={spatial_upsampler_path}")
print(f"[PIPELINE] gemma={gemma_root}")
print(f"[PIPELINE] loras found: {list(LORA_FILES)}")
print("=" * 80)
print("[STARTUP] Pre-compiling default pipeline at root module level to leverage ZeroGPU CUDA emulation...")
try:
    default_lora_path = _resolve_lora_path(DEFAULT_LORA_LABEL, LORA_FILES)
    pipeline = _build_pipeline(default_lora_path, DEFAULT_LORA_STRENGTH)
    print("[STARTUP] Baseline pipeline warmed up successfully!")
except Exception as e:
    print(f"[STARTUP] Warning: Startup compilation failed: {e}")
print("=" * 80)


def log_memory(tag: str):
    if torch.cuda.is_available():
        allocated = torch.cuda.memory_allocated() / 1024**3
        peak = torch.cuda.max_memory_allocated() / 1024**3
        free, total = torch.cuda.mem_get_info()
        print(
            f"[VRAM {tag}] allocated={allocated:.2f}GB peak={peak:.2f}GB "
            f"free={free / 1024**3:.2f}GB total={total / 1024**3:.2f}GB"
        )


def refresh_lora_dropdown():
    global LORA_FILES
    LORA_FILES = scan_lora_files()
    choices = list(LORA_FILES.keys())
    value = _default_lora_label(LORA_FILES) if choices else None
    return gr.update(choices=choices, value=value)


def _extract_last_frame_pil(video_path: str) -> Image.Image | None:
    container = av.open(video_path)
    try:
        stream = container.streams.video[0]
        last_frame = None
        for frame in container.decode(stream):
            last_frame = frame
        if last_frame is None:
            return None
        return Image.fromarray(last_frame.to_rgb().to_ndarray())
    finally:
        container.close()


def _gpu_duration(input_image, prompt, duration, *args, **kwargs) -> int:
    target_video_seconds = float(duration)
    estimated_compute = int((target_video_seconds * 20) + 35)
    return min(estimated_compute, 120)


@spaces.GPU(duration=_gpu_duration)
@torch.inference_mode()
def generate_video(
    input_image,
    prompt: str,
    duration: float,
    lora_label: str,
    lora_strength: float,
    chain_last_frame: bool,
    seed: int,
    randomize_seed: bool,
    progress=gr.Progress(track_tqdm=True),
):
    global pipeline
    current_seed = random.randint(0, MAX_SEED) if randomize_seed else int(seed)
    try:
        torch.cuda.reset_peak_memory_stats()
        log_memory("start")
        
        # Hardcoded clean 64-divisible dimensions (832/64=13, 512/64=8)
        width = 832
        height = 512
        fps = 16.0
        
        lora_path = _resolve_lora_path(lora_label, LORA_FILES)
        pipeline = _build_pipeline(lora_path, lora_strength)
        
        num_frames = int(duration * fps) + 1
        num_frames = ((num_frames - 1 + 7) // 8) * 8 + 1
        print(
            f"Generating: {width}x{height}, {num_frames} frames ({duration}s @ {fps}fps), "
            f"seed={current_seed}, lora={lora_label}@{lora_strength}"
        )
        
        images = []
        if input_image is not None:
            temp_image_path = OUTPUT_DIR / f"temp_input_{current_seed}.jpg"
            if hasattr(input_image, "save"):
                input_image.save(temp_image_path)
            else:
                temp_image_path = Path(input_image)
            images = [ImageConditioningInput(path=str(temp_image_path), frame_idx=0, strength=1.0)]
            
        tiling_config = TilingConfig.default()
        video_chunks_number = get_video_chunks_number(num_frames, tiling_config)
        log_memory("before pipeline call")
        
        video, audio = pipeline(
            prompt=prompt,
            seed=current_seed,
            height=height,
            width=width,
            num_frames=num_frames,
            frame_rate=fps,
            images=images,
            tiling_config=tiling_config,
            enhance_prompt=False,  # Force disabled
        )
        
        log_memory("after pipeline call")
        output_path = tempfile.mktemp(suffix=".mp4")
        encode_video(
            video=video,
            fps=int(round(fps)),
            audio=audio,
            output_path=output_path,
            video_chunks_number=video_chunks_number,
        )
        log_memory("after encode_video")
        
        next_input = input_image
        if chain_last_frame:
            last_frame = _extract_last_frame_pil(output_path)
            if last_frame is not None:
                last_frame.save(LAST_FRAME_PATH)
                next_input = last_frame
                print(f"[CHAIN] saved last frame -> {LAST_FRAME_PATH}")
                
        return str(output_path), current_seed, next_input
    except Exception as e:
        import traceback
        log_memory("on error")
        print(f"Error: {str(e)}\n{traceback.format_exc()}")
        raise gr.Error(str(e)) from e


lora_choices = list(LORA_FILES.keys())
default_lora = DEFAULT_LORA_LABEL or (lora_choices[0] if lora_choices else None)

with gr.Blocks(title="LTX-Naughty-Edition") as demo:
    gr.Markdown(
        "# LTX Naughty Edition\n\n"
        "A Franken-Space. I don't claim credit for underlying parts.<br>"
        "Locked 16:9 480p standards (832x512) @ 16 FPS for maximum speed and priority on ZeroGPU.\n\n"
        "what would random guy on the internet do (wwrdotid). Poor example. Seriously, don't be a creep."
    )
    with gr.Row():
        with gr.Column():
            input_image = gr.Image(label="First frame (optional)", type="pil")
            prompt = gr.Textbox(
                label="Prompt",
                info="For best results - make it as elaborate as possible, anything but default",
                value="animate an erotic scene with the given elements",
                lines=3,
                placeholder="Describe the motion and animation you want...",
            )
            with gr.Row():
                duration = gr.Slider(label="Duration (seconds)", minimum=1.0, maximum=4.0, value=3.0, step=0.1)
            with gr.Row():
                lora_dropdown = gr.Dropdown(
                    label="Distilled LoRA",
                    choices=lora_choices,
                    value=default_lora,
                    interactive=True,
                )
                refresh_loras_btn = gr.Button("Rescan LoRAs", scale=0)
            lora_strength = gr.Slider(
                label="LoRA strength",
                minimum=0.0,
                maximum=1.5,
                value=DEFAULT_LORA_STRENGTH,
                step=0.05,
            )
            with gr.Row():
                chain_last_frame = gr.Checkbox(
                    label="Chain last frame to next input",
                    value=True,
                )
            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=MAX_SEED, value=10, step=1)
                randomize_seed = gr.Checkbox(label="Randomize Seed", value=True)
        with gr.Column():
            output_video = gr.Video(label="Generated Video", autoplay=True)

    refresh_loras_btn.click(fn=refresh_lora_dropdown, inputs=[], outputs=[lora_dropdown])
    generate_btn.click(
        fn=generate_video,
        inputs=[
            input_image,
            prompt,
            duration,
            lora_dropdown,
            lora_strength,
            chain_last_frame,
            seed,
            randomize_seed,
        ],
        outputs=[output_video, seed, input_image],
    )
    demo.load(fn=refresh_lora_dropdown, inputs=[], outputs=[lora_dropdown])

css = """.fillable{max-width: 1200px !important}.progress-text {color: red}"""

if __name__ == "__main__":
    demo.launch(theme=gr.themes.Citrus(), css=css)