Day 19 — YOLOv5 Object Detection, Transfer Learning, and Quantization

Input Image (640x640x3)
        │
        ▼
┌────────────────────┐
│    BACKBONE         │    Feature extraction
│    (CSPDarknet53)   │    Learns "what things look like"
└────────┬───────────┘
         │
         ▼
┌────────────────────┐
│    NECK             │    Feature aggregation
│    (PANet + SPP)    │    Combines features at multiple scales
└────────┬───────────┘
         │
         ▼
┌────────────────────┐
│    HEAD             │    Detection output
│    (Detect Layer)   │    Bounding boxes + classes + confidence
└────────────────────┘
         │
         ▼
  3 scale outputs:
  - 80×80 (small objects)
  - 40×40 (medium objects)
  - 20×20 (large objects)

CSPDarknet53 (simplified):

Input → CBS(3→32) → CBS(32→64) → C3(64, n=1) →
        CBS(64→128) → C3(128, n=2) →
        CBS(128→256) → C3(256, n=3) → [P3: 80×80×256]
        CBS(256→512) → C3(512, n=3) → [P4: 40×40×512]
        CBS(512→1024) → C3(1024, n=1) → SPPF → [P5: 20×20×1024]

P5 (20×20) ──Upsample──► Concat with P4 → C3 → N4 (40×40)
                                                    │
N4 (40×40) ──Upsample──► Concat with P3 → C3 → N3 (80×80)
                                                    │
N3 (80×80) ──Downsample─► Concat with N4 → C3 → N4' (40×40)
                                                    │
N4' (40×40) ──Downsample─► Concat with P5 → C3 → N5' (20×20)

def compute_iou(box1, box2):
    """Compute IoU between two boxes [x1, y1, x2, y2]."""
    x1 = max(box1[0], box2[0])
    y1 = max(box1[1], box2[1])
    x2 = min(box1[2], box2[2])
    y2 = min(box1[3], box2[3])

    intersection = max(0, x2 - x1) * max(0, y2 - y1)
    area1 = (box1[2] - box1[0]) * (box1[3] - box1[1])
    area2 = (box2[2] - box2[0]) * (box2[3] - box2[1])
    union = area1 + area2 - intersection

    return intersection / union if union > 0 else 0.0

Healthy training:
  train_loss ↓  val_loss ↓  mAP ↑  → Keep going

Overfitting:
  train_loss ↓  val_loss ↑  mAP plateau/↓  → Stop or add augmentation

Underfitting:
  train_loss high  val_loss high  mAP low  → More epochs, unfreeze layers

Learning rate too high:
  Losses oscillate wildly  → Reduce lr0

Backbone (layers 0–9):  General visual features
                        → Freeze these for small datasets
Neck (layers 10–17):    Feature aggregation
                        → Can freeze or fine-tune
Head (layers 18–23):    Detection output
                        → Always train these

<class_id> <x_center> <y_center> <width> <height>

# frame_001.txt
0 0.4297 0.4896 0.2344 0.3542

custom_dataset/
├── images/
│   ├── train/
│   │   ├── frame_001.jpg
│   │   ├── frame_002.jpg
│   │   └── ...
│   └── val/
│       ├── frame_100.jpg
│       └── ...
├── labels/
│   ├── train/
│   │   ├── frame_001.txt
│   │   ├── frame_002.txt
│   │   └── ...
│   └── val/
│       ├── frame_100.txt
│       └── ...
└── data.yaml

# data.yaml — Custom dataset configuration
path: /home/user/custom_dataset
train: images/train
val: images/val

nc: 4  # number of classes
names:
  0: stop_sign
  1: speed_limit
  2: pedestrian
  3: traffic_cone

# Clone YOLOv5
git clone https://github.com/ultralytics/yolov5
cd yolov5
pip install -r requirements.txt

# Train with frozen backbone (10 layers)
python train.py \
    --weights yolov5s.pt \
    --data /path/to/custom_dataset/data.yaml \
    --img 640 \
    --batch 16 \
    --epochs 50 \
    --freeze 10 \
    --name custom_model_v1 \
    --patience 10

