AWS SageMaker Image Recognition: From Traditional ML to Deep Learning

# Traditional ML approach for image recognition
import cv2
import numpy as np
from sklearn.svm import SVC
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report
import matplotlib.pyplot as plt

def extract_image_features(image_path):
    """
    Extract traditional features from an image for ML classification
    """
    # Read image
    image = cv2.imread(image_path)
    image = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)

    # Resize for consistency
    image = cv2.resize(image, (128, 128))

    # 1. Histogram of Oriented Gradients (HOG)
    hog = cv2.HOGDescriptor()
    hog_features = hog.compute(image)

    # 2. Color histograms (if using color images)
    # color_image = cv2.imread(image_path)
    # color_hist = []
    # for i in range(3):  # BGR channels
    #     hist = cv2.calcHist([color_image], [i], None, [256], [0, 256])
    #     color_hist.extend(hist.flatten())
    # color_hist = np.array(color_hist)

    # 3. Texture features (Gabor filters)
    gabor_features = []
    for theta in [0, np.pi/4, np.pi/2, 3*np.pi/4]:
        kernel = cv2.getGaborKernel((21, 21), 8.0, theta, 10.0, 0.5, 0, ktype=cv2.CV_32F)
        filtered = cv2.filter2D(image, cv2.CV_8UC3, kernel)
        gabor_features.extend(filtered.flatten()[:1000])  # Limit features

    # 4. Edge detection features
    edges = cv2.Canny(image, 100, 200)
    edge_hist = cv2.calcHist([edges], [0], None, [32], [0, 256]).flatten()

    # Combine all features
    features = np.concatenate([
        hog_features.flatten(),
        np.array(gabor_features),
        edge_hist
    ])

    return features

print("Traditional feature extraction functions ready!")

# SVM implementation for image classification
def train_svm_classifier(X_train, y_train):
    """
    Train SVM classifier for image recognition
    """
    # SVM with RBF kernel (good for image features)
    svm_model = SVC(kernel='rbf', C=1.0, gamma='scale', random_state=42)

    print("Training SVM classifier...")
    svm_model.fit(X_train, y_train)

    return svm_model

# Example usage with CIFAR-10 subset
from sklearn.datasets import fetch_openml

# Load a small subset for demonstration
print("Loading CIFAR-10 dataset (subset)...")
# Note: In practice, you'd use the full dataset
# cifar = fetch_openml('CIFAR-10', version=1)
# For demo, we'll create synthetic features
np.random.seed(42)
n_samples = 1000
n_features = 5000  # Typical feature vector size after extraction

# Synthetic features (in practice, extract from real images)
X = np.random.randn(n_samples, n_features)
y = np.random.randint(0, 10, n_samples)  # 10 classes

# Split data
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# Train SVM
svm_model = train_svm_classifier(X_train, y_train)

# Evaluate
svm_predictions = svm_model.predict(X_test)
print("SVM Classification Report:")
print(classification_report(y_test, svm_predictions))

# Random Forest implementation
def train_rf_classifier(X_train, y_train):
    """
    Train Random Forest classifier for image recognition
    """
    rf_model = RandomForestClassifier(
        n_estimators=100,
        max_depth=20,
        min_samples_split=5,
        min_samples_leaf=2,
        random_state=42,
        n_jobs=-1
    )

    print("Training Random Forest classifier...")
    rf_model.fit(X_train, y_train)

    return rf_model

# Train Random Forest
rf_model = train_rf_classifier(X_train, y_train)

# Evaluate
rf_predictions = rf_model.predict(X_test)
print("Random Forest Classification Report:")
print(classification_report(y_test, rf_predictions))

# Feature importance analysis
feature_importance = rf_model.feature_importances_
print(f"Top 10 most important features: {np.argsort(feature_importance)[-10:]}")

# CNN implementation with TensorFlow/Keras
import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Conv2D, MaxPooling2D, Flatten, Dense, Dropout
from tensorflow.keras.optimizers import Adam
from tensorflow.keras.callbacks import EarlyStopping

