Python Video Pipeline

Python Video Pipeline Skill

Use this skill for end-to-end Python video pipelines that combine decoding, OpenCV/PyAV/frame processing, FFmpeg encoding, serverless execution, GPU acceleration, HLS output, and large-file orchestration.

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. Start by selecting the pipeline architecture: simple OpenCV, FFmpeg plus OpenCV pipes, PyAV frame processing, ffmpegcv/Decord/VidGear, or Modal for scalable execution. 2. Normalize color and shape conventions at every library boundary: OpenCV BGR/HWC, PIL RGB, PyAV RGB, FFmpeg pixel formats, and ML CHW tensors. 3. Probe media metadata before processing so FPS, resolution, frame count, audio presence, and codec assumptions are explicit. 4. Process long videos as streams, batches, or chunks; avoid accumulating all frames unless inputs are small and bounded. 5. Re-mux or preserve audio after frame-level processing, since OpenCV-only workflows usually produce video-only outputs. 6. On Modal or GPU infrastructure, tune batch size, pixel format, decode/encode acceleration, volume usage, and timeout boundaries.

Key Gotchas

• BGR/RGB mismatches silently produce wrong colors across OpenCV, FFmpeg, PyAV, PIL, and ML frameworks.
• Frame dimensions are usually HWC in NumPy/OpenCV but may need CHW for deep learning frameworks.
• OpenCV VideoWriter output may not preserve the source audio; plan an explicit FFmpeg audio re-mux step.
• Parallel frame processing must restore original frame order before reconstruction.
• Chunked processing needs timestamp and concat handling; audio is usually handled after chunk recombination.

Reference Map

• references/video-pipeline-complete-patterns.md - Full original pipeline guide covering library selection, integration gotchas, FFmpeg/OpenCV pipes, ffmpegcv, VidGear, Decord, Modal GPU workflows, chunking, transcoding, HLS, end-to-end workflows, and optimization tips.
• references/modal-video-patterns.md - Additional Modal-specific video 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.

Modal.com Video Processing Patterns Reference

Advanced patterns for video processing on Modal.com serverless infrastructure.

Image Configuration Patterns

Standard Video Processing Image

import modal

# Basic video processing
video_image = (
    modal.Image.debian_slim(python_version="3.12")
    .apt_install(
        "ffmpeg",           # Video I/O
        "libsm6",           # OpenCV dependency
        "libxext6",         # OpenCV dependency
        "libgl1",           # OpenCV dependency
        "libglib2.0-0"      # GLib for video backends
    )
    .pip_install(
        "opencv-python-headless==4.9.0.80",
        "ffmpeg-python==0.2.0",
        "numpy>=1.24.0",
        "Pillow>=10.0.0"
    )
)

GPU-Accelerated Image with NVENC/NVDEC

# GPU video processing with NVIDIA acceleration
gpu_video_image = (
    modal.Image.from_registry(
        "nvidia/cuda:12.1.0-runtime-ubuntu22.04",
        add_python="3.12"
    )
    .apt_install(
        "ffmpeg",
        "libsm6",
        "libxext6",
        "libgl1",
        "libglib2.0-0"
    )
    .pip_install(
        "opencv-python-headless",
        "ffmpeg-python",
        "numpy",
        "torch>=2.0.0",
        "torchvision"
    )
)

app = modal.App("gpu-video", image=gpu_video_image)

@app.function(gpu="A100")
def process_with_gpu():
    import torch
    assert torch.cuda.is_available()
    # GPU processing here

ffmpegcv GPU Image

# Image with ffmpegcv for GPU-accelerated video I/O
ffmpegcv_image = (
    modal.Image.from_registry(
        "nvidia/cuda:12.1.0-runtime-ubuntu22.04",
        add_python="3.12"
    )
    .apt_install("ffmpeg")
    .pip_install(
        "ffmpegcv",
        "numpy",
        "opencv-python-headless"
    )
)

@app.function(gpu="T4", image=ffmpegcv_image)
def gpu_video_io():
    import ffmpegcv

    # GPU decode with NVDEC
    cap = ffmpegcv.VideoCaptureNV('video.mp4', gpu=0)

    # GPU encode with NVENC
    out = ffmpegcv.VideoWriterNV('output.mp4', 'h264_nvenc', fps=30)

    while True:
        ret, frame = cap.read()
        if not ret:
            break
        # Process frame
        out.write(frame)

    cap.release()
    out.release()

Deep Learning Video Image with Decord

# Image optimized for ML video training
ml_video_image = (
    modal.Image.debian_slim()
    .apt_install("ffmpeg", "git")
    .pip_install(
        "decord",
        "torch>=2.0.0",
        "torchvision",
        "numpy",
        "opencv-python-headless"
    )
)

@app.function(gpu="A100", image=ml_video_image)
def train_on_videos(video_paths: list[str]):
    import decord
    from decord import VideoReader
    import torch

    decord.bridge.set_bridge('torch')

    for path in video_paths:
        vr = VideoReader(path, ctx=decord.cpu())

        # Batch random frames for training
        indices = torch.randint(0, len(vr), (32,)).tolist()
        frames = vr.get_batch(indices)  # Returns torch tensor

        # frames shape: (batch, height, width, channels)
        # Train your model...

Storage Patterns

Using Volumes for Video Files

import modal

app = modal.App("video-storage")

# Create persistent volume for videos
video_volume = modal.Volume.from_name("video-library", create_if_missing=True)

@app.function(volumes={"/videos": video_volume})
def process_video(filename: str):
    """Process video from persistent storage."""
    import cv2

    input_path = f"/videos/input/{filename}"
    output_path = f"/videos/output/{filename}"

    cap = cv2.VideoCapture(input_path)
    # Process...
    cap.release()

    # CRITICAL: Commit changes to persist them
    video_volume.commit()

@app.function(volumes={"/videos": video_volume})
def list_videos() -> list[str]:
    """List all videos in storage."""
    import os
    return os.listdir("/videos/input")

Cloud Bucket Mount for Large Video Libraries

import modal

app = modal.App("video-bucket")

# S3 bucket for video storage
s3_credentials = modal.Secret.from_dict({
    "AWS_ACCESS_KEY_ID": "...",
    "AWS_SECRET_ACCESS_KEY": "...",
    "AWS_REGION": "us-east-1"
})

@app.function(
    volumes={
        "/videos": modal.CloudBucketMount(
            "my-video-bucket",
            secret=s3_credentials
        )
    }
)
def process_from_s3(video_key: str):
    """Process video directly from S3 bucket."""
    import cv2

    # S3 path mounted as local filesystem
    local_path = f"/videos/{video_key}"
    cap = cv2.VideoCapture(local_path)

    # Process video...

