AI Video Generation Patterns

AI Video Generation Patterns

Best practices for generating videos with AI tools.

Platform Overview

Runway API Integration

import requests
from dataclasses import dataclass
from typing import Optional
import time

@dataclass
class GenerationRequest:
    prompt: str
    image_url: Optional[str] = None
    duration: int = 4  # seconds
    aspect_ratio: str = "16:9"
    motion_strength: int = 5  # 1-10

@dataclass
class GenerationResult:
    id: str
    status: str
    video_url: Optional[str] = None
    thumbnail_url: Optional[str] = None


class RunwayClient:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.runwayml.com/​v1"
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/​json"
        }

    def generate_video(self, request: GenerationRequest) -> GenerationResult:
        """Generate video from text or image prompt."""
        payload = {
            "model": "gen3a_turbo",
            "prompt": request.prompt,
            "duration": request.duration,
            "aspectRatio": request.aspect_ratio
        }

        if request.image_url:
            payload["imageUrl"] = request.image_url
            payload["mode"] = "image-to-video"
        else:
            payload["mode"] = "text-to-video"

        response = requests.post(
            f"{self.base_url}/​generations",
            headers=self.headers,
            json=payload
        )
        response.raise_for_status()
        data = response.json()

        return GenerationResult(
            id=data["id"],
            status=data["status"]
        )

    def get_generation(self, generation_id: str) -> GenerationResult:
        """Check status of generation."""
        response = requests.get(
            f"{self.base_url}/​generations/​{generation_id}",
            headers=self.headers
        )
        response.raise_for_status()
        data = response.json()

        return GenerationResult(
            id=data["id"],
            status=data["status"],
            video_url=data.get("output", {}).get("videoUrl"),
            thumbnail_url=data.get("output", {}).get("thumbnailUrl")
        )

    def wait_for_completion(
        self,
        generation_id: str,
        poll_interval: float = 5.0,
        timeout: float = 300.0
    ) -> GenerationResult:
        """Poll until generation completes."""
        start_time = time.time()

        while time.time() - start_time < timeout:
            result = self.get_generation(generation_id)

            if result.status == "SUCCEEDED":
                return result
            elif result.status == "FAILED":
                raise Exception(f"Generation failed: {generation_id}")

            time.sleep(poll_interval)

        raise TimeoutError(f"Generation timed out: {generation_id}")

    def generate_and_wait(self, request: GenerationRequest) -> GenerationResult:
        """Generate video and wait for completion."""
        initial = self.generate_video(request)
        return self.wait_for_completion(initial.id)

Pika Labs Integration

class PikaClient:
    """Pika Labs API client (when API available)."""

    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.pika.art/​v1"

    def text_to_video(
        self,
        prompt: str,
        negative_prompt: str = "",
        style: str = "realistic",  # realistic, anime, 3d-animation
        duration: int = 3,
        fps: int = 24
    ) -> dict:
        """Generate video from text prompt."""
        response = requests.post(
            f"{self.base_url}/​generate",
            headers={"Authorization": f"Bearer {self.api_key}"},
            json={
                "prompt": prompt,
                "negativePrompt": negative_prompt,
                "style": style,
                "duration": duration,
                "fps": fps
            }
        )
        return response.json()

    def image_to_video(
        self,
        image_url: str,
        motion_prompt: str,
        motion_strength: float = 1.0
    ) -> dict:
        """Animate an image."""
        response = requests.post(
            f"{self.base_url}/​animate",
            headers={"Authorization": f"Bearer {self.api_key}"},
            json={
                "imageUrl": image_url,
                "motionPrompt": motion_prompt,
                "motionStrength": motion_strength
            }
        )
        return response.json()

Stable Video Diffusion (Local)

import torch
from diffusers import StableVideoDiffusionPipeline
from PIL import Image

class StableVideoGenerator:
    def __init__(self, model_id: str = "stabilityai/​stable-video-diffusion-img2vid-xt"):
        self.pipe = StableVideoDiffusionPipeline.from_pretrained(
            model_id,
            torch_dtype=torch.float16,
            variant="fp16"
        )
        self.pipe.to("cuda")

    def generate(
        self,
        image_path: str,
        num_frames: int = 25,
        fps: int = 7,
        motion_bucket_id: int = 127,
        noise_aug_strength: float = 0.02,
        decode_chunk_size: int = 8
    ) -> list:
        """Generate video from image."""
        image = Image.open(image_path).resize((1024, 576))

        frames = self.pipe(
            image,
            num_frames=num_frames,
            motion_bucket_id=motion_bucket_id,
            noise_aug_strength=noise_aug_strength,
            decode_chunk_size=decode_chunk_size
        ).frames[0]

        return frames

    def save_video(self, frames: list, output_path: str, fps: int = 7):
        """Save frames as video."""
        from diffusers.utils import export_to_video
        export_to_video(frames, output_path, fps=fps)