def create_cnn_model(input_shape=(32, 32, 3), num_classes=10):
    """
    Create a CNN model for image classification
    """
    model = Sequential([
        # Convolutional layers
        Conv2D(32, (3, 3), activation='relu', input_shape=input_shape),
        MaxPooling2D((2, 2)),

        Conv2D(64, (3, 3), activation='relu'),
        MaxPooling2D((2, 2)),

        Conv2D(128, (3, 3), activation='relu'),
        MaxPooling2D((2, 2)),

        # Fully connected layers
        Flatten(),
        Dense(128, activation='relu'),
        Dropout(0.5),
        Dense(num_classes, activation='softmax')
    ])

    return model

# Create and compile model
cnn_model = create_cnn_model()
cnn_model.compile(
    optimizer=Adam(learning_rate=0.001),
    loss='categorical_crossentropy',
    metrics=['accuracy']
)

print("CNN model created!")
print(cnn_model.summary())

# Data preparation for CNN
from tensorflow.keras.utils import to_categorical
from tensorflow.keras.preprocessing.image import ImageDataGenerator

# Convert labels to categorical
y_train_cat = to_categorical(y_train, num_classes=10)
y_test_cat = to_categorical(y_test, num_classes=10)

# Create synthetic image data (32x32x3 RGB images)
X_train_images = np.random.randint(0, 255, (len(X_train), 32, 32, 3), dtype=np.uint8)
X_test_images = np.random.randint(0, 255, (len(X_test), 32, 32, 3), dtype=np.uint8)

# Normalize pixel values
X_train_images = X_train_images.astype('float32') / 255.0
X_test_images = X_test_images.astype('float32') / 255.0

# Data augmentation
datagen = ImageDataGenerator(
    rotation_range=15,
    width_shift_range=0.1,
    height_shift_range=0.1,
    horizontal_flip=True
)

# Training with early stopping
early_stopping = EarlyStopping(
    monitor='val_loss',
    patience=5,
    restore_best_weights=True
)

print("Training CNN...")
history = cnn_model.fit(
    datagen.flow(X_train_images, y_train_cat, batch_size=32),
    epochs=50,
    validation_data=(X_test_images, y_test_cat),
    callbacks=[early_stopping],
    verbose=1
)

# Evaluate CNN
cnn_loss, cnn_accuracy = cnn_model.evaluate(X_test_images, y_test_cat)
print(".4f")
print(".4f")

# Transfer learning with ResNet50
from tensorflow.keras.applications import ResNet50
from tensorflow.keras.models import Model

def create_transfer_learning_model(input_shape=(224, 224, 3), num_classes=10):
    """
    Create a transfer learning model using ResNet50
    """
    # Load pre-trained ResNet50
    base_model = ResNet50(
        weights='imagenet',
        include_top=False,
        input_shape=input_shape
    )

    # Freeze base model layers
    for layer in base_model.layers:
        layer.trainable = False

    # Add custom classification head
    x = base_model.output
    x = tf.keras.layers.GlobalAveragePooling2D()(x)
    x = Dense(512, activation='relu')(x)
    x = Dropout(0.5)(x)
    predictions = Dense(num_classes, activation='softmax')(x)

    # Create final model
    model = Model(inputs=base_model.input, outputs=predictions)

    return model

# Create transfer learning model
tl_model = create_transfer_learning_model()

# Compile with lower learning rate for fine-tuning
tl_model.compile(
    optimizer=Adam(learning_rate=0.0001),
    loss='categorical_crossentropy',
    metrics=['accuracy']
)

print("Transfer learning model created!")
print("Base ResNet50 layers frozen for feature extraction")

# AWS Rekognition example
import boto3
from PIL import Image
import io