# Cloudflare R2 (no egress fees)
r2_credentials = modal.Secret.from_dict({
    "AWS_ACCESS_KEY_ID": "...",
    "AWS_SECRET_ACCESS_KEY": "...",
    "AWS_ENDPOINT_URL": "https://<account_id>.r2.cloudflarestorage.com"
})

@app.function(
    volumes={
        "/videos": modal.CloudBucketMount(
            "my-r2-bucket",
            secret=r2_credentials,
            read_only=True  # For source videos
        )
    }
)
def stream_from_r2(video_key: str):
    pass

Parallel Processing Patterns

Frame-Level Parallelism with map()

import modal

app = modal.App("frame-parallel")
image = modal.Image.debian_slim().pip_install("opencv-python-headless", "numpy")

@app.function(image=image)
def process_frame(frame_data: tuple[int, bytes]) -> tuple[int, bytes]:
    """Process a single frame."""
    import cv2
    import numpy as np

    idx, data = frame_data
    nparr = np.frombuffer(data, np.uint8)
    frame = cv2.imdecode(nparr, cv2.IMREAD_COLOR)

    # Apply processing
    processed = cv2.Canny(frame, 100, 200)
    processed = cv2.cvtColor(processed, cv2.COLOR_GRAY2BGR)

    _, encoded = cv2.imencode('.jpg', processed)
    return idx, encoded.tobytes()

@app.function(image=image)
def parallel_video_process(video_bytes: bytes) -> bytes:
    """Process video with parallel frame processing."""
    import cv2
    import numpy as np
    import tempfile

    # Decode video to frames
    with tempfile.NamedTemporaryFile(suffix='.mp4', delete=False) as f:
        f.write(video_bytes)
        temp_path = f.name

    cap = cv2.VideoCapture(temp_path)
    fps = cap.get(cv2.CAP_PROP_FPS)
    width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))

    frame_data = []
    idx = 0
    while cap.isOpened():
        ret, frame = cap.read()
        if not ret:
            break
        _, encoded = cv2.imencode('.jpg', frame)
        frame_data.append((idx, encoded.tobytes()))
        idx += 1
    cap.release()

    # Process frames in parallel
    results = list(process_frame.map(frame_data))

    # Sort by frame index
    results.sort(key=lambda x: x[0])

    # Reconstruct video
    output_path = temp_path.replace('.mp4', '_out.mp4')
    fourcc = cv2.VideoWriter_fourcc(*'mp4v')
    out = cv2.VideoWriter(output_path, fourcc, fps, (width, height))

    for _, frame_bytes in results:
        nparr = np.frombuffer(frame_bytes, np.uint8)
        frame = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
        out.write(frame)
    out.release()

    with open(output_path, 'rb') as f:
        return f.read()

Video-Level Parallelism with starmap()

import modal

app = modal.App("video-parallel")

@app.function(timeout=600)
def transcode_single(input_path: str, output_path: str, quality: str) -> dict:
    """Transcode a single video."""
    import ffmpeg
    import time

    start = time.time()

    crf_map = {'high': 18, 'medium': 23, 'low': 28}
    crf = crf_map.get(quality, 23)

    (
        ffmpeg
        .input(input_path)
        .output(output_path, vcodec='libx264', crf=crf, acodec='aac')
        .overwrite_output()
        .run(quiet=True)
    )

    return {
        'input': input_path,
        'output': output_path,
        'quality': quality,
        'time': time.time() - start
    }

@app.function()
def batch_transcode(video_list: list[dict]) -> list[dict]:
    """Transcode multiple videos in parallel."""
    tasks = [
        (v['input'], v['output'], v.get('quality', 'medium'))
        for v in video_list
    ]

    results = list(transcode_single.starmap(tasks))
    return results

# Usage
@app.local_entrypoint()
def main():
    videos = [
        {'input': 'video1.mp4', 'output': 'out1.mp4', 'quality': 'high'},
        {'input': 'video2.mp4', 'output': 'out2.mp4', 'quality': 'medium'},
        {'input': 'video3.mp4', 'output': 'out3.mp4', 'quality': 'low'},
    ]
    results = batch_transcode.remote(videos)
    print(results)

Chunk-Based Parallelism for Large Videos

import modal

app = modal.App("chunk-parallel")
vol = modal.Volume.from_name("chunk-storage", create_if_missing=True)

@app.function(gpu="T4", timeout=300)
def process_chunk(chunk_info: dict) -> str:
    """Process a video chunk with GPU."""
    import subprocess
    import cv2

    input_path = chunk_info['input']
    start_time = chunk_info['start_time']
    duration = chunk_info['duration']
    output_path = chunk_info['output']

    # Extract chunk with FFmpeg
    subprocess.run([
        'ffmpeg', '-y',
        '-ss', str(start_time),
        '-i', input_path,
        '-t', str(duration),
        '-c', 'copy',
        f'/tmp/chunk.mp4'
    ], check=True, capture_output=True)

    # Process chunk
    cap = cv2.VideoCapture('/tmp/chunk.mp4')
    fps = cap.get(cv2.CAP_PROP_FPS)
    w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))

    fourcc = cv2.VideoWriter_fourcc(*'mp4v')
    out = cv2.VideoWriter(output_path, fourcc, fps, (w, h))

    while cap.isOpened():
        ret, frame = cap.read()
        if not ret:
            break
        # GPU processing here
        processed = cv2.GaussianBlur(frame, (5, 5), 0)
        out.write(processed)

    cap.release()
    out.release()

    return output_path

@app.function(volumes={"/data": vol}, timeout=7200)
def process_large_video(input_path: str, chunk_duration: float = 30.0) -> str:
    """Process large video by chunking."""
    import ffmpeg
    import subprocess
    import os

    full_path = f"/data/{input_path}"

    # Get video duration
    probe = ffmpeg.probe(full_path)
    duration = float(probe['format']['duration'])

    # Create chunk tasks
    chunks = []
    chunk_idx = 0
    current_time = 0

    while current_time < duration:
        chunk_output = f"/data/chunks/chunk_{chunk_idx:04d}.mp4"
        chunks.append({
            'input': full_path,
            'start_time': current_time,
            'duration': min(chunk_duration, duration - current_time),
            'output': chunk_output
        })
        current_time += chunk_duration
        chunk_idx += 1

    os.makedirs("/data/chunks", exist_ok=True)

    # Process all chunks in parallel
    chunk_outputs = list(process_chunk.map(chunks))

    # Concatenate chunks
    list_file = "/data/chunks/list.txt"
    with open(list_file, 'w') as f:
        for path in sorted(chunk_outputs):
            f.write(f"file '{path}'\n")

    output_path = f"/data/processed_{os.path.basename(input_path)}"
    subprocess.run([
        'ffmpeg', '-y',
        '-f', 'concat',
        '-safe', '0',
        '-i', list_file,
        '-c', 'copy',
        output_path
    ], check=True)

    vol.commit()
    return output_path

