⭐ Must Read: image to video ai ad pipeline steps

October 25, 2026

36 min read

Image to video AI stopped being a demo toy this fall. Kling AI launched Kling 4.0 globally in October 2026, and the Runway Dev platform now sells Kling, Veo, Seedance, Wan and Gemini Omni clips through one credit system where 1 credit equals $0.01. For anyone who turns product photos into short ads, the hard part is no longer getting a clip. It is picking the right model for each shot without burning the budget.

No single model wins every shot. A six-second hero pour with sound is a different job from four seconds of desk b-roll, and the price gap between them runs from $0.09 to $0.157 per second on one platform alone. Lineups also change in weeks, so a script wired to one vendor ages fast. This tutorial builds adforge, a Python tool that takes product stills and a shot list, routes each shot to the cheapest model that fits, rejects bad clips, and finishes the winners with FFmpeg into 16:9, 9:16 and 1:1 ads.

You do not need an API key to follow along. The project ships with offline mock providers that fake camera motion from your stills using FFmpeg, so every command below runs for $0. When you are ready, a config-driven HTTP adapter swaps in a real provider. If you only want one vendor wired up first, our Kling API setup guide covers that piece. This article covers the layer above it: routing, budgets, quality control and finishing.

Google · Preferred Sources

Don't miss new tech stories on Google

Add Tech Insider once in the Google app and our stories appear in your news suggestions.

Add Now

What You Will Build in 75 Minutes

adforge is a small pipeline for image to video AI work. It reads two YAML files and one folder of stills. The shot list says which still to animate, how long the clip runs, what the camera does and what the shot is for. The tool then walks every shot through six stages, and each stage lives in a small module you can swap out later.

Spec: load and validate the shots, then hash each one so reruns skip finished work.
Router: turn a shot’s intent (hero, variation, broll or long) into an ordered list of providers that can handle it.
Budget: check a spend ledger before every paid call.
Provider: call a model, either a mock or an HTTP task API.
QC gate: probe the clip, then check duration, size, frozen frames, black frames and drift from your product photo.
Finish: normalize every clip, crossfade, add an end card, set loudness, export three aspect ratios and cut a contact sheet.

The deliverables are a 31-second master, three social-ready exports, a manifest, a spend ledger and a contact sheet for quick review. Plan on about 75 minutes the first time, most of it typing and reading. On my machine the full run took about 90 seconds, nearly all of it FFmpeg encoding.

A word on scope. This image to video AI tool does not write your ad concept, pick music or run AI upscaling, and it does not replace a human eye on hero shots. If you are still choosing which model deserves your default slot, start with our ranking of the best AI video generator options, then come back and build the harness around your pick.

File	Role	Lines
adforge/spec.py	Shot dataclass, YAML loading, cache key	38
adforge/ledger.py	Thread-safe spend ledger with a hard budget	32
adforge/providers.py	Mock provider and config-driven HTTP task provider	121
adforge/router.py	Intent to ordered provider chain	5
adforge/qc.py	ffprobe checks, freeze, black, SSIM and color drift	87
adforge/finish.py	Normalize, crossfade, end card, loudness, exports	78
adforge/pipeline.py	Orchestrator and command line	118

Image to Video AI Models Worth Routing To in October 2026

The lineup moved fast. Seedance 2.5 became generally available on Runway Dev on August 7, 2026, after ByteDance announced it on July 31. Google released Gemini Omni Flash 1.1 on August 27. Kling 4.0 followed on October 15. Each launch shifts the price-to-quality math for image-to-video work, which is why the router in this tutorial keeps prices in a config file instead of in code.

The table lists the models the sample config routes to. Prices come from the Runway Dev and MiniMax pricing pages as of October 25, 2026. Check each page before you spend, because promos and tiers change.

Model	Sold by	Published price	Notes
Kling 4.0	Kling AI via Runway Dev	$0.157/s list, $0.1256/s promo	Up to 30 s native, 15 reference assets, 10 keyframes
Kling 3.0 Standard	Runway Dev	$0.09/s silent, $0.13/s with audio	Previous generation, still sold
Kling 3.0 Pro	Runway Dev	$0.12/s silent, $0.17/s with audio	Higher quality tier of 3.0
Veo 3.1 Fast	Google via Runway Dev	$0.15/s with audio	Lower price available without audio
Gemini Omni Flash 1.1	Google via Runway Dev	$0.034/s at 360p, $0.10/s at 720p	Released August 27, 2026, native audio
Seedance 2.5	ByteDance via Runway Dev	$0.20/s at 480p, $0.30/s at 720p, $0.68/s at 1080p	$0.80 minimum per generation
Hailuo 2.3 Fast	MiniMax API	$0.19 per 768P 6-second clip	Cheaper batch option
Hailuo 2.3	MiniMax API	$0.28 per 768P 6-second clip	$0.49 for a 1080P 6-second clip

Kling 4.0 is the one to watch for ads. Kling says it generates up to 30 seconds natively and accepts up to 15 multimodal reference assets and up to 10 keyframes. Its API is opening in stages. A partner press release dated October 23 claims the API launched on October 22, but Kling’s own materials describe a gradual rollout to enterprise customers and developers, so confirm access in your own account. Runway Dev shows a promo of $0.1256 per second against the $0.157 list price, 20% off through November 30, 2026. The sample config uses the list price.

Google’s Veo 3.1 and Veo 3.1 Fast are documented on Runway Dev. Direct from Google, developers have reported friction. One weekly roundup says Google shut down the Gemini API preview models for Veo 3.1, Veo 3.1 Fast and Veo 3.1 Lite on October 22 and named Gemini Omni Flash as the replacement. Developers also report quota errors and flaky audio on image-to-video calls. That is a strong argument for fallbacks, and our Veo 3.1 API walkthrough covers the direct route.

Gemini Omni Flash 1.1 is the budget wildcard. Runway Dev lists it at $0.034 per second at 360p, $0.10 at 720p, $0.15 at 1080p and $0.30 at 4K. Runway’s own Gen-4 Turbo handles image-to-video on the same platform, so one account covers several vendors.

Hailuo runs on MiniMax’s own API. The MiniMax model release notes cover the 2.3 family, and the Hailuo site now headlines MiniMax H3. Reports date H3 to late July 2026, with 2K output, clips up to 15 seconds and native stereo audio. Treat those H3 specs as reported until you see them in the docs. Seedance 2.5 costs more per second, but it takes multi-reference input for clips up to 30 seconds, which suits long orbit shots.

Prerequisites and Versions

Everything runs on Linux or macOS. On Windows, use WSL. Nothing here needs a specific version, and a recent build is fine.