Prompt Engineering for Video

from dataclasses import dataclass
from typing import List, Optional

@dataclass
class VideoPrompt:
    subject: str
    action: str
    setting: str
    style: Optional[str] = None
    camera: Optional[str] = None
    lighting: Optional[str] = None
    mood: Optional[str] = None

    def to_prompt(self) -> str:
        """Build optimized prompt string."""
        parts = [
            f"{self.subject} {self.action}",
            f"in {self.setting}" if self.setting else "",
        ]

        modifiers = []
        if self.style:
            modifiers.append(self.style)
        if self.camera:
            modifiers.append(f"{self.camera} shot")
        if self.lighting:
            modifiers.append(f"{self.lighting} lighting")
        if self.mood:
            modifiers.append(f"{self.mood} mood")

        if modifiers:
            parts.append(", ".join(modifiers))

        return ", ".join(filter(None, parts))


class PromptLibrary:
    """Common prompt patterns for video generation."""

    CAMERA_MOVEMENTS = [
        "slow zoom in",
        "slow zoom out",
        "dolly shot",
        "tracking shot",
        "pan left to right",
        "crane shot",
        "steady cam",
        "handheld",
        "aerial drone shot",
        "first person POV"
    ]

    STYLES = [
        "cinematic",
        "photorealistic",
        "anime style",
        "3D animation",
        "stop motion",
        "vintage film grain",
        "documentary style",
        "music video aesthetic",
        "noir",
        "cyberpunk"
    ]

    LIGHTING = [
        "golden hour",
        "blue hour",
        "dramatic shadows",
        "soft diffused",
        "neon lights",
        "natural sunlight",
        "studio lighting",
        "candlelight",
        "moonlight",
        "backlit silhouette"
    ]

    @staticmethod
    def build_cinematic_prompt(
        subject: str,
        action: str,
        setting: str
    ) -> str:
        """Build a cinematic video prompt."""
        prompt = VideoPrompt(
            subject=subject,
            action=action,
            setting=setting,
            style="cinematic, high production value",
            camera="slow tracking",
            lighting="dramatic",
            mood="epic"
        )
        return prompt.to_prompt()

    @staticmethod
    def build_product_prompt(product: str, features: List[str]) -> str:
        """Build product showcase prompt."""
        return f"{product} rotating on clean white background, {', '.join(features)}, professional product photography, soft studio lighting, 4K quality"

    @staticmethod
    def build_nature_prompt(scene: str, atmosphere: str) -> str:
        """Build nature/​landscape prompt."""
        return f"beautiful {scene}, {atmosphere} atmosphere, cinematic wide shot, golden hour lighting, National Geographic style, 8K quality"

Batch Generation

import asyncio
from dataclasses import dataclass
from typing import List
from enum import Enum

class GenerationStatus(Enum):
    PENDING = "pending"
    GENERATING = "generating"
    COMPLETED = "completed"
    FAILED = "failed"

@dataclass
class BatchJob:
    id: str
    prompt: str
    status: GenerationStatus = GenerationStatus.PENDING
    result_url: Optional[str] = None
    error: Optional[str] = None


class BatchGenerator:
    def __init__(self, runway_client: RunwayClient):
        self.client = runway_client
        self.jobs: List[BatchJob] = []

    def add_job(self, prompt: str) -> str:
        """Add generation job to batch."""
        import uuid
        job_id = str(uuid.uuid4())
        job = BatchJob(id=job_id, prompt=prompt)
        self.jobs.append(job)
        return job_id

    async def process_job(self, job: BatchJob, semaphore: asyncio.Semaphore):
        """Process single job with concurrency control."""
        async with semaphore:
            job.status = GenerationStatus.GENERATING
            try:
                request = GenerationRequest(prompt=job.prompt)
                result = await asyncio.to_thread(
                    self.client.generate_and_wait, request
                )
                job.status = GenerationStatus.COMPLETED
                job.result_url = result.video_url
            except Exception as e:
                job.status = GenerationStatus.FAILED
                job.error = str(e)

    async def process_all(self, max_concurrent: int = 3):
        """Process all jobs with concurrency limit."""
        semaphore = asyncio.Semaphore(max_concurrent)

        tasks = [
            self.process_job(job, semaphore)
            for job in self.jobs
        ]

        await asyncio.gather(*tasks)

    def get_results(self) -> dict:
        """Get summary of all jobs."""
        return {
            "total": len(self.jobs),
            "completed": sum(1 for j in self.jobs if j.status == GenerationStatus.COMPLETED),
            "failed": sum(1 for j in self.jobs if j.status == GenerationStatus.FAILED),
            "jobs": [
                {
                    "id": j.id,
                    "prompt": j.prompt[:50] + "...",
                    "status": j.status.value,
                    "url": j.result_url,
                    "error": j.error
                }
                for j in self.jobs
            ]
        }