# Key augmentation hyperparameters
hsv_h: 0.015       # hue shift
hsv_s: 0.7         # saturation shift
hsv_v: 0.4         # value shift
degrees: 0.0       # rotation range
translate: 0.1     # translation fraction
scale: 0.5         # scale range
shear: 0.0         # shear
perspective: 0.0   # perspective distortion
flipud: 0.0        # vertical flip probability
fliplr: 0.5        # horizontal flip probability
mosaic: 1.0        # mosaic augmentation probability
mixup: 0.0         # mixup probability (set > 0 for small datasets)

runs/train/custom_model_v1/
├── weights/
│   ├── best.pt       ← Best mAP checkpoint
│   └── last.pt       ← Last epoch checkpoint
├── results.csv        ← Training curves data
├── results.png        ← Training curves plot
├── confusion_matrix.png
├── F1_curve.png
├── PR_curve.png
├── P_curve.png
├── R_curve.png
└── val_batch0_pred.jpg  ← Sample predictions on validation set

"""Plot training results from results.csv."""
import pandas as pd
import matplotlib.pyplot as plt

df = pd.read_csv("runs/train/custom_model_v1/results.csv",
                  skipinitialspace=True)

fig, axes = plt.subplots(2, 2, figsize=(14, 10))

# Loss curves
axes[0, 0].plot(df["epoch"], df["train/box_loss"], label="Train")
axes[0, 0].plot(df["epoch"], df["val/box_loss"], label="Val")
axes[0, 0].set_title("Box Loss")
axes[0, 0].legend()

axes[0, 1].plot(df["epoch"], df["train/obj_loss"], label="Train")
axes[0, 1].plot(df["epoch"], df["val/obj_loss"], label="Val")
axes[0, 1].set_title("Objectness Loss")
axes[0, 1].legend()

# mAP curves
axes[1, 0].plot(df["epoch"], df["metrics/mAP_0.5"], label="mAP@0.5")
axes[1, 0].plot(df["epoch"], df["metrics/mAP_0.5:0.95"], label="mAP@0.5:0.95")
axes[1, 0].set_title("mAP")
axes[1, 0].legend()

# Precision / Recall
axes[1, 1].plot(df["epoch"], df["metrics/precision"], label="Precision")
axes[1, 1].plot(df["epoch"], df["metrics/recall"], label="Recall")
axes[1, 1].set_title("Precision & Recall")
axes[1, 1].legend()

plt.tight_layout()
plt.savefig("training_analysis.png", dpi=150)
plt.show()

"""
Post-Training Quantization with PyTorch (simplified demonstration).
For actual deployment, the Hailo compiler performs PTQ automatically.
"""

import torch
from torch.quantization import quantize_dynamic, quantize_static

# Method 1: Dynamic Quantization (quantizes weights, activations at runtime)
# Simplest, but less speedup
model_fp32 = torch.load("runs/train/custom_model_v1/weights/best.pt")["model"]
model_int8_dynamic = quantize_dynamic(
    model_fp32.float(),
    {torch.nn.Linear},  # Conv2d not supported for dynamic quantization
    dtype=torch.qint8
)

# Method 2: Static Quantization (quantizes both weights and activations)
# Better performance, requires calibration
model_fp32.eval()
model_fp32.qconfig = torch.quantization.get_default_qconfig("fbgemm")
model_prepared = torch.quantization.prepare(model_fp32)

# Calibrate with representative data
calibration_dataloader = ...  # 100-500 images from your dataset
with torch.no_grad():
    for images, _ in calibration_dataloader:
        model_prepared(images)

model_int8_static = torch.quantization.convert(model_prepared)

Forward pass:
  x_fp32 → Quantize → Dequantize → x_approx_fp32 → Convolution → ...
                     ↑
              Simulates INT8 rounding error

Backward pass:
  Gradients flow through as if Quantize/Dequantize were identity functions
  (Straight-Through Estimator)