Requirement	Version	Why you need it
Python	3.11 or newer (tested on 3.12.3)	Runs the pipeline and uses dataclasses and thread pools
FFmpeg and ffprobe	A current build recommended, such as the 9.0 line	Mock clips, QC probes, normalizing, crossfades, exports
requests	Latest from PyPI	HTTP adapter for real providers
PyYAML	Latest from PyPI	Config and shot list parsing
Pillow	Latest from PyPI	Only for the sample still generator
Provider API key	Optional	Only needed for live runs in Step 13
Disk space	1 GB free	Raw clips, normalized clips and exports

Check your tools first. The filter grep should print one line for each of the nine filters the project relies on.

python3 --version
ffmpeg -version | head -1
ffprobe -version | head -1
ffmpeg -hide_banner -filters | grep -E " (xfade|acrossfade|loudnorm|freezedetect|blackdetect|ssim|tile|drawtext|zoompan) "

Expected output on my machine:

Python 3.12.3
ffmpeg version 6.1.1-3ubuntu5 Copyright (c) 2000-2023 the FFmpeg developers
ffprobe version 6.1.1-3ubuntu5 Copyright (c) 2007-2023 the FFmpeg developers
 ... acrossfade        AA->A      Cross fade two input audio streams.
 ... loudnorm          A->A       EBU R128 loudness normalization
 .S. blackdetect       V->V       Detect video intervals that are (almost) black.
 T.C drawtext          V->V       Draw text on top of video frames using libfreetype library.
 ... freezedetect      V->V       Detects frozen video input.
 TS. ssim              VV->V      Calculate the SSIM between two video streams.
 ...                              (xfade, tile and zoompan follow)

Steps 1-3: Scaffold, Stills and the Shot Spec

Step 1: Create the project and a virtual environment

Make a folder, a virtual environment and the package skeleton. Keep the package name adforge, because the later commands call it with python -m adforge.pipeline.

mkdir -p adforge/adforge adforge/tools adforge/stills && cd adforge
python3 -m venv .venv && source .venv/bin/activate
pip install requests pyyaml pillow
touch adforge/__init__.py
printf "requests\npyyaml\n" > requirements.txt

Step 2: Prepare product stills

Image to video AI models treat your still as frame one, so the quality of the still caps the quality of the clip. Use high-resolution landscape stills, one product per frame, a plain or softly graded background and no motion blur. Keep the product inside the middle third of the frame. The 9:16 export in Step 10 keeps only a narrow slice of a 16:9 frame’s width, and anything outside the center gets cropped away.

If your photos need cleanup first, generate or repair them upstream. Runway Dev lists Nano Banana 2.1 at $0.04 per 1K image, $0.06 at 2K and $0.12 at 4K, and our multi-model AI image pipeline tutorial covers that stage in depth. For this walkthrough, a small script draws three stand-in product shots so you can test without your own photos.

from pathlib import Path

from PIL import Image, ImageDraw, ImageFont

OUT = Path("stills")
OUT.mkdir(exist_ok=True)
FONT = "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf"

SPECS = {
    "can_front": ((235, 120, 60), "EMBER"),
    "can_pour": ((60, 120, 200), "COLD BREW"),
    "can_desk": ((90, 160, 110), "NO SUGAR"),
}

for name, (color, label) in SPECS.items():
    img = Image.new("RGB", (1600, 900))
    d = ImageDraw.Draw(img)
    for y in range(900):
        shade = int(30 + 60 * y / 900)
        d.line([(0, y), (1600, y)], fill=(shade, shade, shade + 8))
    d.rounded_rectangle((650, 150, 950, 760), 40, fill=color)
    d.rectangle((650, 330, 950, 520), fill=(250, 250, 245))
    f = ImageFont.truetype(FONT, 46)
    w = d.textlength(label, font=f)
    d.text((800 - w / 2, 400), label, fill=(20, 20, 20), font=f)
    img.save(OUT / f"{name}.png")
    print("wrote", OUT / f"{name}.png")

Run it once. It writes three PNGs into stills/.

python3 tools/make_stills.py

Step 3: Define the shot spec

A shot is a small dataclass. The important part is key(). It hashes the still’s bytes plus the prompt, duration and audio flag, so editing a prompt or swapping a photo invalidates the cache for that shot only. The loader also fails fast on duplicate ids and missing stills, which saves you from paying for a run that was doomed at line one.

import hashlib
from dataclasses import dataclass
from pathlib import Path

import yaml


@dataclass
class Shot:
    id: str
    still: str
    prompt: str
    seconds: int = 5
    intent: str = "variation"
    audio: bool = False

    def key(self) -> str:
        h = hashlib.sha256()
        h.update(Path(self.still).read_bytes())
        h.update(f"{self.prompt}|{self.seconds}|{self.audio}".encode())
        return h.hexdigest()[:16]


def load_yaml(path):
    with open(path, encoding="utf-8") as f:
        return yaml.safe_load(f)


def load_shots(path):
    raw = load_yaml(path)
    shots = [Shot(**s) for s in raw["shots"]]
    ids = [s.id for s in shots]
    if len(ids) != len(set(ids)):
        raise ValueError("duplicate shot ids in shots file")
    for s in shots:
        if not Path(s.still).exists():
            raise FileNotFoundError(f"{s.id}: still not found: {s.still}")
    return shots, raw.get("campaign", {})

Steps 4-5: Spend Ledger and Provider Adapters

Step 4: Build the spend ledger

The ledger answers one question before every paid call: can we afford this? It keeps a list of charges in JSON, guards the list with a lock because shots run in parallel, and writes through a temp file so a crash never leaves half a ledger on disk. If you want per-call cost tracking at API level, our Gemini cost ledger tutorial goes deeper.

import json
import os
import threading
from pathlib import Path


class BudgetExceeded(Exception):
    pass


class Ledger:
    def __init__(self, path, budget_usd):
        self.path = Path(path)
        self.budget = float(budget_usd)
        self.lock = threading.Lock()
        self.rows = json.loads(self.path.read_text()) if self.path.exists() else []

    @property
    def spent(self):
        return round(sum(r["usd"] for r in self.rows), 4)

    def can_afford(self, usd):
        with self.lock:
            return self.spent + usd <= self.budget

    def charge(self, shot_id, provider, usd, seconds):
        with self.lock:
            self.rows.append({"shot": shot_id, "provider": provider,
                              "usd": round(usd, 4), "seconds": seconds})
            tmp = self.path.with_suffix(".tmp")
            tmp.write_text(json.dumps(self.rows, indent=2))
            os.replace(tmp, self.path)

Step 5: Write the provider adapters

This is the longest file, and it has four jobs. Provider holds the shared rules: an estimate() from the published per-second rate (with an optional audio rate and a minimum charge like Seedance’s $0.80), and an accepts() check for maximum seconds and audio support. MockProvider fakes a model by applying a slow zoom to your still with FFmpeg. Each mock can declare a different size, frame rate and motion direction, which mimics the mess real providers hand you.

HttpTaskProvider speaks the submit-then-poll pattern most video APIs use. It templates the request body from config, retries 429 and 5xx responses with exponential backoff, polls until a status in your done list appears, and streams the finished file to disk. The _dig helper reads values from nested JSON by dotted path, so output.0 means the first item of the output list.