def analyze_image_with_rekognition(image_path, rekognition_client):
    """
    Analyze image using AWS Rekognition
    """
    # Read image
    with open(image_path, 'rb') as image_file:
        image_bytes = image_file.read()

    # Detect labels
    labels_response = rekognition_client.detect_labels(
        Image={'Bytes': image_bytes},
        MaxLabels=10,
        MinConfidence=70
    )

    # Detect faces
    faces_response = rekognition_client.detect_faces(
        Image={'Bytes': image_bytes},
        Attributes=['ALL']
    )

    # Detect text
    text_response = rekognition_client.detect_text(
        Image={'Bytes': image_bytes}
    )

    return {
        'labels': labels_response['Labels'],
        'faces': faces_response['FaceDetails'],
        'text': text_response['TextDetections']
    }

# Initialize Rekognition client (requires AWS credentials)
# rekognition = boto3.client('rekognition', region_name='us-east-1')

# Example usage (commented out - requires actual image and AWS setup)
# results = analyze_image_with_rekognition('path/to/image.jpg', rekognition)
# print("Detected labels:", [label['Name'] for label in results['labels']])

# Custom model training with Rekognition Custom Labels
def create_rekognition_custom_model(project_name, bucket_name, s3_client, rekognition_client):
    """
    Create and train a custom Rekognition model
    """
    # Create project
    project_response = rekognition_client.create_project(
        ProjectName=project_name
    )
    project_arn = project_response['ProjectArn']

    # Assume training data is organized in S3 bucket
    # Structure: s3://bucket-name/training-data/class1/, class2/, etc.

    # Create dataset
    dataset_response = rekognition_client.create_dataset(
        DatasetType='TRAIN',
        ProjectArn=project_arn
    )

    # In practice, you'd upload images to S3 and create manifest files
    # Then train the model...

    return project_arn

# Example project creation (requires proper S3 setup)
# project_arn = create_rekognition_custom_model('my-custom-model', 'my-bucket', s3_client, rekognition_client)

# Compare algorithm performance
import pandas as pd

# Synthetic performance data (in practice, use real results)
performance_data = {
    'Algorithm': ['SVM', 'Random Forest', 'CNN (Custom)', 'Transfer Learning', 'AWS Rekognition'],
    'Accuracy': [0.72, 0.78, 0.85, 0.92, 0.89],
    'Training Time (hours)': [0.5, 2, 24, 8, 0],
    'Dataset Size Needed': ['Small', 'Medium', 'Large', 'Medium', 'Any'],
    'Interpretability': ['Medium', 'High', 'Low', 'Low', 'Medium'],
    'Setup Complexity': ['Low', 'Low', 'High', 'Medium', 'Low']
}

performance_df = pd.DataFrame(performance_data)
print("Algorithm Comparison:")
print(performance_df.to_string(index=False))

# Deploy CNN model to SageMaker endpoint
import sagemaker
from sagemaker.tensorflow import TensorFlowModel

def deploy_cnn_to_sagemaker(model_path, role_arn):
    """
    Deploy trained CNN model to SageMaker endpoint
    """
    # Create SageMaker model
    sagemaker_model = TensorFlowModel(
        model_data=model_path,
        role=role_arn,
        framework_version='2.8'
    )

    # Deploy to endpoint
    predictor = sagemaker_model.deploy(
        initial_instance_count=1,
        instance_type='ml.m5.large'
    )

    return predictor

# Example deployment (requires model artifacts in S3)
# predictor = deploy_cnn_to_sagemaker('s3://my-bucket/models/cnn-model.tar.gz', role_arn)

# Production Rekognition integration
def process_images_batch(image_paths, rekognition_client, output_bucket):
    """
    Process batch of images with Rekognition
    """
    results = []

    for image_path in image_paths:
        try:
            # Analyze image
            analysis = analyze_image_with_rekognition(image_path, rekognition_client)

            # Store results
            result = {
                'image_path': image_path,
                'labels': analysis['labels'],
                'faces': len(analysis['faces']),
                'text_detected': len(analysis['text']) > 0
            }

            results.append(result)

        except Exception as e:
            print(f"Error processing {image_path}: {e}")

    return results

# Batch processing example
# image_batch = ['image1.jpg', 'image2.jpg', 'image3.jpg']
# batch_results = process_images_batch(image_batch, rekognition_client, 'results-bucket')