Introduction
This is a summary of the entire course. You will build a complete CV system — from data collection to deployment to production.
🎯 Goal: Build a system to detect and classify defective products on the production line (Manufacturing Defect Detection).
Why choose this project?
✅ Ứng dụng thực tế đang có nhu cầu rất cao
✅ Cover hết: detection + classification + segmentation
✅ Cần optimize cho real-time (edge deployment)
✅ Cần monitoring & retraining pipeline
1. System Architecture
📸 Camera Stream
│
▼
┌────────────────────┐
│ Image Capture │ ← Capture & preprocessing
│ (OpenCV/GStreamer) │
└────────┬───────────┘
│
▼
┌────────────────────┐
│ Defect Detection │ ← YOLOv8 (TensorRT)
│ (Object Detection)│
└────────┬───────────┘
│
▼
┌────────────────────┐
│ Classification │ ← MobileNet (crop → classify)
│ (Defect Type) │
└────────┬───────────┘
│
▼
┌─────────┬──────────┐
│ API │ Dashboard│ ← FastAPI + WebSocket
│ Server │ (React) │
└─────────┴──────────┘
│
▼
┌────────────────────┐
│ Logging & │ ← MLflow + Prometheus
│ Monitoring │
└────────────────────┘
2. Phase 1: Data Pipeline
2.1 Data Collection
"""Data collection từ camera hoặc folder ảnh"""
import cv2
import os
from pathlib import Path
from datetime import datetime
class DataCollector:
def __init__(self, save_dir="data/raw"):
self.save_dir = Path(save_dir)
self.save_dir.mkdir(parents=True, exist_ok=True)
def capture_from_camera(self, camera_id=0, interval_sec=2):
"""Chụp ảnh từ camera theo interval"""
cap = cv2.VideoCapture(camera_id)
count = 0
while True:
ret, frame = cap.read()
if not ret:
break
# Hiển thị
cv2.imshow("Capture", frame)
# Lưu theo interval
key = cv2.waitKey(int(interval_sec * 1000))
if key == ord('s'): # Press 's' to save
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"img_{timestamp}_{count:04d}.jpg"
cv2.imwrite(str(self.save_dir / filename), frame)
count += 1
print(f"📸 Saved: {filename} (total: {count})")
elif key == ord('q'):
break
cap.release()
print(f"✅ Collected {count} images")
def augment_dataset(self, src_dir, dst_dir, multiplier=5):
"""Augment dataset để tăng dữ liệu"""
import albumentations as A
transform = A.Compose([
A.RandomRotate90(p=0.5),
A.HorizontalFlip(p=0.5),
A.VerticalFlip(p=0.3),
A.RandomBrightnessContrast(p=0.5),
A.GaussNoise(p=0.3),
A.Blur(blur_limit=3, p=0.2),
A.CLAHE(p=0.3),
])
dst = Path(dst_dir)
dst.mkdir(parents=True, exist_ok=True)
for img_path in Path(src_dir).glob("*.jpg"):
img = cv2.imread(str(img_path))
# Save original
cv2.imwrite(str(dst / img_path.name), img)
# Augmented copies
for i in range(multiplier - 1):
augmented = transform(image=img)["image"]
aug_name = f"{img_path.stem}_aug{i}{img_path.suffix}"
cv2.imwrite(str(dst / aug_name), augmented)
print(f"✅ Augmented: {multiplier}x")
2.2 Data Labeling
"""Setup labeling project — Dùng Label Studio hoặc CVAT"""
# Option 1: Label Studio (recommended)
# pip install label-studio
# label-studio start --port 8080
# Option 2: Roboflow (web-based, free tier)
# https://roboflow.com → Upload → Annotate → Export YOLO format
# Option 3: Script tạo dataset structure
def prepare_yolo_dataset(labeled_dir, output_dir, train_ratio=0.8):
"""Chia dataset thành train/val theo YOLO format"""
import random
import shutil
images = list(Path(labeled_dir).glob("*.jpg"))
random.shuffle(images)
split = int(len(images) * train_ratio)
train_images = images[:split]
val_images = images[split:]
for split_name, split_images in [("train", train_images), ("val", val_images)]:
img_dir = Path(output_dir) / split_name / "images"
lbl_dir = Path(output_dir) / split_name / "labels"
img_dir.mkdir(parents=True, exist_ok=True)
lbl_dir.mkdir(parents=True, exist_ok=True)
for img_path in split_images:
shutil.copy(img_path, img_dir / img_path.name)
label_path = img_path.with_suffix(".txt")
if label_path.exists():
shutil.copy(label_path, lbl_dir / label_path.name)
# Tạo data.yaml
data_yaml = {
"path": str(output_dir),
"train": "train/images",
"val": "val/images",
"names": {
0: "scratch",
1: "dent",
2: "discoloration",
3: "crack",
4: "missing_part",
},
}
import yaml
with open(Path(output_dir) / "data.yaml", "w") as f:
yaml.dump(data_yaml, f)
print(f"✅ Dataset: {len(train_images)} train, {len(val_images)} val")
3. Phase 2: Model Training
3.1 Detection Model (YOLOv8)
"""Train YOLOv8 cho defect detection"""
from ultralytics import YOLO
# Load pretrained
model = YOLO("yolov8s.pt") # Small variant — tốt cho edge
# Train
results = model.train(
data="dataset/data.yaml",
epochs=100,
imgsz=640,
batch=16,
device=0,
# Optimization
lr0=0.01,
lrf=0.01,
momentum=0.937,
weight_decay=0.0005,
warmup_epochs=3,
# Augmentation
hsv_h=0.015,
hsv_s=0.7,
hsv_v=0.4,
flipud=0.5,
mosaic=1.0,
mixup=0.1,
# Callbacks
project="runs/defect_detection",
name="yolov8s_v1",
save_period=10,
)
# Evaluate
metrics = model.val()
print(f"mAP@50: {metrics.box.map50:.3f}")
print(f"mAP@50-95: {metrics.box.map:.3f}")
3.2 Classification Model (for crop from detection)
"""Train classifier cho defect type"""
import torch
import torchvision
from torchvision import transforms
from torch.utils.data import DataLoader
# Dataset cho classification (crop từ detection)
transform_train = transforms.Compose([
transforms.Resize((224, 224)),
transforms.RandomHorizontalFlip(),
transforms.RandomRotation(15),
transforms.ColorJitter(brightness=0.2, contrast=0.2),
transforms.ToTensor(),
transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225]),
])
train_dataset = torchvision.datasets.ImageFolder(
"dataset/classification/train",
transform=transform_train,
)
train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True)
# Model: MobileNetV3 (lightweight cho edge)
model = torchvision.models.mobilenet_v3_small(pretrained=True)
model.classifier[-1] = torch.nn.Linear(1024, 5) # 5 defect types
# Train
optimizer = torch.optim.AdamW(model.parameters(), lr=1e-3)
scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=30)
criterion = torch.nn.CrossEntropyLoss()
for epoch in range(30):
model.train()
total_loss = 0
correct = 0
total = 0
for images, labels in train_loader:
images, labels = images.cuda(), labels.cuda()
outputs = model(images)
loss = criterion(outputs, labels)
optimizer.zero_grad()
loss.backward()
optimizer.step()
total_loss += loss.item()
correct += (outputs.argmax(1) == labels).sum().item()
total += labels.size(0)
scheduler.step()
acc = correct / total
print(f"Epoch {epoch+1}: loss={total_loss/len(train_loader):.4f}, acc={acc:.2%}")
torch.save(model.state_dict(), "models/defect_classifier.pt")
4. Phase 3: Optimization & Export
"""Optimize models cho deployment"""
from ultralytics import YOLO
# === Detection: YOLOv8 → TensorRT ===
model = YOLO("runs/defect_detection/yolov8s_v1/weights/best.pt")
model.export(
format="engine", # TensorRT
half=True, # FP16
imgsz=640,
device=0,
simplify=True,
)
# Output: best.engine
# === Classification: MobileNetV3 → ONNX ===
import torch
classifier = torchvision.models.mobilenet_v3_small()
classifier.classifier[-1] = torch.nn.Linear(1024, 5)
classifier.load_state_dict(torch.load("models/defect_classifier.pt"))
classifier.eval()
dummy = torch.randn(1, 3, 224, 224)
torch.onnx.export(
classifier, dummy,
"models/defect_classifier.onnx",
opset_version=17,
input_names=["image"],
output_names=["logits"],
)
5. Phase 4: API Server
"""FastAPI server cho inference"""
from fastapi import FastAPI, File, UploadFile, WebSocket
from fastapi.responses import JSONResponse
import cv2
import numpy as np
from ultralytics import YOLO
import onnxruntime as ort
from datetime import datetime
import asyncio
import json
app = FastAPI(title="Defect Detection API")
# Load models
detector = YOLO("models/best.engine")
classifier_session = ort.InferenceSession("models/defect_classifier.onnx")
DEFECT_NAMES = ["scratch", "dent", "discoloration", "crack", "missing_part"]
def classify_crop(crop_img):
"""Classify một crop region"""
img = cv2.resize(crop_img, (224, 224))
img = img.astype(np.float32) / 255.0
img = (img - [0.485, 0.456, 0.406]) / [0.229, 0.224, 0.225]
img = np.transpose(img, (2, 0, 1))[np.newaxis, ...]
logits = classifier_session.run(None, {"image": img.astype(np.float32)})[0]
probs = np.exp(logits) / np.exp(logits).sum()
idx = probs.argmax()
return DEFECT_NAMES[idx], float(probs[0][idx])
@app.post("/detect")
async def detect_defects(file: UploadFile = File(...)):
"""Detect defects trong ảnh upload"""
contents = await file.read()
nparr = np.frombuffer(contents, np.uint8)
img = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
# Detection
results = detector(img, conf=0.5)
defects = []
for box in results[0].boxes:
x1, y1, x2, y2 = box.xyxy[0].cpu().numpy().astype(int)
# Crop & classify
crop = img[y1:y2, x1:x2]
defect_type, confidence = classify_crop(crop)
defects.append({
"type": defect_type,
"confidence": confidence,
"bbox": [int(x1), int(y1), int(x2), int(y2)],
"timestamp": datetime.now().isoformat(),
})
return JSONResponse({
"total_defects": len(defects),
"defects": defects,
"image_size": list(img.shape[:2]),
})
@app.websocket("/ws/stream")
async def websocket_stream(websocket: WebSocket):
"""WebSocket cho real-time camera stream"""
await websocket.accept()
cap = cv2.VideoCapture(0)
try:
while True:
ret, frame = cap.read()
if not ret:
break
results = detector(frame, conf=0.5)
defects = []
for box in results[0].boxes:
x1, y1, x2, y2 = box.xyxy[0].cpu().numpy().astype(int)
crop = frame[y1:y2, x1:x2]
defect_type, conf = classify_crop(crop)
defects.append({
"type": defect_type,
"confidence": conf,
"bbox": [int(x1), int(y1), int(x2), int(y2)],
})
await websocket.send_json({
"defects": defects,
"fps": detector.speed,
})
await asyncio.sleep(0.033) # ~30 FPS
except Exception:
pass
finally:
cap.release()
# Run: uvicorn server:app --host 0.0.0.0 --port 8000
6. Phase 5: Monitoring & Retraining
"""Monitoring pipeline"""
import mlflow
from prometheus_client import Counter, Histogram, start_http_server
import logging
# === Prometheus metrics ===
INFERENCE_COUNT = Counter('inference_total', 'Total inferences')
DEFECT_COUNT = Counter('defects_detected', 'Defects detected', ['type'])
INFERENCE_LATENCY = Histogram('inference_seconds', 'Inference latency')
# === MLflow tracking ===
mlflow.set_tracking_uri("http://mlflow-server:5000")
mlflow.set_experiment("defect_detection_production")
class ProductionMonitor:
def __init__(self):
self.predictions = []
self.low_confidence = []
def log_prediction(self, defects, latency_ms):
"""Log mỗi prediction"""
INFERENCE_COUNT.inc()
INFERENCE_LATENCY.observe(latency_ms / 1000)
for d in defects:
DEFECT_COUNT.labels(type=d["type"]).inc()
# Flag low confidence cho review
if d["confidence"] < 0.7:
self.low_confidence.append(d)
logging.warning(
f"⚠️ Low confidence: {d['type']} ({d['confidence']:.0%})"
)
def check_data_drift(self, recent_images, reference_stats):
"""Detect data drift → trigger retraining"""
# So sánh distribution của brightness, contrast, etc.
recent_brightness = np.mean([img.mean() for img in recent_images])
ref_brightness = reference_stats["mean_brightness"]
drift_score = abs(recent_brightness - ref_brightness) / ref_brightness
if drift_score > 0.2: # >20% drift
logging.error(f"🚨 Data drift detected! Score: {drift_score:.2%}")
self.trigger_retraining()
def trigger_retraining(self):
"""Trigger retraining pipeline"""
with mlflow.start_run(run_name="auto_retrain"):
mlflow.log_param("trigger", "data_drift")
mlflow.log_param("n_new_samples", len(self.low_confidence))
# ... trigger training job
logging.info("🔄 Retraining triggered!")
# Start Prometheus metrics server
start_http_server(9090)
7. Docker Deployment
# Dockerfile
FROM nvidia/cuda:12.1-runtime-ubuntu22.04
# System deps
RUN apt-get update && apt-get install -y \
python3-pip libgl1-mesa-glx libglib2.0-0 \
&& rm -rf /var/lib/apt/lists/*
# Python deps
COPY requirements.txt .
RUN pip3 install --no-cache-dir -r requirements.txt
# App
WORKDIR /app
COPY models/ models/
COPY server.py .
EXPOSE 8000
CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8000"]
# docker-compose.yml
services:
api:
build: .
ports:
- "8000:8000"
deploy:
resources:
reservations:
devices:
- capabilities: [gpu]
mlflow:
image: ghcr.io/mlflow/mlflow:latest
ports:
- "5000:5000"
command: mlflow server --host 0.0.0.0
prometheus:
image: prom/prometheus
ports:
- "9090:9090"
volumes:
- ./prometheus.yml:/etc/prometheus/prometheus.yml
grafana:
image: grafana/grafana
ports:
- "3000:3000"
depends_on:
- prometheus
8. Project Checklist
Phase 1: Data
✅ Thu thập ≥500 ảnh
✅ Label với bounding boxes
✅ Augmentation → 2500+ ảnh
✅ Train/Val split 80/20
Phase 2: Training
✅ YOLOv8s detection: mAP@50 > 0.85
✅ MobileNetV3 classification: accuracy > 90%
✅ Validation trên holdout set
Phase 3: Optimization
✅ Export TensorRT (FP16)
✅ Benchmark: <20ms per frame
✅ Model size < 50MB
Phase 4: Deployment
✅ FastAPI server + WebSocket
✅ Docker containerized
✅ API documentation (Swagger)
Phase 5: Monitoring
✅ Prometheus metrics
✅ Grafana dashboard
✅ MLflow experiment tracking
✅ Data drift detection
✅ Auto-retrain pipeline
Course summary
| Article | Topics | Key knowledge |
|---|---|---|
| 1 | What is CV? | Image basics, OpenCV, PIL |
| 2 | CNN Deep Dive | Conv, Pool, Architectures |
| 3 | Transfer Learning | Pretrained models, fine-tuning |
| 4 | YOLO Detection | YOLOv8 inference |
| 5 | Train YOLO Custom | Custom dataset training |
| 6 | Real-time Detection | Video, webcam, tracking |
| 7 | Image Segmentation | Semantic, Instance, Panoptic |
| 8 | SAM | Segment Anything Model |
| 9 | Image Generation | Stable Diffusion, ControlNet |
| 10 | ViT & CLIP | Vision Transformer, zero-shot |
| 11 | OCR | Tesseract, PaddleOCR, LayoutLM |
| 12 | Multimodal | GPT-4o Vision, Gemini |
| 13 | Edge Deploy | ONNX, TensorRT, TFLite |
| 14 | Capstone | End-to-End System |
Capstone exercises
-
Choose a real problem (may not be defect detection):
- People counting system
- Vehicle license plate recognition
- Waste classification
- Check the quality of agricultural products
-
Apply all 5 phases above.
-
Deliverables:
- Source code on GitHub
- Docker Compose to deploy
- README with benchmark results
- Demo video in action
🎉 Congratulations on completing the Computer Vision with Deep Learning course!