Web Endpoint Patterns

Video Upload and Process API

import modal

app = modal.App("video-api")
vol = modal.Volume.from_name("api-videos", create_if_missing=True)

image = (
    modal.Image.debian_slim()
    .apt_install("ffmpeg")
    .pip_install(
        "fastapi",
        "python-multipart",
        "opencv-python-headless",
        "ffmpeg-python",
        "numpy"
    )
)

@app.function(image=image, volumes={"/data": vol}, timeout=600)
@modal.web_endpoint(method="POST")
def upload_and_process(request: dict):
    """Upload video and start processing."""
    import base64
    import uuid
    import os

    video_data = base64.b64decode(request['video_base64'])
    job_id = str(uuid.uuid4())

    input_path = f"/data/uploads/{job_id}/input.mp4"
    os.makedirs(os.path.dirname(input_path), exist_ok=True)

    with open(input_path, 'wb') as f:
        f.write(video_data)

    vol.commit()

    # Start async processing
    process_video_async.spawn(job_id)

    return {"job_id": job_id, "status": "processing"}

@app.function(image=image, volumes={"/data": vol}, timeout=3600)
def process_video_async(job_id: str):
    """Process video asynchronously."""
    import cv2
    import os

    input_path = f"/data/uploads/{job_id}/input.mp4"
    output_path = f"/data/outputs/{job_id}/processed.mp4"

    os.makedirs(os.path.dirname(output_path), exist_ok=True)

    cap = cv2.VideoCapture(input_path)
    fps = cap.get(cv2.CAP_PROP_FPS)
    w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))

    fourcc = cv2.VideoWriter_fourcc(*'mp4v')
    out = cv2.VideoWriter(output_path, fourcc, fps, (w, h))

    while cap.isOpened():
        ret, frame = cap.read()
        if not ret:
            break
        processed = cv2.GaussianBlur(frame, (5, 5), 0)
        out.write(processed)

    cap.release()
    out.release()

    # Mark job as complete
    with open(f"/data/outputs/{job_id}/status.txt", 'w') as f:
        f.write("complete")

    vol.commit()

@app.function(image=image, volumes={"/data": vol})
@modal.web_endpoint(method="GET")
def check_status(job_id: str):
    """Check processing status."""
    import os

    status_file = f"/data/outputs/{job_id}/status.txt"
    output_file = f"/data/outputs/{job_id}/processed.mp4"

    if os.path.exists(status_file):
        with open(status_file) as f:
            status = f.read().strip()

        if status == "complete" and os.path.exists(output_file):
            return {
                "job_id": job_id,
                "status": "complete",
                "download_ready": True
            }

    return {"job_id": job_id, "status": "processing"}

Streaming Response for Processed Video

import modal

app = modal.App("video-stream")
vol = modal.Volume.from_name("stream-videos", create_if_missing=True)

image = (
    modal.Image.debian_slim()
    .apt_install("ffmpeg")
    .pip_install("fastapi", "opencv-python-headless")
)

@app.function(image=image, volumes={"/data": vol})
@modal.web_endpoint(method="GET")
def stream_video(job_id: str):
    """Stream processed video."""
    from fastapi.responses import StreamingResponse
    import os

    output_path = f"/data/outputs/{job_id}/processed.mp4"

    if not os.path.exists(output_path):
        return {"error": "Video not found"}

    def iter_file():
        with open(output_path, 'rb') as f:
            while chunk := f.read(1024 * 1024):  # 1MB chunks
                yield chunk

    return StreamingResponse(
        iter_file(),
        media_type="video/mp4",
        headers={
            "Content-Disposition": f"attachment; filename=processed_{job_id}.mp4"
        }
    )

Cost Optimization Patterns

GPU Selection Based on Task

import modal

app = modal.App("optimized-video")

# T4: Best for light transcoding ($0.59/hr)
@app.function(gpu="T4")
def transcode_small(video_path: str):
    """Use T4 for videos < 1080p or simple effects."""
    pass

# A10G: Good balance for HD processing ($1.10/hr)
@app.function(gpu="A10G")
def process_hd(video_path: str):
    """Use A10G for 1080p processing."""
    pass

# A100-40GB: For 4K or ML inference ($3.24/hr)
@app.function(gpu="A100-40GB")
def process_4k(video_path: str):
    """Use A100 for 4K or heavy ML workloads."""
    pass

# H100: For maximum throughput ($4.93/hr)
@app.function(gpu="H100")
def process_batch_heavy(video_paths: list[str]):
    """Use H100 for high-throughput batch processing."""
    pass

CPU-Only for Simple Tasks

import modal

app = modal.App("cpu-video")

# Use CPU for metadata extraction, simple transcoding
@app.function(cpu=2.0, memory=4096)  # No GPU
def extract_metadata(video_path: str) -> dict:
    """CPU is sufficient for ffprobe."""
    import ffmpeg
    return ffmpeg.probe(video_path)

@app.function(cpu=4.0, memory=8192)  # No GPU
def simple_transcode(input_path: str, output_path: str):
    """CPU transcoding for non-time-critical tasks."""
    import ffmpeg

    (
        ffmpeg
        .input(input_path)
        .output(output_path, vcodec='libx264', preset='slow', crf=23)
        .run()
    )

Spot Instances for Batch Processing

import modal

app = modal.App("batch-video")

# Use spot for non-urgent batch jobs (cheaper but interruptible)
@app.function(
    gpu="A100",
    timeout=3600,
    retries=3  # Retry on preemption
)
def batch_process_with_retry(video_path: str) -> str:
    """Process with automatic retry on spot preemption."""
    import cv2

    # Save progress periodically
    checkpoint_interval = 100

    cap = cv2.VideoCapture(video_path)
    frame_count = 0

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

        # Process frame
        frame_count += 1

        # Checkpoint every N frames
        if frame_count % checkpoint_interval == 0:
            save_checkpoint(frame_count)

    return "complete"

Error Handling Patterns

Robust Video Processing

import modal

app = modal.App("robust-video")

