はじめに
これがコース全体の概要です。データ収集から導入、実稼働に至るまで完全な CV システムを構築します。
🎯 目標: 生産ラインで欠陥製品を検出および分類するシステムを構築します (製造欠陥検出)。
このプロジェクトを選んだ理由は何ですか?
✅ Ứ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. システムアーキテクチャ
📸 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. フェーズ 1: データ パイプライン
2.1 データ収集
"""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 データのラベル付け
"""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. フェーズ 2: モデルのトレーニング
3.1 検出モデル (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 分類モデル (検出から作物用)
"""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. フェーズ 3: 最適化とエクスポート
"""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. フェーズ 4: API サーバー
"""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. フェーズ 5: モニタリングと再トレーニング
"""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 のデプロイメント
# 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. プロジェクトのチェックリスト
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
コース概要
| 記事 | トピックス | 重要な知識 |
|---|---|---|
| 1 | 履歴書とは何ですか? | 画像の基礎、OpenCV、PIL |
| 2 | CNN の詳細 | コンバージョン、プール、アーキテクチャ |
| 3 | 転移学習 | 事前トレーニング済みモデル、微調整 |
| 4 | YOLO 検出 | YOLOv8 推論 |
| 5 | 鉄道YOLOカスタム | カスタム データセットのトレーニング |
| 6 | リアルタイム検出 | ビデオ、ウェブカメラ、追跡 |
| 7 | 画像のセグメンテーション | セマンティック、インスタンス、パノプティック |
| 8 | サム | 何でもモデルをセグメント化する |
| 9 | 画像生成 | 安定した拡散、ControlNet |
| 10 | ViT&CLIP | ビジョントランスフォーマー、ゼロショット |
| 11 | OCR | Tesseract、PaddleOCR、LayoutLM |
| 12 | マルチモーダル | GPT-4o ビジョン、ジェミニ |
| 13 | エッジ展開 | ONNX、TensorRT、TFLite |
| 14 | キャップストーン | エンドツーエンド システム |
キャップストーン演習
-
実際の問題を選択してください (欠陥検出ではない場合があります):
- 人数カウントシステム
- 車のナンバープレートの認識
- 廃棄物の分類
- 農産物の品質をチェックする
-
上記の 5 つのフェーズをすべて適用します。
-
成果物:
- GitHub 上のソースコード
- デプロイする Docker Compose
- ベンチマーク結果を含む README
- 動作中のデモビデオ
🎉 深層学習を使用したコンピューター ビジョン コースを修了されました、おめでとうございます!