Streamlined AI/ML Image Processing with Cloud Functions and Cloud Vision API: Automating Image Annotation at Scale

Introduction:

In the world of artificial intelligence and machine learning (AI/ML), processing and classifying images is a common task for applications in e-commerce, security, and even content moderation. Google Cloud offers a powerful solution to automate image annotation and classification using serverless technologies like Cloud Functions and Cloud Vision API.

This blog will walk you through how to implement a serverless, highly scalable solution for image recognition and classification. We’ll deploy Cloud Functions to handle both individual image uploads via an HTTP-triggered REST API and batch image uploads through Cloud Storage. With Cloud Vision API, we’ll automate the process of labeling and annotating images. The annotations will be stored back in Cloud Storage as JSON, ready for further analysis or downstream consumption.

Overview of the Workflow:

1. REST API:
2. Batch Upload:

Step-by-Step Guide

Step 1: Enable Necessary APIs

Before we start coding, make sure you have enabled the required APIs in your Google Cloud project:

gcloud services enable cloudfunctions.googleapis.com vision.googleapis.com storage.googleapis.com        

Step 2: Create a Cloud Storage Bucket

We’ll use Cloud Storage for batch image uploads, so create a bucket to store your images.

gsutil mb gs://your-bucket-name        

Step 3: Write the Cloud Function to Handle Image Processing

The following Python code defines a Cloud Function that handles both the REST API calls for single image annotations and batch image uploads triggered by Cloud Storage events. It interacts with the Cloud Vision API to annotate the images.

cloud_function_main.py

import os
from google.cloud import vision
from google.cloud import storage
from flask import Flask, request, jsonify

app = Flask(__name__)
vision_client = vision.ImageAnnotatorClient()
storage_client = storage.Client()

def annotate_image(image_content, features=None):
    image = vision.Image(content=image_content)
    
    # Use default feature (label detection) if no specific features are provided
    response = vision_client.annotate_image({
        'image': image,
        'features': features or [{'type': vision.Feature.Type.LABEL_DETECTION}]
    })
    
    return response

# REST API to process single image
@app.route('/process-image', methods=['POST'])
def process_image():
    data = request.get_json()
    image_url = data.get('image_url')

    if not image_url:
        return jsonify({"error": "Image URL not provided"}), 400

    bucket_name, file_name = parse_gcs_url(image_url)
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(file_name)
    image_content = blob.download_as_bytes()

    # Call the annotation function
    features = parse_features(data.get('features'))
    response = annotate_image(image_content, features)

    return jsonify(response)

def parse_gcs_url(gcs_url):
    """Parses a Cloud Storage URL into bucket and file name."""
    gcs_url = gcs_url.replace("gs://", "")
    bucket_name, file_name = gcs_url.split("/", 1)
    return bucket_name, file_name

def parse_features(features_list):
    """Parses and returns specific features for Cloud Vision API."""
    if not features_list:
        return None
    
    features = []
    for feature_name in features_list:
        if feature_name == 'LABEL_DETECTION':
            features.append({'type': vision.Feature.Type.LABEL_DETECTION})
        elif feature_name == 'TEXT_DETECTION':
            features.append({'type': vision.Feature.Type.TEXT_DETECTION})
        # Add more features as needed
    return features

# Cloud Function triggered by Cloud Storage for batch processing
def process_image_from_storage(event, context):
    """Triggered by Cloud Storage when a new file is uploaded."""
    bucket_name = event['bucket']
    file_name = event['name']

    # Download the image from Cloud Storage
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(file_name)
    image_content = blob.download_as_bytes()

    # Annotate the image using Cloud Vision
    response = annotate_image(image_content)

    # Store the annotation as a JSON file in Cloud Storage
    annotation_blob = bucket.blob(f"annotations/{file_name}.json")
    annotation_blob.upload_from_string(response)

    print(f"Processed and stored annotations for {file_name}")

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080)
        

Step 4: Deploy the Cloud Functions

1. Deploy the REST API Cloud Function for individual image processing:

gcloud functions deploy process-image \
--runtime python310 \
--trigger-http \
--allow-unauthenticated        

2. Deploy the Cloud Function for batch image processing, triggered by Cloud Storage uploads:

gcloud functions deploy process_image_from_storage \
--runtime python310 \
--trigger-resource=your-bucket-name \
--trigger-event=google.storage.object.finalize        

Step 5: Test the REST API

Once your Cloud Functions are deployed, you can test the REST API by making a POST request. You can pass the Cloud Storage URL of the image you want to process, along with optional annotation features (like label detection or text detection).

curl -X POST https://REGION-PROJECT_ID.cloudfunctions.net/process-image \
-H "Content-Type: application/json" \
-d '{
  "image_url": "gs://your-bucket-name/your-image.jpg",
  "features": ["LABEL_DETECTION", "TEXT_DETECTION"]
}'        

This will return a JSON response containing the image annotations.

Step 6: Test Batch Processing via Cloud Storage

To test the batch processing functionality, simply upload an image to the Cloud Storage bucket.

gsutil cp your-image.jpg gs://your-bucket-name/        

The Cloud Function will be triggered automatically, and it will annotate the image using Cloud Vision. The annotations will be saved in the annotations/ folder in the same Cloud Storage bucket as a JSON file.

Terraform Option: Automate Deployment

You can also automate the deployment of this entire solution using Terraform. Google Cloud offers a Jump Start solution that you can directly deploy through the console or via GitHub.

Jump Start Solution Repo: https://github.com/GoogleCloudPlatform/terraform-ml-image-annotation-gcf/tree/sic-jss/infra

To deploy via Terraform:

git clone https://github.com/GoogleCloudPlatform/terraform-ml-image-annotation-gcf.git
cd terraform-ml-image-annotation-gcf/infra
terraform init
terraform apply        

Conclusion

By integrating Cloud Functions and the Cloud Vision API, we’ve built a highly scalable, serverless solution for image recognition and annotation. Whether you're handling individual image uploads via a REST API or batch uploads through Cloud Storage, this architecture allows you to automate the entire process with minimal setup. Additionally, the use of Terraform makes deployment quick and efficient, allowing you to focus on building solutions instead of infrastructure management.

This solution is ideal for e-commerce platforms, content moderation systems, or any use case that requires scalable image processing with AI/ML.