handler.py

"""
handler.py — Hugging Face Inference Endpoint custom handler
Outputs: GIF, WebM, ZIP(frames)

This version maintains UNIVERSAL compatibility:
- Defensive argument guessing (num_frames vs video_length)
- Robust output shape parsing (TBL, BCTHW, etc.)
- Adds Support for Image-to-Video via `image` input (base64)
"""

from __future__ import annotations

# ---------------------------------------------------------------------
# 0) VERY EARLY RUNTIME PATCHES (must happen before model/toolkit imports)
# ---------------------------------------------------------------------

import os
import sys
import types

os.environ.setdefault("IMAGEIO_NO_INTERNET", "1")
os.environ.setdefault("HF_HUB_DISABLE_TELEMETRY", "1")

def _patch_hf_toolkit_ffmpeg_plugin() -> dict:
    """
    Best-effort patching so huggingface_inference_toolkit won't crash if something
    tries to resolve plugin name "ffmpeg".
    """
    diag = {"patched": False, "details": []}

    try:
        import huggingface_inference_toolkit as hfit  # type: ignore
        diag["details"].append("imported huggingface_inference_toolkit")
    except Exception as e:
        diag["details"].append(f"huggingface_inference_toolkit not importable: {e}")
        return diag

    # Registry-like dicts on root module
    registry_candidates = []
    for name in dir(hfit):
        if any(k in name.lower() for k in ("plugin", "registry", "registries", "plugins")):
            try:
                obj = getattr(hfit, name)
                if isinstance(obj, dict):
                    registry_candidates.append((name, obj))
            except Exception:
                pass

    for name, reg in registry_candidates:
        if "ffmpeg" not in reg:
            try:
                reg["ffmpeg"] = object()
                diag["patched"] = True
                diag["details"].append(f"added ffmpeg to registry dict: hfit.{name}")
            except Exception as e:
                diag["details"].append(f"failed adding to hfit.{name}: {e}")

    # Wrap resolver-like functions
    fn_names = [n for n in dir(hfit) if any(k in n.lower() for k in ("get_plugin", "resolve_plugin", "load_plugin"))]
    for fn_name in fn_names:
        try:
            fn = getattr(hfit, fn_name)
            if not callable(fn):
                continue
            if getattr(fn, "__ffmpeg_patched__", False):
                continue

            def _wrap(original):
                def wrapped(*args, **kwargs):
                    if args and isinstance(args[0], str) and args[0].lower() == "ffmpeg":
                        return object()
                    return original(*args, **kwargs)
                wrapped.__ffmpeg_patched__ = True  # type: ignore
                return wrapped

            setattr(hfit, fn_name, _wrap(fn))
            diag["patched"] = True
            diag["details"].append(f"wrapped callable: hfit.{fn_name} to accept ffmpeg")
        except Exception as e:
            diag["details"].append(f"failed wrapping {fn_name}: {e}")

    # Dummy module injection (helps if toolkit tries importing a plugin module)
    dummy_mod_name = "huggingface_inference_toolkit.plugins.ffmpeg"
    if dummy_mod_name not in sys.modules:
        dummy = types.ModuleType(dummy_mod_name)
        dummy.__dict__["__doc__"] = "Dummy ffmpeg plugin injected by handler.py to avoid registry errors."
        sys.modules[dummy_mod_name] = dummy
        diag["details"].append(f"injected sys.modules['{dummy_mod_name}'] (dummy module)")

    return diag

HF_TOOLKIT_PATCH_DIAG = _patch_hf_toolkit_ffmpeg_plugin()

# ---------------------------------------------------------------------
# 1) Normal imports
# ---------------------------------------------------------------------

import base64
import io
import time
import tempfile
import zipfile
from dataclasses import dataclass
from typing import Any, Dict, List, Optional, Tuple

import numpy as np
from PIL import Image

import imageio

try:
    import imageio_ffmpeg  # type: ignore
    _FFMPEG_EXE = imageio_ffmpeg.get_ffmpeg_exe()
    # Force absolute ffmpeg path via env var; do NOT use imageio executable= param (not supported here).
    os.environ["IMAGEIO_FFMPEG_EXE"] = _FFMPEG_EXE
except Exception:
    _FFMPEG_EXE = ""