"""
Compare FP32 vs INT8 model accuracy.
Run YOLOv5 validation on the same dataset with both models.
"""

# FP32 baseline
# python val.py --weights best.pt --data data.yaml --img 640

# INT8 (after ONNX export + quantization)
# python val.py --weights best_int8.onnx --data data.yaml --img 640

PyTorch (.pt) → ONNX (.onnx) → Hailo Compiler → .hef (INT8)
                                    ↑
                         Calibration images needed

# Step 1: Clone and install
git clone https://github.com/ultralytics/yolov5
cd yolov5
pip install -r requirements.txt

# Step 2: Prepare dataset (assuming labeled with labelImg)
# Verify structure:
ls /path/to/custom_dataset/images/train/ | head
ls /path/to/custom_dataset/labels/train/ | head
cat /path/to/custom_dataset/data.yaml

# Step 3: Train with frozen backbone
python train.py \
    --weights yolov5s.pt \
    --data /path/to/custom_dataset/data.yaml \
    --img 640 \
    --batch 16 \
    --epochs 50 \
    --freeze 10 \
    --name track_signs_v1 \
    --patience 10

# Step 4: Validate
python val.py \
    --weights runs/train/track_signs_v1/weights/best.pt \
    --data /path/to/custom_dataset/data.yaml \
    --img 640 \
    --verbose

# Step 5: Test on images
python detect.py \
    --weights runs/train/track_signs_v1/weights/best.pt \
    --source /path/to/test_images/ \
    --img 640 \
    --conf-thres 0.5

"""
Lab: Analyze training results and generate report.
"""

import pandas as pd
import matplotlib.pyplot as plt
from pathlib import Path


def analyze_training(run_dir):
    """Complete analysis of a YOLOv5 training run."""
    run_path = Path(run_dir)

    # Read results
    df = pd.read_csv(run_path / "results.csv", skipinitialspace=True)

    # Best epoch
    best_epoch = df["metrics/mAP_0.5"].idxmax()
    best_map50 = df.loc[best_epoch, "metrics/mAP_0.5"]
    best_map5095 = df.loc[best_epoch, "metrics/mAP_0.5:0.95"]

    print(f"Best epoch: {best_epoch}")
    print(f"  mAP@0.5:     {best_map50:.4f}")
    print(f"  mAP@0.5:0.95: {best_map5095:.4f}")
    print(f"  Precision:    {df.loc[best_epoch, 'metrics/precision']:.4f}")
    print(f"  Recall:       {df.loc[best_epoch, 'metrics/recall']:.4f}")

    # Check for overfitting
    final_train_loss = df["train/box_loss"].iloc[-1]
    final_val_loss = df["val/box_loss"].iloc[-1]
    min_val_loss = df["val/box_loss"].min()
    min_val_epoch = df["val/box_loss"].idxmin()

    if df["val/box_loss"].iloc[-1] > min_val_loss * 1.1:
        print(f"\n[WARNING] Possible overfitting detected!")
        print(f"  Min val loss at epoch {min_val_epoch}: {min_val_loss:.4f}")
        print(f"  Final val loss: {final_val_loss:.4f}")
    else:
        print(f"\nNo overfitting detected. Training looks healthy.")

    return df


# Run analysis
df = analyze_training("runs/train/track_signs_v1")

# Export to ONNX for deployment
python export.py \
    --weights runs/train/track_signs_v1/weights/best.pt \
    --img 640 \
    --batch 1 \
    --include onnx \
    --simplify

"""
Lab: Run YOLOv5 inference using OpenCV DNN backend.
Measures FPS on CPU for baseline comparison.
"""

import cv2
import numpy as np
import time


