Deep Learning CNN for Image Recognition: From Theory to Production

import tensorflow as tf
from tensorflow.keras import layers, models

def explain_cnn_layers():
    """
    Demonstrate the purpose of each CNN layer type.
    """

    # Convolutional Layer: Detects local patterns
    conv_layer = layers.Conv2D(
        filters=32,           # Number of feature detectors
        kernel_size=(3, 3),   # Size of sliding window
        strides=(1, 1),       # Step size
        padding='same',       # Preserve spatial dimensions
        activation='relu'     # Non-linearity
    )

    # Pooling Layer: Reduces spatial dimensions
    pool_layer = layers.MaxPooling2D(
        pool_size=(2, 2),     # Downsampling factor
        strides=(2, 2)        # Non-overlapping windows
    )

    # Batch Normalization: Stabilizes training
    bn_layer = layers.BatchNormalization()

    # Dropout: Prevents overfitting
    dropout_layer = layers.Dropout(rate=0.5)

    # Dense Layer: Classification head
    dense_layer = layers.Dense(
        units=128,
        activation='relu'
    )

    return conv_layer, pool_layer, bn_layer, dropout_layer, dense_layer

import tensorflow as tf
from tensorflow.keras import layers, models, regularizers

def build_image_classifier(
    input_shape=(224, 224, 3),
    num_classes=10,
    dropout_rate=0.5
):
    """
    Build a production-ready CNN for image classification.

    Architecture follows VGG-style design with modern enhancements:
    - Batch normalization after convolutions
    - Dropout for regularization
    - Global average pooling instead of flatten

    Args:
        input_shape: Tuple of (height, width, channels)
        num_classes: Number of classification categories
        dropout_rate: Dropout probability

    Returns:
        Compiled Keras model
    """

    model = models.Sequential([
        # Input layer
        layers.Input(shape=input_shape),

        # Block 1: Initial feature extraction
        layers.Conv2D(64, (3, 3), padding='same'),
        layers.BatchNormalization(),
        layers.Activation('relu'),
        layers.Conv2D(64, (3, 3), padding='same'),
        layers.BatchNormalization(),
        layers.Activation('relu'),
        layers.MaxPooling2D((2, 2)),
        layers.Dropout(0.25),

        # Block 2: Intermediate features
        layers.Conv2D(128, (3, 3), padding='same'),
        layers.BatchNormalization(),
        layers.Activation('relu'),
        layers.Conv2D(128, (3, 3), padding='same'),
        layers.BatchNormalization(),
        layers.Activation('relu'),
        layers.MaxPooling2D((2, 2)),
        layers.Dropout(0.25),

        # Block 3: Complex patterns
        layers.Conv2D(256, (3, 3), padding='same'),
        layers.BatchNormalization(),
        layers.Activation('relu'),
        layers.Conv2D(256, (3, 3), padding='same'),
        layers.BatchNormalization(),
        layers.Activation('relu'),
        layers.Conv2D(256, (3, 3), padding='same'),
        layers.BatchNormalization(),
        layers.Activation('relu'),
        layers.MaxPooling2D((2, 2)),
        layers.Dropout(0.25),

        # Block 4: High-level features
        layers.Conv2D(512, (3, 3), padding='same'),
        layers.BatchNormalization(),
        layers.Activation('relu'),
        layers.Conv2D(512, (3, 3), padding='same'),
        layers.BatchNormalization(),
        layers.Activation('relu'),
        layers.MaxPooling2D((2, 2)),
        layers.Dropout(0.25),

        # Classification head
        layers.GlobalAveragePooling2D(),
        layers.Dense(512, activation='relu',
                    kernel_regularizer=regularizers.l2(0.01)),
        layers.Dropout(dropout_rate),
        layers.Dense(num_classes, activation='softmax')
    ])

    return model

# Create the model
model = build_image_classifier(
    input_shape=(224, 224, 3),
    num_classes=10
)

model.summary()

import tensorflow as tf

