How to Build an AI Image Upscaling Pipeline

An ai image upscaling pipeline to enhance resolution rescues low-quality images that would otherwise be unusable. I build these for businesses with legacy product photos, old screenshots, or user-submitted images that need to look professional. The AI fills in detail that was never there, turning 400x400 images into sharp 1600x1600 outputs.

This runs as a batch pipeline. Drop images in, get enhanced versions out.

What You Need Before Starting

• Python 3.8+ with Real-ESRGAN or a cloud upscaling API
• Low-resolution source images
• Storage for processed images
• Quality benchmarks (target resolution and sharpness)

Step 1: Set Up the Upscaling Model

from PIL import Image
import subprocess
import os

def upscale_with_realesrgan(input_path, output_path, scale=4):
    """Use Real-ESRGAN for high-quality upscaling."""
    subprocess.run([
        "python", "realesrgan/inference_realesrgan.py",
        "-n", "RealESRGAN_x4plus",
        "-i", input_path,
        "-o", output_path,
        "-s", str(scale)
    ], check=True)
    return output_path

def upscale_with_api(input_path, output_path, api_key):
    """Use a cloud API for upscaling."""
    import requests
    with open(input_path, "rb") as f:
        response = requests.post(
            "https://api.upscale.example.com/v1/upscale",
            headers={"Authorization": f"Bearer {api_key}"},
            files={"image": f},
            data={"scale": 4}
        )
    with open(output_path, "wb") as f:
        f.write(response.content)
    return output_path

Step 2: Build the Processing Pipeline

def process_image(input_path, output_folder, target_size=None):
    filename = os.path.basename(input_path)
    name, ext = os.path.splitext(filename)

    # Check if upscaling is needed
    img = Image.open(input_path)
    if min(img.size) >= 1080 and target_size is None:
        return {"status": "skipped", "reason": "already high resolution"}

    output_path = os.path.join(output_folder, f"{name}_upscaled.png")
    upscale_with_realesrgan(input_path, output_path)

    # Resize to target if specified
    if target_size:
        upscaled = Image.open(output_path)
        upscaled = upscaled.resize(target_size, Image.LANCZOS)
        upscaled.save(output_path)

    return {"status": "processed", "output": output_path}

Step 3: Batch Process a Folder

def batch_upscale(input_folder, output_folder, target_size=None):
    os.makedirs(output_folder, exist_ok=True)
    results = {"processed": 0, "skipped": 0, "failed": 0}

    for filename in os.listdir(input_folder):
        if not filename.lower().endswith((".png", ".jpg", ".jpeg", ".webp")):
            continue

        input_path = os.path.join(input_folder, filename)
        try:
            result = process_image(input_path, output_folder, target_size)
            if result["status"] == "processed":
                results["processed"] += 1
            else:
                results["skipped"] += 1
        except Exception as e:
            results["failed"] += 1
            log_error(filename, str(e))

    return results

Step 4: Add Quality Checks

def check_quality(original_path, upscaled_path):
    original = Image.open(original_path)
    upscaled = Image.open(upscaled_path)

    return {
        "original_size": original.size,
        "upscaled_size": upscaled.size,
        "scale_factor": upscaled.size[0] / original.size[0],
        "file_size_kb": os.path.getsize(upscaled_path) / 1024,
        "meets_minimum": min(upscaled.size) >= 1080
    }

Step 5: Serve via API

from flask import Flask, request, jsonify, send_file

app = Flask(__name__)

@app.route("/api/upscale", methods=["POST"])
def upscale_endpoint():
    if "image" not in request.files:
        return jsonify({"error": "No image provided"}), 400

    file = request.files["image"]
    input_path = f"temp/{file.filename}"
    file.save(input_path)

    scale = request.form.get("scale", 4, type=int)
    output_path = f"temp/{os.path.splitext(file.filename)[0]}_upscaled.png"

    upscale_with_realesrgan(input_path, output_path, scale=scale)
    quality = check_quality(input_path, output_path)

    return jsonify({"quality": quality, "download_url": f"/api/download/{os.path.basename(output_path)}"})

What to Build Next

Add format-specific optimization. JPEG images need denoising before upscaling. Screenshots need sharpening. Product photos need color correction. Detect the image type and apply the right preprocessing for the best results.

Related Reading

• AI for Content Creation at Scale - image quality as part of content production
• Build vs Buy: The AI Framework - building a custom pipeline vs using upscaling services
• The Pipeline Architecture - image processing as a pipeline pattern