""" Leonardo AI Video Provider -- Image-to-Video via Leonardo Motion 2.0 API. Uses the official Leonardo AI REST API for video generation. Supports Motion Control elements for camera movements. Env vars: LEONARDO_API_KEY - API key from Leonardo AI dashboard Docs: https://docs.leonardo.ai/reference/createimagetovideogeneration """ from __future__ import annotations import asyncio import json import logging import os import time from collections.abc import Callable from pathlib import Path from typing import Any, Optional import httpx from providers.ports import VideoProvider logger = logging.getLogger(__name__) LEONARDO_API_BASE = "https://cloud.leonardo.ai/api/rest/v1" # Motion Control element UUIDs MOTION_ELEMENTS = { "bullet_time": "fbed015e-594e-4f78-b4be-3b07142aaa1e", "crane_down": "5a1d2a6a-7709-4097-9158-1b7ae6c9e647", "crane_over_head": "1054d533-168c-4821-bd3d-a56182afa4f3", "crane_up": "c765bd57-cdc5-4317-a600-69a8bd6c4ce6", "crash_zoom_in": "b0191ad1-a723-439c-a4bc-a3f5d5884db3", "crash_zoom_out": "1975ac74-92ca-46b3-81b3-6f191a9ae438", "disintegration": "a51e2e8d-ba5e-44f2-9e00-3d86fd93c9bc", "dolly_in": "ece8c6a9-3deb-430e-8c93-4d5061b6adbf", "dolly_left": "f507880a-3fa8-4c3a-96bb-3ce3b70ac53b", "dolly_out": "772cb36a-7d18-4250-b4aa-0c3f1a8431a0", "dolly_right": "587a0109-30be-4781-a18e-e353b580fd10", "explosion": "65da803d-c015-495a-8d5c-e969a79c9894", "eyes_in": "148b50d0-2040-4524-a36f-6e330f9e362e", "flood": "a12c150e-95e9-469b-ba9b-6d5323ac5a09", "handheld": "75722d13-108f-4cea-9471-cb7e5fc049fe", "lens_crack": "193da194-2632-4f6a-a1df-d03ca9ae0ea9", "medium_zoom_in": "f46d8e7f-e0ca-4f6a-90ab-141d731f47ae", "orbit_left": "74bea0cc-9942-4d45-9977-28c25078bfd4", "orbit_right": "aec24e36-a2e8-4fae-920c-127d276bbe4b", "robo_arm": "8df55fe2-5c6f-4dbf-8ade-eb997807ca0d", "super_dolly_in": "a3992d78-34fc-44c6-b157-e2755d905197", "super_dolly_out": "906b93f2-beb3-42be-9283-92236cc90ed6", "tilt_down": "a1923b1b-854a-46a1-9e26-07c435098b87", "tilt_up": "6ad6de1f-bd15-4d0b-ae0e-81d1a4c6c085", } # Video generation timeout VIDEO_GEN_TIMEOUT = 300 VIDEO_POLL_INTERVAL = 5 class LeonardoVideoProvider(VideoProvider): """Generates video clips from images via Leonardo AI Motion 2.0 API.""" name = "leonardo" def __init__(self): self.api_key = os.getenv("LEONARDO_API_KEY", "") if not self.api_key: raise EnvironmentError( "LEONARDO_API_KEY is required. " "Get it from https://app.leonardo.ai/settings → API" ) self.client = httpx.Client( timeout=120, headers={ "accept": "application/json", "content-type": "application/json", "authorization": f"Bearer {self.api_key}", }, ) def close(self): """Close HTTP client.""" if self.client: self.client.close() def _upload_image(self, image_path: Path) -> str: """Upload an image to Leonardo AI and return the image ID. Returns: The uploaded image ID for use in video generation. """ # Step 1: Initialize upload init_resp = self.client.post( f"{LEONARDO_API_BASE}/init-image", json={"extension": image_path.suffix.lstrip(".")}, ) init_resp.raise_for_status() init_data = init_resp.json() upload_info = init_data["uploadInitImage"] upload_url = upload_info["url"] fields_str = upload_info["fields"] image_id = upload_info["id"] # Parse fields - it's a JSON string if isinstance(fields_str, str): fields = json.loads(fields_str) else: fields = fields_str # Step 2: Upload to S3 with open(image_path, "rb") as f: files = {"file": (image_path.name, f, "image/png")} # Use a separate client without auth headers for S3 with httpx.Client(timeout=60) as s3_client: upload_resp = s3_client.post(upload_url, data=fields, files=files) upload_resp.raise_for_status() logger.info("Uploaded image to Leonardo AI: %s", image_id) return image_id def _get_motion_element(self, video_prompt: str) -> Optional[dict]: """Parse video prompt to extract motion control element. Looks for keywords in the prompt to determine camera movement. Returns element dict with akUUID and weight, or None. """ prompt_lower = video_prompt.lower() # Map keywords to motion elements keyword_map = { "dolly in": "dolly_in", "push in": "dolly_in", "move closer": "dolly_in", "dolly out": "dolly_out", "pull out": "dolly_out", "move away": "dolly_out", "dolly left": "dolly_left", "pan left": "dolly_left", "dolly right": "dolly_right", "pan right": "dolly_right", "zoom in": "medium_zoom_in", "crash zoom": "crash_zoom_in", "zoom out": "crash_zoom_out", "orbit left": "orbit_left", "orbit right": "orbit_right", "tilt up": "tilt_up", "look up": "tilt_up", "tilt down": "tilt_down", "look down": "tilt_down", "crane up": "crane_up", "crane down": "crane_down", "handheld": "handheld", "shaky": "handheld", "explosion": "explosion", "disintegrat": "disintegration", } for keyword, motion_key in keyword_map.items(): if keyword in prompt_lower: return { "akUUID": MOTION_ELEMENTS[motion_key], "weight": 1.0, } # Default: subtle dolly in for engagement return { "akUUID": MOTION_ELEMENTS["dolly_in"], "weight": 0.5, } def _poll_for_video(self, generation_id: str) -> str: """Poll Leonardo API until video is ready. Returns: URL of the generated video """ start_time = time.time() while time.time() - start_time < VIDEO_GEN_TIMEOUT: resp = self.client.get( f"{LEONARDO_API_BASE}/generations/{generation_id}", ) resp.raise_for_status() data = resp.json() gen_data = data.get("generations_by_pk", {}) status = gen_data.get("status", "") if status == "COMPLETE": # Get video URL from generated images images = gen_data.get("generated_images", []) for img in images: video_url = img.get("motionMP4URL") if video_url: return video_url raise RuntimeError("Generation complete but no video URL found") elif status == "FAILED": raise RuntimeError(f"Leonardo video generation failed: {gen_data}") logger.debug("Leonardo generation status: %s", status) time.sleep(VIDEO_POLL_INTERVAL) raise TimeoutError( f"Leonardo video generation timed out after {VIDEO_GEN_TIMEOUT}s" ) def _download_video(self, video_url: str, output_path: Path): """Download video from URL to local path.""" with httpx.Client(timeout=120) as dl_client: resp = dl_client.get(video_url) resp.raise_for_status() output_path.write_bytes(resp.content) async def generate( self, image_path: Path, video_prompt: str, output_path: Path, on_progress: Callable[[int], None] | None = None, **kwargs: Any, ) -> Path: """Async wrapper for image-to-video generation. Extra kwargs (resolution, frame_interpolation, prompt_enhance) are passed to _generate_sync. """ return await asyncio.to_thread( self._generate_sync, image_path, video_prompt, output_path, kwargs.get('resolution', '720'), kwargs.get('frame_interpolation', True), kwargs.get('prompt_enhance', True), ) def _generate_sync( self, image_path: Path, video_prompt: str, output_path: Path, resolution: str = "720", frame_interpolation: bool = True, prompt_enhance: bool = True, ) -> Path: """Sync implementation of generate.""" output_path = Path(output_path) output_path.parent.mkdir(parents=True, exist_ok=True) # Upload image first image_id = self._upload_image(Path(image_path)) # Get motion element from prompt motion_element = self._get_motion_element(video_prompt) # Build request payload for Motion 2.0 endpoint (generations-image-to-video) payload = { "imageType": "UPLOADED", # UPLOADED for init images, GENERATED for Leonardo-generated images "imageId": image_id, "isPublic": False, "resolution": f"RESOLUTION_{resolution}", "prompt": video_prompt or "subtle natural movement", "frameInterpolation": frame_interpolation, # Smooth Video "promptEnhance": prompt_enhance, } # Add motion control element if detected from prompt if motion_element: payload["elements"] = [motion_element] logger.info( "Generating video via Leonardo AI (Motion 2.0): prompt=%s, resolution=%s, element=%s", video_prompt[:50] if video_prompt else "(no prompt)", resolution, motion_element.get("akUUID", "none")[:8] if motion_element else "none", ) # Start generation using Motion 2.0 endpoint gen_resp = self.client.post( f"{LEONARDO_API_BASE}/generations-image-to-video", json=payload, ) gen_resp.raise_for_status() gen_data = gen_resp.json() # Response key can be "imageToVideoGeneration" or "motionVideoGenerationJob" generation_id = ( gen_data.get("imageToVideoGeneration", {}).get("generationId") or gen_data.get("motionVideoGenerationJob", {}).get("generationId") ) if not generation_id: raise RuntimeError(f"No generation ID returned: {gen_data}") logger.info("Leonardo video generation started: %s", generation_id) # Poll for completion video_url = self._poll_for_video(generation_id) # Download video self._download_video(video_url, output_path) logger.info("Leonardo video saved: %s", output_path) return output_path