def create_data_pipeline(
    data_dir,
    batch_size=32,
    image_size=(224, 224),
    augment=True
):
    """
    Create an efficient data pipeline with augmentation.

    Uses tf.data for optimal GPU utilization.
    """

    # Load dataset from directory structure
    train_ds = tf.keras.utils.image_dataset_from_directory(
        data_dir,
        validation_split=0.2,
        subset="training",
        seed=42,
        image_size=image_size,
        batch_size=batch_size
    )

    val_ds = tf.keras.utils.image_dataset_from_directory(
        data_dir,
        validation_split=0.2,
        subset="validation",
        seed=42,
        image_size=image_size,
        batch_size=batch_size
    )

    # Get class names
    class_names = train_ds.class_names
    print(f"Classes: {class_names}")

    # Normalization layer
    normalization_layer = layers.Rescaling(1./255)

    # Data augmentation for training
    data_augmentation = tf.keras.Sequential([
        layers.RandomFlip("horizontal"),
        layers.RandomRotation(0.1),
        layers.RandomZoom(0.1),
        layers.RandomContrast(0.1),
        layers.RandomTranslation(0.1, 0.1),
    ])

    def prepare_train(image, label):
        image = normalization_layer(image)
        if augment:
            image = data_augmentation(image, training=True)
        return image, label

    def prepare_val(image, label):
        image = normalization_layer(image)
        return image, label

    # Apply preprocessing with prefetching
    AUTOTUNE = tf.data.AUTOTUNE

    train_ds = train_ds.map(prepare_train, num_parallel_calls=AUTOTUNE)
    train_ds = train_ds.cache().shuffle(1000).prefetch(AUTOTUNE)

    val_ds = val_ds.map(prepare_val, num_parallel_calls=AUTOTUNE)
    val_ds = val_ds.cache().prefetch(AUTOTUNE)

    return train_ds, val_ds, class_names

# Advanced augmentation with Albumentations
import albumentations as A
from albumentations.tensorflow import ToTensorV2

def get_advanced_augmentation():
    """
    Advanced augmentation pipeline using Albumentations.
    """
    return A.Compose([
        A.RandomResizedCrop(224, 224, scale=(0.8, 1.0)),
        A.HorizontalFlip(p=0.5),
        A.ShiftScaleRotate(
            shift_limit=0.1,
            scale_limit=0.1,
            rotate_limit=15,
            p=0.5
        ),
        A.OneOf([
            A.GaussNoise(var_limit=(10.0, 50.0)),
            A.GaussianBlur(blur_limit=(3, 7)),
            A.MotionBlur(blur_limit=7),
        ], p=0.3),
        A.ColorJitter(
            brightness=0.2,
            contrast=0.2,
            saturation=0.2,
            hue=0.1,
            p=0.5
        ),
        A.CoarseDropout(
            max_holes=8,
            max_height=32,
            max_width=32,
            p=0.3
        ),
        A.Normalize(
            mean=[0.485, 0.456, 0.406],
            std=[0.229, 0.224, 0.225]
        ),
        ToTensorV2()
    ])

from tensorflow.keras.callbacks import (
    ModelCheckpoint, EarlyStopping,
    ReduceLROnPlateau, TensorBoard
)
from tensorflow.keras.optimizers import Adam
from tensorflow.keras.losses import SparseCategoricalCrossentropy

def train_model(model, train_ds, val_ds, epochs=100):
    """
    Train the CNN with production-grade configuration.
    """

    # Compile model
    model.compile(
        optimizer=Adam(learning_rate=0.001),
        loss=SparseCategoricalCrossentropy(),
        metrics=['accuracy', 'top_k_categorical_accuracy']
    )

    # Define callbacks
    callbacks = [
        ModelCheckpoint(
            'best_model.keras',
            monitor='val_accuracy',
            save_best_only=True,
            mode='max',
            verbose=1
        ),
        EarlyStopping(
            monitor='val_loss',
            patience=15,
            restore_best_weights=True,
            verbose=1
        ),
        ReduceLROnPlateau(
            monitor='val_loss',
            factor=0.5,
            patience=5,
            min_lr=1e-7,
            verbose=1
        ),
        TensorBoard(
            log_dir='./logs',
            histogram_freq=1,
            write_graph=True,
            write_images=True
        )
    ]

    # Train
    history = model.fit(
        train_ds,
        validation_data=val_ds,
        epochs=epochs,
        callbacks=callbacks,
        verbose=1
    )

    return history

