"""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