Python Ffmpeg

Python FFmpeg Skill

Use this skill for Python-driven FFmpeg work: encoding, filtering, audio processing, metadata probing, streaming, thumbnails, PyAV frame access, subprocess integration, and production troubleshooting.

When to Use This Skill

Use when the user asks for tasks covered by the frontmatter triggers, especially implementation guidance, debugging, architecture choices, production hardening, or performance-sensitive decisions in this domain. Start from this orchestrator, then load the focused reference file that matches the requested detail level.

Core Workflow

1. Choose the integration layer first: ffmpeg-python for readable filter graphs, subprocess for full CLI parity, PyAV for frame-level access, and moviepy only for simple edits. 2. Probe inputs before processing so codec, duration, resolution, FPS, audio streams, and metadata are known. 3. Preserve audio explicitly whenever a video filter is applied; filtered video streams do not automatically carry audio. 4. Select the output codec and acceleration path based on compatibility, compression, and deployment hardware. 5. Use overwrite/error-handling patterns consistently: overwrite_output() with ffmpeg-python or -y with subprocess, plus captured stderr for diagnostics. 6. For large media, stream frames or use temp files instead of reading all stdout or frames into memory.

Key Gotchas

• Video filters commonly drop audio unless the audio stream is passed to ffmpeg.output(...) or copied explicitly.
• Subprocess pipes can deadlock if stdout/stderr are not drained or frame reads are not sized correctly.
• Windows paths are safest as Path(...).as_posix() or as subprocess argument-list entries, not hand-quoted command strings.
• CRF values are codec-specific: H.265 CRF 28 is roughly comparable to H.264 CRF 23.
• Hardware encoders trade compression efficiency for speed; validate availability before choosing NVENC, QSV, AMF, or VAAPI.

Reference Map

• references/ffmpeg-complete-recipes.md - Full original recipe guide covering encoding, hardware acceleration, audio/audio filters, video filters, trimming, concatenation, streaming, probing, thumbnails, PyAV, subprocess patterns, common transformations, errors, and performance tips.
• references/ffmpeg-advanced-patterns.md - Additional advanced FFmpeg patterns already maintained for this skill.

Response Guidance

• Preserve the user's existing framework, library, and tooling choices unless there is a clear compatibility or performance reason to suggest an alternative.
• Give copy-pasteable code only for the exact task at hand; otherwise point to the relevant reference section.
• Call out tradeoffs, failure modes, and verification steps for production workflows.
• Prefer accessible, maintainable, measurable solutions over clever micro-optimizations.

FFmpeg Advanced Patterns Reference

Advanced FFmpeg patterns for complex video processing workflows.

Complex Filter Graphs

Multiple Inputs and Outputs

import ffmpeg

# Split video into multiple outputs with different qualities
input_file = ffmpeg.input('input.mp4')

# Split the stream
split = input_file.video.filter_multi_output('split', 3)

# Create different quality outputs
high = split[0].filter('scale', 1920, 1080)
medium = split[1].filter('scale', 1280, 720)
low = split[2].filter('scale', 640, 360)

# Output all three
(
    ffmpeg
    .merge_outputs(
        ffmpeg.output(high, input_file.audio, 'high.mp4', vcodec='libx264', crf=18),
        ffmpeg.output(medium, input_file.audio, 'medium.mp4', vcodec='libx264', crf=23),
        ffmpeg.output(low, input_file.audio, 'low.mp4', vcodec='libx264', crf=28)
    )
    .overwrite_output()
    .run()
)

Side-by-Side Video Comparison

video1 = ffmpeg.input('original.mp4')
video2 = ffmpeg.input('processed.mp4')

# Stack horizontally
(
    ffmpeg
    .filter([video1, video2], 'hstack')
    .output('comparison.mp4')
    .overwrite_output()
    .run()
)

# Stack vertically
(
    ffmpeg
    .filter([video1, video2], 'vstack')
    .output('comparison.mp4')
    .overwrite_output()
    .run()
)

# 2x2 Grid
v1 = ffmpeg.input('video1.mp4').filter('scale', 640, 360)
v2 = ffmpeg.input('video2.mp4').filter('scale', 640, 360)
v3 = ffmpeg.input('video3.mp4').filter('scale', 640, 360)
v4 = ffmpeg.input('video4.mp4').filter('scale', 640, 360)

row1 = ffmpeg.filter([v1, v2], 'hstack')
row2 = ffmpeg.filter([v3, v4], 'hstack')
(
    ffmpeg
    .filter([row1, row2], 'vstack')
    .output('grid.mp4')
    .overwrite_output()
    .run()
)

Crossfade Transitions

# Crossfade between two videos
video1 = ffmpeg.input('video1.mp4')
video2 = ffmpeg.input('video2.mp4')

# Video crossfade (last 1 second of video1 fades into first 1 second of video2)
v1 = video1.video.filter('trim', start=0, end=9).filter('setpts', 'PTS-STARTPTS')
v2 = video2.video.filter('trim', start=0, end=10).filter('setpts', 'PTS-STARTPTS')

# Apply xfade filter
video_out = ffmpeg.filter(
    [v1, v2],
    'xfade',
    transition='fade',
    duration=1,
    offset=8  # Start fade at 8 seconds
)

# Audio crossfade
a1 = video1.audio.filter('atrim', start=0, end=9).filter('asetpts', 'PTS-STARTPTS')
a2 = video2.audio.filter('atrim', start=0, end=10).filter('asetpts', 'PTS-STARTPTS')

audio_out = ffmpeg.filter(
    [a1, a2],
    'acrossfade',
    duration=1
)

(
    ffmpeg
    .output(video_out, audio_out, 'output.mp4')
    .overwrite_output()
    .run()
)

# Transition types: fade, wipeleft, wiperight, wipeup, wipedown,
# slideleft, slideright, slideup, slidedown, circlecrop, rectcrop,
# distance, fadeblack, fadewhite, radial, smoothleft, smoothright,
# smoothup, smoothdown, circleopen, circleclose, vertopen, vertclose,
# horzopen, horzclose, dissolve, pixelize, diagtl, diagtr, diagbl, diagbr

Picture-in-Picture with Animation

main = ffmpeg.input('main.mp4')
pip = ffmpeg.input('pip.mp4')

# Scale PIP video
pip_scaled = pip.video.filter('scale', 320, 180)

# Animate PIP position (slide in from right)
# Uses expressions for dynamic positioning
video_out = ffmpeg.overlay(
    main.video,
    pip_scaled,
    x="if(lt(t,1), W, W-w-10+((W-w-10)/1)*(1-t))",  # Slide in during first second
    y=10
)

(
    ffmpeg
    .output(video_out, main.audio, 'output.mp4')
    .overwrite_output()
    .run()
)

Advanced Audio Processing

Multiple Audio Tracks

video = ffmpeg.input('video.mp4')
audio1 = ffmpeg.input('music.mp3')
audio2 = ffmpeg.input('voiceover.mp3')

# Mix multiple audio tracks
mixed_audio = ffmpeg.filter(
    [video.audio, audio1, audio2],
    'amix',
    inputs=3,
    duration='first',  # Match duration of first input
    dropout_transition=2
)

(
    ffmpeg
    .output(video.video, mixed_audio, 'output.mp4')
    .overwrite_output()
    .run()
)

# Adjust individual volumes before mixing
video = ffmpeg.input('video.mp4')
music = ffmpeg.input('music.mp3').filter('volume', 0.3)  # 30% volume
voice = ffmpeg.input('voiceover.mp3').filter('volume', 1.5)  # 150% volume