# Train the model
history = train_model(model, train_ds, val_ds, epochs=100)

import tensorflow as tf
import math

def cosine_decay_with_warmup(
    global_step,
    learning_rate_base,
    total_steps,
    warmup_steps=1000
):
    """
    Cosine decay learning rate schedule with linear warmup.
    """
    if global_step < warmup_steps:
        # Linear warmup
        lr = learning_rate_base * (global_step / warmup_steps)
    else:
        # Cosine decay
        progress = (global_step - warmup_steps) / (total_steps - warmup_steps)
        lr = learning_rate_base * 0.5 * (1 + math.cos(math.pi * progress))

    return lr

class CosineDecayWarmup(tf.keras.optimizers.schedules.LearningRateSchedule):
    """Custom learning rate schedule with warmup."""

    def __init__(self, learning_rate_base, total_steps, warmup_steps=1000):
        super().__init__()
        self.learning_rate_base = learning_rate_base
        self.total_steps = total_steps
        self.warmup_steps = warmup_steps

    def __call__(self, step):
        return cosine_decay_with_warmup(
            step,
            self.learning_rate_base,
            self.total_steps,
            self.warmup_steps
        )

# Usage
lr_schedule = CosineDecayWarmup(
    learning_rate_base=0.001,
    total_steps=10000,
    warmup_steps=1000
)

optimizer = Adam(learning_rate=lr_schedule)

from tensorflow.keras.applications import (
    ResNet50, EfficientNetB0, VGG16
)

def build_transfer_model(
    base_model_name='resnet50',
    input_shape=(224, 224, 3),
    num_classes=10,
    trainable_layers=20
):
    """
    Build a transfer learning model using pre-trained weights.

    Args:
        base_model_name: One of 'resnet50', 'efficientnet', 'vgg16'
        input_shape: Input image dimensions
        num_classes: Number of output classes
        trainable_layers: Number of layers to fine-tune

    Returns:
        Compiled Keras model
    """

    # Select base model
    base_models = {
        'resnet50': ResNet50,
        'efficientnet': EfficientNetB0,
        'vgg16': VGG16
    }

    BaseModel = base_models[base_model_name]

    # Load pre-trained model without top layers
    base_model = BaseModel(
        weights='imagenet',
        include_top=False,
        input_shape=input_shape
    )

    # Freeze early layers
    for layer in base_model.layers[:-trainable_layers]:
        layer.trainable = False

    # Build classification head
    model = models.Sequential([
        base_model,
        layers.GlobalAveragePooling2D(),
        layers.Dense(512, activation='relu'),
        layers.Dropout(0.5),
        layers.Dense(256, activation='relu'),
        layers.Dropout(0.3),
        layers.Dense(num_classes, activation='softmax')
    ])

    return model

# Create transfer learning model
transfer_model = build_transfer_model(
    base_model_name='efficientnet',
    num_classes=10,
    trainable_layers=30
)

import numpy as np
from sklearn.metrics import (
    classification_report, confusion_matrix
)
import seaborn as sns
import matplotlib.pyplot as plt

def evaluate_model(model, test_ds, class_names):
    """
    Comprehensive model evaluation with visualizations.
    """

    # Get predictions
    y_true = []
    y_pred = []
    y_pred_proba = []

    for images, labels in test_ds:
        predictions = model.predict(images, verbose=0)
        y_true.extend(labels.numpy())
        y_pred.extend(np.argmax(predictions, axis=1))
        y_pred_proba.extend(predictions)

    y_true = np.array(y_true)
    y_pred = np.array(y_pred)
    y_pred_proba = np.array(y_pred_proba)

    # Classification report
    print("Classification Report:")
    print(classification_report(y_true, y_pred, target_names=class_names))

    # Confusion matrix
    cm = confusion_matrix(y_true, y_pred)

    plt.figure(figsize=(12, 10))
    sns.heatmap(
        cm, annot=True, fmt='d', cmap='Blues',
        xticklabels=class_names,
        yticklabels=class_names
    )
    plt.xlabel('Predicted')
    plt.ylabel('Actual')
    plt.title('Confusion Matrix')
    plt.tight_layout()
    plt.savefig('confusion_matrix.png', dpi=150)

    # Per-class accuracy
    per_class_acc = cm.diagonal() / cm.sum(axis=1)

    print("\nPer-Class Accuracy:")
    for name, acc in zip(class_names, per_class_acc):
        print(f"  {name}: {acc:.2%}")

    return {
        'y_true': y_true,
        'y_pred': y_pred,
        'y_pred_proba': y_pred_proba,
        'confusion_matrix': cm
    }