import base64
import mimetypes
import subprocess
import time
from pathlib import Path

import requests


class ProviderError(Exception):
    pass


def _dig(obj, path):
    for part in path.split("."):
        obj = obj[int(part)] if part.isdigit() else obj[part]
    return obj


def _subst(node, ctx):
    if isinstance(node, dict):
        return {k: _subst(v, ctx) for k, v in node.items()}
    if isinstance(node, list):
        return [_subst(v, ctx) for v in node]
    if isinstance(node, str):
        if node.startswith("{") and node.endswith("}") and node[1:-1] in ctx:
            return ctx[node[1:-1]]
        return node.format_map(ctx) if "{" in node else node
    return node


class Provider:
    def __init__(self, name, cfg):
        self.name, self.cfg = name, cfg
        self.max_seconds = cfg.get("max_seconds", 10)
        self.has_audio = cfg.get("audio", False)

    def estimate(self, shot):
        rate = self.cfg["usd_per_second"]
        if shot.audio and "usd_per_second_audio" in self.cfg:
            rate = self.cfg["usd_per_second_audio"]
        return max(rate * shot.seconds, self.cfg.get("min_usd", 0.0))

    def accepts(self, shot):
        return shot.seconds <= self.max_seconds and (self.has_audio or not shot.audio)


class MockProvider(Provider):
    def generate(self, shot, still, out):
        w, h = self.cfg.get("size", [1920, 1080])
        fps = self.cfg.get("fps", 30)
        frames = shot.seconds * fps
        zoom = "min(zoom+0.0012,1.2)" if self.cfg.get("motion", "in") == "in" else "max(1.2-on*0.0012,1.0)"
        vf = (f"scale=2400:-1,zoompan=z='{zoom}':x='iw/2-(iw/zoom/2)':y='ih/2-(ih/zoom/2)'"
              f":d={frames}:s={w}x{h}:fps={fps},format=yuv420p")
        cmd = ["ffmpeg", "-y", "-loglevel", "error", "-i", str(still)]
        if shot.audio:
            cmd += ["-f", "lavfi", "-i", "sine=frequency=220:sample_rate=48000", "-shortest"]
        cmd += ["-vf", vf, "-t", str(shot.seconds), "-c:v", "libx264", "-crf", "20"]
        cmd += ["-c:a", "aac"] if shot.audio else ["-an"]
        cmd.append(str(out))
        r = subprocess.run(cmd, capture_output=True, text=True)
        if r.returncode:
            raise ProviderError(r.stderr[-400:])


class HttpTaskProvider(Provider):
    def __init__(self, name, cfg, env):
        super().__init__(name, cfg)
        self.base = env[cfg["base_url_env"]].rstrip("/")
        self.key = env[cfg["key_env"]]

    def _headers(self):
        h = {"Authorization": f"Bearer {self.key}"}
        h.update(self.cfg.get("headers", {}))
        return h

    def _request(self, method, url, **kw):
        for attempt in range(4):
            r = requests.request(method, url, headers=self._headers(), timeout=60, **kw)
            if r.status_code in (429, 500, 502, 503):
                time.sleep(2 ** attempt)
                continue
            if r.status_code >= 400:
                raise ProviderError(f"{r.status_code} {r.text[:300]}")
            return r
        raise ProviderError("retries exhausted")

    def generate(self, shot, still, out):
        mime = mimetypes.guess_type(str(still))[0] or "image/png"
        b64 = base64.b64encode(Path(still).read_bytes()).decode()
        ctx = {"prompt": shot.prompt, "seconds": shot.seconds, "audio": shot.audio,
               "image_data_uri": f"data:{mime};base64,{b64}", "model": self.cfg["model"]}
        sub = self.cfg["submit"]
        r = self._request("POST", self.base + sub["path"], json=_subst(sub["body"], ctx))
        f, poll = self.cfg["fields"], self.cfg["poll"]
        task_id = _dig(r.json(), f["task_id"])
        deadline = time.time() + poll.get("timeout", 600)
        while time.time() < deadline:
            time.sleep(poll.get("interval", 5))
            body = self._request("GET", self.base + poll["path"].format(task_id=task_id)).json()
            status = _dig(body, f["status"])
            if status in f["done"]:
                url = _dig(body, f["output"])
                break
            if status in f["failed"]:
                raise ProviderError(f"task {task_id} {status}: {body}")
        else:
            raise ProviderError(f"task {task_id} timed out")
        with requests.get(url, stream=True, timeout=120) as dl:
            dl.raise_for_status()
            with open(out, "wb") as fh:
                for chunk in dl.iter_content(1 << 20):
                    fh.write(chunk)


def build_providers(cfg, env):
    out = {}
    for name, p in cfg["providers"].items():
        out[name] = MockProvider(name, p) if p["kind"] == "mock" else HttpTaskProvider(name, p, env)
    return out

Steps 6-7: Router and Configuration

Step 6: Write the router

The router is five lines on purpose. Given a shot, it looks up the provider chain for the shot’s intent, drops providers that cannot handle the duration or audio request, and returns the rest with their cost estimates, in order. Policy lives in config, not here.

def candidates(shot, cfg, providers):
    """Ordered (provider, estimated_usd) pairs for one shot."""
    names = cfg["routes"].get(shot.intent) or cfg["routes"]["default"]
    return [(providers[n], providers[n].estimate(shot))
            for n in names if n in providers and providers[n].accepts(shot)]

Step 7: Write the configuration

The config has five blocks: budget, providers, routes, QC rules and finishing. The provider rates are the published Runway Dev numbers from the table above, so the cost estimates mean something. The max_seconds values are conservative placeholders, so confirm each one on the model’s own docs page before a live run. In the sample file every provider is a mock, which keeps the whole tutorial free.

budget_usd: 6.00
output_dir: out

providers:
  kling_std:    {kind: mock, usd_per_second: 0.09, usd_per_second_audio: 0.13, max_seconds: 15, audio: true,  size: [1280, 720],  fps: 24, motion: in}
  kling_pro:    {kind: mock, usd_per_second: 0.12, usd_per_second_audio: 0.17, max_seconds: 15, audio: true,  size: [1280, 720],  fps: 24, motion: out}
  kling_4:      {kind: mock, usd_per_second: 0.157, max_seconds: 30, audio: true, size: [1920, 1080], fps: 30, motion: in}
  veo_fast:     {kind: mock, usd_per_second: 0.15, max_seconds: 8,  audio: true,  size: [1920, 1080], fps: 24, motion: in}
  omni_720:     {kind: mock, usd_per_second: 0.10, max_seconds: 10, audio: true,  size: [1280, 720],  fps: 30, motion: out}
  seedance_720: {kind: mock, usd_per_second: 0.30, min_usd: 0.80, max_seconds: 30, audio: false, size: [1280, 720], fps: 24, motion: in}