mixed = ffmpeg.filter([video.audio, music, voice], 'amix', inputs=3)

Audio Ducking (Voice Over Music)

# Duck music when voice is present
video = ffmpeg.input('video.mp4')
music = ffmpeg.input('music.mp3')
voice = ffmpeg.input('voiceover.mp3')

# Use sidechaincompress to duck music based on voice level
ducked_music = ffmpeg.filter(
    [music, voice],
    'sidechaincompress',
    threshold='0.02',
    ratio='10',
    attack='200',
    release='1000'
)

# Mix ducked music with voice
mixed = ffmpeg.filter([ducked_music, voice], 'amix', inputs=2)

(
    ffmpeg
    .output(video.video, mixed, 'output.mp4')
    .overwrite_output()
    .run()
)

Audio Visualization

# Create audio waveform visualization
audio = ffmpeg.input('audio.mp3')

# Showwaves filter
(
    ffmpeg
    .filter(audio, 'showwaves', s='1920x200', mode='line', rate=30)
    .output('waveform.mp4', vcodec='libx264')
    .overwrite_output()
    .run()
)

# Spectrum visualization
(
    ffmpeg
    .filter(audio, 'showspectrum', s='1920x1080', mode='combined', color='intensity', slide='scroll')
    .output('spectrum.mp4', vcodec='libx264')
    .overwrite_output()
    .run()
)

# Volume meter
(
    ffmpeg
    .filter(audio, 'showvolume', f=1, c='0xff0000', b=4, w=1920, h=100)
    .output('volume.mp4', vcodec='libx264')
    .overwrite_output()
    .run()
)

Advanced Video Effects

Stabilization

import subprocess

# Step 1: Analyze video and generate transform data
subprocess.run([
    'ffmpeg', '-i', 'shaky.mp4',
    '-vf', 'vidstabdetect=shakiness=10:accuracy=15',
    '-f', 'null', '-'
])

# Step 2: Apply stabilization
(
    ffmpeg
    .input('shaky.mp4')
    .filter('vidstabtransform', smoothing=30)
    .output('stable.mp4', vcodec='libx264', crf=18)
    .overwrite_output()
    .run()
)

# One-pass (lower quality but simpler)
(
    ffmpeg
    .input('shaky.mp4')
    .filter('deshake')
    .output('stable.mp4')
    .overwrite_output()
    .run()
)

Deinterlacing

# High-quality deinterlacing with yadif
(
    ffmpeg
    .input('interlaced.mp4')
    .filter('yadif', mode=0, parity=-1, deint=0)
    .output('progressive.mp4')
    .overwrite_output()
    .run()
)

# Even better quality with bwdif
(
    ffmpeg
    .input('interlaced.mp4')
    .filter('bwdif')
    .output('progressive.mp4')
    .overwrite_output()
    .run()
)

Noise Reduction

# Video noise reduction
(
    ffmpeg
    .input('noisy.mp4')
    .filter('nlmeans', s=3.0, p=7, pc=5, r=15)  # Non-local means
    .output('clean.mp4')
    .overwrite_output()
    .run()
)

# Faster but lower quality
(
    ffmpeg
    .input('noisy.mp4')
    .filter('hqdn3d', luma_spatial=4, chroma_spatial=3, luma_tmp=6, chroma_tmp=4.5)
    .output('clean.mp4')
    .overwrite_output()
    .run()
)

# Audio noise reduction
(
    ffmpeg
    .input('noisy.mp3')
    .filter('afftdn', nf=-25)  # FFT-based denoiser
    .output('clean.mp3')
    .overwrite_output()
    .run()
)

Green Screen / Chroma Key

# Remove green screen and overlay on background
foreground = ffmpeg.input('greenscreen.mp4')
background = ffmpeg.input('background.mp4')

# Remove green (adjust color values as needed)
keyed = foreground.filter(
    'chromakey',
    color='0x00ff00',  # Green
    similarity=0.1,
    blend=0.1
)

(
    ffmpeg
    .overlay(background, keyed)
    .output('composited.mp4')
    .overwrite_output()
    .run()
)

# Alternative with colorkey (better edge handling)
keyed = foreground.filter(
    'colorkey',
    color='green',
    similarity=0.3,
    blend=0.1
)

Motion Blur

# Add motion blur effect
(
    ffmpeg
    .input('input.mp4')
    .filter('minterpolate', fps=120, mi_mode='mci', mc_mode='aobmc', me_mode='bidir', vsbmc=1)
    .filter('tblend', all_mode='average', all_opacity=0.5)
    .filter('fps', fps=30)
    .output('motion_blur.mp4')
    .overwrite_output()
    .run()
)

Slow Motion with Frame Interpolation

# High-quality slow motion using motion interpolation
(
    ffmpeg
    .input('input.mp4')
    .filter('minterpolate', fps=120, mi_mode='mci', mc_mode='aobmc', me_mode='bidir')
    .filter('setpts', '4*PTS')  # 4x slower
    .filter('fps', fps=30)
    .output('slowmo.mp4')
    .overwrite_output()
    .run()
)

Time Lapse from Video

# Create time lapse (keep every 30th frame = 30x speed)
(
    ffmpeg
    .input('input.mp4')
    .filter('select', 'not(mod(n,30))')
    .filter('setpts', 'N/30/TB')  # Adjust timestamps
    .output('timelapse.mp4', r=30)
    .overwrite_output()
    .run()
)

Batch Processing

Process Multiple Files

from pathlib import Path
import ffmpeg
from concurrent.futures import ThreadPoolExecutor, as_completed

def convert_video(input_path: Path, output_dir: Path) -> bool:
    """Convert a single video file."""
    output_path = output_dir / f"{input_path.stem}.mp4"
    try:
        (
            ffmpeg
            .input(str(input_path))
            .output(str(output_path),
                vcodec='libx264',
                crf=23,
                preset='medium',
                acodec='aac',
                audio_bitrate='128k'
            )
            .overwrite_output()
            .run(quiet=True)
        )
        return True
    except ffmpeg.Error as e:
        print(f"Error processing {input_path}: {e.stderr.decode()}")
        return False

def batch_convert(input_dir: str, output_dir: str, max_workers: int = 4):
    """Batch convert all videos in a directory."""
    input_path = Path(input_dir)
    output_path = Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)

    video_files = list(input_path.glob('**/*.mov')) + \
                  list(input_path.glob('**/*.avi')) + \
                  list(input_path.glob('**/*.mkv'))

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(convert_video, f, output_path): f
            for f in video_files
        }

        for future in as_completed(futures):
            file = futures[future]
            if future.result():
                print(f"Converted: {file.name}")
            else:
                print(f"Failed: {file.name}")

# Usage
batch_convert('input_videos/', 'output_videos/')

Progress Reporting

import re
import subprocess
from pathlib import Path

def transcode_with_progress(input_path: str, output_path: str):
    """Transcode video with progress reporting."""
    # Get duration first
    probe = ffmpeg.probe(input_path)
    duration = float(probe['format']['duration'])

    process = subprocess.Popen(
        [
            'ffmpeg', '-y',
            '-i', input_path,
            '-c:v', 'libx264',
            '-crf', '23',
            '-c:a', 'aac',
            '-progress', 'pipe:1',  # Output progress to stdout
            output_path
        ],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        universal_newlines=True
    )

    time_pattern = re.compile(r'out_time_ms=(\d+)')

    for line in process.stdout:
        match = time_pattern.search(line)
        if match:
            time_ms = int(match.group(1))
            time_s = time_ms / 1_000_000
            progress = min(100, (time_s / duration) * 100)
            print(f"\rProgress: {progress:.1f}%", end='', flush=True)

    process.wait()
    print("\nDone!")

    if process.returncode != 0:
        raise RuntimeError(f"FFmpeg error: {process.stderr.read()}")

