Chapter 20: Applied Deep Learning

Problem Statement

Indian agriculture loses 15-25% of crop yield annually to plant diseases. An average farmer in Maharashtra or Punjab cannot afford an agronomist visit (₹2,000-5,000). You will build a model that photographs a leaf and identifies the disease within 3 seconds — running entirely on a ₹10,000 smartphone with no internet required.

Dataset: PlantVillage

Architecture: ResNet50 + Custom Head

Full PyTorch Implementation

Python
import torch
import torch.nn as nn
import torch.optim as optim
from torchvision import models, transforms, datasets
from torch.utils.data import DataLoader
from sklearn.metrics import classification_report, confusion_matrix
import numpy as np

# ── Device Setup ──
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

# ── Data Augmentation (Indian crop-aware) ──
train_transforms = transforms.Compose([
    transforms.RandomResizedCrop(224, scale=(0.7, 1.0)),
    transforms.RandomHorizontalFlip(),
    transforms.RandomVerticalFlip(),
    transforms.RandomRotation(30),
    transforms.ColorJitter(
        brightness=0.3, contrast=0.3,
        saturation=0.3, hue=0.1  # Simulate Indian soil/lighting
    ),
    transforms.RandomAffine(degrees=0, translate=(0.1, 0.1)),
    transforms.GaussianBlur(kernel_size=3),  # Phone camera blur
    transforms.ToTensor(),
    transforms.Normalize([0.485, 0.456, 0.406],
                         [0.229, 0.224, 0.225])
])

val_transforms = transforms.Compose([
    transforms.Resize(256),
    transforms.CenterCrop(224),
    transforms.ToTensor(),
    transforms.Normalize([0.485, 0.456, 0.406],
                         [0.229, 0.224, 0.225])
])

# ── Dataset Loading ──
train_dataset = datasets.ImageFolder("plantvillage/train", train_transforms)
val_dataset   = datasets.ImageFolder("plantvillage/val", val_transforms)

train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True,
                          num_workers=4, pin_memory=True)
val_loader   = DataLoader(val_dataset, batch_size=32, shuffle=False,
                          num_workers=4, pin_memory=True)

# ── Model: ResNet50 with Custom Head ──
class CropDiseaseNet(nn.Module):
    def __init__(self, num_classes=38, pretrained=True):
        super().__init__()
        self.backbone = models.resnet50(
            weights=models.ResNet50_Weights.IMAGENET1K_V2 if pretrained else None
        )
        # Freeze backbone initially
        for param in self.backbone.parameters():
            param.requires_grad = False

        # Replace classifier head
        in_features = self.backbone.fc.in_features  # 2048
        self.backbone.fc = nn.Sequential(
            nn.Dropout(0.4),
            nn.Linear(in_features, 512),
            nn.ReLU(),
            nn.BatchNorm1d(512),
            nn.Dropout(0.3),
            nn.Linear(512, num_classes)
        )

    def unfreeze_backbone(self, layers="layer4"):
        """Gradually unfreeze backbone layers for fine-tuning."""
        for name, param in self.backbone.named_parameters():
            if layers in name:
                param.requires_grad = True

    def forward(self, x):
        return self.backbone(x)

model = CropDiseaseNet(num_classes=38).to(device)

# ── Training Loop with 2-Phase Strategy ──
def train_one_epoch(model, loader, criterion, optimizer, device):
    model.train()
    running_loss, correct, total = 0.0, 0, 0
    for images, labels in loader:
        images, labels = images.to(device), labels.to(device)
        optimizer.zero_grad()
        outputs = model(images)
        loss = criterion(outputs, labels)
        loss.backward()
        optimizer.step()
        running_loss += loss.item() * images.size(0)
        _, preds = outputs.max(1)
        correct += preds.eq(labels).sum().item()
        total += labels.size(0)
    return running_loss / total, correct / total