@app.function(timeout=600, retries=2)
def robust_process(video_path: str) -> dict:
    """Process video with comprehensive error handling."""
    import cv2
    import ffmpeg
    import os

    result = {
        'status': 'unknown',
        'input': video_path,
        'output': None,
        'error': None
    }

    # Validate input
    if not os.path.exists(video_path):
        result['status'] = 'error'
        result['error'] = 'Input file not found'
        return result

    # Probe video
    try:
        probe = ffmpeg.probe(video_path)
    except ffmpeg.Error as e:
        result['status'] = 'error'
        result['error'] = f'Invalid video file: {e.stderr.decode()}'
        return result

    # Check for video stream
    video_streams = [s for s in probe['streams'] if s['codec_type'] == 'video']
    if not video_streams:
        result['status'] = 'error'
        result['error'] = 'No video stream found'
        return result

    # Process video
    try:
        cap = cv2.VideoCapture(video_path)

        if not cap.isOpened():
            result['status'] = 'error'
            result['error'] = 'Failed to open video with OpenCV'
            return result

        fps = cap.get(cv2.CAP_PROP_FPS)
        if fps <= 0:
            fps = 30  # Default fallback

        width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
        height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))

        if width <= 0 or height <= 0:
            result['status'] = 'error'
            result['error'] = 'Invalid video dimensions'
            cap.release()
            return result

        output_path = video_path.replace('.mp4', '_processed.mp4')
        fourcc = cv2.VideoWriter_fourcc(*'mp4v')
        out = cv2.VideoWriter(output_path, fourcc, fps, (width, height))

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

            # Process frame with error handling
            try:
                processed = cv2.GaussianBlur(frame, (5, 5), 0)
                out.write(processed)
                frame_count += 1
            except Exception as e:
                # Log but continue on frame errors
                print(f"Frame {frame_count} error: {e}")
                continue

        cap.release()
        out.release()

        if frame_count == 0:
            result['status'] = 'error'
            result['error'] = 'No frames processed'
            return result

        result['status'] = 'success'
        result['output'] = output_path
        result['frames_processed'] = frame_count

    except Exception as e:
        result['status'] = 'error'
        result['error'] = str(e)

    return result

Monitoring and Logging

Progress Tracking

import modal

app = modal.App("monitored-video")
vol = modal.Volume.from_name("progress-tracking", create_if_missing=True)

@app.function(volumes={"/data": vol}, timeout=3600)
def process_with_progress(job_id: str, video_path: str):
    """Track processing progress."""
    import cv2
    import json
    import os

    progress_file = f"/data/progress/{job_id}.json"
    os.makedirs(os.path.dirname(progress_file), exist_ok=True)

    def update_progress(current_frame: int, total_frames: int, status: str):
        progress = {
            'job_id': job_id,
            'current_frame': current_frame,
            'total_frames': total_frames,
            'percentage': round(current_frame / total_frames * 100, 2),
            'status': status
        }
        with open(progress_file, 'w') as f:
            json.dump(progress, f)
        vol.commit()

    cap = cv2.VideoCapture(video_path)
    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))

    update_progress(0, total_frames, 'processing')

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

        # Process frame...
        frame_idx += 1

        # Update progress every 100 frames
        if frame_idx % 100 == 0:
            update_progress(frame_idx, total_frames, 'processing')

    cap.release()
    update_progress(total_frames, total_frames, 'complete')

@app.function(volumes={"/data": vol})
@modal.web_endpoint(method="GET")
def get_progress(job_id: str):
    """Get processing progress."""
    import json
    import os

    progress_file = f"/data/progress/{job_id}.json"

    if os.path.exists(progress_file):
        with open(progress_file) as f:
            return json.load(f)

    return {"error": "Job not found"}

Additional Resources

• Modal GPU Docs
• Modal Volumes Docs
• Modal Cloud Bucket Mounts
• Modal Parallel Execution
• Modal Web Endpoints

Python Video Pipeline Complete Patterns

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 Video Processing Pipeline Skill

Comprehensive guide to building video processing pipelines combining FFmpeg, OpenCV, and Modal.com for scalable, GPU-accelerated workflows.

Quick Reference: Library Selection

Critical Integration Gotchas

1. Color Format Mismatch (MOST COMMON BUG)

# CRITICAL: Different libraries use different color formats!

# OpenCV uses BGR
import cv2
frame_bgr = cv2.imread('image.jpg')  # BGR format

# FFmpeg outputs RGB by default
import ffmpeg
# When piping to OpenCV, specify bgr24
process = (
    ffmpeg
    .input('video.mp4')
    .output('pipe:', format='rawvideo', pix_fmt='bgr24')  # NOT rgb24!
    .run_async(pipe_stdout=True)
)

# PyAV outputs RGB
import av
container = av.open('video.mp4')
for frame in container.decode(video=0):
    rgb_array = frame.to_ndarray(format='rgb24')
    bgr_array = rgb_array[:, :, ::-1]  # Convert to BGR for OpenCV

# PIL/Pillow uses RGB
from PIL import Image
pil_image = Image.open('image.jpg')  # RGB
opencv_image = cv2.cvtColor(np.array(pil_image), cv2.COLOR_RGB2BGR)

2. Frame Dimension Order

# NumPy/OpenCV: (height, width, channels) - img[y, x]
# Some ML frameworks: (channels, height, width) - img[c, y, x]

import numpy as np

# OpenCV frame shape
frame = cv2.imread('image.jpg')
height, width, channels = frame.shape  # (1080, 1920, 3)

# Access pixel: frame[row, col] = frame[y, x]
pixel = frame[100, 200]  # Row 100, Column 200

# Transpose for CHW format (PyTorch, etc.)
chw_frame = frame.transpose(2, 0, 1)  # (3, 1080, 1920)

# Or use np.moveaxis
chw_frame = np.moveaxis(frame, -1, 0)

3. Audio Stream Loss in Pipelines

import ffmpeg

# BAD: Processing video loses audio!
input_file = ffmpeg.input('input.mp4')
processed = input_file.filter('scale', 1280, 720)
ffmpeg.output(processed, 'output.mp4').run()  # NO AUDIO!

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

# When processing with OpenCV, re-mux audio separately
# Step 1: Process video frames with OpenCV
# Step 2: Extract original audio
ffmpeg.input('input.mp4').output('audio.aac', vn=None, acodec='copy').run()
# Step 3: Combine processed video with original audio
ffmpeg.input('processed_video.mp4').input('audio.aac').output(
    'final.mp4', vcodec='copy', acodec='copy'
).run()

4. Memory Management with Large Videos

# BAD: Loading all frames into memory
frames = []
cap = cv2.VideoCapture('large_video.mp4')
while True:
    ret, frame = cap.read()
    if not ret:
        break
    frames.append(frame)  # OOM for large videos!