class YOLOv5OpenCV:
    """YOLOv5 inference using OpenCV DNN module."""

    def __init__(self, onnx_path, conf_thresh=0.5, iou_thresh=0.45,
                 input_size=640):
        self.net = cv2.dnn.readNetFromONNX(onnx_path)
        self.net.setPreferableBackend(cv2.dnn.DNN_BACKEND_OPENCV)
        self.net.setPreferableTarget(cv2.dnn.DNN_TARGET_CPU)

        self.conf_thresh = conf_thresh
        self.iou_thresh = iou_thresh
        self.input_size = input_size

    def preprocess(self, frame):
        """Letterbox resize + normalize."""
        h, w = frame.shape[:2]
        scale = min(self.input_size / h, self.input_size / w)
        new_w, new_h = int(w * scale), int(h * scale)

        resized = cv2.resize(frame, (new_w, new_h))

        # Pad to square
        canvas = np.full((self.input_size, self.input_size, 3), 114,
                         dtype=np.uint8)
        dw = (self.input_size - new_w) // 2
        dh = (self.input_size - new_h) // 2
        canvas[dh:dh + new_h, dw:dw + new_w] = resized

        blob = cv2.dnn.blobFromImage(canvas, 1.0 / 255.0,
                                      (self.input_size, self.input_size),
                                      swapRB=True, crop=False)
        return blob, scale, dw, dh

    def postprocess(self, output, scale, dw, dh, orig_h, orig_w):
        """Extract detections from network output."""
        # output shape: [1, num_detections, 5 + num_classes]
        detections = output[0]

        boxes = []
        confidences = []
        class_ids = []

        for det in detections:
            scores = det[5:]
            class_id = np.argmax(scores)
            confidence = scores[class_id] * det[4]  # class_score * objectness

            if confidence < self.conf_thresh:
                continue

            # Center format → corner format, undo letterbox
            cx, cy, bw, bh = det[0], det[1], det[2], det[3]
            x1 = int((cx - bw / 2 - dw) / scale)
            y1 = int((cy - bh / 2 - dh) / scale)
            x2 = int((cx + bw / 2 - dw) / scale)
            y2 = int((cy + bh / 2 - dh) / scale)

            # Clamp to image bounds
            x1 = max(0, min(x1, orig_w))
            y1 = max(0, min(y1, orig_h))
            x2 = max(0, min(x2, orig_w))
            y2 = max(0, min(y2, orig_h))

            boxes.append([x1, y1, x2 - x1, y2 - y1])
            confidences.append(float(confidence))
            class_ids.append(class_id)

        # NMS
        indices = cv2.dnn.NMSBoxes(boxes, confidences, self.conf_thresh,
                                    self.iou_thresh)

        results = []
        for i in indices:
            idx = i if isinstance(i, int) else i[0]
            results.append({
                "box": boxes[idx],
                "confidence": confidences[idx],
                "class_id": class_ids[idx],
            })

        return results

    def detect(self, frame):
        """Run full detection pipeline on one frame."""
        h, w = frame.shape[:2]
        blob, scale, dw, dh = self.preprocess(frame)

        self.net.setInput(blob)
        output = self.net.forward()

        return self.postprocess(output, scale, dw, dh, h, w)


def benchmark_fps(detector, source, n_frames=100):
    """Measure average FPS over n_frames."""
    cap = cv2.VideoCapture(source)
    if not cap.isOpened():
        print(f"Cannot open {source}")
        return

    times = []
    for i in range(n_frames):
        ret, frame = cap.read()
        if not ret:
            cap.set(cv2.CAP_PROP_POS_FRAMES, 0)
            ret, frame = cap.read()

        t_start = time.perf_counter()
        results = detector.detect(frame)
        t_end = time.perf_counter()

        times.append(t_end - t_start)

        if (i + 1) % 10 == 0:
            avg_ms = np.mean(times[-10:]) * 1000
            fps = 1000 / avg_ms
            print(f"Frame {i+1}/{n_frames}: {avg_ms:.1f} ms ({fps:.1f} FPS), "
                  f"{len(results)} detections")

    cap.release()

    avg_time = np.mean(times)
    avg_fps = 1.0 / avg_time

    print(f"\n--- Benchmark Results ---")
    print(f"Average inference time: {avg_time * 1000:.1f} ms")
    print(f"Average FPS: {avg_fps:.1f}")
    print(f"Min time: {min(times) * 1000:.1f} ms")
    print(f"Max time: {max(times) * 1000:.1f} ms")

    return avg_fps