routes:
  hero:      [veo_fast, kling_pro, omni_720]
  variation: [kling_std, omni_720, seedance_720]
  broll:     [omni_720, kling_std]
  long:      [kling_4, seedance_720]
  default:   [kling_std, omni_720]

qc:
  duration_tolerance_s: 0.3
  min_height: 700
  max_frozen_s: 0.5
  max_black_s: 0.3
  min_ssim_first: 0.55
  max_color_delta: 25

finish:
  crossfade_s: 0.5
  end_card_seconds: 2.5
  lufs: -14
  font: /usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf
  grade:
    default: "eq=contrast=1.04:saturation=1.05"
    kling_pro: "eq=contrast=1.02:saturation=0.97"
    omni_720: "eq=contrast=1.06:saturation=1.02:gamma=0.98"
  formats:
    landscape: {size: [1920, 1080]}
    vertical:  {size: [1080, 1920], focus_x: 0.5}
    square:    {size: [1080, 1080], focus_x: 0.5}

The routes encode a simple philosophy: spend where the viewer looks, save everywhere else.

Intent	Chain	Why
hero	veo_fast, kling_pro, omni_720	Needs audio and the best look, so the pricier models go first
variation	kling_std, omni_720, seedance_720	Silent clips, cheapest first, expensive Seedance as a last resort
broll	omni_720, kling_std	Lowest rate wins because nobody studies b-roll
long	kling_4, seedance_720	Only providers allowed more than 15 seconds can take a 12-second orbit
default	kling_std, omni_720	Anything without an intent falls back to the cheap pair

Steps 8-9: Shot List, Dry Run and Quality Gate

Step 8: Write the shot list and run a dry plan

The shot list is the creative brief in YAML. Notice that two shots reuse the same still with different prompts, which is a cheap way to get variety from one photo session.

campaign:
  name: ember
  cta: "EMBER COLD BREW\nTRY IT TODAY"

shots:
  - {id: s01_hero,   still: stills/can_front.png, intent: hero,      seconds: 6, audio: true,  prompt: "Slow push-in on the can, condensation beads, soft rim light"}
  - {id: s02_pour,   still: stills/can_pour.png,  intent: variation, seconds: 5, prompt: "Cold brew pours over ice, camera holds steady"}
  - {id: s03_desk,   still: stills/can_desk.png,  intent: broll,     seconds: 4, prompt: "Desk scene, gentle handheld drift, morning light"}
  - {id: s04_detail, still: stills/can_front.png, intent: variation, seconds: 4, prompt: "Macro on the label, shallow depth of field"}
  - {id: s05_long,   still: stills/can_pour.png,  intent: long,      seconds: 12, prompt: "One continuous orbit around the can, label stays readable"}

Before spending anything, run the planner. It prints each shot’s chain with per-provider estimates and exits with code 1 if the first-choice total exceeds your budget, so you can also run it in CI to block an expensive pull request.

python3 -m adforge.pipeline plan

s01_hero      hero       6s  veo_fast($0.90) > kling_pro($1.02) > omni_720($0.60)
s02_pour      variation  5s  kling_std($0.45) > omni_720($0.50) > seedance_720($1.50)
s03_desk      broll      4s  omni_720($0.40) > kling_std($0.36)
s04_detail    variation  4s  kling_std($0.36) > omni_720($0.40) > seedance_720($1.20)
s05_long      long      12s  kling_4($1.88) > seedance_720($3.60)
first-choice estimate: $3.99  budget: $6.00

Step 9: Add the quality gate

A clip that looks fine in a thumbnail can still be unusable. The QC module runs five automated checks after every download. It compares the real duration against the request, rejects low resolution, measures frozen and black seconds with FFmpeg’s freezedetect and blackdetect filters, and compares the clip’s first frame to your source still. That last check uses two signals, SSIM for structure and a center-region color delta, because each misses what the other catches.

import json
import re
import subprocess


def _run(cmd):
    return subprocess.run(cmd, capture_output=True, text=True)


def probe(path):
    r = _run(["ffprobe", "-v", "error", "-print_format", "json", "-show_streams",
              "-show_format", str(path)])
    if r.returncode:
        raise ValueError(f"ffprobe failed: {r.stderr[-200:]}")
    info = json.loads(r.stdout)
    v = next((s for s in info["streams"] if s["codec_type"] == "video"), None)
    if v is None:
        raise ValueError("no video stream")
    num, den = v["r_frame_rate"].split("/")
    return {"width": v["width"], "height": v["height"], "fps": float(num) / float(den),
            "duration": float(info["format"]["duration"]),
            "audio": any(s["codec_type"] == "audio" for s in info["streams"])}


def frozen_seconds(path):
    r = _run(["ffmpeg", "-hide_banner", "-i", str(path), "-vf", "freezedetect=n=0.003:d=0.8",
              "-an", "-f", "null", "-"])
    return sum(float(x) for x in re.findall(r"freeze_duration: ([\d.]+)", r.stderr))


def black_seconds(path):
    r = _run(["ffmpeg", "-hide_banner", "-i", str(path), "-vf", "blackdetect=d=0.4:pix_th=0.10",
              "-an", "-f", "null", "-"])
    return sum(float(x) for x in re.findall(r"black_duration:([\d.]+)", r.stderr))


def first_frame_ssim(clip, still):
    fc = ("[0:v]trim=end_frame=1,scale=640:360:force_original_aspect_ratio=increase,crop=640:360[a];"
          "[1:v]scale=640:360:force_original_aspect_ratio=increase,crop=640:360[b];[a][b]ssim")
    r = _run(["ffmpeg", "-hide_banner", "-i", str(clip), "-i", str(still),
              "-filter_complex", fc, "-f", "null", "-"])
    m = re.search(r"All:([\d.]+)", r.stderr)
    return float(m.group(1)) if m else 0.0


def _center_rgb(path):
    vf = "crop=iw/5:ih/2,scale=1:1,format=rgb24"
    cmd = ["ffmpeg", "-v", "error", "-i", str(path), "-vf", vf, "-frames:v", "1",
           "-f", "rawvideo", "-"]
    out = subprocess.run(cmd, capture_output=True).stdout
    return list(out[:3]) if len(out) >= 3 else [0, 0, 0]


def center_color_delta(clip, still):
    a, b = _center_rgb(clip), _center_rgb(still)
    return max(abs(x - y) for x, y in zip(a, b))