# ---------------------------------------------------------------------
# 2) Helpers
# ---------------------------------------------------------------------

def _now_ms() -> int:
    return int(time.time() * 1000)


def _b64(data: bytes) -> str:
    return base64.b64encode(data).decode("utf-8")


def _b64_to_pil(b64_str: str) -> Image.Image:
    if "," in b64_str:
        b64_str = b64_str.split(",")[1]
    data = base64.b64decode(b64_str)
    return Image.open(io.BytesIO(data)).convert("RGB")


def _clamp_uint8_frame(frame: np.ndarray) -> np.ndarray:
    """
    Normalize a frame into uint8 RGB (H,W,3).
    Accepts float [0,1] or [-1,1], uint8, grayscale, RGBA, or CHW.
    """
    if not isinstance(frame, np.ndarray):
        frame = np.array(frame)

    # squeeze batch dim (rare)
    if frame.ndim == 4 and frame.shape[0] == 1:
        frame = frame[0]

    # grayscale -> rgb
    if frame.ndim == 2:
        frame = np.stack([frame, frame, frame], axis=-1)

    if frame.ndim != 3:
        raise ValueError(f"Frame must be 2D or 3D array; got shape {frame.shape}")

    # channel fixups
    if frame.shape[-1] == 4:
        frame = frame[..., :3]
    elif frame.shape[-1] == 1:
        frame = np.repeat(frame, 3, axis=-1)
    elif frame.shape[-1] != 3:
        # maybe CHW
        if frame.shape[0] == 3:
            frame = np.transpose(frame, (1, 2, 0))
        else:
            raise ValueError(f"Unsupported channels shape: {frame.shape}")

    if frame.dtype == np.uint8:
        return frame

    f = frame.astype(np.float32)
    if f.min() < 0.0:
        f = (f + 1.0) / 2.0
    f = np.clip(f, 0.0, 1.0)
    return (f * 255.0).round().astype(np.uint8)


def _encode_gif(frames: List[np.ndarray], fps: int) -> bytes:
    if not frames:
        raise ValueError("No frames to encode GIF.")
    pil_frames = [Image.fromarray(_clamp_uint8_frame(f)) for f in frames]
    duration_ms = int(1000 / max(1, fps))
    buf = io.BytesIO()
    pil_frames[0].save(
        buf,
        format="GIF",
        save_all=True,
        append_images=pil_frames[1:],
        duration=duration_ms,
        loop=0,
        optimize=False,
        disposal=2,
    )
    return buf.getvalue()


def _encode_webm(frames: List[np.ndarray], fps: int, quality: str = "good") -> bytes:
    """
    Encode WebM (VP9) via imageio.
    """
    if not frames:
        raise ValueError("No frames to encode WebM.")

    quality = (quality or "good").lower()
    if quality == "fast":
        crf = 42
        preset = "veryfast"
    elif quality == "best":
        crf = 28
        preset = "slow"
    else:
        crf = 34
        preset = "medium"

    with tempfile.NamedTemporaryFile(suffix=".webm", delete=False) as tmp:
        out_path = tmp.name

    try:
        writer = imageio.get_writer(
            out_path,
            fps=max(1, fps),
            format="FFMPEG",
            codec="libvpx-vp9",
            ffmpeg_params=[
                "-pix_fmt", "yuv420p",
                "-crf", str(crf),
                "-b:v", "0",
                "-preset", preset,
            ],
        )
        try:
            for f in frames:
                writer.append_data(_clamp_uint8_frame(f))
        finally:
            writer.close()

        with open(out_path, "rb") as f:
            return f.read()
    finally:
        try:
            os.remove(out_path)
        except Exception:
            pass


def _encode_zip_frames(frames: List[np.ndarray]) -> bytes:
    if not frames:
        raise ValueError("No frames to ZIP.")

    buf = io.BytesIO()
    with zipfile.ZipFile(buf, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=6) as zf:
        for i, f in enumerate(frames):
            arr = _clamp_uint8_frame(f)
            im = Image.fromarray(arr)
            frame_buf = io.BytesIO()
            im.save(frame_buf, format="PNG", optimize=True)
            zf.writestr(f"frame_{i:06d}.png", frame_buf.getvalue())
    return buf.getvalue()


# ---------------------------------------------------------------------
# 3) Request schema
# ---------------------------------------------------------------------