if __name__ == "__main__":
    CLASS_NAMES = ["stop_sign", "speed_limit", "pedestrian", "traffic_cone"]

    detector = YOLOv5OpenCV(
        onnx_path="runs/train/track_signs_v1/weights/best.onnx",
        conf_thresh=0.5,
        iou_thresh=0.45,
        input_size=640,
    )

    # Benchmark
    fps = benchmark_fps(detector, "test_video.mp4", n_frames=100)

    # Visual test
    cap = cv2.VideoCapture("test_video.mp4")
    while True:
        ret, frame = cap.read()
        if not ret:
            break

        results = detector.detect(frame)

        for r in results:
            x, y, w, h = r["box"]
            label = f"{CLASS_NAMES[r['class_id']]} {r['confidence']:.2f}"
            cv2.rectangle(frame, (x, y), (x + w, y + h), (0, 255, 0), 2)
            cv2.putText(frame, label, (x, y - 10),
                        cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 255, 0), 2)

        cv2.imshow("YOLOv5 Detection", frame)
        if cv2.waitKey(1) & 0xFF == ord('q'):
            break

    cap.release()
    cv2.destroyAllWindows()

"""
Lab: Compare FP32 vs INT8 quantized model.
Uses ONNX Runtime for fair comparison on CPU.
"""

import onnxruntime as ort
from onnxruntime.quantization import quantize_dynamic, QuantType
import numpy as np
import time


def quantize_onnx_model(input_path, output_path):
    """Apply dynamic INT8 quantization to ONNX model."""
    quantize_dynamic(
        input_path,
        output_path,
        weight_type=QuantType.QInt8,
    )
    print(f"Quantized model saved to {output_path}")


def benchmark_onnx(model_path, input_shape=(1, 3, 640, 640), n_runs=50):
    """Benchmark ONNX model inference time."""
    session = ort.InferenceSession(model_path)
    input_name = session.get_inputs()[0].name

    # Warm up
    dummy = np.random.randn(*input_shape).astype(np.float32)
    for _ in range(5):
        session.run(None, {input_name: dummy})

    # Benchmark
    times = []
    for _ in range(n_runs):
        t0 = time.perf_counter()
        session.run(None, {input_name: dummy})
        t1 = time.perf_counter()
        times.append(t1 - t0)

    avg_ms = np.mean(times) * 1000
    fps = 1000 / avg_ms
    return avg_ms, fps


def main():
    fp32_path = "runs/train/track_signs_v1/weights/best.onnx"
    int8_path = "runs/train/track_signs_v1/weights/best_int8.onnx"

    # Quantize
    quantize_onnx_model(fp32_path, int8_path)

    # Benchmark both
    fp32_ms, fp32_fps = benchmark_onnx(fp32_path)
    int8_ms, int8_fps = benchmark_onnx(int8_path)

    # Report
    import os
    fp32_size = os.path.getsize(fp32_path) / 1e6
    int8_size = os.path.getsize(int8_path) / 1e6

    print(f"\n{'='*50}")
    print(f"{'Metric':<25} {'FP32':>10} {'INT8':>10}")
    print(f"{'='*50}")
    print(f"{'Model size (MB)':<25} {fp32_size:>10.1f} {int8_size:>10.1f}")
    print(f"{'Inference time (ms)':<25} {fp32_ms:>10.1f} {int8_ms:>10.1f}")
    print(f"{'FPS':<25} {fp32_fps:>10.1f} {int8_fps:>10.1f}")
    print(f"{'Size reduction':<25} {'1.0x':>10} {f'{fp32_size/int8_size:.1f}x':>10}")
    print(f"{'Speedup':<25} {'1.0x':>10} {f'{fp32_ms/int8_ms:.1f}x':>10}")
    print(f"{'='*50}")


if __name__ == "__main__":
    main()