def evaluate(model, loader, criterion, device):
    model.eval()
    running_loss, correct, total = 0.0, 0, 0
    all_preds, all_labels = [], []
    with torch.no_grad():
        for images, labels in loader:
            images, labels = images.to(device), labels.to(device)
            outputs = model(images)
            loss = criterion(outputs, labels)
            running_loss += loss.item() * images.size(0)
            _, preds = outputs.max(1)
            correct += preds.eq(labels).sum().item()
            total += labels.size(0)
            all_preds.extend(preds.cpu().numpy())
            all_labels.extend(labels.cpu().numpy())
    return running_loss / total, correct / total, all_preds, all_labels

# ── Phase 1: Train head only (5 epochs) ──
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.backbone.fc.parameters(), lr=1e-3)
scheduler = optim.lr_scheduler.StepLR(optimizer, step_size=3, gamma=0.5)

for epoch in range(5):
    train_loss, train_acc = train_one_epoch(model, train_loader,
                                            criterion, optimizer, device)
    val_loss, val_acc, _, _ = evaluate(model, val_loader, criterion, device)
    scheduler.step()
    print(f"Phase1 Epoch {epoch+1}: Train Acc={train_acc:.4f}, Val Acc={val_acc:.4f}")

# ── Phase 2: Unfreeze layer4 + fine-tune (15 epochs) ──
model.unfreeze_backbone("layer4")
optimizer = optim.Adam(filter(lambda p: p.requires_grad, model.parameters()),
                       lr=1e-5)  # Much lower LR!
scheduler = optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=15)

best_val_acc = 0
for epoch in range(15):
    train_loss, train_acc = train_one_epoch(model, train_loader,
                                            criterion, optimizer, device)
    val_loss, val_acc, preds, labels = evaluate(model, val_loader,
                                                criterion, device)
    scheduler.step()
    print(f"Phase2 Epoch {epoch+1}: Train={train_acc:.4f}, Val={val_acc:.4f}")
    if val_acc > best_val_acc:
        best_val_acc = val_acc
        torch.save(model.state_dict(), "crop_disease_best.pth")

# ── Evaluation ──
print(classification_report(labels, preds, target_names=train_dataset.classes))

Expected Results

Grad-CAM Visualization

Python
import torch.nn.functional as F
import matplotlib.pyplot as plt

def grad_cam(model, image_tensor, target_class, target_layer):
    """Generate Grad-CAM heatmap for a given image and class."""
    model.eval()
    activations, gradients = {}, {}

    # Register hooks on target layer
    def forward_hook(module, input, output):
        activations['value'] = output.detach()

    def backward_hook(module, grad_in, grad_out):
        gradients['value'] = grad_out[0].detach()

    handle_f = target_layer.register_forward_hook(forward_hook)
    handle_b = target_layer.register_full_backward_hook(backward_hook)

    # Forward pass
    output = model(image_tensor.unsqueeze(0).to(device))
    # Backward pass for target class
    model.zero_grad()
    output[0, target_class].backward()

    # Compute weights (global average pooling of gradients)
    weights = gradients['value'].mean(dim=[2, 3], keepdim=True)  # α_k^c
    # Weighted combination + ReLU
    cam = F.relu((weights * activations['value']).sum(dim=1, keepdim=True))
    # Upsample to input size
    cam = F.interpolate(cam, size=(224, 224), mode='bilinear', align_corners=False)
    cam = cam.squeeze().cpu().numpy()
    cam = (cam - cam.min()) / (cam.max() - cam.min() + 1e-8)

    handle_f.remove()
    handle_b.remove()
    return cam

# Usage
target_layer = model.backbone.layer4[-1]  # Last bottleneck in layer4
heatmap = grad_cam(model, sample_image, predicted_class, target_layer)
plt.imshow(original_image)
plt.imshow(heatmap, alpha=0.5, cmap='jet')
plt.title(f"Grad-CAM: {class_names[predicted_class]}")
plt.show()

Deployment: ONNX Export

Python
# Export to ONNX for mobile deployment
dummy_input = torch.randn(1, 3, 224, 224).to(device)
torch.onnx.export(model, dummy_input, "crop_disease.onnx",
                  input_names=["image"], output_names=["prediction"],
                  dynamic_axes={"image": {0: "batch"}})