Live Streaming

Stream Desktop to RTMP

import subprocess
import platform

def stream_desktop(rtmp_url: str):
    """Stream desktop to RTMP server."""
    system = platform.system()

    if system == 'Windows':
        # Use gdigrab for Windows
        cmd = [
            'ffmpeg',
            '-f', 'gdigrab',
            '-framerate', '30',
            '-i', 'desktop',
            '-f', 'dshow',
            '-i', 'audio=Microphone',  # Adjust device name
            '-c:v', 'libx264',
            '-preset', 'ultrafast',
            '-tune', 'zerolatency',
            '-c:a', 'aac',
            '-b:a', '128k',
            '-f', 'flv',
            rtmp_url
        ]
    elif system == 'Linux':
        cmd = [
            'ffmpeg',
            '-f', 'x11grab',
            '-framerate', '30',
            '-s', '1920x1080',
            '-i', ':0.0',
            '-f', 'pulse',
            '-i', 'default',
            '-c:v', 'libx264',
            '-preset', 'ultrafast',
            '-tune', 'zerolatency',
            '-c:a', 'aac',
            '-b:a', '128k',
            '-f', 'flv',
            rtmp_url
        ]
    elif system == 'Darwin':
        cmd = [
            'ffmpeg',
            '-f', 'avfoundation',
            '-framerate', '30',
            '-i', '1:0',  # Screen:Audio device indices
            '-c:v', 'libx264',
            '-preset', 'ultrafast',
            '-tune', 'zerolatency',
            '-c:a', 'aac',
            '-b:a', '128k',
            '-f', 'flv',
            rtmp_url
        ]

    subprocess.run(cmd)

Webcam Streaming

def stream_webcam(rtmp_url: str, device: str = '0'):
    """Stream webcam to RTMP server."""
    import platform
    system = platform.system()

    if system == 'Windows':
        input_device = f'video={device}'
        input_format = 'dshow'
    elif system == 'Linux':
        input_device = f'/dev/video{device}'
        input_format = 'v4l2'
    elif system == 'Darwin':
        input_device = device
        input_format = 'avfoundation'

    cmd = [
        'ffmpeg',
        '-f', input_format,
        '-framerate', '30',
        '-video_size', '1280x720',
        '-i', input_device,
        '-c:v', 'libx264',
        '-preset', 'ultrafast',
        '-tune', 'zerolatency',
        '-b:v', '2500k',
        '-maxrate', '2500k',
        '-bufsize', '5000k',
        '-g', '60',  # Keyframe interval
        '-f', 'flv',
        rtmp_url
    ]

    subprocess.run(cmd)

Subtitle Processing

Burn Subtitles into Video

# From SRT file
(
    ffmpeg
    .input('video.mp4')
    .output('output.mp4',
        vf="subtitles=subtitles.srt:force_style='FontSize=24,FontName=Arial'"
    )
    .overwrite_output()
    .run()
)

# From ASS file (preserves styling)
(
    ffmpeg
    .input('video.mp4')
    .output('output.mp4',
        vf="ass=subtitles.ass"
    )
    .overwrite_output()
    .run()
)

# Embedded subtitles
(
    ffmpeg
    .input('video.mp4')
    .output('output.mp4',
        vf="subtitles=video.mp4:si=0"  # si = subtitle stream index
    )
    .overwrite_output()
    .run()
)

Extract Subtitles

# Extract to SRT
(
    ffmpeg
    .input('video.mp4')
    .output('subtitles.srt',
        map='0:s:0',  # First subtitle stream
        c='srt'
    )
    .overwrite_output()
    .run()
)

# List available subtitle streams
probe = ffmpeg.probe('video.mp4')
for stream in probe['streams']:
    if stream['codec_type'] == 'subtitle':
        print(f"Index: {stream['index']}, Codec: {stream['codec_name']}, "
              f"Language: {stream.get('tags', {}).get('language', 'unknown')}")

Add Subtitle Track

video = ffmpeg.input('video.mp4')
subs = ffmpeg.input('subtitles.srt')

(
    ffmpeg
    .output(video, subs, 'output.mkv',
        **{'c:v': 'copy', 'c:a': 'copy', 'c:s': 'srt'}
    )
    .overwrite_output()
    .run()
)

HDR and Color Processing

HDR to SDR Conversion (Tone Mapping)

(
    ffmpeg
    .input('hdr_video.mp4')
    .filter('zscale', t='linear', npl=100)
    .filter('format', 'gbrpf32le')
    .filter('zscale', p='bt709')
    .filter('tonemap', tonemap='hable', desat=0)
    .filter('zscale', t='bt709', m='bt709', r='tv')
    .filter('format', 'yuv420p')
    .output('sdr_video.mp4', vcodec='libx264', crf=18)
    .overwrite_output()
    .run()
)

Color Space Conversion

# Convert to BT.709 (standard HD)
(
    ffmpeg
    .input('input.mp4')
    .filter('colorspace', all='bt709')
    .output('output.mp4')
    .overwrite_output()
    .run()
)

# Convert to BT.2020 (HDR/UHD)
(
    ffmpeg
    .input('input.mp4')
    .filter('colorspace', all='bt2020')
    .output('output.mp4')
    .overwrite_output()
    .run()
)

LUT Application

# Apply 3D LUT for color grading
(
    ffmpeg
    .input('input.mp4')
    .filter('lut3d', file='color_grade.cube')
    .output('graded.mp4')
    .overwrite_output()
    .run()
)

GPU Processing with CUDA

CUDA-Accelerated Scaling

# Requires FFmpeg compiled with --enable-cuda-nvcc
(
    ffmpeg
    .input('input.mp4', hwaccel='cuda')
    .filter('scale_cuda', 1920, 1080)
    .output('output.mp4',
        vcodec='h264_nvenc',
        preset='p4'
    )
    .overwrite_output()
    .run()
)

Full GPU Pipeline

(
    ffmpeg
    .input('input.mp4',
        hwaccel='cuda',
        hwaccel_output_format='cuda'
    )
    .filter('scale_cuda', 1920, 1080)
    .filter('yadif_cuda')  # GPU deinterlacing
    .output('output.mp4',
        vcodec='h264_nvenc',
        preset='p4',
        cq=23
    )
    .overwrite_output()
    .run()
)

Error Recovery and Repair

Fix Broken Video

# Re-mux without re-encoding to fix container issues
(
    ffmpeg
    .input('broken.mp4')
    .output('fixed.mp4', c='copy')
    .overwrite_output()
    .run()
)

# Force keyframe at start
(
    ffmpeg
    .input('broken.mp4')
    .output('fixed.mp4',
        vcodec='libx264',
        crf=18,
        force_key_frames='expr:gte(t,0)'
    )
    .overwrite_output()
    .run()
)

# Recover from corrupt file (skip errors)
subprocess.run([
    'ffmpeg',
    '-err_detect', 'ignore_err',
    '-i', 'corrupt.mp4',
    '-c', 'copy',
    'recovered.mp4'
])

Validate Video File

def validate_video(filepath: str) -> dict:
    """Validate video file integrity."""
    try:
        # Quick probe
        probe = ffmpeg.probe(filepath)

        # Full decode test (slower but thorough)
        process = subprocess.run(
            ['ffmpeg', '-v', 'error', '-i', filepath, '-f', 'null', '-'],
            capture_output=True,
            text=True
        )

        return {
            'valid': process.returncode == 0,
            'duration': float(probe['format'].get('duration', 0)),
            'streams': len(probe['streams']),
            'errors': process.stderr if process.stderr else None
        }
    except ffmpeg.Error as e:
        return {
            'valid': False,
            'error': str(e)
        }