# Evaluate
results = evaluate_model(model, test_ds, class_names)

def visualize_predictions(model, test_ds, class_names, num_samples=16):
    """
    Visualize model predictions on sample images.
    """
    images, labels = next(iter(test_ds.take(1)))
    predictions = model.predict(images)

    fig, axes = plt.subplots(4, 4, figsize=(16, 16))

    for i, ax in enumerate(axes.flat):
        if i >= num_samples:
            break

        img = images[i].numpy()
        true_label = class_names[labels[i]]
        pred_label = class_names[np.argmax(predictions[i])]
        confidence = np.max(predictions[i])

        ax.imshow(img)

        color = 'green' if true_label == pred_label else 'red'
        ax.set_title(
            f"True: {true_label}\nPred: {pred_label} ({confidence:.2%})",
            color=color
        )
        ax.axis('off')

    plt.tight_layout()
    plt.savefig('prediction_samples.png', dpi=150)

visualize_predictions(model, test_ds, class_names)

# Save in TensorFlow SavedModel format
model.save('saved_model/image_classifier')

# Convert to TensorFlow Lite for mobile deployment
converter = tf.lite.TFLiteConverter.from_keras_model(model)
converter.optimizations = [tf.lite.Optimize.DEFAULT]
tflite_model = converter.convert()

with open('model.tflite', 'wb') as f:
    f.write(tflite_model)

# Export to ONNX for cross-platform deployment
import tf2onnx

spec = (tf.TensorSpec((None, 224, 224, 3), tf.float32, name="input"),)
onnx_model, _ = tf2onnx.convert.from_keras(model, input_signature=spec)

with open('model.onnx', 'wb') as f:
    f.write(onnx_model.SerializeToString())

from flask import Flask, request, jsonify
import tensorflow as tf
import numpy as np
from PIL import Image
import io

app = Flask(__name__)

# Load model
model = tf.keras.models.load_model('saved_model/image_classifier')

# Class names
class_names = ['class_0', 'class_1', 'class_2', ...]  # Your classes

def preprocess_image(image_bytes):
    """Preprocess image for model inference."""
    image = Image.open(io.BytesIO(image_bytes))
    image = image.resize((224, 224))
    image = np.array(image) / 255.0
    image = np.expand_dims(image, axis=0)
    return image

@app.route('/predict', methods=['POST'])
def predict():
    """Image classification endpoint."""
    if 'image' not in request.files:
        return jsonify({'error': 'No image provided'}), 400

    image_file = request.files['image']
    image_bytes = image_file.read()

    # Preprocess
    image = preprocess_image(image_bytes)

    # Predict
    predictions = model.predict(image)[0]

    # Format response
    results = [
        {'class': class_names[i], 'confidence': float(predictions[i])}
        for i in range(len(class_names))
    ]
    results.sort(key=lambda x: x['confidence'], reverse=True)

    return jsonify({
        'prediction': results[0]['class'],
        'confidence': results[0]['confidence'],
        'all_predictions': results[:5]
    })

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

# Post-training quantization
converter = tf.lite.TFLiteConverter.from_keras_model(model)
converter.optimizations = [tf.lite.Optimize.DEFAULT]

# Representative dataset for calibration
def representative_dataset():
    for images, _ in train_ds.take(100):
        yield [images]

converter.representative_dataset = representative_dataset
converter.target_spec.supported_ops = [tf.lite.OpsSet.TFLITE_BUILTINS_INT8]
converter.inference_input_type = tf.uint8
converter.inference_output_type = tf.uint8

quantized_model = converter.convert()