def check(clip, still, shot, rules):
    """Return (ok, report). Rules come from config.yaml."""
    try:
        p = probe(clip)
    except ValueError as e:
        return False, {"error": str(e)}
    rep = {"duration": round(p["duration"], 2), "size": f'{p["width"]}x{p["height"]}',
           "fps": round(p["fps"], 2), "audio": p["audio"]}
    rep["frozen_s"] = round(frozen_seconds(clip), 2)
    rep["black_s"] = round(black_seconds(clip), 2)
    rep["ssim_first"] = round(first_frame_ssim(clip, still), 3)
    rep["color_delta"] = center_color_delta(clip, still)
    problems = []
    if abs(p["duration"] - shot.seconds) > rules["duration_tolerance_s"]:
        problems.append("duration")
    if p["height"] < rules["min_height"]:
        problems.append("resolution")
    if rep["frozen_s"] > rules["max_frozen_s"]:
        problems.append("frozen")
    if rep["black_s"] > rules["max_black_s"]:
        problems.append("black")
    if rep["ssim_first"] < rules["min_ssim_first"]:
        problems.append("product-drift")
    if rep["color_delta"] > rules["max_color_delta"]:
        problems.append("color-drift")
    if shot.audio and not p["audio"]:
        problems.append("missing-audio")
    rep["problems"] = problems
    return not problems, rep

The thresholds in the config are starting points from my tests, not industry standards, so tune them against your own accepted clips. When you want to rank models head to head instead of gating single clips, pair this module with the benchmark harness we built earlier. I fed the checker three clips to confirm it works:

ok          True   ssim_first 0.994  color_delta 2    problems []
wrong can   False  ssim_first 0.973  color_delta 65   problems ['color-drift']
black start False  ssim_first 0.597  color_delta 230  problems ['frozen', 'black', 'color-drift']

The second row matters most. I recolored the can from orange to green, and SSIM still scored 0.973. Only the color delta caught it.

Steps 10-11: FFmpeg Finishing and the Orchestrator

Step 10: Write the finishing module

Providers return clips in different sizes, frame rates and audio layouts, so the first rule of finishing is to normalize before you combine anything. normalize() forces 1920×1080 at 30 fps, applies a per-provider color grade from config, and guarantees a stereo 48 kHz audio track by inserting silence when a clip has none. assemble() chains xfade and acrossfade transitions, where each offset equals the running length minus the fade duration. loudness() applies single-pass loudnorm at -14 LUFS, my default for social, so adjust it to your platform’s spec.

export() makes the three ratios. For 9:16 and 1:1 it crops the 16:9 master with crop=ih*W/H:ih:(iw-ow)*focus_x:0, where focus_x slides the crop window left or right. The end card is generated natively with drawtext and sized to survive the vertical crop.

import subprocess
from pathlib import Path

from .qc import probe


def ff(args):
    r = subprocess.run(["ffmpeg", "-y", "-loglevel", "error"] + args, capture_output=True, text=True)
    if r.returncode:
        raise RuntimeError(r.stderr[-600:])


def normalize(src, dst, grade, W=1920, H=1080, fps=30):
    """Common spec: size, fps, pixel format, grade, stereo 48 kHz audio (silent if absent)."""
    has_audio = probe(src)["audio"]
    vf = (f"fps={fps},scale={W}:{H}:force_original_aspect_ratio=increase,crop={W}:{H},"
          f"setsar=1,{grade},format=yuv420p")
    args = ["-i", str(src)]
    if not has_audio:
        args += ["-f", "lavfi", "-i", "anullsrc=r=48000:cl=stereo"]
    args += ["-vf", vf, "-af", "aresample=48000,aformat=channel_layouts=stereo",
             "-map", "0:v", "-map", "0:a" if has_audio else "1:a", "-shortest",
             "-c:v", "libx264", "-crf", "18", "-preset", "medium", "-c:a", "aac", "-b:a", "192k", str(dst)]
    ff(args)


def end_card(dst, text, seconds, font, W=1920, H=1080, fps=30, bg="0x101820"):
    lines = text.replace("\\n", "\n")
    textfile = Path(dst).with_suffix(".txt")
    textfile.write_text(lines, encoding="utf-8")
    vf = (f"drawtext=fontfile={font}:textfile={textfile}:fontcolor=white:fontsize=56:"
          f"line_spacing=18:x=(w-text_w)/2:y=(h-text_h)/2,format=yuv420p")
    ff(["-f", "lavfi", "-i", f"color=c={bg}:s={W}x{H}:r={fps}:d={seconds}",
        "-f", "lavfi", "-i", "anullsrc=r=48000:cl=stereo", "-vf", vf, "-t", str(seconds),
        "-c:v", "libx264", "-crf", "18", "-c:a", "aac", "-b:a", "192k", str(dst)])


def assemble(clips, dst, fade=0.5):
    durs = [probe(c)["duration"] for c in clips]
    args, vparts, aparts = [], [], []
    for c in clips:
        args += ["-i", str(c)]
    vlast, alast, acc = "[0:v]", "[0:a]", durs[0]
    for i in range(1, len(clips)):
        off = round(acc - fade, 3)
        vparts.append(f"{vlast}[{i}:v]xfade=transition=fade:duration={fade}:offset={off}[v{i}]")
        aparts.append(f"{alast}[{i}:a]acrossfade=d={fade}[a{i}]")
        vlast, alast = f"[v{i}]", f"[a{i}]"
        acc += durs[i] - fade
    fc = ";".join(vparts + aparts)
    ff(args + ["-filter_complex", fc, "-map", vlast, "-map", alast,
               "-c:v", "libx264", "-crf", "18", "-c:a", "aac", "-b:a", "192k", str(dst)])


def loudness(src, dst, lufs=-14):
    ff(["-i", str(src), "-af", f"loudnorm=I={lufs}:TP=-1.5:LRA=11", "-c:v", "copy",
        "-c:a", "aac", "-b:a", "192k", "-ar", "48000", str(dst)])


def export(master, outdir, name, formats):
    outs = []
    for label, f in formats.items():
        W, H = f["size"]
        fx = f.get("focus_x", 0.5)
        crop = f"crop=ih*{W}/{H}:ih:(iw-ow)*{fx}:0" if W / H < 16 / 9 else "null"
        dst = Path(outdir) / f"{name}_{label}.mp4"
        ff(["-i", str(master), "-vf", f"{crop},scale={W}:{H}:flags=lanczos,setsar=1",
            "-c:v", "libx264", "-crf", "19", "-preset", "slow", "-c:a", "copy",
            "-movflags", "+faststart", str(dst)])
        outs.append(dst)
    return outs


def contact_sheet(master, dst, cols=4, rows=3):
    d = probe(master)["duration"]
    step = max(d / (cols * rows), 0.5)
    ff(["-i", str(master), "-vf", f"fps=1/{step:.3f},scale=320:-1,tile={cols}x{rows}",
        "-frames:v", "1", str(dst)])

Step 11: Wire the orchestrator

The orchestrator ties everything together. Shots run in a thread pool, one thread per in-flight generation, and each thread follows the same loop: skip if cached, otherwise try providers in order, skip any the ledger cannot afford, generate, charge, run QC, then either record the clip or move to the next provider. It charges before QC because providers bill for generation whether or not you like the result. If any shot fails every candidate, the run stops with exit code 1 and tells you how much it spent. A rerun picks up where it left off.