# GOOD: Generator pattern
def read_frames(video_path: str):
    """Generator that yields frames one at a time."""
    cap = cv2.VideoCapture(video_path)
    try:
        while cap.isOpened():
            ret, frame = cap.read()
            if not ret:
                break
            yield frame
    finally:
        cap.release()

# Process frames without storing all in memory
for frame in read_frames('large_video.mp4'):
    process(frame)

# GOOD: Batch processing with fixed memory
def read_frame_batches(video_path: str, batch_size: int = 32):
    """Yield batches of frames."""
    batch = []
    for frame in read_frames(video_path):
        batch.append(frame)
        if len(batch) >= batch_size:
            yield np.stack(batch)
            batch = []
    if batch:
        yield np.stack(batch)

FFmpeg + OpenCV Integration

Pattern 1: FFmpeg Decode → OpenCV Process → FFmpeg Encode

import subprocess
import cv2
import numpy as np

def process_with_ffmpeg_opencv(
    input_path: str,
    output_path: str,
    width: int,
    height: int,
    fps: int = 30
):
    """
    Use FFmpeg for I/O, OpenCV for processing.
    Best of both worlds: FFmpeg codec support + OpenCV algorithms.
    """
    # FFmpeg reader process
    reader = subprocess.Popen(
        [
            'ffmpeg',
            '-i', input_path,
            '-f', 'rawvideo',
            '-pix_fmt', 'bgr24',  # OpenCV format
            '-s', f'{width}x{height}',
            'pipe:1'
        ],
        stdout=subprocess.PIPE,
        stderr=subprocess.DEVNULL
    )

    # FFmpeg writer process
    writer = 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',
            '-preset', 'fast',
            '-crf', '23',
            '-pix_fmt', 'yuv420p',
            output_path
        ],
        stdin=subprocess.PIPE,
        stderr=subprocess.DEVNULL
    )

    frame_size = width * height * 3

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

            # Convert to numpy array
            frame = np.frombuffer(raw_frame, dtype=np.uint8)
            frame = frame.reshape((height, width, 3))

            # OpenCV processing
            processed = cv2.GaussianBlur(frame, (5, 5), 0)
            # Add more processing here...

            # Write processed frame
            writer.stdin.write(processed.tobytes())
    finally:
        reader.stdout.close()
        writer.stdin.close()
        reader.wait()
        writer.wait()

Pattern 2: Using ffmpegcv (OpenCV-Compatible API)

import ffmpegcv

# Drop-in replacement for cv2.VideoCapture
# Supports more codecs, GPU acceleration, network streams

# Basic usage (same as OpenCV)
cap = ffmpegcv.VideoCapture('video.mp4')
while True:
    ret, frame = cap.read()
    if not ret:
        break
    # frame is BGR numpy array, just like OpenCV
    cv2.imshow('Frame', frame)
cap.release()

# GPU-accelerated decoding (NVIDIA only)
cap = ffmpegcv.VideoCaptureNV('video.mp4')  # Uses NVDEC

# GPU decoding with specific GPU
cap = ffmpegcv.VideoCaptureNV('video.mp4', gpu=0)

# Network streams
cap = ffmpegcv.VideoCapture('rtsp://192.168.1.100:554/stream')

# With resize during decode (more efficient than post-resize)
cap = ffmpegcv.VideoCapture('video.mp4', resize=(1280, 720))

# ROI cropping during decode
cap = ffmpegcv.VideoCapture('video.mp4', crop_xywh=(100, 100, 640, 480))

# Writing with GPU encoding
out = ffmpegcv.VideoWriterNV('output.mp4', 'h264_nvenc', fps=30)
out.write(frame)
out.release()

# Direct to CUDA memory (for deep learning)
cap = ffmpegcv.VideoCaptureNV('video.mp4', pix_fmt='cuda')
cuda_frame = cap.read()  # Returns GPU memory pointer

Pattern 3: Using VidGear for Streaming

from vidgear.gears import CamGear, WriteGear

# High-performance capture with multi-threading
stream = CamGear(
    source='rtsp://192.168.1.100:554/stream',
    stream_mode=True,  # Enable network stream mode
    logging=True
).start()

# Capture from YouTube live stream
stream = CamGear(
    source='https://youtu.be/live_stream_id',
    stream_mode=True,
    STREAM_RESOLUTION="1080p"
).start()

while True:
    frame = stream.read()
    if frame is None:
        break

    # OpenCV processing
    processed = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)

    cv2.imshow('Stream', processed)
    if cv2.waitKey(1) == ord('q'):
        break

stream.stop()

# Writing with FFmpeg backend
output_params = {
    '-vcodec': 'libx264',
    '-crf': 23,
    '-preset': 'fast'
}
writer = WriteGear(output='output.mp4', **output_params)

for frame in frames:
    writer.write(frame)

writer.close()

# RTMP streaming output
output_params = {
    '-vcodec': 'libx264',
    '-preset': 'ultrafast',
    '-tune': 'zerolatency',
    '-f': 'flv'
}
writer = WriteGear(output='rtmp://server/live/stream', **output_params)

Pattern 4: Using Decord for Deep Learning

import decord
from decord import VideoReader, gpu

# CPU decoding
decord.bridge.set_bridge('torch')  # or 'mxnet', 'tensorflow'
vr = VideoReader('video.mp4', ctx=decord.cpu())

# GPU decoding (requires build from source with CUDA)
vr = VideoReader('video.mp4', ctx=decord.gpu(0))

# Get video info
print(f"Frames: {len(vr)}, FPS: {vr.get_avg_fps()}")

# Random access (efficient!)
frame_10 = vr[10]  # Get frame 10
frames = vr[10:20]  # Get frames 10-19

# Batch loading for training (most efficient)
frame_indices = [0, 10, 20, 30, 40]  # Non-sequential access
batch = vr.get_batch(frame_indices)  # Returns stacked tensor

# VideoLoader for training with shuffling
from decord import VideoLoader

# Multiple videos, shuffled for training
vl = VideoLoader(
    ['video1.mp4', 'video2.mp4', 'video3.mp4'],
    ctx=decord.cpu(),
    shape=(8, 224, 224, 3),  # (batch, height, width, channels)
    interval=1,  # Sample every frame
    skip=0,
    shuffle=1  # 1=shuffle filenames, 2=random order, 3=random frames
)

for batch in vl:
    # batch shape: (batch_size, num_frames, H, W, C)
    train_on_batch(batch)

Modal.com Video Processing

Basic Setup: FFmpeg + OpenCV on Modal

import modal