print("✅ Exported! ONNX model size:",
      os.path.getsize("crop_disease.onnx") / 1e6, "MB")

Problem Statement

Post-demonetization (Nov 2016), India introduced new ₹500 and ₹2000 notes. The RBI seized ₹8.26 crore in counterfeit currency in FY2023. You will build a CNN that analyzes texture patterns, watermark regions, and security thread features to classify notes as genuine or counterfeit — a binary classification problem with critical precision requirements.

Dataset Engineering

No public dataset exists for Indian counterfeit notes (for obvious security reasons). You'll create a synthetic pipeline:

Architecture: Multi-Region CNN

Python
class CurrencyAuthNet(nn.Module):
    """Multi-region CNN for Indian currency authentication.
    Analyzes watermark, security thread, latent image, and texture
    regions separately, then fuses features for final prediction."""

    def __init__(self):
        super().__init__()
        # Shared feature extractor for each region
        def make_branch():
            return nn.Sequential(
                nn.Conv2d(3, 32, 3, padding=1), nn.BatchNorm2d(32), nn.ReLU(),
                nn.MaxPool2d(2),
                nn.Conv2d(32, 64, 3, padding=1), nn.BatchNorm2d(64), nn.ReLU(),
                nn.MaxPool2d(2),
                nn.Conv2d(64, 128, 3, padding=1), nn.BatchNorm2d(128), nn.ReLU(),
                nn.AdaptiveAvgPool2d((4, 4)),
                nn.Flatten(),
                nn.Linear(128 * 4 * 4, 256), nn.ReLU(), nn.Dropout(0.3)
            )

        self.watermark_branch  = make_branch()  # Region 1: Watermark area
        self.thread_branch     = make_branch()  # Region 2: Security thread
        self.latent_branch     = make_branch()  # Region 3: Latent image
        self.texture_branch    = make_branch()  # Region 4: Overall texture

        # Fusion classifier
        self.classifier = nn.Sequential(
            nn.Linear(256 * 4, 512), nn.ReLU(), nn.BatchNorm1d(512),
            nn.Dropout(0.4),
            nn.Linear(512, 128), nn.ReLU(),
            nn.Linear(128, 2)  # genuine vs counterfeit
        )

    def forward(self, watermark, thread, latent, texture):
        f1 = self.watermark_branch(watermark)
        f2 = self.thread_branch(thread)
        f3 = self.latent_branch(latent)
        f4 = self.texture_branch(texture)
        fused = torch.cat([f1, f2, f3, f4], dim=1)
        return self.classifier(fused)

