Keras

graph TD
    A[Keras] --> B[High-Level API]
    A --> C[Backend Integration]
    A --> D[Model Types]
    A --> E[Training Utilities]

    B --> B1[Layers]
    B --> B2[Models]
    B --> B3[Optimizers]
    B --> B4[Losses]
    B --> B5[Metrics]

    C --> C1[TensorFlow]
    C --> C2[Theano]
    C --> C3[CNTK]

    D --> D1[Sequential]
    D --> D2[Functional API]
    D --> D3[Model Subclassing]

    E --> E1[Callbacks]
    E --> E2[Data Utilities]
    E --> E3[Preprocessing]
    E --> E4[Deployment]

    style A fill:#ff6b6b,stroke:#333
    style B fill:#4ecdc4,stroke:#333
    style C fill:#f9ca24,stroke:#333
    style D fill:#6c5ce7,stroke:#333
    style E fill:#a0e7e5,stroke:#333

import numpy as np
import matplotlib.pyplot as plt
from tensorflow import keras
from tensorflow.keras import layers
from tensorflow.keras.datasets import mnist

# 1. Load and prepare data
print("Loading and preparing data...")
(x_train, y_train), (x_test, y_test) = mnist.load_data()

# Normalize pixel values to [0, 1]
x_train = x_train.astype("float32") / 255.0
x_test = x_test.astype("float32") / 255.0

# Reshape images to include channel dimension
x_train = np.expand_dims(x_train, -1)
x_test = np.expand_dims(x_test, -1)

# Convert labels to one-hot encoding
num_classes = 10
y_train = keras.utils.to_categorical(y_train, num_classes)
y_test = keras.utils.to_categorical(y_test, num_classes)

# 2. Build the model using Sequential API
print("Building model with Sequential API...")
model = keras.Sequential([
    layers.Conv2D(32, kernel_size=(3, 3), activation="relu", input_shape=(28, 28, 1)),
    layers.MaxPooling2D(pool_size=(2, 2)),
    layers.Conv2D(64, kernel_size=(3, 3), activation="relu"),
    layers.MaxPooling2D(pool_size=(2, 2)),
    layers.Flatten(),
    layers.Dropout(0.5),
    layers.Dense(num_classes, activation="softmax")
])

model.summary()

# 3. Compile the model
print("Compiling the model...")
model.compile(
    loss="categorical_crossentropy",
    optimizer="adam",
    metrics=["accuracy"]
)

# 4. Train the model
print("Training the model...")
batch_size = 128
epochs = 5

history = model.fit(
    x_train, y_train,
    batch_size=batch_size,
    epochs=epochs,
    validation_split=0.1
)

# 5. Evaluate the model
print("Evaluating the model...")
score = model.evaluate(x_test, y_test, verbose=0)
print(f"Test loss: {score[0]:.4f}")
print(f"Test accuracy: {score[1]:.4f}")

# 6. Make predictions
print("Making predictions...")
predictions = model.predict(x_test[:5])
predicted_classes = np.argmax(predictions, axis=1)
true_classes = np.argmax(y_test[:5], axis=1)

print("\nSample predictions:")
for i in range(5):
    print(f"Predicted: {predicted_classes[i]}, True: {true_classes[i]}")

# 7. Visualize training history
plt.figure(figsize=(12, 4))

plt.subplot(1, 2, 1)
plt.plot(history.history['accuracy'], label='Training Accuracy')
plt.plot(history.history['val_accuracy'], label='Validation Accuracy')
plt.title('Training and Validation Accuracy')
plt.xlabel('Epoch')
plt.ylabel('Accuracy')
plt.legend()

plt.subplot(1, 2, 2)
plt.plot(history.history['loss'], label='Training Loss')
plt.plot(history.history['val_loss'], label='Validation Loss')
plt.title('Training and Validation Loss')
plt.xlabel('Epoch')
plt.ylabel('Loss')
plt.legend()

plt.tight_layout()
plt.show()

# Functional API example - more flexible than Sequential
print("\nBuilding model with Functional API...")

# Define input layer
inputs = keras.Input(shape=(28, 28, 1), name="input_layer")