Integration with OpenCV

Read Video Frames with FFmpeg, Process with OpenCV

import cv2
import numpy as np
import subprocess

def read_with_ffmpeg(filepath: str, width: int, height: int):
    """Read video frames using FFmpeg subprocess (faster than cv2.VideoCapture for some formats)."""
    process = subprocess.Popen(
        [
            'ffmpeg',
            '-i', filepath,
            '-f', 'rawvideo',
            '-pix_fmt', 'bgr24',  # OpenCV uses BGR
            '-s', f'{width}x{height}',
            'pipe:1'
        ],
        stdout=subprocess.PIPE,
        stderr=subprocess.DEVNULL
    )

    frame_size = width * height * 3

    while True:
        raw_frame = process.stdout.read(frame_size)
        if len(raw_frame) != frame_size:
            break

        frame = np.frombuffer(raw_frame, np.uint8).reshape((height, width, 3))
        yield frame

    process.terminate()

# Usage
for frame in read_with_ffmpeg('video.mp4', 1920, 1080):
    # Process with OpenCV
    gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
    # ...

Write Processed Frames with FFmpeg

import cv2
import numpy as np
import subprocess

def write_with_ffmpeg(output_path: str, width: int, height: int, fps: int = 30):
    """Create a writer context that accepts OpenCV frames."""
    process = subprocess.Popen(
        [
            'ffmpeg',
            '-y',
            '-f', 'rawvideo',
            '-vcodec', 'rawvideo',
            '-s', f'{width}x{height}',
            '-pix_fmt', 'bgr24',
            '-r', str(fps),
            '-i', 'pipe:0',
            '-c:v', 'libx264',
            '-pix_fmt', 'yuv420p',
            '-crf', '18',
            output_path
        ],
        stdin=subprocess.PIPE,
        stderr=subprocess.PIPE
    )
    return process

# Usage
cap = cv2.VideoCapture('input.mp4')
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
fps = int(cap.get(cv2.CAP_PROP_FPS))

writer = write_with_ffmpeg('output.mp4', width, height, fps)

try:
    while cap.isOpened():
        ret, frame = cap.read()
        if not ret:
            break

        # Process frame with OpenCV
        processed = cv2.GaussianBlur(frame, (5, 5), 0)

        # Write to FFmpeg
        writer.stdin.write(processed.tobytes())
finally:
    cap.release()
    writer.stdin.close()
    writer.wait()

Performance Optimization

Memory-Efficient Processing

import ffmpeg
import tempfile
from pathlib import Path

def process_large_video(input_path: str, output_path: str, chunk_duration: int = 60):
    """Process large video in chunks to manage memory."""
    probe = ffmpeg.probe(input_path)
    duration = float(probe['format']['duration'])

    chunks = []
    temp_dir = Path(tempfile.mkdtemp())

    try:
        # Process in chunks
        for start in range(0, int(duration), chunk_duration):
            chunk_path = temp_dir / f"chunk_{start:06d}.mp4"
            (
                ffmpeg
                .input(input_path, ss=start, t=chunk_duration)
                .filter('scale', 1280, 720)  # Your processing here
                .output(str(chunk_path), vcodec='libx264', crf=23)
                .overwrite_output()
                .run(quiet=True)
            )
            chunks.append(chunk_path)

        # Concatenate chunks
        with open(temp_dir / 'list.txt', 'w') as f:
            for chunk in chunks:
                f.write(f"file '{chunk}'\n")

        (
            ffmpeg
            .input(str(temp_dir / 'list.txt'), f='concat', safe=0)
            .output(output_path, c='copy')
            .overwrite_output()
            .run()
        )
    finally:
        # Cleanup
        import shutil
        shutil.rmtree(temp_dir)

Parallel Encoding

from concurrent.futures import ThreadPoolExecutor
import ffmpeg

def encode_quality(input_path: str, output_path: str, height: int, crf: int):
    """Encode a single quality level."""
    (
        ffmpeg
        .input(input_path)
        .filter('scale', -2, height)
        .output(output_path, vcodec='libx264', crf=crf, preset='slow')
        .overwrite_output()
        .run(quiet=True)
    )

def create_quality_ladder(input_path: str):
    """Create multiple quality versions in parallel."""
    qualities = [
        ('1080p.mp4', 1080, 18),
        ('720p.mp4', 720, 23),
        ('480p.mp4', 480, 26),
        ('360p.mp4', 360, 28),
    ]

    with ThreadPoolExecutor(max_workers=4) as executor:
        futures = [
            executor.submit(encode_quality, input_path, output, height, crf)
            for output, height, crf in qualities
        ]

        for future in futures:
            future.result()  # Wait for completion

create_quality_ladder('input.mp4')

Additional Resources

• FFmpeg Filters Documentation
• FFmpeg Codecs
• ffmpeg-python Source
• PyAV Cookbook
• FFmpeg Wiki - Encoding Guide

Python FFmpeg Complete Recipes

This reference preserves the detailed guide content extracted from SKILL.md so the top-level skill can remain a lean orchestrator while retaining all examples, tables, recipes, and troubleshooting guidance.

---

Python FFmpeg Skill

Comprehensive guide to using FFmpeg with Python for video/audio processing, encoding, streaming, and media manipulation.

Quick Reference

Installation

# System FFmpeg (REQUIRED for ffmpeg-python and subprocess)
# Windows: winget install ffmpeg
# macOS: brew install ffmpeg
# Ubuntu: sudo apt install ffmpeg

# Python libraries
uv add ffmpeg-python  # Wrapper library
uv add av             # PyAV - Cython bindings
uv add moviepy        # High-level editing

Critical Gotchas

1. Audio Stream Loss (MOST COMMON BUG)

import ffmpeg

# BAD: Video filters can drop audio!
(
    ffmpeg
    .input('input.mp4')
    .filter('scale', 1280, 720)
    .output('output.mp4')
    .run()
)
# Result: Video only, NO AUDIO!

# GOOD: Explicitly handle audio stream
input_file = ffmpeg.input('input.mp4')
video = input_file.video.filter('scale', 1280, 720)
audio = input_file.audio
(
    ffmpeg
    .output(video, audio, 'output.mp4')
    .run()
)

# ALTERNATIVE: Copy audio stream
(
    ffmpeg
    .input('input.mp4')
    .output('output.mp4', vf='scale=1280:720', acodec='copy')
    .run()
)

2. Subprocess Deadlock

import subprocess

# BAD: Can deadlock with large outputs!
process = subprocess.Popen(
    ['ffmpeg', '-i', 'input.mp4', '-f', 'rawvideo', 'pipe:1'],
    stdout=subprocess.PIPE
)
output = process.stdout.read()  # DEADLOCK if output > buffer size

# GOOD: Use communicate() for small outputs
process = subprocess.Popen(
    ['ffmpeg', '-i', 'input.mp4', '-f', 'null', '-'],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE
)
stdout, stderr = process.communicate()

# GOOD: Stream processing for large outputs
process = subprocess.Popen(
    ['ffmpeg', '-i', 'input.mp4', '-f', 'rawvideo', '-pix_fmt', 'rgb24', 'pipe:1'],
    stdout=subprocess.PIPE,
    stderr=subprocess.DEVNULL
)
while True:
    # Read frame by frame (width * height * 3 bytes for RGB24)
    frame_data = process.stdout.read(1920 * 1080 * 3)
    if not frame_data:
        break
    # Process frame...

