src/animate.py

"""Turn a generated still into a short clip via fal xai/grok-imagine-video."""

from __future__ import annotations

from pathlib import Path

import fal_client
import requests

from src.gemini_analyze import FrameSpec
from src.regenerate import _scrub, require_fal_key, upload_image

GROK_VIDEO_MODEL = "xai/grok-imagine-video/image-to-video"
DEFAULT_CLIP_SECONDS = 4
MIN_CLIP_SECONDS = 1
MAX_CLIP_SECONDS = 15
RESOLUTIONS = ("480p", "720p")
DEFAULT_RESOLUTION = "720p"
# Narrower than the image endpoint's enum — no 2:1, 1:2, 20:9, 9:20 here.
VIDEO_ASPECT_RATIOS = ("auto", "16:9", "4:3", "3:2", "1:1", "2:3", "3:4", "9:16")

# fal list price per second of output, for the in-app estimate only. Update if fal changes it.
PRICE_PER_SECOND = {"480p": 0.08, "720p": 0.14}


def estimate_cost(n_clips: int, seconds: int, resolution: str) -> float:
    """Rough USD estimate — clips are the expensive part of a run, so it is shown up front."""
    return n_clips * seconds * PRICE_PER_SECOND.get(resolution, PRICE_PER_SECOND["720p"])


def clamp_seconds(seconds: int | float | None) -> int:
    """fal takes an integer 1..15 seconds."""
    try:
        value = int(round(float(seconds)))
    except (TypeError, ValueError):
        return DEFAULT_CLIP_SECONDS
    return max(MIN_CLIP_SECONDS, min(MAX_CLIP_SECONDS, value))


def normalize_video_aspect_ratio(aspect_ratio: str | None) -> str:
    ar = (aspect_ratio or "auto").strip() or "auto"
    return ar if ar in VIDEO_ASPECT_RATIOS else "auto"


def motion_prompt(frame: FrameSpec | None, extra: str = "") -> str:
    """Ask for gentle, believable movement — this animates an ad still, not a music video."""
    hint = ""
    if frame is not None:
        hint = _scrub(frame.regen_brief or "") or _scrub(frame.description or "")
    parts = [
        "Subtle, natural motion from this photo: slight handheld camera drift, small "
        "lifelike movement from the subject, gentle ambient motion in the background.",
        "Keep the same person, wardrobe, setting and low-grade phone-video look.",
        "No new text, captions, logos or watermarks. No scene cuts.",
    ]
    if hint:
        parts.insert(1, f"Scene: {hint}.")
    if extra.strip():
        parts.insert(1, f"Motion: {extra.strip()[:300]}.")
    return " ".join(parts)


def _extract_video_url(output: object) -> str | None:
    """fal image-to-video responds with ``{"video": {"url": ...}}``."""
    if isinstance(output, dict):
        video = output.get("video")
        if isinstance(video, dict):
            url = video.get("url")
            return str(url) if url else None
        if isinstance(video, str) and video.startswith("http"):
            return video
        url = output.get("url")
        return str(url) if url else None
    if isinstance(output, str) and output.startswith("http"):
        return output
    return None


def animate_frame(
    image_path: Path | str,
    out_path: Path | str,
    *,
    frame: FrameSpec | None = None,
    seconds: int = DEFAULT_CLIP_SECONDS,
    aspect_ratio: str = "auto",
    resolution: str = DEFAULT_RESOLUTION,
    instruction: str = "",
    api_token: str | None = None,
) -> Path:
    """Generate one clip from one still. A single, and expensive, API call."""
    src = Path(image_path)
    if not src.is_file():
        raise FileNotFoundError(f"Image not found: {src}")
    require_fal_key(api_token)

    dest = Path(out_path)
    dest.parent.mkdir(parents=True, exist_ok=True)

    args = {
        "prompt": motion_prompt(frame, instruction),
        "image_url": upload_image(src),
        "duration": clamp_seconds(seconds),
        "resolution": resolution if resolution in RESOLUTIONS else DEFAULT_RESOLUTION,
        "aspect_ratio": normalize_video_aspect_ratio(aspect_ratio),
    }
    output = fal_client.subscribe(GROK_VIDEO_MODEL, arguments=args)
    url = _extract_video_url(output)
    if not url:
        raise RuntimeError(f"fal returned no video URL: {output!r}")
    resp = requests.get(url, timeout=600)
    resp.raise_for_status()
    dest.write_bytes(resp.content)
    if dest.stat().st_size == 0:
        raise RuntimeError(f"fal returned an empty video for {src.name}")
    return dest