# Define model architecture
x = layers.Conv2D(32, (3, 3), activation="relu", name="conv1")(inputs)
x = layers.MaxPooling2D((2, 2), name="pool1")(x)
x = layers.Conv2D(64, (3, 3), activation="relu", name="conv2")(x)
x = layers.MaxPooling2D((2, 2), name="pool2")(x)
x = layers.Flatten(name="flatten")(x)
x = layers.Dense(128, activation="relu", name="dense1")(x)
x = layers.Dropout(0.5, name="dropout")(x)

# Define multiple outputs
outputs = layers.Dense(num_classes, activation="softmax", name="output_layer")(x)

# Create model
functional_model = keras.Model(inputs=inputs, outputs=outputs, name="functional_mnist_model")

functional_model.summary()

# Compile and train
print("Compiling and training Functional API model...")
functional_model.compile(
    loss="categorical_crossentropy",
    optimizer="adam",
    metrics=["accuracy"]
)

history_functional = functional_model.fit(
    x_train, y_train,
    batch_size=batch_size,
    epochs=epochs,
    validation_split=0.1
)

# Evaluate
score_functional = functional_model.evaluate(x_test, y_test, verbose=0)
print(f"Functional API Test loss: {score_functional[0]:.4f}")
print(f"Functional API Test accuracy: {score_functional[1]:.4f}")

# Model subclassing example - most flexible approach
print("\nBuilding model with Subclassing API...")

class MNISTModel(keras.Model):
    def __init__(self, num_classes=10):
        super(MNISTModel, self).__init__(name="subclassed_mnist_model")
        self.conv1 = layers.Conv2D(32, (3, 3), activation="relu")
        self.pool1 = layers.MaxPooling2D((2, 2))
        self.conv2 = layers.Conv2D(64, (3, 3), activation="relu")
        self.pool2 = layers.MaxPooling2D((2, 2))
        self.flatten = layers.Flatten()
        self.dense1 = layers.Dense(128, activation="relu")
        self.dropout = layers.Dropout(0.5)
        self.dense2 = layers.Dense(num_classes, activation="softmax")

    def call(self, inputs, training=False):
        x = self.conv1(inputs)
        x = self.pool1(x)
        x = self.conv2(x)
        x = self.pool2(x)
        x = self.flatten(x)
        x = self.dense1(x)
        if training:
            x = self.dropout(x, training=training)
        return self.dense2(x)

    def get_config(self):
        # For serialization
        return {"num_classes": self.dense2.units}

# Create and train model
subclassed_model = MNISTModel(num_classes)
subclassed_model.build(input_shape=(None, 28, 28, 1))  # Build with input shape
subclassed_model.summary()

print("Compiling and training Subclassed model...")
subclassed_model.compile(
    loss="categorical_crossentropy",
    optimizer="adam",
    metrics=["accuracy"]
)

history_subclassed = subclassed_model.fit(
    x_train, y_train,
    batch_size=batch_size,
    epochs=epochs,
    validation_split=0.1
)

# Evaluate
score_subclassed = subclassed_model.evaluate(x_test, y_test, verbose=0)
print(f"Subclassed API Test loss: {score_subclassed[0]:.4f}")
print(f"Subclassed API Test accuracy: {score_subclassed[1]:.4f}")

# Transfer learning example using Keras Applications
print("\nTransfer learning with Keras Applications...")

from tensorflow.keras.applications import VGG16
from tensorflow.keras.preprocessing import image
from tensorflow.keras.applications.vgg16 import preprocess_input, decode_predictions

# Load pre-trained VGG16 model
base_model = VGG16(weights='imagenet', include_top=True)

# Load and preprocess an example image
img_path = 'example.jpg'  # Replace with actual image path
img = image.load_img(img_path, target_size=(224, 224))
x = image.img_to_array(img)
x = np.expand_dims(x, axis=0)
x = preprocess_input(x)

# Make prediction
print("Making prediction with pre-trained VGG16...")
preds = base_model.predict(x)
decoded_preds = decode_predictions(preds, top=5)[0]

print("Top 5 predictions:")
for i, (imagenet_id, label, prob) in enumerate(decoded_preds):
    print(f"{i+1}: {label} ({prob:.4f})")

# Transfer learning example - using base model for custom task
print("\nTransfer learning for custom task...")