3. Overwrite Without Prompt

# BAD: FFmpeg prompts for overwrite, blocking script
ffmpeg.input('input.mp4').output('output.mp4').run()

# GOOD: Always use overwrite_output()
ffmpeg.input('input.mp4').output('output.mp4').overwrite_output().run()

# Or with subprocess
subprocess.run(['ffmpeg', '-y', '-i', 'input.mp4', 'output.mp4'])

4. Error Handling

import ffmpeg

# BAD: No error handling
ffmpeg.input('input.mp4').output('output.mp4').run()

# GOOD: Catch and log errors
try:
    ffmpeg.input('input.mp4').output('output.mp4').overwrite_output().run(
        capture_stdout=True,
        capture_stderr=True
    )
except ffmpeg.Error as e:
    print(f"FFmpeg error: {e.stderr.decode()}")
    raise

# Check if file exists first
from pathlib import Path
if not Path('input.mp4').exists():
    raise FileNotFoundError("Input file not found")

5. Path Handling on Windows

from pathlib import Path

# BAD: Backslashes can cause issues
path = "C:\\Users\\video.mp4"

# GOOD: Use forward slashes or Path
path = "C:/Users/video.mp4"
# Or
path = str(Path("C:\\Users\\video.mp4").as_posix())

# GOOD: Quote paths with spaces
ffmpeg.input('my video.mp4')  # ffmpeg-python handles quoting
subprocess.run(['ffmpeg', '-i', 'my video.mp4', 'output.mp4'])  # List handles quoting

Video Encoding

H.264 (Best Compatibility)

import ffmpeg

# CRF encoding (recommended for quality)
# CRF 0 = lossless, 23 = default, 51 = worst
# Lower = better quality, larger file
(
    ffmpeg
    .input('input.mp4')
    .output('output.mp4',
        vcodec='libx264',
        crf=23,           # Quality (18-28 typical range)
        preset='medium',  # Speed vs compression
        acodec='aac',
        audio_bitrate='192k'
    )
    .overwrite_output()
    .run()
)

# Presets (fastest to slowest, worst to best compression):
# ultrafast, superfast, veryfast, faster, fast, medium, slow, slower, veryslow, placebo
# Use 'medium' or 'slow' for best quality/time tradeoff

# Two-pass encoding for target file size
import os

# Calculate bitrate for target size (in kbps)
duration_seconds = 120
target_size_mb = 50
video_bitrate = (target_size_mb * 8192 / duration_seconds) - 192  # Subtract audio bitrate

# Pass 1
(
    ffmpeg
    .input('input.mp4')
    .output('NUL' if os.name == 'nt' else '/dev/null',
        vcodec='libx264',
        video_bitrate=f'{int(video_bitrate)}k',
        preset='slow',
        **{'pass': 1},
        f='null'
    )
    .overwrite_output()
    .run()
)

# Pass 2
(
    ffmpeg
    .input('input.mp4')
    .output('output.mp4',
        vcodec='libx264',
        video_bitrate=f'{int(video_bitrate)}k',
        preset='slow',
        **{'pass': 2},
        acodec='aac',
        audio_bitrate='192k'
    )
    .overwrite_output()
    .run()
)

H.265/HEVC (Better Compression)

# CRF 28 for H.265 ≈ CRF 23 for H.264
(
    ffmpeg
    .input('input.mp4')
    .output('output.mp4',
        vcodec='libx265',
        crf=28,
        preset='medium',
        acodec='aac',
        audio_bitrate='128k',
        **{'tag:v': 'hvc1'}  # For Apple compatibility
    )
    .overwrite_output()
    .run()
)

VP9/WebM (Web Streaming)

(
    ffmpeg
    .input('input.mp4')
    .output('output.webm',
        vcodec='libvpx-vp9',
        crf=30,
        video_bitrate='0',  # Required for CRF mode
        acodec='libopus',
        audio_bitrate='128k'
    )
    .overwrite_output()
    .run()
)

AV1 (Best Compression, Slowest)

(
    ffmpeg
    .input('input.mp4')
    .output('output.mp4',
        vcodec='libaom-av1',
        crf=30,
        cpu_used=4,  # 0-8, higher = faster but lower quality
        acodec='libopus'
    )
    .overwrite_output()
    .run()
)

Hardware Acceleration

NVIDIA NVENC

# Check if NVENC is available
import subprocess
result = subprocess.run(
    ['ffmpeg', '-encoders'],
    capture_output=True,
    text=True
)
nvenc_available = 'h264_nvenc' in result.stdout

if nvenc_available:
    (
        ffmpeg
        .input('input.mp4')
        .output('output.mp4',
            vcodec='h264_nvenc',
            preset='p4',  # p1 (fastest) to p7 (best quality)
            cq=23,        # Constant quality (like CRF)
            acodec='copy'
        )
        .overwrite_output()
        .run()
    )

Intel Quick Sync (QSV)

(
    ffmpeg
    .input('input.mp4', hwaccel='qsv')
    .output('output.mp4',
        vcodec='h264_qsv',
        preset='medium',
        global_quality=23,
        acodec='copy'
    )
    .overwrite_output()
    .run()
)

AMD AMF (Windows) / VAAPI (Linux)

# Linux VAAPI
(
    ffmpeg
    .input('input.mp4', hwaccel='vaapi', hwaccel_device='/dev/dri/renderD128')
    .output('output.mp4',
        vcodec='h264_vaapi',
        acodec='copy'
    )
    .overwrite_output()
    .run()
)

Audio Processing

Extract Audio

# Extract to MP3
(
    ffmpeg
    .input('video.mp4')
    .output('audio.mp3',
        acodec='libmp3lame',
        audio_bitrate='320k',
        vn=None  # No video
    )
    .overwrite_output()
    .run()
)

# Extract to WAV (lossless)
(
    ffmpeg
    .input('video.mp4')
    .output('audio.wav',
        acodec='pcm_s16le',
        ar=44100,  # Sample rate
        ac=2,      # Channels (stereo)
        vn=None
    )
    .overwrite_output()
    .run()
)

# Extract to FLAC (lossless compressed)
(
    ffmpeg
    .input('video.mp4')
    .output('audio.flac',
        acodec='flac',
        vn=None
    )
    .overwrite_output()
    .run()
)

Audio Conversion

# MP3 with variable bitrate (recommended)
(
    ffmpeg
    .input('audio.wav')
    .output('audio.mp3',
        acodec='libmp3lame',
        q=2  # VBR quality 0-9 (0 = best, ~245kbps; 2 ≈ 190kbps)
    )
    .overwrite_output()
    .run()
)

# AAC (best for video)
(
    ffmpeg
    .input('audio.wav')
    .output('audio.m4a',
        acodec='aac',
        audio_bitrate='256k'
    )
    .overwrite_output()
    .run()
)

# Opus (best quality per bitrate)
(
    ffmpeg
    .input('audio.wav')
    .output('audio.opus',
        acodec='libopus',
        audio_bitrate='128k'
    )
    .overwrite_output()
    .run()
)

Audio Filters

# Volume adjustment
(
    ffmpeg
    .input('audio.mp3')
    .filter('volume', 1.5)  # 1.5x volume
    .output('louder.mp3')
    .overwrite_output()
    .run()
)

# Normalize audio (loudnorm)
(
    ffmpeg
    .input('audio.mp3')
    .filter('loudnorm', I=-16, TP=-1.5, LRA=11)
    .output('normalized.mp3')
    .overwrite_output()
    .run()
)