@dataclass
class GenParams:
    prompt: str
    negative_prompt: str
    num_frames: int
    fps: int
    height: int
    width: int
    seed: Optional[int]
    num_inference_steps: int
    guidance_scale: float
    image_b64: Optional[str] = None


def _unwrap_inputs(payload: Dict[str, Any]) -> Dict[str, Any]:
    if isinstance(payload, dict) and "inputs" in payload and isinstance(payload["inputs"], dict):
        return payload["inputs"]
    return payload


def _parse_request(payload: Dict[str, Any]) -> Tuple[GenParams, List[str], bool, Dict[str, Any]]:
    data = _unwrap_inputs(payload)

    prompt = str(data.get("prompt") or data.get("inputs") or "").strip()
    if not prompt and "image" not in data:
         pass

    negative_prompt = str(data.get("negative_prompt") or "").strip()

    num_frames = int(data.get("num_frames") or data.get("frames") or 32)
    fps = int(data.get("fps") or 12)
    height = int(data.get("height") or 512)
    width = int(data.get("width") or 512)
    seed = data.get("seed")
    seed = int(seed) if seed is not None and str(seed).strip() != "" else None

    # Image input for I2V
    image_b64 = data.get("image") or data.get("image_base64")

    num_inference_steps = int(data.get("num_inference_steps") or 30)
    guidance_scale = float(data.get("guidance_scale") or 7.5)

    outputs = data.get("outputs") or ["gif"]
    if isinstance(outputs, str):
        outputs = [outputs]
    outputs = [str(x).lower() for x in outputs]
    allowed = {"gif", "webm", "zip"}
    outputs = [o for o in outputs if o in allowed]
    if not outputs:
        outputs = ["gif"]

    return_base64 = bool(data.get("return_base64", True))

    out_cfg = data.get("output_config") or {}
    for k in ("gif", "webm", "zip"):
        if k in data and isinstance(data[k], dict):
            out_cfg[k] = data[k]

    params = GenParams(
        prompt=prompt,
        negative_prompt=negative_prompt,
        num_frames=max(1, num_frames),
        fps=max(1, fps),
        height=max(64, height),
        width=max(64, width),
        seed=seed,
        num_inference_steps=max(1, num_inference_steps),
        guidance_scale=guidance_scale,
        image_b64=image_b64
    )
    return params, outputs, return_base64, out_cfg


# ---------------------------------------------------------------------
# 4) EndpointHandler
# ---------------------------------------------------------------------