import argparse
import json
import os
import sys
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

from . import finish, qc
from .ledger import Ledger
from .providers import ProviderError, build_providers
from .router import candidates
from .spec import load_shots, load_yaml


def run_shot(shot, cfg, providers, ledger, manifest, mlock, out):
    key = shot.key()
    with mlock:
        done = manifest.get(shot.id)
    if done and done["key"] == key and Path(done["path"]).exists():
        return shot.id, "cached", done
    tried = []
    for prov, est in candidates(shot, cfg, providers):
        if not ledger.can_afford(est):
            tried.append((prov.name, "over-budget"))
            continue
        clip = Path(out) / "raw" / f"{shot.id}__{prov.name}.mp4"
        clip.parent.mkdir(parents=True, exist_ok=True)
        try:
            prov.generate(shot, shot.still, clip)
        except ProviderError as e:
            tried.append((prov.name, f"error: {str(e)[:120]}"))
            continue
        ledger.charge(shot.id, prov.name, est, shot.seconds)
        ok, rep = qc.check(clip, shot.still, shot, cfg["qc"])
        if ok:
            rec = {"key": key, "provider": prov.name, "path": str(clip), "usd": round(est, 4), "qc": rep}
            with mlock:
                manifest[shot.id] = rec
                (Path(out) / "manifest.json").write_text(json.dumps(manifest, indent=2))
            return shot.id, "ok", rec
        tried.append((prov.name, f"qc: {','.join(rep.get('problems', ['?']))}"))
    return shot.id, "failed", {"tried": tried}


def cmd_plan(cfg, shots, providers):
    total = 0.0
    for s in shots:
        c = candidates(s, cfg, providers)
        first = c[0] if c else None
        est = first[1] if first else 0.0
        total += est
        chain = " > ".join(f"{p.name}(${e:.2f})" for p, e in c) or "NO ROUTE"
        print(f"{s.id:<14}{s.intent:<10}{s.seconds:>2}s  {chain}")
    print(f"first-choice estimate: ${total:.2f}  budget: ${cfg['budget_usd']:.2f}")
    return 1 if total > cfg["budget_usd"] else 0


def cmd_run(cfg, shots, campaign, providers, workers):
    out = Path(cfg["output_dir"])
    out.mkdir(parents=True, exist_ok=True)
    ledger = Ledger(out / "ledger.json", cfg["budget_usd"])
    mpath = out / "manifest.json"
    manifest = json.loads(mpath.read_text()) if mpath.exists() else {}
    mlock, results = threading.Lock(), {}
    with ThreadPoolExecutor(max_workers=workers) as ex:
        futs = [ex.submit(run_shot, s, cfg, providers, ledger, manifest, mlock, out) for s in shots]
        for f in as_completed(futs):
            sid, status, rec = f.result()
            results[sid] = (status, rec)
            note = rec.get("provider", rec.get("tried", ""))
            print(f"[{status:>6}] {sid}  {note}")
    failed = [sid for sid, (st, _) in results.items() if st == "failed"]
    if failed:
        print(f"FAILED shots: {failed}. Spent ${ledger.spent:.2f}. Fix and re-run; finished shots are cached.")
        return 1
    fin = cfg["finish"]
    work = out / "work"
    work.mkdir(exist_ok=True)
    normed = []
    for s in shots:
        rec = manifest[s.id]
        grade = fin["grade"].get(rec["provider"], fin["grade"]["default"])
        dst = work / f"{s.id}.mp4"
        finish.normalize(rec["path"], dst, grade)
        normed.append(dst)
    card = work / "endcard.mp4"
    finish.end_card(card, campaign["cta"], fin["end_card_seconds"], fin["font"])
    joined = work / "joined.mp4"
    finish.assemble(normed + [card], joined, fin["crossfade_s"])
    master = out / f"{campaign['name']}_master.mp4"
    finish.loudness(joined, master, fin["lufs"])
    files = finish.export(master, out, campaign["name"], fin["formats"])
    finish.contact_sheet(master, out / f"{campaign['name']}_contact.png")
    print(f"master: {master}")
    for f in files:
        print(f"export: {f}")
    print(f"total spend: ${ledger.spent:.2f} of ${cfg['budget_usd']:.2f}")
    return 0


def main(argv=None):
    ap = argparse.ArgumentParser(prog="adforge")
    ap.add_argument("command", choices=["plan", "run"])
    ap.add_argument("--config", default="config.yaml")
    ap.add_argument("--shots", default="shots.yaml")
    ap.add_argument("--workers", type=int, default=3)
    a = ap.parse_args(argv)
    cfg = load_yaml(a.config)
    shots, campaign = load_shots(a.shots)
    providers = build_providers(cfg, os.environ)
    if a.command == "plan":
        return cmd_plan(cfg, shots, providers)
    return cmd_run(cfg, shots, campaign, providers, a.workers)


if __name__ == "__main__":
    sys.exit(main())

Steps 12-13: Run, Verify and Cut Over to a Live API

Step 12: Run the pipeline

Run it. Shots finish in thread order, so your lines may print in a different sequence than mine.

python3 -m adforge.pipeline run

[    ok] s02_pour  kling_std
[    ok] s03_desk  omni_720
[    ok] s01_hero  veo_fast
[    ok] s04_detail  kling_std
[    ok] s05_long  kling_4
master: out/ember_master.mp4
export: out/ember_landscape.mp4
export: out/ember_vertical.mp4
export: out/ember_square.mp4
total spend: $3.99 of $6.00

Now rerun the same command. Every shot should say cached and the ledger should not move. Generation is skipped, but finishing is not cached in this version, so my second run still took about 83 seconds. Advanced Tips covers the fix.

Step 13: Cut over to a live API

Every provider that has kind: http is driven entirely by config. You supply the base URL and key through environment variables and describe the request in YAML. The submit path, body fields, status strings and output path differ per vendor and change over time, so copy them from the model’s own docs page. For Kling, start at KlingAI’s API documentation. For models hosted on Runway Dev, each model has its own page with access, pricing and code samples.

For a Runway-hosted route, our Runway API setup guide covers keys and task polling, and the Seedance 2.5 API setup shows a multi-reference model in action.

To prove the adapter works before you touch a real account, this repo includes a 61-line fake task API. It rejects bad keys, returns a 429 on the first submit, reports RUNNING once, then SUCCEEDED, and fails any prompt containing the word FAIL. Here is the adapter config the test uses:

budget_usd: 3.00
output_dir: out_http

providers:
  fake_task_api:
    kind: http
    base_url_env: VIDEO_API_BASE
    key_env: VIDEO_API_KEY
    model: fake-model-1
    usd_per_second: 0.10
    max_seconds: 10
    audio: false
    submit:
      path: /v1/image_to_video
      body:
        model: "{model}"
        promptImage: "{image_data_uri}"
        promptText: "{prompt}"
        duration: "{seconds}"
    poll: {path: "/v1/tasks/{task_id}", interval: 1, timeout: 60}
    fields:
      task_id: id
      status: status
      output: output.0
      done: [SUCCEEDED]
      failed: [FAILED, CANCELLED]