# Load base model without top layers
base_model = VGG16(weights='imagenet', include_top=False, input_shape=(32, 32, 3))

# Freeze the base model
base_model.trainable = False

# Create new model on top
inputs = keras.Input(shape=(32, 32, 3))
x = layers.UpSampling2D(size=(7, 7))(inputs)  # Resize 32x32 to 224x224
x = base_model(x, training=False)
x = layers.Flatten()(x)
x = layers.Dense(256, activation='relu')(x)
x = layers.Dropout(0.5)(x)
outputs = layers.Dense(10, activation='softmax')(x)

transfer_model = keras.Model(inputs, outputs)

transfer_model.summary()

# Compile and train (would need appropriate dataset)
print("Transfer learning model ready for training with appropriate dataset.")

# Callbacks example - powerful tools for training optimization
print("\nTraining with callbacks...")

from tensorflow.keras.callbacks import (
    EarlyStopping,
    ModelCheckpoint,
    ReduceLROnPlateau,
    TensorBoard,
    CSVLogger
)

# Define callbacks
callbacks = [
    EarlyStopping(
        monitor='val_loss',
        patience=3,
        restore_best_weights=True,
        verbose=1
    ),
    ModelCheckpoint(
        filepath='best_model.h5',
        monitor='val_accuracy',
        save_best_only=True,
        save_weights_only=False,
        verbose=1
    ),
    ReduceLROnPlateau(
        monitor='val_loss',
        factor=0.1,
        patience=2,
        min_lr=1e-6,
        verbose=1
    ),
    TensorBoard(
        log_dir='./logs',
        histogram_freq=1,
        write_graph=True,
        write_images=True
    ),
    CSVLogger(
        filename='training_log.csv',
        separator=',',
        append=False
    )
]

# Build and compile model
model = keras.Sequential([
    layers.Conv2D(32, (3, 3), activation='relu', input_shape=(28, 28, 1)),
    layers.MaxPooling2D((2, 2)),
    layers.Conv2D(64, (3, 3), activation='relu'),
    layers.MaxPooling2D((2, 2)),
    layers.Flatten(),
    layers.Dense(128, activation='relu'),
    layers.Dropout(0.5),
    layers.Dense(10, activation='softmax')
])

model.compile(
    loss='categorical_crossentropy',
    optimizer='adam',
    metrics=['accuracy']
)

# Train with callbacks
history = model.fit(
    x_train, y_train,
    batch_size=128,
    epochs=10,  # Increased epochs to demonstrate callbacks
    validation_split=0.1,
    callbacks=callbacks
)

print("Training with callbacks completed!")

# Custom callback example
class CustomCallback(keras.callbacks.Callback):
    def __init__(self, threshold=0.95):
        super(CustomCallback, self).__init__()
        self.threshold = threshold

    def on_epoch_end(self, epoch, logs=None):
        val_acc = logs.get('val_accuracy')
        if val_acc and val_acc > self.threshold:
            print(f"\nReached {val_acc:.4f} validation accuracy (> {self.threshold}), stopping training!")
            self.model.stop_training = True

    def on_batch_end(self, batch, logs=None):
        # Example: log learning rate
        lr = float(keras.backend.get_value(self.model.optimizer.lr))
        logs['lr'] = lr

    def on_train_begin(self, logs=None):
        print("Starting training with custom callback...")

    def on_train_end(self, logs=None):
        print("Training completed with custom callback!")

# Train with custom callback
print("\nTraining with custom callback...")
model = keras.Sequential([
    layers.Conv2D(32, (3, 3), activation='relu', input_shape=(28, 28, 1)),
    layers.MaxPooling2D((2, 2)),
    layers.Conv2D(64, (3, 3), activation='relu'),
    layers.MaxPooling2D((2, 2)),
    layers.Flatten(),
    layers.Dense(128, activation='relu'),
    layers.Dropout(0.5),
    layers.Dense(10, activation='softmax')
])

model.compile(
    loss='categorical_crossentropy',
    optimizer='adam',
    metrics=['accuracy']
)

history = model.fit(
    x_train, y_train,
    batch_size=128,
    epochs=10,
    validation_split=0.1,
    callbacks=[CustomCallback(threshold=0.98)]
)