class EndpointHandler:
    def __init__(self, path: str = "") -> None:
        self.repo_path = path or ""
        self.pipe = None
        self.init_error: Optional[str] = None

        print("=== CUSTOM handler.py LOADED (Universal Mode) ===", flush=True)
        print(f"=== HF toolkit patch diag: {HF_TOOLKIT_PATCH_DIAG} ===", flush=True)
        print(f"=== imageio-ffmpeg exe: {_FFMPEG_EXE} ===", flush=True)

        try:
            import torch  # type: ignore
            from diffusers import DiffusionPipeline, LTXConditionPipeline

            device = "cuda" if torch.cuda.is_available() else "cpu"
            dtype = torch.float16 if device == "cuda" else torch.float32

            subdir = os.getenv("HF_MODEL_SUBDIR", "").strip()
            model_path = self.repo_path if not subdir else os.path.join(self.repo_path, subdir)

            # --- Attempt to load LTXConditionPipeline first (for I2V Support) ---
            # If that fails (e.g. model isn't LTX or diffusers version old), fallback to generic.
            try:
                print("Attempting to load LTXConditionPipeline...", flush=True)
                self.pipe = LTXConditionPipeline.from_pretrained(model_path, torch_dtype=dtype)
            except Exception as e:
                print(f"LTXConditionPipeline load failed ({e}), falling back to generic DiffusionPipeline...", flush=True)
                self.pipe = DiffusionPipeline.from_pretrained(model_path, torch_dtype=dtype)

            try:
                self.pipe.to(device)
            except Exception:
                pass

            try:
                if hasattr(self.pipe, "enable_vae_slicing"):
                    self.pipe.enable_vae_slicing()
            except Exception:
                pass
            
            # Optimization for LTX / newer diffusers
            if hasattr(self.pipe, "vae") and hasattr(self.pipe.vae, "enable_tiling"):
                self.pipe.vae.enable_tiling()

        except Exception as e:
            self.init_error = str(e)
            self.pipe = None
            print(f"=== PIPELINE INIT FAILED: {self.init_error} ===", flush=True)

    def __call__(self, payload: Dict[str, Any]) -> Dict[str, Any]:
        t0 = _now_ms()

        try:
            params, outputs, return_b64, out_cfg = _parse_request(payload)

            frames, gen_diag = self._generate_frames(params)

            t1 = _now_ms()
            result_outputs: Dict[str, Any] = {}

            # GIF
            if "gif" in outputs:
                gif_fps = int((out_cfg.get("gif") or {}).get("fps") or params.fps)
                gif_bytes = _encode_gif(frames, fps=gif_fps)
                result_outputs["gif_base64" if return_b64 else "gif_bytes"] = _b64(gif_bytes) if return_b64 else gif_bytes

            t2 = _now_ms()

            # WebM
            if "webm" in outputs:
                webm_cfg = out_cfg.get("webm") or {}
                webm_fps = int(webm_cfg.get("fps") or params.fps)
                webm_quality = str(webm_cfg.get("quality") or "good")
                webm_bytes = _encode_webm(frames, fps=webm_fps, quality=webm_quality)
                result_outputs["webm_base64" if return_b64 else "webm_bytes"] = _b64(webm_bytes) if return_b64 else webm_bytes

            t3 = _now_ms()

            # ZIP
            if "zip" in outputs:
                zip_bytes = _encode_zip_frames(frames)
                result_outputs["zip_base64" if return_b64 else "zip_bytes"] = _b64(zip_bytes) if return_b64 else zip_bytes

            t4 = _now_ms()

            return {
                "ok": True,
                "outputs": result_outputs,
                "diagnostics": {
                    "timing_ms": {
                        "total": t4 - t0,
                        "generate": t1 - t0,
                        "gif": (t2 - t1) if "gif" in outputs else 0,
                        "webm": (t3 - t2) if "webm" in outputs else 0,
                        "zip": (t4 - t3) if "zip" in outputs else 0,
                    },
                    "generator": gen_diag,
                    "ffmpeg_exe": _FFMPEG_EXE,
                    "hf_toolkit_patch": HF_TOOLKIT_PATCH_DIAG,
                    "init_error": self.init_error,
                },
            }

        except Exception as e:
            import traceback
            traceback.print_exc()
            return {
                "ok": False,
                "error": str(e),
                "diagnostics": {
                    "ffmpeg_exe": _FFMPEG_EXE,
                    "hf_toolkit_patch": HF_TOOLKIT_PATCH_DIAG,
                    "init_error": self.init_error,
                },
            }

    # ----------------------------
    # Frame generation
    # ----------------------------

    def _generate_frames(self, params: GenParams) -> Tuple[List[np.ndarray], Dict[str, Any]]:
        if self.pipe is None:
            raise RuntimeError(f"Model pipeline not initialized. Init error: {self.init_error}")

        # Seed (best effort)
        generator = None
        try:
            import torch  # type: ignore
            if params.seed is not None:
                device = "cuda" if torch.cuda.is_available() else "cpu"
                generator = torch.Generator(device=device).manual_seed(params.seed)
        except Exception:
            generator = None

        kwargs: Dict[str, Any] = {
            "prompt": params.prompt,
            "negative_prompt": params.negative_prompt if params.negative_prompt else None,
            "height": params.height,
            "width": params.width,
            "num_inference_steps": params.num_inference_steps,
            "guidance_scale": params.guidance_scale,
            # "num_frames" is intentionally OMITTED here to be handled by the loop below
        }
        
        # Handle Image-to-Video
        # Use simple argument passing if pipeline supports it (LTXConditionPipeline does)
        # If image is present, we pass it.
        if params.image_b64:
            print("Received image input, performing Image-to-Video.", flush=True)
            pil_image = _b64_to_pil(params.image_b64)
            kwargs["image"] = pil_image

        # Try common frame arg names across video pipelines
        output = None
        last_err: Optional[Exception] = None
        
        # UNIVERSAL LOOP: Try all known frame arguments
        for frame_arg in ("num_frames", "video_length", "num_video_frames"):
            try:
                call_kwargs = dict(kwargs)
                call_kwargs[frame_arg] = params.num_frames
                if generator is not None:
                    call_kwargs["generator"] = generator
                
                # Filter out None values just in case
                clean_kwargs = {k: v for k, v in call_kwargs.items() if v is not None}
                
                output = self.pipe(**clean_kwargs)
                break
            except Exception as e:
                last_err = e
                # Don't print spam, just try next arg
                continue

        if output is None:
            raise RuntimeError(f"Pipeline call failed. Last error: {last_err}")

        frames: List[np.ndarray] = []

        # UNIVERSAL OUTPUT PARSING: Handle all known shapes
        
        # 1) output.frames — may be list OR ndarray/tensor-like
        if hasattr(output, "frames") and getattr(output, "frames") is not None:
            frames_raw = getattr(output, "frames")
            arr = np.array(frames_raw)

            # (T,H,W,C)
            if arr.ndim == 4:
                frames = [arr[t] for t in range(arr.shape[0])]
            # (B,T,H,W,C)
            elif arr.ndim == 5:
                arr = arr[0]
                frames = [arr[t] for t in range(arr.shape[0])]
            # list-of-frames
            elif isinstance(frames_raw, list):
                frames = [np.array(f) for f in frames_raw]
            else:
                raise ValueError(f"Unexpected frames shape: {arr.shape}")

        # 2) output.videos
        elif hasattr(output, "videos") and getattr(output, "videos") is not None:
            vids = getattr(output, "videos")
            arr = None
            try:
                import torch  # type: ignore
                if isinstance(vids, torch.Tensor):
                    arr = vids.detach().cpu().numpy()
                else:
                    arr = np.array(vids)
            except Exception:
                arr = np.array(vids)

            # (B,T,C,H,W) or (B,T,H,W,C) or (T,C,H,W) or (T,H,W,C)
            if arr.ndim == 5:
                arr = arr[0]

            # (T,C,H,W) -> (T,H,W,C)
            if arr.ndim == 4 and arr.shape[1] in (1, 3, 4):
                arr = np.transpose(arr, (0, 2, 3, 1))

            if arr.ndim != 4:
                raise ValueError(f"Unexpected video tensor shape: {arr.shape}")

            frames = [arr[t] for t in range(arr.shape[0])]

        # 3) output.images (single frame or list)
        elif hasattr(output, "images") and getattr(output, "images") is not None:
            imgs = getattr(output, "images\")
            if isinstance(imgs, list):
                frames = [np.array(im) for im in imgs]
            else:
                frames = [np.array(imgs)]

        # 4) dict fallback
        elif isinstance(output, dict):
            for key in ("frames", "videos", "images"):
                if key in output and output[key] is not None:
                    v = output[key]
                    if key == "frames":
                        arr = np.array(v)
                        if arr.ndim == 4:
                            frames = [arr[t] for t in range(arr.shape[0])]
                        elif arr.ndim == 5:
                            arr = arr[0]
                            frames = [arr[t] for t in range(arr.shape[0])]
                        elif isinstance(v, list):
                            frames = [np.array(x) for x in v]
                        else:
                            raise ValueError(f"Unexpected dict frames shape: {arr.shape}")
                    elif key == "videos":
                        arr = np.array(v)
                        if arr.ndim == 5:
                            arr = arr[0]
                        if arr.ndim == 4 and arr.shape[1] in (1, 3, 4):
                            arr = np.transpose(arr, (0, 2, 3, 1))
                        frames = [arr[t] for t in range(arr.shape[0])]
                    else:
                        if isinstance(v, list):
                            frames = [np.array(x) for x in v]
                        else:
                            frames = [np.array(v)]
                    break

        if not frames:
            raise RuntimeError("Could not extract frames from pipeline output (no frames/videos/images found).")

        frames_u8 = [_clamp_uint8_frame(f) for f in frames]

        diag = {
            "prompt_len": len(params.prompt),
            "negative_prompt_len": len(params.negative_prompt),
            "num_frames": len(frames_u8),
            "height": int(frames_u8[0].shape[0]),
            "width": int(frames_u8[0].shape[1]),
            "mode": "i2v" if params.image_b64 else "t2v"
        }
        return frames_u8, diag