# Define image with FFmpeg and OpenCV
image = (
    modal.Image.debian_slim(python_version="3.12")
    .apt_install("ffmpeg", "libsm6", "libxext6", "libgl1")  # System deps
    .pip_install(
        "opencv-python-headless",  # Headless for servers
        "ffmpeg-python",
        "numpy"
    )
)

app = modal.App("video-processing", image=image)
vol = modal.Volume.from_name("video-storage", create_if_missing=True)

@app.function(volumes={"/data": vol})
def process_video(input_path: str, output_path: str):
    """Process a single video file."""
    import cv2
    import ffmpeg

    cap = cv2.VideoCapture(f"/data/{input_path}")
    width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
    fps = cap.get(cv2.CAP_PROP_FPS)

    fourcc = cv2.VideoWriter_fourcc(*'mp4v')
    out = cv2.VideoWriter(f"/data/{output_path}", fourcc, fps, (width, height))

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

        # Process frame
        processed = cv2.GaussianBlur(frame, (5, 5), 0)
        out.write(processed)

    cap.release()
    out.release()

    # Commit changes to volume
    vol.commit()
    return output_path

@app.local_entrypoint()
def main():
    process_video.remote("input.mp4", "output.mp4")

GPU-Accelerated Processing on Modal

import modal

# Image with CUDA + OpenCV + FFmpeg
gpu_image = (
    modal.Image.from_registry("nvidia/cuda:12.1.0-runtime-ubuntu22.04")
    .apt_install("ffmpeg", "python3-pip", "libsm6", "libxext6", "libgl1")
    .run_commands("pip install opencv-python-headless numpy torch ffmpeg-python")
)

app = modal.App("gpu-video-processing", image=gpu_image)

@app.function(gpu="A100", timeout=3600)
def process_video_gpu(video_bytes: bytes) -> bytes:
    """Process video using GPU acceleration."""
    import torch
    import cv2
    import numpy as np
    import tempfile

    # Write input to temp file
    with tempfile.NamedTemporaryFile(suffix='.mp4', delete=False) as f:
        f.write(video_bytes)
        input_path = f.name

    # Read video
    cap = cv2.VideoCapture(input_path)
    frames = []

    while cap.isOpened():
        ret, frame = cap.read()
        if not ret:
            break
        frames.append(frame)
    cap.release()

    # Process on GPU
    device = torch.device('cuda')
    frames_tensor = torch.from_numpy(np.stack(frames)).to(device)

    # GPU processing (example: normalize)
    frames_tensor = frames_tensor.float() / 255.0
    # Add your GPU processing here...
    frames_tensor = (frames_tensor * 255).byte()

    processed_frames = frames_tensor.cpu().numpy()

    # Write output
    output_path = input_path.replace('.mp4', '_processed.mp4')
    fourcc = cv2.VideoWriter_fourcc(*'mp4v')
    fps = cap.get(cv2.CAP_PROP_FPS) or 30
    h, w = processed_frames[0].shape[:2]
    out = cv2.VideoWriter(output_path, fourcc, fps, (w, h))

    for frame in processed_frames:
        out.write(frame)
    out.release()

    with open(output_path, 'rb') as f:
        return f.read()

Parallel Frame Processing with Modal

import modal

image = (
    modal.Image.debian_slim()
    .apt_install("ffmpeg")
    .pip_install("opencv-python-headless", "numpy", "ffmpeg-python")
)

app = modal.App("parallel-video", image=image)
vol = modal.Volume.from_name("video-frames", create_if_missing=True)

@app.function()
def process_frame(frame_data: bytes, frame_idx: int) -> tuple[int, bytes]:
    """Process a single frame."""
    import cv2
    import numpy as np

    # Decode frame
    nparr = np.frombuffer(frame_data, np.uint8)
    frame = cv2.imdecode(nparr, cv2.IMREAD_COLOR)

    # Process (example: edge detection)
    edges = cv2.Canny(frame, 100, 200)
    edges_bgr = cv2.cvtColor(edges, cv2.COLOR_GRAY2BGR)

    # Encode result
    _, encoded = cv2.imencode('.png', edges_bgr)
    return frame_idx, encoded.tobytes()

@app.function(volumes={"/data": vol}, timeout=3600)
def process_video_parallel(input_path: str) -> str:
    """Process video frames in parallel using Modal map."""
    import cv2
    import ffmpeg
    import numpy as np

    # Extract frames
    cap = cv2.VideoCapture(f"/data/{input_path}")
    fps = cap.get(cv2.CAP_PROP_FPS)
    width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))

    frame_data_list = []
    frame_idx = 0

    while cap.isOpened():
        ret, frame = cap.read()
        if not ret:
            break
        _, encoded = cv2.imencode('.png', frame)
        frame_data_list.append((encoded.tobytes(), frame_idx))
        frame_idx += 1
    cap.release()

    # Process frames in parallel
    results = list(process_frame.starmap(frame_data_list))

    # Sort by frame index
    results.sort(key=lambda x: x[0])

    # Reconstruct video
    output_path = input_path.replace('.mp4', '_processed.mp4')
    fourcc = cv2.VideoWriter_fourcc(*'mp4v')
    out = cv2.VideoWriter(f"/data/{output_path}", fourcc, fps, (width, height))

    for _, frame_bytes in results:
        nparr = np.frombuffer(frame_bytes, np.uint8)
        frame = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
        out.write(frame)

    out.release()
    vol.commit()

    return output_path

Chunk-Based Video Processing for Large Files

import modal

image = (
    modal.Image.debian_slim()
    .apt_install("ffmpeg")
    .pip_install("opencv-python-headless", "numpy", "ffmpeg-python")
)

app = modal.App("chunked-video", image=image)
vol = modal.Volume.from_name("video-chunks", create_if_missing=True)

@app.function(gpu="T4", timeout=600)
def process_chunk(
    input_path: str,
    output_path: str,
    start_frame: int,
    end_frame: int
) -> str:
    """Process a chunk of video frames."""
    import cv2
    import subprocess

    # Use FFmpeg to extract specific frame range
    chunk_input = f"/tmp/chunk_{start_frame}.mp4"
    subprocess.run([
        'ffmpeg', '-y',
        '-i', input_path,
        '-vf', f'select=between(n\\,{start_frame}\\,{end_frame}),setpts=PTS-STARTPTS',
        '-an',  # No audio for chunks
        chunk_input
    ], check=True, capture_output=True)

    # Process chunk
    cap = cv2.VideoCapture(chunk_input)
    fps = cap.get(cv2.CAP_PROP_FPS)
    width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))

    chunk_output = f"/tmp/processed_{start_frame}.mp4"
    fourcc = cv2.VideoWriter_fourcc(*'mp4v')
    out = cv2.VideoWriter(chunk_output, fourcc, fps, (width, height))

    while cap.isOpened():
        ret, frame = cap.read()
        if not ret:
            break
        # GPU processing here
        processed = cv2.GaussianBlur(frame, (5, 5), 0)
        out.write(processed)

    cap.release()
    out.release()

    return chunk_output