# Fade in/out
(
    ffmpeg
    .input('audio.mp3')
    .filter('afade', type='in', duration=3)
    .filter('afade', type='out', start_time=57, duration=3)
    .output('faded.mp3')
    .overwrite_output()
    .run()
)

# Resample
(
    ffmpeg
    .input('audio.mp3')
    .filter('aresample', 48000)
    .output('resampled.mp3')
    .overwrite_output()
    .run()
)

Video Filters

Scaling/Resizing

# Scale to specific size
(
    ffmpeg
    .input('input.mp4')
    .filter('scale', 1920, 1080)
    .output('output.mp4', acodec='copy')
    .overwrite_output()
    .run()
)

# Scale maintaining aspect ratio
(
    ffmpeg
    .input('input.mp4')
    .filter('scale', 1280, -1)  # -1 = auto-calculate
    .output('output.mp4', acodec='copy')
    .overwrite_output()
    .run()
)

# Scale with padding (letterbox/pillarbox)
(
    ffmpeg
    .input('input.mp4')
    .filter('scale', 1920, 1080, force_original_aspect_ratio='decrease')
    .filter('pad', 1920, 1080, '(ow-iw)/2', '(oh-ih)/2')
    .output('output.mp4', acodec='copy')
    .overwrite_output()
    .run()
)

# Scale algorithms
# fast_bilinear, bilinear, bicubic, lanczos (best for downscaling)
(
    ffmpeg
    .input('input.mp4')
    .filter('scale', 1280, 720, flags='lanczos')
    .output('output.mp4', acodec='copy')
    .overwrite_output()
    .run()
)

Cropping

# Crop to size (width:height:x:y)
(
    ffmpeg
    .input('input.mp4')
    .filter('crop', 1280, 720, 100, 50)  # 1280x720 starting at (100, 50)
    .output('output.mp4', acodec='copy')
    .overwrite_output()
    .run()
)

# Center crop
(
    ffmpeg
    .input('input.mp4')
    .filter('crop', 1280, 720, '(in_w-1280)/2', '(in_h-720)/2')
    .output('output.mp4', acodec='copy')
    .overwrite_output()
    .run()
)

# Crop to aspect ratio (16:9)
(
    ffmpeg
    .input('input.mp4')
    .filter('crop', 'min(iw,ih*16/9)', 'min(ih,iw*9/16)')
    .output('output.mp4', acodec='copy')
    .overwrite_output()
    .run()
)

Rotation

# Rotate 90 degrees clockwise
(
    ffmpeg
    .input('input.mp4')
    .filter('transpose', 1)  # 0=ccw+vflip, 1=cw, 2=ccw, 3=cw+vflip
    .output('output.mp4', acodec='copy')
    .overwrite_output()
    .run()
)

# Rotate arbitrary angle
(
    ffmpeg
    .input('input.mp4')
    .filter('rotate', 'PI/6')  # 30 degrees in radians
    .output('output.mp4', acodec='copy')
    .overwrite_output()
    .run()
)

# Horizontal flip
(
    ffmpeg
    .input('input.mp4')
    .filter('hflip')
    .output('output.mp4', acodec='copy')
    .overwrite_output()
    .run()
)

# Vertical flip
(
    ffmpeg
    .input('input.mp4')
    .filter('vflip')
    .output('output.mp4', acodec='copy')
    .overwrite_output()
    .run()
)

Text Overlay

# Simple text
(
    ffmpeg
    .input('input.mp4')
    .filter('drawtext',
        text='Hello World',
        fontsize=48,
        fontcolor='white',
        x='(w-text_w)/2',  # Center horizontally
        y='h-th-20'        # Bottom with padding
    )
    .output('output.mp4', acodec='copy')
    .overwrite_output()
    .run()
)

# Text with background box
(
    ffmpeg
    .input('input.mp4')
    .filter('drawtext',
        text='Watermark',
        fontsize=24,
        fontcolor='white',
        box=1,
        boxcolor='black@0.5',
        boxborderw=5,
        x=10,
        y=10
    )
    .output('output.mp4', acodec='copy')
    .overwrite_output()
    .run()
)

# Timestamp
(
    ffmpeg
    .input('input.mp4')
    .filter('drawtext',
        text='%{pts\\:hms}',  # Escape colons
        fontsize=24,
        fontcolor='white',
        x=10,
        y=10
    )
    .output('output.mp4', acodec='copy')
    .overwrite_output()
    .run()
)

# Custom font (specify path)
(
    ffmpeg
    .input('input.mp4')
    .filter('drawtext',
        text='Custom Font',
        fontfile='/path/to/font.ttf',
        fontsize=48,
        fontcolor='yellow'
    )
    .output('output.mp4', acodec='copy')
    .overwrite_output()
    .run()
)

Image Overlay (Watermark)

# Overlay image in corner
main = ffmpeg.input('video.mp4')
logo = ffmpeg.input('logo.png')

(
    ffmpeg
    .overlay(main, logo, x=10, y=10)  # Top-left
    .output('output.mp4', acodec='copy')
    .overwrite_output()
    .run()
)

# Overlay with scaling
main = ffmpeg.input('video.mp4')
logo = ffmpeg.input('logo.png').filter('scale', 100, -1)

(
    ffmpeg
    .overlay(main, logo, x='main_w-overlay_w-10', y=10)  # Top-right
    .output('output.mp4', acodec='copy')
    .overwrite_output()
    .run()
)

# Semi-transparent overlay
logo = ffmpeg.input('logo.png').filter('format', 'rgba').filter('colorchannelmixer', aa=0.5)

Color Adjustments

# Brightness, contrast, saturation
(
    ffmpeg
    .input('input.mp4')
    .filter('eq',
        brightness=0.1,    # -1 to 1, default 0
        contrast=1.2,      # 0 to 2, default 1
        saturation=1.3     # 0 to 3, default 1
    )
    .output('output.mp4', acodec='copy')
    .overwrite_output()
    .run()
)

# Gamma correction
(
    ffmpeg
    .input('input.mp4')
    .filter('eq', gamma=1.5)
    .output('output.mp4', acodec='copy')
    .overwrite_output()
    .run()
)

# Grayscale
(
    ffmpeg
    .input('input.mp4')
    .filter('format', 'gray')
    .output('output.mp4', acodec='copy')
    .overwrite_output()
    .run()
)

# Color balance
(
    ffmpeg
    .input('input.mp4')
    .filter('colorbalance',
        rs=0.1,  # Red shadows
        gm=0.1,  # Green midtones
        bh=0.1   # Blue highlights
    )
    .output('output.mp4', acodec='copy')
    .overwrite_output()
    .run()
)

Trimming and Concatenation

Trimming

# Trim by time
(
    ffmpeg
    .input('input.mp4', ss=10, t=30)  # Start at 10s, duration 30s
    .output('output.mp4', c='copy')   # Stream copy (fast, no re-encode)
    .overwrite_output()
    .run()
)

# Trim to end time
(
    ffmpeg
    .input('input.mp4', ss=10, to=40)  # Start at 10s, end at 40s
    .output('output.mp4', c='copy')
    .overwrite_output()
    .run()
)

# Accurate trimming (re-encode, slower but precise)
(
    ffmpeg
    .input('input.mp4')
    .trim(start=10, end=40)
    .setpts('PTS-STARTPTS')  # Reset timestamps
    .output('output.mp4', vcodec='libx264', crf=18)
    .overwrite_output()
    .run()
)

