Turn a photo into a video

import { randomUUID } from "node:crypto";
import { readFile, writeFile } from "node:fs/promises";

const API = "https://api.tryleap.ai";
const KEY = process.env.LEAP_API_KEY;
if (!KEY) throw new Error("Set LEAP_API_KEY first.");

// Sends a request with your key, and throws the API's message on an error.
async function leap(path, init = {}) {
  const response = await fetch(`${API}${path}`, {
    ...init,
    headers: { "x-api-key": KEY, ...init.headers },
  });
  const body = await response.json();
  if (!response.ok) throw new Error(`${body.error.code}: ${body.error.message}`);
  return body;
}

// 1. Upload the photo. The body is the image itself.
const file = await leap("/v1/files?filename=product.jpg", {
  method: "POST",
  headers: { "content-type": "image/jpeg" },
  body: await readFile("product.jpg"),
});
console.log(`Uploaded ${file.id}, ${file.width}x${file.height}`);

// 2. Start the clip from the photo, with a new idempotency key.
let generation = await leap("/v1/generations", {
  method: "POST",
  headers: { "content-type": "application/json", "idempotency-key": randomUUID() },
  body: JSON.stringify({
    model: "google/veo-3.1-fast",
    input: {
      prompt: "A slow push in on the product as soft morning light moves across it",
      first_frame: file.id,
      duration: 4,
      aspect_ratio: "9:16",
    },
  }),
});
console.log(`Started ${generation.id}`);

// 3. Long-poll until it's done, for up to 15 minutes: each read waits up to
// 60 seconds for the clip to finish.
const deadline = Date.now() + 15 * 60_000;
while (!["succeeded", "failed", "canceled"].includes(generation.status)) {
  if (Date.now() > deadline) {
    throw new Error(`${generation.id} is still ${generation.status}. Read it again later.`);
  }
  await new Promise((resolve) => setTimeout(resolve, 2_000));
  generation = await leap(`/v1/generations/${generation.id}`, {
    headers: { prefer: "wait=60" },
    signal: AbortSignal.timeout(75_000),
  });
}

if (generation.status !== "succeeded") {
  throw new Error(`${generation.status}: ${generation.error?.message ?? "no message"}`);
}

// 4. Download the clip. The signed link needs no API key.
const video = await fetch(generation.output[0].url);
if (!video.ok) throw new Error(`Download failed: ${video.status}`);
await writeFile("product.mp4", Buffer.from(await video.arrayBuffer()));
console.log(`Saved product.mp4 for $${generation.usage.cost_usd}`);

import os
import time
import uuid

import requests

API = "https://api.tryleap.ai"
HEADERS = {"x-api-key": os.environ["LEAP_API_KEY"]}


def check(response):
    """Returns the answer's JSON, or raises the API's message on an error."""
    body = response.json()
    if not response.ok:
        raise RuntimeError(f"{body['error']['code']}: {body['error']['message']}")
    return body


# 1. Upload the photo. The body is the image itself.
with open("product.jpg", "rb") as f:
    file = check(
        requests.post(
            f"{API}/v1/files",
            params={"filename": "product.jpg"},
            headers={**HEADERS, "content-type": "image/jpeg"},
            data=f.read(),
            timeout=60,
        )
    )
print(f"Uploaded {file['id']}, {file['width']}x{file['height']}")

# 2. Start the clip from the photo, with a new idempotency key.
generation = check(
    requests.post(
        f"{API}/v1/generations",
        headers={**HEADERS, "idempotency-key": str(uuid.uuid4())},
        json={
            "model": "google/veo-3.1-fast",
            "input": {
                "prompt": "A slow push in on the product as soft morning light moves across it",
                "first_frame": file["id"],
                "duration": 4,
                "aspect_ratio": "9:16",
            },
        },
        timeout=60,
    )
)
print(f"Started {generation['id']}")

# 3. Long-poll until it's done, for up to 15 minutes: each read waits up to
# 60 seconds for the clip to finish.
deadline = time.monotonic() + 15 * 60
while generation["status"] not in {"succeeded", "failed", "canceled"}:
    if time.monotonic() > deadline:
        raise TimeoutError(f"{generation['id']} is still {generation['status']}. Read it again later.")
    time.sleep(2)
    generation = check(
        requests.get(
            f"{API}/v1/generations/{generation['id']}",
            headers={**HEADERS, "prefer": "wait=60"},
            timeout=75,
        )
    )

if generation["status"] != "succeeded":
    message = (generation["error"] or {}).get("message", "no message")
    raise RuntimeError(f"{generation['status']}: {message}")

# 4. Download the clip. The signed link needs no API key.
video = requests.get(generation["output"][0]["url"], timeout=300)
video.raise_for_status()
with open("product.mp4", "wb") as f:
    f.write(video.content)
print(f"Saved product.mp4 for ${generation['usage']['cost_usd']}")