runway-ml

pip install requests python-dotenv
export RUNWAY_API_KEY="your_api_key_here"

import os
import time
import requests

API_KEY = os.environ["RUNWAY_API_KEY"]
BASE_URL = "https://api.dev.runwayml.com/v1"
HEADERS = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json",
    "X-Runway-Version": "2024-11-06"
}

def text_to_video(
    prompt_text: str,
    duration: int = 5,
    ratio: str = "1280:720",
    seed: int = None
) -> str:
    """Submit a text-to-video task and return the task_id."""
    payload = {
        "model": "gen3a_turbo",
        "promptText": prompt_text,
        "duration": duration,
        "ratio": ratio
    }
    if seed is not None:
        payload["seed"] = seed

    r = requests.post(f"{BASE_URL}/image_to_video", json=payload, headers=HEADERS)
    r.raise_for_status()
    return r.json()["id"]

task_id = text_to_video(
    prompt_text="A drone shot flying over a misty mountain valley at golden hour, cinematic, slow motion",
    duration=5,
    ratio="1280:720"
)
print(f"Task submitted: {task_id}")

import base64
from pathlib import Path

def image_to_video(
    image_path: str,
    prompt_text: str = "",
    duration: int = 5,
    ratio: str = "1280:720",
    seed: int = None
) -> str:
    """Animate an image into video. image_path can be a local file or URL."""

    if image_path.startswith("http"):
        prompt_image = image_path
    else:
        # Encode local file as data URI
        img_bytes = Path(image_path).read_bytes()
        ext = Path(image_path).suffix.lstrip(".").lower()
        mime = {"jpg": "image/jpeg", "jpeg": "image/jpeg", "png": "image/png", "webp": "image/webp"}.get(ext, "image/png")
        b64 = base64.b64encode(img_bytes).decode()
        prompt_image = f"data:{mime};base64,{b64}"

    payload = {
        "model": "gen3a_turbo",
        "promptImage": prompt_image,
        "promptText": prompt_text,
        "duration": duration,
        "ratio": ratio
    }
    if seed is not None:
        payload["seed"] = seed

    r = requests.post(f"{BASE_URL}/image_to_video", json=payload, headers=HEADERS)
    r.raise_for_status()
    return r.json()["id"]

task_id = image_to_video(
    image_path="product_shot.png",
    prompt_text="The product slowly rotates, sparkling particles float around it, luxury feel",
    duration=5,
    ratio="1280:720"
)
print(f"Task submitted: {task_id}")

def get_task(task_id: str) -> dict:
    r = requests.get(f"{BASE_URL}/tasks/{task_id}", headers=HEADERS)
    r.raise_for_status()
    return r.json()

def wait_for_task(task_id: str, poll_interval: int = 5, timeout: int = 600) -> list[str]:
    """Poll until task completes; return list of output video URLs."""
    start = time.time()
    while True:
        task = get_task(task_id)
        status = task["status"]
        progress = task.get("progress", 0)
        print(f"[{int(time.time()-start)}s] Status: {status} ({int(progress*100)}%)")

        if status == "SUCCEEDED":
            return task["output"]  # list of video URLs
        elif status in ("FAILED", "CANCELLED"):
            raise RuntimeError(f"Task {status}: {task.get('failure', '')}")
        elif time.time() - start > timeout:
            raise TimeoutError(f"Task not done after {timeout}s")

        time.sleep(poll_interval)

output_urls = wait_for_task(task_id)
print(f"Video(s) ready: {output_urls}")

def download_video(url: str, output_path: str = "output.mp4") -> str:
    r = requests.get(url, stream=True)
    r.raise_for_status()
    with open(output_path, "wb") as f:
        for chunk in r.iter_content(chunk_size=8192):
            f.write(chunk)
    size_mb = os.path.getsize(output_path) / 1024 / 1024
    print(f"Saved: {output_path} ({size_mb:.1f} MB)")
    return output_path

download_video(output_urls[0], "mountain_valley.mp4")

def generate_and_download(prompt: str, output_path: str = "output.mp4", **kwargs) -> str:
    """One-shot: generate video from text and download it."""
    print(f"Generating: {prompt[:80]}...")
    task_id = text_to_video(prompt, **kwargs)
    urls = wait_for_task(task_id)
    return download_video(urls[0], output_path)

# Generate a product ad clip
generate_and_download(
    prompt="Close-up of a sleek smartphone on a white desk, screen lights up, smooth camera pull-back",
    output_path="product_ad.mp4",
    duration=5,
    ratio="1280:720"
)