AI-Powered Serverless Image Processing Pipeline — Part 5: Orchestrating Tasks with Shell Scripts & S3

  Bucket Name
  Region
  Purpose




  `image-source‑yourname‑2025`
  eu‑west‑1
  Raw uploads from users or external systems.


  `image-processed‑yourname‑2025`
  eu‑west‑1
  AI‑enhanced, resized, or annotated images.

#!/usr/bin/env bash
#
# orchestrate.sh – Detect, process, and move images across S3 buckets.
# Author: Vijay Vinoth (Lead Programmer Analyst)
# Date:   2026‑09‑04
#
# Usage:
#   ./orchestrate.sh            # Run once (ideal for cron)
#   ./orchestrate.sh --watch    # Run as a daemon (checks every 30 s)
#
# Exit on any unhandled error
set -euo pipefail

# ----------------------------------------------------------------------
# Configuration – adapt these values to your environment
# ----------------------------------------------------------------------
SOURCE_BUCKET="image-source-yourname-2025"
PROCESSED_BUCKET="image-processed-yourname-2025"
SQS_URL="https://sqs.eu-west-1.amazonaws.com/123456789012/ImagePipelineQueue"
LAMBDA_ARN="arn:aws:lambda:eu-west-1:123456789012:function:PostProcessMetadata"
TMP_DIR="/tmp/image-pipeline"
MAX_PARALLEL=8               # Tune based on CPU / memory
WATCH_INTERVAL=30            # Seconds between polls in watch mode

# ----------------------------------------------------------------------
# Helper functions
# ----------------------------------------------------------------------
log() {
    local level=$1
    shift
    printf "[%s] %s: %s\n" "$(date +%Y-%m-%dT%H:%M:%S%z)" "$level" "$*"
}

error_exit() {
    log "ERROR" "$*"
    exit 1
}

# Ensure temporary workspace exists
mkdir -p "$TMP_DIR"

# ----------------------------------------------------------------------
# Step 1 – List new objects in the source bucket
# ----------------------------------------------------------------------
list_new_objects() {
    # We keep a simple checkpoint file that stores the latest processed
    # object key. This makes the script idempotent across restarts.
    local checkpoint="${TMP_DIR}/last_key.txt"
    local start_after=""

    if [[ -f "$checkpoint" ]]; then
        start_after=$(0].Key" \
        --output text
}

# ----------------------------------------------------------------------
# Step 2 – Process a single image (download → AI → upload)
# ----------------------------------------------------------------------
process_one() {
    local key=$1
    local local_raw="${TMP_DIR}/$(basename "$key")"
    local local_out="${TMP_DIR}/processed-$(basename "$key")"

    # 1️⃣ Download raw image
    log "INFO" "Downloading s3://$SOURCE_BUCKET/$key"
    aws s3 cp "s3://$SOURCE_BUCKET/$key" "$local_raw" >/dev/null

    # 2️⃣ Run the Python AI wrapper (see process_image.py)
    log "INFO" "Running AI model on $local_raw"
    python3 process_image.py "$local_raw" "$local_out"

    # 3️⃣ Upload processed image
    local processed_key="processed/${key}"
    log "INFO" "Uploading processed image to s3://$PROCESSED_BUCKET/$processed_key"
    aws s3 cp "$local_out" "s3://$PROCESSED_BUCKET/$processed_key" >/dev/null

    # 4️⃣ Notify downstream Lambda via SQS (metadata extraction)
    local payload=$(jq -n \
        --arg src "s3://$SOURCE_BUCKET/$key" \
        --arg dst "s3://$PROCESSED_BUCKET/$processed_key" \
        '{source: $src, destination: $dst, timestamp: now}')
    log "INFO" "Sending SQS message"
    aws sqs send-message --queue-url "$SQS_URL" --message-body "$payload" >/dev/null

    # 5️⃣ Optional: directly invoke a Lambda for real‑time hooks
    # aws lambda invoke --function-name "$LAMBDA_ARN" --payload "$payload" /dev/null

    # Clean up local files
    rm -f "$local_raw" "$local_out"

    # Update checkpoint – the key we just processed becomes the new start point.
    echo "$key" > "${TMP_DIR}/last_key.txt"
    log "INFO" "Finished processing $key"
}

export -f log error_exit process_one
export SOURCE_BUCKET PROCESSED_BUCKET SQS_URL TMP_DIR LAMBDA_ARN

# ----------------------------------------------------------------------
# Main driver – either one‑shot or watch mode
# ----------------------------------------------------------------------
run_once() {
    mapfile -t keys  Image.Image:
    """
    Simulate an AI operation: enhance contrast, apply a subtle sharpening filter,
    and add a watermark that mentions the model name.
    """
    # Contrast boost
    enhancer = ImageEnhance.Contrast(img)
    img = enhancer.enhance(1.25)

    # Sharpen
    img = img.filter(ImageFilter.UnsharpMask(radius=2, percent=150, threshold=3))

    # Watermark
    watermark = Image.new("RGBA", img.size)
    # In a real scenario you would render text with a library like Pillow‑ImageDraw.
    # Here we just embed a tiny transparent overlay to keep the example simple.
    return img

def process(in_path: Path, out_path: Path) -> None:
    if not in_path.is_file():
        raise FileNotFoundError(f"Input file not found: {in_path}")

    # Load image
    img = Image.open(in_path).convert("RGB")

    # Run the AI model (replace dummy with real inference)
    processed = dummy_ai_transform(img)

    # Save with reasonable compression
    processed.save(out_path, format="JPEG", quality=85, optimize=True)

if __name__ == "__main__":
    if len(sys.argv) != 3:
        sys.stderr.write("Usage: process_image.py  \\n")
        sys.exit(1)

    input_path = Path(sys.argv[1])
    output_path = Path(sys.argv[2])

    try:
        process(input_path, output_path)
        print(f"Processed {input_path.name} → {output_path.name}")
    except Exception as exc:
        sys.stderr.write(f"Error processing image: {exc}\\n")
        sys.exit(1)

chmod +x orchestrate.sh process_image.py

# 1️⃣ Upload a test image to the source bucket
aws s3 cp ./sample.jpg "s3://image-source-yourname-2025/test/sample.jpg"

# 2️⃣ Run the orchestrator once (ideal for a cron job)
./orchestrate.sh

# 3️⃣ Verify the processed image appears in the destination bucket
aws s3 ls "s3://image-processed-yourname-2025/processed/test/"

# 4️⃣ (Optional) Start watch mode for continuous processing
./orchestrate.sh --watch

  Metric
  Typical Value (per 10 k images)
  Cost Implication




  Average image size
  1 MiB
  S3 storage ≈ $0.23 (10 GB)


  Lambda invocations (metadata step)
  10 k
  Free tier covers first 1 M requests


  CPU on orchestrator host
  ~2 vCPU, 4 GiB RAM
  EC2 spot instance ≈ $0.011 /h


  Data transfer (S3 ↔ EC2)
  20 GiB (in + out)
  Free within same region

User Upload → S3 (source bucket) ──► orchestrate.sh (watch/cron) ──►
   download → Python AI model → upload → S3 (processed bucket)
   └─► SQS message ──► Lambda (metadata, DB write, notifications)