# Trim with audio (both streams)
input_file = ffmpeg.input('input.mp4')
video = input_file.video.trim(start=10, end=40).setpts('PTS-STARTPTS')
audio = input_file.audio.filter('atrim', start=10, end=40).filter('asetpts', 'PTS-STARTPTS')
(
    ffmpeg
    .output(video, audio, 'output.mp4')
    .overwrite_output()
    .run()
)

Concatenation

# Method 1: Concat demuxer (same codec, fast)
# Create file list
with open('filelist.txt', 'w') as f:
    f.write("file 'video1.mp4'\n")
    f.write("file 'video2.mp4'\n")
    f.write("file 'video3.mp4'\n")

(
    ffmpeg
    .input('filelist.txt', f='concat', safe=0)
    .output('output.mp4', c='copy')
    .overwrite_output()
    .run()
)

# Method 2: Concat filter (different codecs, re-encodes)
video1 = ffmpeg.input('video1.mp4')
video2 = ffmpeg.input('video2.mp4')

(
    ffmpeg
    .concat(video1, video2, v=1, a=1)  # v=video streams, a=audio streams
    .output('output.mp4')
    .overwrite_output()
    .run()
)

# Concatenate multiple videos
videos = [ffmpeg.input(f'video{i}.mp4') for i in range(1, 5)]
(
    ffmpeg
    .concat(*videos, v=1, a=1)
    .output('output.mp4')
    .overwrite_output()
    .run()
)

Streaming

HLS (HTTP Live Streaming)

# Create HLS stream
(
    ffmpeg
    .input('input.mp4')
    .output('stream.m3u8',
        vcodec='libx264',
        crf=23,
        preset='fast',
        acodec='aac',
        audio_bitrate='128k',
        f='hls',
        hls_time=10,           # Segment duration
        hls_list_size=0,       # Keep all segments
        hls_segment_filename='segment_%03d.ts'
    )
    .overwrite_output()
    .run()
)

# Multi-bitrate HLS (adaptive streaming)
input_file = ffmpeg.input('input.mp4')

# Create multiple quality streams
streams = []
for height, bitrate in [(1080, '5000k'), (720, '2500k'), (480, '1000k')]:
    stream = (
        input_file
        .filter('scale', -2, height)
        .output(f'stream_{height}p.m3u8',
            vcodec='libx264',
            video_bitrate=bitrate,
            preset='fast',
            acodec='aac',
            f='hls',
            hls_time=10,
            hls_segment_filename=f'segment_{height}p_%03d.ts'
        )
    )
    streams.append(stream)

# Create master playlist manually

DASH (Dynamic Adaptive Streaming)

(
    ffmpeg
    .input('input.mp4')
    .output('manifest.mpd',
        vcodec='libx264',
        acodec='aac',
        f='dash',
        seg_duration=10,
        adaptation_sets='id=0,streams=v id=1,streams=a'
    )
    .overwrite_output()
    .run()
)

RTMP Streaming

# Stream to RTMP server
(
    ffmpeg
    .input('input.mp4', re=None)  # Read at native rate
    .output('rtmp://server/live/stream',
        vcodec='libx264',
        preset='ultrafast',
        tune='zerolatency',
        acodec='aac',
        f='flv'
    )
    .overwrite_output()
    .run()
)

# Stream from webcam
(
    ffmpeg
    .input('0', f='dshow', i='video="Webcam Name"')  # Windows
    # .input('/dev/video0', f='v4l2')  # Linux
    .output('rtmp://server/live/stream',
        vcodec='libx264',
        preset='ultrafast',
        tune='zerolatency',
        f='flv'
    )
    .run()
)

Metadata and Probing

Get Video Information

import ffmpeg
import json

# Probe file
probe = ffmpeg.probe('video.mp4')

# Get format info
format_info = probe['format']
duration = float(format_info['duration'])
size_bytes = int(format_info['size'])
bitrate = int(format_info['bit_rate'])

print(f"Duration: {duration:.2f}s")
print(f"Size: {size_bytes / 1024 / 1024:.2f}MB")
print(f"Bitrate: {bitrate / 1000:.0f}kbps")

# Get video stream info
video_stream = next(
    (s for s in probe['streams'] if s['codec_type'] == 'video'),
    None
)
if video_stream:
    width = video_stream['width']
    height = video_stream['height']
    fps = eval(video_stream['r_frame_rate'])  # e.g., "30/1"
    codec = video_stream['codec_name']
    print(f"Resolution: {width}x{height}")
    print(f"FPS: {fps:.2f}")
    print(f"Codec: {codec}")

# Get audio stream info
audio_stream = next(
    (s for s in probe['streams'] if s['codec_type'] == 'audio'),
    None
)
if audio_stream:
    sample_rate = audio_stream['sample_rate']
    channels = audio_stream['channels']
    audio_codec = audio_stream['codec_name']
    print(f"Audio: {audio_codec}, {sample_rate}Hz, {channels}ch")

Using ffprobe Directly

import subprocess
import json

def get_video_info(filepath: str) -> dict:
    """Get detailed video information using ffprobe."""
    cmd = [
        'ffprobe',
        '-v', 'quiet',
        '-print_format', 'json',
        '-show_format',
        '-show_streams',
        filepath
    ]
    result = subprocess.run(cmd, capture_output=True, text=True)
    return json.loads(result.stdout)

# Get specific field
def get_duration(filepath: str) -> float:
    """Get video duration in seconds."""
    cmd = [
        'ffprobe',
        '-v', 'error',
        '-show_entries', 'format=duration',
        '-of', 'default=noprint_wrappers=1:nokey=1',
        filepath
    ]
    result = subprocess.run(cmd, capture_output=True, text=True)
    return float(result.stdout.strip())

# Get frame count
def get_frame_count(filepath: str) -> int:
    """Get total frame count."""
    cmd = [
        'ffprobe',
        '-v', 'error',
        '-select_streams', 'v:0',
        '-count_frames',
        '-show_entries', 'stream=nb_read_frames',
        '-of', 'default=noprint_wrappers=1:nokey=1',
        filepath
    ]
    result = subprocess.run(cmd, capture_output=True, text=True)
    return int(result.stdout.strip())

Set Metadata

# Set video metadata
(
    ffmpeg
    .input('input.mp4')
    .output('output.mp4',
        c='copy',
        **{
            'metadata': 'title=My Video',
            'metadata:s:v:0': 'title=Video Track',
            'metadata:s:a:0': 'title=Audio Track',
        }
    )
    .overwrite_output()
    .run()
)

# Copy all metadata from another file
(
    ffmpeg
    .input('input.mp4')
    .output('output.mp4',
        c='copy',
        map_metadata=0
    )
    .overwrite_output()
    .run()
)

# Strip all metadata
(
    ffmpeg
    .input('input.mp4')
    .output('output.mp4',
        c='copy',
        map_metadata=-1
    )
    .overwrite_output()
    .run()
)

Thumbnail Generation

Single Thumbnail

# Extract frame at specific time
(
    ffmpeg
    .input('video.mp4', ss=10)  # Seek to 10 seconds
    .output('thumbnail.jpg',
        vframes=1,
        q=2  # Quality 1-31 (lower is better)
    )
    .overwrite_output()
    .run()
)

# Extract at percentage
import ffmpeg

probe = ffmpeg.probe('video.mp4')
duration = float(probe['format']['duration'])
timestamp = duration * 0.25  # 25% into video

(
    ffmpeg
    .input('video.mp4', ss=timestamp)
    .output('thumbnail.jpg', vframes=1)
    .overwrite_output()
    .run()
)

Multiple Thumbnails