@app.function(volumes={"/data": vol}, timeout=7200)
def process_large_video(input_path: str, chunk_size: int = 1000) -> str:
    """Process large video by splitting into chunks."""
    import cv2
    import subprocess

    full_path = f"/data/{input_path}"

    # Get video info
    cap = cv2.VideoCapture(full_path)
    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    fps = cap.get(cv2.CAP_PROP_FPS)
    cap.release()

    # Create chunks
    chunks = []
    for start in range(0, total_frames, chunk_size):
        end = min(start + chunk_size - 1, total_frames - 1)
        chunks.append((full_path, f"/tmp/chunk_{start}.mp4", start, end))

    # Process chunks in parallel
    chunk_outputs = list(process_chunk.starmap(chunks))

    # Concatenate chunks with FFmpeg
    list_file = "/tmp/chunks.txt"
    with open(list_file, 'w') as f:
        for path in chunk_outputs:
            f.write(f"file '{path}'\n")

    output_path = input_path.replace('.mp4', '_processed.mp4')
    subprocess.run([
        'ffmpeg', '-y',
        '-f', 'concat',
        '-safe', '0',
        '-i', list_file,
        '-c', 'copy',
        f"/data/{output_path}"
    ], check=True)

    vol.commit()
    return output_path

Video Transcoding Pipeline on Modal

import modal

image = (
    modal.Image.debian_slim()
    .apt_install("ffmpeg")
    .pip_install("ffmpeg-python")
)

app = modal.App("transcoding-pipeline", image=image)
vol = modal.Volume.from_name("transcoded-videos", create_if_missing=True)

# Quality presets
QUALITY_PRESETS = {
    '4k': {'width': 3840, 'height': 2160, 'bitrate': '15M', 'crf': 18},
    '1080p': {'width': 1920, 'height': 1080, 'bitrate': '5M', 'crf': 23},
    '720p': {'width': 1280, 'height': 720, 'bitrate': '2.5M', 'crf': 26},
    '480p': {'width': 854, 'height': 480, 'bitrate': '1M', 'crf': 28},
}

@app.function(gpu="T4", timeout=3600)  # Use GPU for NVENC
def transcode_video(
    input_path: str,
    output_path: str,
    quality: str,
    use_hardware: bool = True
) -> dict:
    """Transcode video to specific quality."""
    import ffmpeg
    import time

    preset = QUALITY_PRESETS[quality]

    start_time = time.time()

    input_stream = ffmpeg.input(input_path)
    video = input_stream.video.filter('scale', preset['width'], preset['height'])
    audio = input_stream.audio

    if use_hardware:
        # NVIDIA NVENC encoding
        output = ffmpeg.output(
            video, audio, output_path,
            vcodec='h264_nvenc',
            preset='p4',  # Quality preset
            cq=preset['crf'],  # Constant quality
            acodec='aac',
            audio_bitrate='192k'
        )
    else:
        # CPU encoding
        output = ffmpeg.output(
            video, audio, output_path,
            vcodec='libx264',
            preset='medium',
            crf=preset['crf'],
            acodec='aac',
            audio_bitrate='192k'
        )

    output.overwrite_output().run(quiet=True)

    elapsed = time.time() - start_time

    return {
        'quality': quality,
        'output_path': output_path,
        'encoding_time': elapsed
    }

@app.function(volumes={"/data": vol})
def create_quality_ladder(input_path: str) -> list[dict]:
    """Create multiple quality versions (adaptive streaming ready)."""
    import os

    full_path = f"/data/{input_path}"
    base_name = os.path.splitext(input_path)[0]

    # Transcode to all qualities in parallel
    tasks = [
        (full_path, f"/data/{base_name}_{q}.mp4", q, True)
        for q in ['1080p', '720p', '480p']
    ]

    results = list(transcode_video.starmap(tasks))

    vol.commit()
    return results

HLS Streaming Generation on Modal

import modal

image = (
    modal.Image.debian_slim()
    .apt_install("ffmpeg")
    .pip_install("ffmpeg-python")
)

app = modal.App("hls-generator", image=image)
vol = modal.Volume.from_name("hls-streams", create_if_missing=True)

@app.function(gpu="T4", volumes={"/data": vol}, timeout=3600)
def generate_hls(input_path: str, output_dir: str) -> dict:
    """Generate HLS stream with multiple quality levels."""
    import ffmpeg
    import os

    os.makedirs(f"/data/{output_dir}", exist_ok=True)

    input_stream = ffmpeg.input(f"/data/{input_path}")

    # Create multiple quality streams
    qualities = [
        ('1080p', 1920, 1080, '5000k'),
        ('720p', 1280, 720, '2500k'),
        ('480p', 854, 480, '1000k'),
    ]

    for name, w, h, bitrate in qualities:
        stream_dir = f"/data/{output_dir}/{name}"
        os.makedirs(stream_dir, exist_ok=True)

        video = input_stream.video.filter('scale', w, h)
        audio = input_stream.audio

        output = ffmpeg.output(
            video, audio,
            f"{stream_dir}/stream.m3u8",
            vcodec='h264_nvenc',
            video_bitrate=bitrate,
            acodec='aac',
            audio_bitrate='128k',
            f='hls',
            hls_time=10,
            hls_list_size=0,
            hls_segment_filename=f"{stream_dir}/segment_%03d.ts"
        )
        output.overwrite_output().run(quiet=True)

    # Create master playlist
    master_playlist = f"/data/{output_dir}/master.m3u8"
    with open(master_playlist, 'w') as f:
        f.write("#EXTM3U\n")
        f.write("#EXT-X-VERSION:3\n")
        for name, w, h, bitrate in qualities:
            bandwidth = int(bitrate.replace('k', '')) * 1000
            f.write(f"#EXT-X-STREAM-INF:BANDWIDTH={bandwidth},RESOLUTION={w}x{h}\n")
            f.write(f"{name}/stream.m3u8\n")

    vol.commit()

    return {
        'master_playlist': f"{output_dir}/master.m3u8",
        'qualities': [q[0] for q in qualities]
    }

Complete Pipeline Example

End-to-End: Upload → Process → Transcode → HLS

import modal