routes:
  default: [fake_task_api]

qc:
  duration_tolerance_s: 0.3
  min_height: 700
  max_frozen_s: 0.5
  max_black_s: 0.3
  min_ssim_first: 0.55

finish: {}

And the fake server:

"""Tiny task-style video API for testing the HTTP adapter offline."""
import base64
import json
import re
import subprocess
import tempfile
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path

TASKS, POLLS, HITS = {}, {}, {"n": 0}
FILES = Path(tempfile.mkdtemp(prefix="fakeapi_"))


class H(BaseHTTPRequestHandler):
    def _send(self, code, body, ctype="application/json"):
        data = body if isinstance(body, bytes) else json.dumps(body).encode()
        self.send_response(code)
        self.send_header("Content-Type", ctype)
        self.send_header("Content-Length", str(len(data)))
        self.end_headers()
        self.wfile.write(data)

    def do_POST(self):
        if self.headers.get("Authorization") != "Bearer test-key":
            return self._send(401, {"error": "bad key"})
        HITS["n"] += 1
        if HITS["n"] == 1:
            return self._send(429, {"error": "slow down"})
        body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
        tid = f"task{len(TASKS) + 1}"
        TASKS[tid], POLLS[tid] = body, 0
        self._send(200, {"id": tid})

    def do_GET(self):
        m = re.match(r"/v1/tasks/(\w+)", self.path)
        if m:
            tid = m.group(1)
            POLLS[tid] += 1
            if "FAIL" in TASKS[tid]["promptText"]:
                return self._send(200, {"status": "FAILED", "failure": "moderation"})
            if POLLS[tid] < 2:
                return self._send(200, {"status": "RUNNING"})
            img = FILES / f"{tid}.png"
            img.write_bytes(base64.b64decode(TASKS[tid]["promptImage"].split(",", 1)[1]))
            mp4 = FILES / f"{tid}.mp4"
            subprocess.run(["ffmpeg", "-y", "-loglevel", "error", "-loop", "1", "-i", str(img),
                            "-vf", "scale=1280:720:force_original_aspect_ratio=increase,crop=1280:720,format=yuv420p",
                            "-t", str(TASKS[tid]["duration"]), "-r", "24", "-c:v", "libx264", str(mp4)], check=True)
            return self._send(200, {"status": "SUCCEEDED", "output": [f"http://127.0.0.1:8099/files/{tid}.mp4"]})
        m = re.match(r"/files/(\w+\.mp4)", self.path)
        if m:
            return self._send(200, (FILES / m.group(1)).read_bytes(), "video/mp4")
        self._send(404, {"error": "not found"})

    def log_message(self, *a):
        pass


if __name__ == "__main__":
    print("fake API on http://127.0.0.1:8099")
    HTTPServer(("127.0.0.1", 8099), H).serve_forever()

Start the server in one terminal. In another, make a two-shot test file where the second prompt contains FAIL, then run against it:

python3 tools/fake_api.py &
cat > shots.http.yaml <<'EOF'
campaign: {name: httptest, cta: "x"}
shots:
  - {id: h01, still: stills/can_front.png, seconds: 3, prompt: "Slow push-in"}
  - {id: h02, still: stills/can_pour.png, seconds: 3, prompt: "FAIL this one"}
EOF
VIDEO_API_BASE=http://127.0.0.1:8099 VIDEO_API_KEY=test-key \
  python3 -m adforge.pipeline run --config config.http.yaml --shots shots.http.yaml --workers 1

[    ok] h01  fake_task_api
[failed] h02  [('fake_task_api', "error: task task2 FAILED: {'status': 'FAILED', 'failure': 'moderation'}")]
FAILED shots: ['h02']. Spent $0.30. Fix and re-run; finished shots are cached.

That output proves four behaviors at once: the 429 retry worked, polling reached SUCCEEDED, a failed task charged nothing, and the run exited with a clear message. Swap the config values for a real model and the same code drives it.

What the Output Should Look Like

A good run leaves a tidy output folder. Probe the exports with ffprobe and you should see the sizes below. The master runs 31.0 seconds because the five clips total 31 seconds, the end card adds 2.5, and five 0.5-second crossfades subtract 2.5.

File	Resolution	Duration	Audio
ember_master.mp4	1920×1080 at 30 fps	31.0 s	AAC stereo
ember_landscape.mp4	1920×1080	31.0 s	AAC stereo
ember_vertical.mp4	1080×1920	31.0 s	AAC stereo
ember_square.mp4	1080×1080	31.0 s	AAC stereo
ember_contact.png	1280×540 tile sheet	Still image	None

Measuring the master with FFmpeg’s ebur128 filter reported integrated loudness of -14.1 LUFS, which is within 0.1 of the -14 target. The contact sheet is a 4×3 grid sampled evenly through the master, ending on the dark CTA card. Scan it for product drift, odd crops and any shot that looks like a different brand. Each record in manifest.json shows which provider won and what QC measured:

{
  "s02_pour": {
    "key": "b1c301e008c7846a",
    "provider": "kling_std",
    "path": "out/raw/s02_pour__kling_std.mp4",
    "usd": 0.45,
    "qc": {"duration": 5.0, "size": "1280x720", "fps": 24.0, "audio": false,
           "frozen_s": 0, "black_s": 0, "ssim_first": 0.995, "problems": []}
  }
}

Cost Math: Where the Money Goes

The ledger recorded these charges. They are estimates from published rates, not a bill, because the mock providers cost nothing. Live, expect your invoice to differ if a provider rounds up, adds audio surcharges or applies a promo.

Shot	Provider	Seconds	Rate per second	Cost
s01_hero	veo_fast	6	$0.15	$0.90
s02_pour	kling_std	5	$0.09	$0.45
s03_desk	omni_720	4	$0.10	$0.40
s04_detail	kling_std	4	$0.09	$0.36
s05_long	kling_4	12	$0.157	Not listed
Total		31		Not listed

Here is what the same 31 seconds would cost on one model, using the same published rates (my arithmetic). All on Kling 3.0 Standard with audio: $4.03. All on Kling 4.0 at the list price: $4.87. All on Kling 3.0 Pro with audio: $5.27. All on Seedance 2.5 at 720p: $9.30. So routing is not a big discount on a 31-second spot. The cheapest single-model run lands close to the routed total. What routing buys is placement. The premium audio model lands on the hero, the 30-second-capable model takes the 12-second orbit, and cheap tiers handle b-roll. The savings grow once a catalog holds hundreds of shots and the cheap tier handles most of them.