# Extract every N seconds
(
    ffmpeg
    .input('video.mp4')
    .filter('fps', fps=1/10)  # 1 frame every 10 seconds
    .output('thumb_%04d.jpg', q=2)
    .overwrite_output()
    .run()
)

# Extract specific number of thumbnails
probe = ffmpeg.probe('video.mp4')
duration = float(probe['format']['duration'])
num_thumbnails = 10

(
    ffmpeg
    .input('video.mp4')
    .filter('fps', fps=num_thumbnails/duration)
    .output('thumb_%04d.jpg', q=2)
    .overwrite_output()
    .run()
)

# Create thumbnail grid/sprite sheet
(
    ffmpeg
    .input('video.mp4')
    .filter('fps', fps=1/5)
    .filter('scale', 160, -1)
    .filter('tile', '5x5')
    .output('sprite_%04d.jpg')
    .overwrite_output()
    .run()
)

PyAV (High Performance)

PyAV provides direct access to FFmpeg's libraries via Cython bindings, offering better performance for frame-level operations.

import av
import numpy as np

# Read video frames
def read_frames(filepath: str):
    """Generator yielding video frames as numpy arrays."""
    container = av.open(filepath)
    for frame in container.decode(video=0):
        yield frame.to_ndarray(format='rgb24')

# Process video frame by frame
with av.open('input.mp4') as input_container:
    with av.open('output.mp4', 'w') as output_container:
        # Create output stream matching input
        input_stream = input_container.streams.video[0]
        output_stream = output_container.add_stream('libx264', rate=input_stream.rate)
        output_stream.width = input_stream.width
        output_stream.height = input_stream.height
        output_stream.pix_fmt = 'yuv420p'

        for frame in input_container.decode(video=0):
            # Convert to numpy, process, convert back
            img = frame.to_ndarray(format='rgb24')
            # Process image...
            processed = av.VideoFrame.from_ndarray(img, format='rgb24')
            processed.pts = frame.pts

            # Encode and write
            for packet in output_stream.encode(processed):
                output_container.mux(packet)

        # Flush encoder
        for packet in output_stream.encode():
            output_container.mux(packet)

# Extract audio samples
with av.open('video.mp4') as container:
    audio_stream = container.streams.audio[0]
    for frame in container.decode(audio=0):
        # frame.to_ndarray() returns audio samples
        samples = frame.to_ndarray()
        # Shape: (channels, samples)

Subprocess Patterns

For full FFmpeg control when libraries don't support specific features.

import subprocess
from pathlib import Path
from typing import Optional

def run_ffmpeg(
    input_path: str,
    output_path: str,
    options: list[str],
    timeout: Optional[int] = None
) -> tuple[int, str, str]:
    """Run FFmpeg with proper error handling."""
    cmd = [
        'ffmpeg',
        '-y',  # Overwrite output
        '-i', input_path,
        *options,
        output_path
    ]

    result = subprocess.run(
        cmd,
        capture_output=True,
        text=True,
        timeout=timeout
    )

    return result.returncode, result.stdout, result.stderr

# Example usage
returncode, stdout, stderr = run_ffmpeg(
    'input.mp4',
    'output.mp4',
    ['-c:v', 'libx264', '-crf', '23', '-c:a', 'aac']
)

if returncode != 0:
    print(f"Error: {stderr}")

# Pipe frames to FFmpeg
def write_frames_to_video(
    frames: list[np.ndarray],
    output_path: str,
    fps: int = 30,
    width: int = 1920,
    height: int = 1080
):
    """Write numpy array frames to video file."""
    process = subprocess.Popen(
        [
            'ffmpeg',
            '-y',
            '-f', 'rawvideo',
            '-vcodec', 'rawvideo',
            '-s', f'{width}x{height}',
            '-pix_fmt', 'rgb24',
            '-r', str(fps),
            '-i', 'pipe:0',
            '-c:v', 'libx264',
            '-pix_fmt', 'yuv420p',
            '-crf', '18',
            output_path
        ],
        stdin=subprocess.PIPE,
        stderr=subprocess.PIPE
    )

    for frame in frames:
        process.stdin.write(frame.tobytes())

    process.stdin.close()
    process.wait()

    if process.returncode != 0:
        raise RuntimeError(f"FFmpeg error: {process.stderr.read().decode()}")

Common Patterns

Video to GIF

# Basic GIF
(
    ffmpeg
    .input('video.mp4', ss=0, t=5)
    .filter('fps', fps=10)
    .filter('scale', 480, -1)
    .output('output.gif')
    .overwrite_output()
    .run()
)

# High quality GIF with palette
# Step 1: Generate palette
(
    ffmpeg
    .input('video.mp4', ss=0, t=5)
    .filter('fps', fps=10)
    .filter('scale', 480, -1, flags='lanczos')
    .filter('palettegen')
    .output('palette.png')
    .overwrite_output()
    .run()
)

# Step 2: Use palette
input_video = ffmpeg.input('video.mp4', ss=0, t=5)
palette = ffmpeg.input('palette.png')
(
    ffmpeg
    .filter([input_video, palette], 'paletteuse')
    .filter('fps', fps=10)
    .filter('scale', 480, -1, flags='lanczos')
    .output('output.gif')
    .overwrite_output()
    .run()
)

Speed Change

# Speed up 2x (video and audio)
input_file = ffmpeg.input('input.mp4')
video = input_file.video.filter('setpts', '0.5*PTS')  # Halve presentation timestamps
audio = input_file.audio.filter('atempo', 2.0)        # Double audio speed
(
    ffmpeg
    .output(video, audio, 'output.mp4')
    .overwrite_output()
    .run()
)

# Slow down 0.5x
input_file = ffmpeg.input('input.mp4')
video = input_file.video.filter('setpts', '2*PTS')
audio = input_file.audio.filter('atempo', 0.5)
(
    ffmpeg
    .output(video, audio, 'output.mp4')
    .overwrite_output()
    .run()
)

# Speed change > 2x (chain atempo filters)
# atempo only supports 0.5 to 2.0
audio = input_file.audio.filter('atempo', 2.0).filter('atempo', 2.0)  # 4x speed

Picture-in-Picture

main = ffmpeg.input('main.mp4')
pip = ffmpeg.input('pip.mp4').filter('scale', 320, 180)

(
    ffmpeg
    .overlay(main, pip, x='main_w-overlay_w-10', y=10)
    .output('output.mp4', acodec='copy')
    .overwrite_output()
    .run()
)

Blur Detection / Quality Check

def check_blur(filepath: str) -> float:
    """Check video for blur using Laplacian variance."""
    import cv2
    import numpy as np

    cap = cv2.VideoCapture(filepath)
    blur_scores = []

    try:
        while cap.isOpened():
            ret, frame = cap.read()
            if not ret:
                break

            gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
            laplacian_var = cv2.Laplacian(gray, cv2.CV_64F).var()
            blur_scores.append(laplacian_var)
    finally:
        cap.release()

    return np.mean(blur_scores)  # Higher = sharper

Error Reference

Performance Tips

1. Use stream copy when possible: -c copy avoids re-encoding 2. Use hardware acceleration: NVENC, QSV, VAAPI for encoding 3. Use PyAV for frame-level operations: 10-50x faster than ffmpeg-python for frame access 4. Avoid pipes for large files: Use temporary files instead 5. Use appropriate presets: ultrafast for real-time, slow for final output 6. Process in parallel: Split video and process chunks simultaneously 7. Match input/output formats: Reduces transcoding overhead

Additional Resources

• FFmpeg Documentation
• ffmpeg-python Examples
• PyAV Documentation
• FFmpeg Wiki
• FFmpeg Filters