Video Enhancement

class VideoEnhancer:
    """Post-process AI-generated videos."""

    def __init__(self):
        pass

    def upscale(
        self,
        input_path: str,
        output_path: str,
        scale: int = 2
    ):
        """Upscale video using AI."""
        import subprocess

        # Using Real-ESRGAN for upscaling
        subprocess.run([
            "realesrgan-ncnn-vulkan",
            "-i", input_path,
            "-o", output_path,
            "-s", str(scale),
            "-n", "realesrgan-x4plus"
        ])

    def interpolate_frames(
        self,
        input_path: str,
        output_path: str,
        target_fps: int = 60
    ):
        """Interpolate frames for smoother video."""
        import subprocess

        # Using RIFE for frame interpolation
        subprocess.run([
            "ffmpeg",
            "-i", input_path,
            "-vf", f"minterpolate=fps={target_fps}:mi_mode=mci:mc_mode=aobmc:me_mode=bidir:vsbmc=1",
            "-c:v", "libx264",
            "-preset", "slow",
            "-crf", "18",
            output_path
        ])

    def remove_artifacts(
        self,
        input_path: str,
        output_path: str
    ):
        """Remove common AI artifacts."""
        import subprocess

        subprocess.run([
            "ffmpeg",
            "-i", input_path,
            "-vf", "unsharp=5:5:0.8:3:3:0.4,hqdn3d=3:3:6:6",
            "-c:v", "libx264",
            "-crf", "18",
            output_path
        ])

    def add_audio(
        self,
        video_path: str,
        audio_path: str,
        output_path: str
    ):
        """Add audio track to silent AI video."""
        import subprocess

        subprocess.run([
            "ffmpeg",
            "-i", video_path,
            "-i", audio_path,
            "-c:v", "copy",
            "-c:a", "aac",
            "-shortest",
            output_path
        ])

Storyboard to Video

from dataclasses import dataclass
from typing import List

@dataclass
class StoryboardScene:
    description: str
    duration: int  # seconds
    transition: str = "cut"  # cut, fade, dissolve
    audio: Optional[str] = None

@dataclass
class Storyboard:
    title: str
    scenes: List[StoryboardScene]


class StoryboardRenderer:
    def __init__(self, video_client: RunwayClient):
        self.client = video_client

    async def render_storyboard(self, storyboard: Storyboard) -> List[str]:
        """Generate video for each scene."""
        video_paths = []

        for i, scene in enumerate(storyboard.scenes):
            print(f"Generating scene {i + 1}/​{len(storyboard.scenes)}")

            request = GenerationRequest(
                prompt=scene.description,
                duration=scene.duration
            )

            result = self.client.generate_and_wait(request)

            # Download video
            output_path = f"scene_{i:03d}.mp4"
            self._download_video(result.video_url, output_path)
            video_paths.append(output_path)

        return video_paths

    def _download_video(self, url: str, output_path: str):
        """Download video from URL."""
        import requests

        response = requests.get(url, stream=True)
        with open(output_path, 'wb') as f:
            for chunk in response.iter_content(chunk_size=8192):
                f.write(chunk)

    def concatenate_scenes(
        self,
        video_paths: List[str],
        output_path: str
    ):
        """Combine scenes into final video."""
        import subprocess

        # Create file list
        with open('scenes.txt', 'w') as f:
            for path in video_paths:
                f.write(f"file '{path}'\n")

        subprocess.run([
            "ffmpeg",
            "-f", "concat",
            "-safe", "0",
            "-i", "scenes.txt",
            "-c", "copy",
            output_path
        ])

References

• Runway API Documentation
• Stable Video Diffusion
• Pika Labs
• Luma Dream Machine
• Kling AI