🚨 Phát hiện Tai nạn Giao thông với AI
· 5 phút để đọc
Một trong những tính năng quan trọng nhất của UIP là khả năng phát hiện tai nạn giao thông tự động trong vòng vài giây. Bài viết này chia sẻ cách chúng tôi xây dựng hệ thống này.
🎯 Vấn đề cần giải quyết
Tại TP.HCM, mỗi năm có hàng nghìn vụ tai nạn giao thông. Thời gian phản ứng nhanh là yếu tố quyết định để:
- 🚑 Cứu sống nạn nhân
- 🚗 Giảm ùn tắc giao thông kéo dài
- 📊 Thu thập dữ liệu chính xác
Mục tiêu: Phát hiện tai nạn trong < 3 giây với độ chính xác > 90%
🧠 Lựa chọn Model: Tại sao YOLOX?
So sánh các model
| Model | mAP | Speed (FPS) | Size |
|---|---|---|---|
| YOLOv5 | 50.7% | 140 | 27MB |
| YOLOv7 | 51.2% | 120 | 37MB |
| YOLOX | 51.1% | 155 | 25MB |
| YOLOR | 52.0% | 80 | 45MB |
YOLOX là lựa chọn tối ưu vì:
- ✅ Anchor-free design - đơn giản hơn
- ✅ Decoupled head - train nhanh hơn
- ✅ SimOTA label assignment - chính xác hơn
- ✅ Strong augmentation - robust hơn
🏗️ Kiến trúc hệ thống
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ Camera │────▶│ Image │────▶│ YOLOX │
│ Stream │ │ Buffer │ │ Model │
└─────────────┘ └─────────────┘ └──────┬──────┘
│
▼
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ Alert │◀────│ Post │◀────│ Detection │
│ System │ │ Process │ │ Results │
└─────────────┘ └─────────────┘ └─────────────┘
💻 Implementation
1. Model Loading
import torch
from yolox.exp import get_exp
from yolox.utils import postprocess
class AccidentDetector:
def __init__(self, model_path: str, device: str = "cuda"):
self.device = device
self.exp = get_exp("yolox_s")
self.model = self.exp.get_model()
# Load pretrained weights
checkpoint = torch.load(model_path, map_location=device)
self.model.load_state_dict(checkpoint["model"])
self.model.to(device)
self.model.eval()
# Accident-related classes
self.accident_classes = [
"car_crash",
"motorcycle_fallen",
"person_injured",
"vehicle_damage"
]
2. Inference Pipeline
async def detect_accidents(self, images: List[np.ndarray]) -> List[Detection]:
"""Process batch of images for accident detection"""
detections = []
# Preprocess
batch = self.preprocess_batch(images)
with torch.no_grad():
# Forward pass
outputs = self.model(batch)
# Post-process
outputs = postprocess(
outputs,
num_classes=len(self.classes),
conf_thre=0.7,
nms_thre=0.45
)
for i, output in enumerate(outputs):
if output is None:
continue
for det in output:
cls = int(det[6])
conf = float(det[4] * det[5])
if self.classes[cls] in self.accident_classes:
detections.append(Detection(
image_id=i,
class_name=self.classes[cls],
confidence=conf,
bbox=det[:4].tolist(),
timestamp=datetime.now()
))
return detections
3. Alert Dispatch
async def dispatch_alert(self, detection: Detection, camera: Camera):
"""Send accident alert through multiple channels"""
alert = AccidentAlert(
camera_id=camera.id,
location=camera.location,
severity=self.calculate_severity(detection),
confidence=detection.confidence,
timestamp=detection.timestamp,
image_url=await self.capture_snapshot(camera)
)
# Multi-channel dispatch
await asyncio.gather(
self.send_websocket_alert(alert),
self.send_email_alert(alert),
self.send_sms_alert(alert),
self.store_in_database(alert),
self.publish_ngsi_ld_entity(alert)
)
📊 Training Data
Dataset Composition
| Category | Images | Annotations |
|---|---|---|
| Car crashes | 5,000 | 12,000 |
| Motorcycle falls | 3,500 | 8,000 |
| Pedestrian incidents | 2,000 | 4,500 |
| Multi-vehicle | 1,500 | 5,000 |
| Total | 12,000 | 29,500 |
Data Augmentation
from albumentations import (
Compose, RandomBrightnessContrast,
RandomRain, RandomFog, RandomSunFlare,
HorizontalFlip, RandomScale
)
transform = Compose([
RandomBrightnessContrast(p=0.5),
RandomRain(p=0.3), # Mưa
RandomFog(p=0.2), # Sương mù
RandomSunFlare(p=0.2), # Chói sáng
HorizontalFlip(p=0.5),
RandomScale(scale_limit=0.2, p=0.5)
])
📈 Performance Results
Accuracy Metrics
| Metric | Value |
|---|---|
| mAP@0.5 | 91.2% |
| mAP@0.5:0.95 | 72.8% |
| Precision | 89.5% |
| Recall | 93.1% |
| F1-Score | 91.3% |
Speed Metrics
| Hardware | FPS | Latency |
|---|---|---|
| RTX 3080 | 155 | 6.5ms |
| RTX 3060 | 120 | 8.3ms |
| T4 (Cloud) | 95 | 10.5ms |
| CPU (i7-12700) | 15 | 66ms |
🚀 Deployment
GPU Optimization
# TensorRT optimization
import tensorrt as trt
def optimize_for_tensorrt(model_path: str, output_path: str):
"""Convert ONNX model to TensorRT"""
logger = trt.Logger(trt.Logger.INFO)
builder = trt.Builder(logger)
config = builder.create_builder_config()
config.set_flag(trt.BuilderFlag.FP16) # Half precision
config.max_workspace_size = 1 << 30 # 1GB
network = builder.create_network(
1 << int(trt.NetworkDefinitionCreationFlag.EXPLICIT_BATCH)
)
parser = trt.OnnxParser(network, logger)
with open(model_path, 'rb') as f:
parser.parse(f.read())
engine = builder.build_serialized_network(network, config)
with open(output_path, 'wb') as f:
f.write(engine)
Batch Processing
async def process_camera_batch(cameras: List[Camera], batch_size: int = 32):
"""Process cameras in batches for efficiency"""
for i in range(0, len(cameras), batch_size):
batch = cameras[i:i + batch_size]
# Fetch images concurrently
images = await asyncio.gather(*[
fetch_camera_image(cam) for cam in batch
])
# Batch inference
detections = await detector.detect_accidents(images)
# Process detections
for detection in detections:
if detection.is_accident:
await dispatch_alert(detection, batch[detection.image_id])
📊 Real-world Results
Sau 3 tháng triển khai thử nghiệm:
| Metric | Value |
|---|---|
| Tai nạn phát hiện | 127 vụ |
| Thời gian phát hiện TB | 2.3 giây |
| False positives | 8 (6.3%) |
| Thời gian phản ứng giảm | 45% |
🎓 Lessons Learned
- Lighting matters - Augmentation với các điều kiện ánh sáng khác nhau rất quan trọng
- Context helps - Kết hợp thông tin camera position cải thiện accuracy
- Edge cases - Xe máy nhỏ khó phát hiện hơn ô tô
- False positives - Cần filtering thông minh để giảm false alarms
🔜 Future Work
- Multi-camera tracking
- Severity estimation
- Incident type classification
- Integration với emergency services
Bạn muốn đóng góp? Xem Contributing Guide để bắt đầu!
Nguyễn Nhật Quang & Nguyễn Việt Hoàng - UIP Team