image = (
    modal.Image.debian_slim()
    .apt_install("ffmpeg", "libsm6", "libxext6", "libgl1")
    .pip_install(
        "opencv-python-headless",
        "ffmpeg-python",
        "numpy",
        "boto3"  # For S3 integration
    )
)

app = modal.App("video-pipeline", image=image)
vol = modal.Volume.from_name("pipeline-storage", create_if_missing=True)

# S3 credentials
s3_secret = modal.Secret.from_name("aws-credentials")

@app.function()
def analyze_video(video_bytes: bytes) -> dict:
    """Analyze video metadata."""
    import ffmpeg
    import tempfile

    with tempfile.NamedTemporaryFile(suffix='.mp4', delete=False) as f:
        f.write(video_bytes)
        temp_path = f.name

    probe = ffmpeg.probe(temp_path)
    video_info = next(s for s in probe['streams'] if s['codec_type'] == 'video')

    return {
        'duration': float(probe['format']['duration']),
        'width': video_info['width'],
        'height': video_info['height'],
        'fps': eval(video_info['r_frame_rate']),
        'codec': video_info['codec_name'],
        'size_mb': int(probe['format']['size']) / (1024 * 1024)
    }

@app.function(gpu="A100")
def apply_cv_effects(frame_batch: list[bytes]) -> list[bytes]:
    """Apply computer vision effects to a batch of frames."""
    import cv2
    import numpy as np

    processed = []
    for frame_bytes in frame_batch:
        nparr = np.frombuffer(frame_bytes, np.uint8)
        frame = cv2.imdecode(nparr, cv2.IMREAD_COLOR)

        # Apply effects
        frame = cv2.GaussianBlur(frame, (5, 5), 0)
        frame = cv2.Canny(frame, 100, 200)
        frame = cv2.cvtColor(frame, cv2.COLOR_GRAY2BGR)

        _, encoded = cv2.imencode('.png', frame)
        processed.append(encoded.tobytes())

    return processed

@app.function(
    volumes={"/data": vol},
    secrets=[s3_secret],
    timeout=7200
)
def run_pipeline(
    s3_input_key: str,
    s3_output_prefix: str,
    apply_effects: bool = True
) -> dict:
    """Complete video processing pipeline."""
    import boto3
    import cv2
    import ffmpeg
    import numpy as np
    import os

    # Download from S3
    s3 = boto3.client('s3')
    bucket = os.environ['S3_BUCKET']
    local_input = "/data/input.mp4"

    s3.download_file(bucket, s3_input_key, local_input)

    # Analyze
    with open(local_input, 'rb') as f:
        metadata = analyze_video.remote(f.read())

    # Extract and process frames if needed
    if apply_effects:
        cap = cv2.VideoCapture(local_input)
        fps = metadata['fps']
        width = metadata['width']
        height = metadata['height']

        # Batch frames for parallel processing
        batch_size = 100
        batches = []
        current_batch = []

        while cap.isOpened():
            ret, frame = cap.read()
            if not ret:
                break
            _, encoded = cv2.imencode('.png', frame)
            current_batch.append(encoded.tobytes())

            if len(current_batch) >= batch_size:
                batches.append(current_batch)
                current_batch = []

        if current_batch:
            batches.append(current_batch)

        cap.release()

        # Process batches in parallel
        processed_batches = list(apply_cv_effects.map(batches))

        # Reconstruct video
        local_processed = "/data/processed.mp4"
        fourcc = cv2.VideoWriter_fourcc(*'mp4v')
        out = cv2.VideoWriter(local_processed, fourcc, fps, (width, height))

        for batch in processed_batches:
            for frame_bytes in batch:
                nparr = np.frombuffer(frame_bytes, np.uint8)
                frame = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
                out.write(frame)

        out.release()

        # Add back audio
        ffmpeg.input(local_processed).input(local_input).output(
            "/data/with_audio.mp4",
            vcodec='copy',
            acodec='copy',
            map=['0:v:0', '1:a:0']
        ).overwrite_output().run()

        local_input = "/data/with_audio.mp4"

    # Generate HLS
    hls_result = generate_hls.remote(
        os.path.basename(local_input),
        "hls_output"
    )

    # Upload results to S3
    for root, dirs, files in os.walk("/data/hls_output"):
        for file in files:
            local_path = os.path.join(root, file)
            s3_key = f"{s3_output_prefix}/{os.path.relpath(local_path, '/data/hls_output')}"
            s3.upload_file(local_path, bucket, s3_key)

    vol.commit()

    return {
        'metadata': metadata,
        'hls': hls_result,
        's3_output': f"s3://{bucket}/{s3_output_prefix}/"
    }

Performance Optimization Tips

1. Use Hardware Acceleration When Available

# Check for NVENC support
import subprocess
result = subprocess.run(['ffmpeg', '-encoders'], capture_output=True, text=True)
has_nvenc = 'h264_nvenc' in result.stdout

# Use appropriate encoder
vcodec = 'h264_nvenc' if has_nvenc else 'libx264'

2. Optimize Batch Sizes for GPU Memory

import torch

def get_optimal_batch_size(frame_shape: tuple, gpu_memory_gb: float = 40) -> int:
    """Calculate optimal batch size for GPU memory."""
    h, w, c = frame_shape
    bytes_per_frame = h * w * c * 4  # float32
    available_bytes = gpu_memory_gb * 1e9 * 0.8  # 80% utilization
    return int(available_bytes / bytes_per_frame)

3. Use Efficient Pixel Formats

# For deep learning: Use fp32 CHW directly on GPU
import ffmpegcv

# Skip CPU conversion, go straight to GPU
cap = ffmpegcv.VideoCaptureNV(
    'video.mp4',
    pix_fmt='cuda',
    resize=(224, 224)
)
cuda_frame = cap.read()  # Already on GPU, CHW format

4. Stream Processing for Large Videos

# Never load entire video into memory
def process_streaming(input_path: str, output_path: str):
    """Process video frame-by-frame without memory accumulation."""
    import cv2

    cap = cv2.VideoCapture(input_path)
    fourcc = cv2.VideoWriter_fourcc(*'mp4v')
    out = None

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

            if out is None:
                h, w = frame.shape[:2]
                fps = cap.get(cv2.CAP_PROP_FPS)
                out = cv2.VideoWriter(output_path, fourcc, fps, (w, h))

            processed = process_single_frame(frame)
            out.write(processed)
    finally:
        cap.release()
        if out:
            out.release()

Additional Resources

• ffmpegcv GitHub
• VidGear Documentation
• Decord GitHub
• Modal.com Docs - GPU
• Modal.com Docs - Volumes
• OpenCV CUDA Module
• FFmpeg Hardware Acceleration