# ── Region Extraction Utility ──
def extract_regions(note_image):
    """Extract 4 security-feature regions from a currency note image.
    Coordinates calibrated for ₹500/₹2000 note dimensions."""
    h, w = note_image.shape[1:]
    watermark = note_image[:, :h//2, :w//3]       # Top-left quadrant
    thread    = note_image[:, :, w//3:w//3+w//10]  # Vertical strip
    latent    = note_image[:, h//2:, :w//3]       # Bottom-left
    texture   = note_image                          # Full note for texture
    # Resize all to 64×64 for uniform processing
    resize = transforms.Resize((64, 64))
    return resize(watermark), resize(thread), resize(latent), resize(texture)

# ── Training ──
model = CurrencyAuthNet().to(device)
criterion = nn.CrossEntropyLoss(weight=torch.tensor([1.0, 2.0]).to(device))
# Weight=2.0 for counterfeit class — missing a fake note is worse!
optimizer = optim.Adam(model.parameters(), lr=1e-3, weight_decay=1e-4)

for epoch in range(30):
    model.train()
    for batch in train_loader:
        wm, th, lt, tx, labels = [b.to(device) for b in batch]
        optimizer.zero_grad()
        outputs = model(wm, th, lt, tx)
        loss = criterion(outputs, labels)
        loss.backward()
        optimizer.step()

Problem Statement

India has 4.6 lakh road accidents annually — the highest in the world. Indian traffic signs are fundamentally different from the German Traffic Sign Recognition Benchmark (GTSRB) used in most research: they're multilingual (Hindi + English + regional), have different color conventions, and are often occluded by trees, ads, or dust. You will build a real-time classifier for Indian road signs.

Indian vs. German Signs: Key Differences

Architecture: MobileNetV2 for Edge Speed

Python
import torch
import torch.nn as nn
from torchvision import models, transforms

class IndianTrafficSignNet(nn.Module):
    """MobileNetV2-based Indian traffic sign classifier.
    Optimized for real-time inference on edge devices (Jetson Nano, phones).
    Handles 50 Indian sign categories including multilingual signs."""

    def __init__(self, num_classes=50):
        super().__init__()
        self.backbone = models.mobilenet_v2(
            weights=models.MobileNet_V2_Weights.IMAGENET1K_V2
        )
        # Freeze first 14 layers (of 19 inverted residual blocks)
        for i, (name, param) in enumerate(self.backbone.features.named_parameters()):
            if i < 100:  # Approx first 14 blocks
                param.requires_grad = False

        # Replace classifier
        self.backbone.classifier = nn.Sequential(
            nn.Dropout(0.3),
            nn.Linear(1280, 256),
            nn.ReLU(),
            nn.BatchNorm1d(256),
            nn.Dropout(0.2),
            nn.Linear(256, num_classes)
        )

    def forward(self, x):
        return self.backbone(x)

# ── Indian-specific augmentation ──
indian_sign_transforms = transforms.Compose([
    transforms.RandomResizedCrop(224, scale=(0.6, 1.0)),  # Partial occlusion
    transforms.RandomRotation(15),          # Tilted signs
    transforms.ColorJitter(
        brightness=0.4, contrast=0.4,
        saturation=0.2, hue=0.05
    ),                                      # Dust/sun fading
    transforms.RandomPerspective(
        distortion_scale=0.3, p=0.5
    ),                                      # Viewing angle variation
    transforms.GaussianBlur(5, sigma=(0.1, 2.0)),  # Rain/fog
    transforms.ToTensor(),
    transforms.Normalize([0.485, 0.456, 0.406],
                         [0.229, 0.224, 0.225]),
    transforms.RandomErasing(p=0.3, scale=(0.02, 0.15))  # Sticker occlusion
])

model = IndianTrafficSignNet(num_classes=50).to(device)
optimizer = optim.AdamW(filter(lambda p: p.requires_grad, model.parameters()),
                        lr=3e-4, weight_decay=0.01)
scheduler = optim.lr_scheduler.OneCycleLR(
    optimizer, max_lr=3e-3, epochs=25,
    steps_per_epoch=len(train_loader)
)

# ── Training with OneCycleLR ──
for epoch in range(25):
    model.train()
    for images, labels in train_loader:
        images, labels = images.to(device), labels.to(device)
        optimizer.zero_grad()
        loss = nn.CrossEntropyLoss()(model(images), labels)
        loss.backward()
        torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
        optimizer.step()
        scheduler.step()

Problem Statement

India has only 1 radiologist per 100,000 people (vs. 1 per 10,000 in the US). A single radiologist at a district hospital in Jharkhand reads 200+ chest X-rays daily. Fatigue-related misdiagnosis is a real risk. You will build a pneumonia detection system that serves as a "second opinion" — not a replacement — for radiologists.

Dataset: NIH Chest X-Ray / Kermany

Architecture: DenseNet121 with Grad-CAM

Python
import torch
import torch.nn as nn
from torchvision import models, transforms
from sklearn.metrics import roc_auc_score, roc_curve

class PneumoniaNet(nn.Module):
    """DenseNet121-based chest X-ray pneumonia detector.
    DenseNet chosen because:
    1. Feature reuse via dense connections → better with limited data
    2. Smaller model than ResNet50 (8M vs 25M params)
    3. CheXNet (Rajpurkar et al., 2017) validated on 14 pathologies"""

    def __init__(self):
        super().__init__()
        self.densenet = models.densenet121(
            weights=models.DenseNet121_Weights.IMAGENET1K_V1
        )
        # DenseNet121 final features: 1024 channels
        in_features = self.densenet.classifier.in_features
        self.densenet.classifier = nn.Sequential(
            nn.Linear(in_features, 256), nn.ReLU(),
            nn.Dropout(0.3),
            nn.Linear(256, 1)  # Binary: sigmoid output
        )

    def forward(self, x):
        return self.densenet(x)

model = PneumoniaNet().to(device)

# ── Weighted BCE for class imbalance ──
# Pneumonia:Normal = 4273:1583 → weight Normal higher
pos_weight = torch.tensor([1583/4273]).to(device)
criterion = nn.BCEWithLogitsLoss(pos_weight=pos_weight)

# ── Medical-appropriate augmentation (conservative!) ──
medical_transforms = transforms.Compose([
    transforms.Resize((256, 256)),
    transforms.RandomCrop(224),
    transforms.RandomHorizontalFlip(),  # X-rays can be flipped
    transforms.RandomRotation(10),     # Slight rotation only!
    transforms.RandomAffine(
        degrees=0, translate=(0.05, 0.05)
    ),  # Small translation
    # NO color jitter — X-rays are grayscale!
    # NO aggressive crops — might remove pathology!
    transforms.ToTensor(),
    transforms.Normalize([0.485], [0.229])  # Single channel norms
])

# ── Training with sensitivity-focused early stopping ──
optimizer = optim.Adam(model.parameters(), lr=1e-4, weight_decay=1e-5)
scheduler = optim.lr_scheduler.ReduceLROnPlateau(
    optimizer, mode='max', factor=0.5, patience=3,
    verbose=True
)  # Monitor recall, not loss!

best_recall = 0
for epoch in range(20):
    model.train()
    for images, labels in train_loader:
        images = images.to(device)
        labels = labels.float().unsqueeze(1).to(device)
        optimizer.zero_grad()
        outputs = model(images)
        loss = criterion(outputs, labels)
        loss.backward()
        optimizer.step()

    # Evaluate with medical metrics
    model.eval()
    all_probs, all_labels = [], []
    with torch.no_grad():
        for images, labels in val_loader:
            probs = torch.sigmoid(model(images.to(device)))
            all_probs.extend(probs.cpu().numpy().flatten())
            all_labels.extend(labels.numpy())

    # Find threshold that gives recall ≥ 0.95
    fpr, tpr, thresholds = roc_curve(all_labels, all_probs)
    auc = roc_auc_score(all_labels, all_probs)
    # Choose threshold where TPR (recall) ≥ 0.95
    idx = np.argmin(np.abs(tpr - 0.95))
    optimal_threshold = thresholds[idx]

    preds = (np.array(all_probs) >= optimal_threshold).astype(int)
    recall = np.sum((preds == 1) & (np.array(all_labels) == 1)) / \
             np.sum(np.array(all_labels) == 1)
    print(f"Epoch {epoch+1}: AUC={auc:.4f}, Recall={recall:.4f}, "
          f"Threshold={optimal_threshold:.3f}")

    scheduler.step(recall)
    if recall > best_recall:
        best_recall = recall
        torch.save(model.state_dict(), "pneumonia_best.pth")

Grad-CAM for Medical Explainability

Python
def medical_grad_cam(model, image, target_layer):
    """Generate Grad-CAM for chest X-ray interpretation.
    The heatmap must highlight lung regions where pathology is detected.
    If it highlights bones, borders, or text — the model is wrong!"""

    model.eval()
    activations, gradients = {}, {}

    def fwd_hook(m, i, o): activations['val'] = o.detach()
    def bwd_hook(m, gi, go): gradients['val'] = go[0].detach()

    h1 = target_layer.register_forward_hook(fwd_hook)
    h2 = target_layer.register_full_backward_hook(bwd_hook)

    output = model(image.unsqueeze(0).to(device))
    model.zero_grad()
    output.backward()  # Binary — no class selection needed

    weights = gradients['val'].mean(dim=[2, 3], keepdim=True)
    cam = torch.relu((weights * activations['val']).sum(dim=1))
    cam = nn.functional.interpolate(
        cam.unsqueeze(0), size=(224, 224),
        mode='bilinear', align_corners=False
    ).squeeze().cpu().numpy()
    cam = (cam - cam.min()) / (cam.max() - cam.min() + 1e-8)

    h1.remove(); h2.remove()
    return cam

# Validate: Check if Grad-CAM focuses on lung regions
# If attention is on diaphragm/text/borders → model learned shortcuts!
target_layer = model.densenet.features.denseblock4
cam = medical_grad_cam(model, test_image, target_layer)

Problem Statement

Self-driving car companies training on US/European data fail spectacularly on Indian roads. Why? Because their models have never seen an auto-rickshaw, a cow sitting on the highway median, or 4 people riding a single two-wheeler. You will train YOLOv8 to detect India-specific objects in real-time traffic video.

Indian Traffic Object Classes

YOLOv8: Architecture Overview

Full Implementation with Ultralytics

Python
# ── Install: pip install ultralytics ──
from ultralytics import YOLO
import cv2
import yaml

# ── Step 1: Prepare dataset config (YOLO format) ──
dataset_config = {
    'path': 'indian_traffic_dataset',
    'train': 'images/train',
    'val': 'images/val',
    'test': 'images/test',
    'nc': 10,  # Number of classes
    'names': [
        'auto_rickshaw', 'cow', 'pedestrian',
        'two_wheeler', 'truck', 'bus', 'car',
        'street_dog', 'cart', 'road_barrier'
    ]
}
with open('indian_traffic.yaml', 'w') as f:
    yaml.dump(dataset_config, f)

# ── Step 2: Load pretrained YOLOv8 and fine-tune ──
model = YOLO('yolov8m.pt')  # Medium model — good speed/accuracy balance

# ── Step 3: Train on Indian traffic data ──
results = model.train(
    data='indian_traffic.yaml',
    epochs=100,
    imgsz=640,
    batch=16,
    lr0=0.01,
    lrf=0.01,       # Final LR = lr0 × lrf
    momentum=0.937,
    weight_decay=0.0005,
    warmup_epochs=3,
    warmup_momentum=0.8,
    augment=True,   # Mosaic + MixUp + HSV jitter
    mosaic=1.0,     # Mosaic probability
    mixup=0.1,
    close_mosaic=10,  # Disable mosaic last 10 epochs
    device='0',
    project='indian_traffic',
    name='yolov8m_exp1'
)

# ── Step 4: Evaluate ──
metrics = model.val()
print(f"mAP@0.5:     {metrics.box.map50:.4f}")
print(f"mAP@0.5:0.95: {metrics.box.map:.4f}")

# Per-class AP
for i, name in enumerate(dataset_config['names']):
    print(f"  {name:20s}: AP50={metrics.box.ap50[i]:.3f}")

# ── Step 5: Real-time inference on Indian dashcam video ──
model = YOLO('indian_traffic/yolov8m_exp1/weights/best.pt')

cap = cv2.VideoCapture('indian_highway_dashcam.mp4')
while cap.isOpened():
    ret, frame = cap.read()
    if not ret: break

    results = model(frame, conf=0.5, iou=0.45)
    annotated = results[0].plot()  # Draw boxes + labels

    cv2.imshow('Indian Traffic Detection', annotated)
    if cv2.waitKey(1) & 0xFF == ord('q'): break

cap.release()
cv2.destroyAllWindows()

# ── Step 6: Export for edge deployment ──
model.export(format='onnx', simplify=True, dynamic=True)
model.export(format='engine', half=True, device='0')  # TensorRT FP16

Per-Class Detection Performance

results = model.train(
    data='indian_traffic.yaml',
    epochs=50,
    imgsz=320,       # Bug 1: ???
    batch=2,          # Bug 2: ???
    lr0=0.1,          # Bug 3: ???
    augment=False,
    mosaic=0.0,
)