The bigger lesson comes from the failure case. With min_height set to 1000, every 720p provider would fail QC, so the pipeline keeps falling through the chain and the ledger keeps logging spend before the run gives up. A strict gate with a long fallback chain can cost more than no gate at all, so cap retries per shot if your budget is tight.

Common Pitfalls

Billing for clips that fail QC. Providers charge for generation, not for your opinion of the result. This project charges the moment a clip downloads. Mirror that in your own tooling, or your budget math will drift low.
Hardcoding model IDs. Preview models get retired. The reported October 22 shutdown of Gemini API preview models for Veo 3.1 is a fresh example. Keep IDs in config and keep a fallback in every chain. Our guide to migrating from Sora 2 to the Veo 3.1 API shows what a forced move looks like.
Letting the vertical crop eat the product. A 9:16 window keeps only a narrow slice of a 16:9 frame’s width. Frame the product in the center third and set focus_x per campaign if it sits off-center.
Trusting SSIM alone. My recolored can scored 0.973. SSIM measures structure, so label text, logo color and packaging tint can drift unseen. Keep the color check and eyeball hero shots anyway.
Mixing frame rates and sizes before xfade. FFmpeg’s xfade needs both inputs to share size, frame rate and timebase. Normalize every clip first, as Step 10 does.
Skipping the audio track. acrossfade fails if one clip has no audio stream. Insert silence with anullsrc so every clip carries stereo 48 kHz audio.
Polling too fast. A one-second poll interval across many parallel shots is a quick route to 429 errors. Start at five seconds in production and keep the worker count low until you know your rate limits.
Letting promo prices rot in config. Kling 4.0’s 20% promo ends November 30, 2026. Add a comment with the price date next to every rate so stale numbers stand out.

Troubleshooting

Symptom	Likely cause	Fix
`ffmpeg: command not found`	FFmpeg is not installed or not on PATH	Install it with your package manager and confirm `ffmpeg -version` works in the same shell
`Error while filtering: Invalid argument`	Inputs to a combining filter differ in size or frame rate	Run every clip through `normalize()` before assembling
`moov atom not found`	A download saved an error page or a partial file as .mp4	Delete the file in `out/raw`, check the provider’s status response and rerun
`KeyError: 'VIDEO_API_BASE'`	Env var named in config is not set	Export both the base URL and key variables in the shell that runs the pipeline
HTTP 401 from a provider	Wrong key, wrong header scheme or a missing version header	Match the docs exactly and add any required headers under `headers:` in config
Repeated 429 responses	Too many parallel shots or a short poll interval	Lower `--workers`, raise `poll.interval`, and keep the built-in backoff
Task polls until timeout	Status strings in `done` do not match the API’s wording	Print one poll response and copy the exact success and failure strings into config
Every shot fails `qc: resolution`	`min_height` exceeds what the providers return	Lower it to the lowest resolution you will ship and rerun
`drawtext` error or blank end card	Missing libfreetype or a wrong font path	Check `ffmpeg -filters` for drawtext and point `finish.font` at a real .ttf file
Product cut off in the 9:16 export	Subject sits left or right of center	Adjust `focus_x` in the format config or reshoot the still
`FileNotFoundError` on a still	Relative path wrong for the directory you run from	Run from the project root or use absolute paths in shots.yaml
`ValueError: duplicate shot ids`	Two shots share an id	Rename one so each id is unique

Advanced Tips

Use references and keyframes. Kling 4.0 accepts up to 15 reference assets and 10 keyframes, and Seedance 2.5 is built around multi-reference input. Add a refs list to the Shot dataclass and pass it through the body template to keep a product consistent across a long take.
Reserve 4K for the hero. Kling 3.0’s 4K tier costs $0.43 per second against $0.09 for Standard silent, about 4.8 times more. Render one 4K hero and ship the rest at 720p or 1080p.
Know what FFmpeg upscaling is. A Lanczos scale makes a bigger file, not new detail. It is fine for a final nudge to the platform’s required size but it is no substitute for a provider’s native 4K.
Cache the finishing stage. My rerun spent about 83 seconds re-encoding unchanged clips. Hash the manifest and skip normalize() when the source clip and grade have not changed.
Gate pull requests on the plan. Because plan exits 1 when the estimate beats the budget, a CI job can block a PR that quietly doubles ad spend.
Test agent access. Kling’s launch coverage mentions MCP and CLI integration, which could let an AI agent drive generation. Worth a trial once your API access is confirmed.
Add a human hold for heroes. Have the pipeline write hero clips to a review folder and pause until a person approves them. A hero that drifts wastes more than the clip’s price.

A simple Lanczos upscale with light sharpening looks like this:

ffmpeg -i out/ember_master.mp4 -vf "scale=3840:2160:flags=lanczos,unsharp=5:5:0.6" -c:v libx264 -crf 18 -c:a copy out/ember_master_4k.mp4

If you are still comparing tools, our roundup of the best text-to-video generator tools maps the field, and the AI and machine learning hub tracks new model releases as they land.

FAQ

What is the best image to video AI for product ads?

No single model wins. Kling 4.0 targets long, reference-heavy shots, Veo 3.1 Fast suits audio-led hero clips, and Gemini Omni Flash 1.1 is the cheapest option at 720p. Routing by shot intent beats picking one favorite.

Can I follow this tutorial without paying for an API?

Yes. The sample config uses mock providers that animate your stills with FFmpeg, so the full pipeline, QC gate and exports run offline at no cost. You only need keys when you switch a provider to kind: http.

Is the Kling 4.0 API available to everyone?

Not confirmed. Kling says API access is opening in stages for enterprise customers and developers, and a partner announced an October 22 launch. Check your own dashboard before you build on it.

How long should each image-to-video clip be?

Keep most shots between 4 and 6 seconds. Short clips cost less, drift less and cut together faster. Use the 12-second long-shot route only for continuous moves like an orbit.

Does FFmpeg upscale video with AI?

No. FFmpeg’s scale filter resamples pixels with methods like Lanczos and adds no new detail. For real resolution gains, pay for a provider’s native 4K tier on the shots that matter.

Why route across several providers instead of using one?

Prices differ by more than 50% between tiers, models launch and retire within weeks, and failures happen. A router with fallbacks lowers cost and keeps campaigns moving when one vendor stumbles.

Can I use these clips commercially?

Check each provider’s terms. Commercial rights can differ by vendor, model, plan and API tier, and a promo plan may not carry the same license as a paid one. Read the license before a client sees the work.

Nadia Dubois

AI & Innovation Editor

Nadia Dubois is the AI & Innovation Editor at Tech Insider, where she tracks the rapid evolution of artificial intelligence, from foundation models to real-world enterprise deployment. She previously covered AI and startups for La Tribune and contributed to MIT Technology Review's European coverage. Nadia specializes in generative AI, AI regulation, and the intersection of technology and European industrial policy. She holds a dual degree in Computational Linguistics and Journalism from Sciences Po Paris.

View all articles