Uçta Bilgisayarlı Görme: Mobil ve Gömülü Cihazlar için Optimizasyon
Bilgisayarla görme modellerini Raspberry Pi, NVIDIA Jetson, akıllı telefonlar gibi uç cihazlara dağıtın. ARM mikrokontrolörleri - ve bulut dağıtımından tamamen farklı bir mühendislik sorunu veya GPU sunucusu. Kaynaklar sınırlıdır: birkaç watt'lık tüketim, onlarca yerine gigabaytlarca RAM, özel GPU veya giriş seviyesi GPU yok. Ancak milyonlarca uygulama çıkarım gerektiriyor yerel: çevrimdışı gözetim, robot teknolojisi, taşınabilir tıbbi cihazlar, endüstriyel otomasyon Bağlantının olmadığı ortamlarda.
Bu makalede uç dağıtımlar için optimizasyon tekniklerini inceleyeceğiz: niceleme, budama, bilginin ayrıştırılması, optimize edilmiş formatlar (ONNX, TFLite, NCNN) ve gerçek kıyaslamalar Raspberry Pi 5 ve NVIDIA Jetson Orin'de.
Ne Öğreneceksiniz
- Edge donanımına genel bakış: Raspberry Pi, Jetson Nano/Orin, Coral TPU, Hailo
- Niceleme: INT8, FP16 - teori ve pratik uygulama
- Parametreleri azaltmak için yapılandırılmış ve yapılandırılmamış budama
- Bilgi Ayrımı: Büyük modellerden küçük modelleri eğitme
- TFLite ve NCNN: ARM cihazlarına dağıtım
- TensorRT: NVIDIA GPU'da (Jetson) maksimum hız
- CPU ve NPU optimizasyonlarıyla ONNX Çalışma Zamanı
- Raspberry Pi 5'te YOLO26: kıyaslama ve eksiksiz konfigürasyon
- Jetson Orin Nano'da gerçek zamanlı video hattı
1. Bilgisayarla Görme için Uç Donanım
Uç Donanım Karşılaştırması 2026
| Cihaz | İşlemci | GPU/NPU | Veri deposu | TDP | YOLOv8n FPS |
|---|---|---|---|---|---|
| Ahududu Pi 5 | ARM Cortex-A76 4 çekirdekli | VideoCore VII | 8GB | 15W | ~5 FPS |
| Jetson Nano (2 GB) | ARM A57 4 çekirdekli | 128 CUDA çekirdeği | 2GB | 10W | ~20 FPS |
| Jetson Orin Nano | ARM Cortex-A78AE 6 çekirdekli | 1024 CUDA + DLA | 8GB | 25W | ~80FPS |
| Jetson AGX Orin | ARM Cortex-A78AE 12 çekirdekli | 2048 CUDA + DLA | 64GB | 60W | ~200FPS |
| Google Mercan TPU | ARM Cortex-A53 4 çekirdekli | 4 ÜST Kenar TPU | 1 GB | 4W | ~30 FPS (TFLite) |
| Hailo-8 | - (PCIe hızlandırıcı) | 26 TOPS Sinir Motoru | - | 5W | ~120FPS |
2. Niceleme: FP32'den INT8'e
La nicemleme ağırlıkların ve aktivasyonların sayısal kesinliğini azaltır şablonun: float32'den (32 bit) float16'ya (16 bit) veya int8'e (8 bit). Pratik etki: INT8 ile 4 kat daha küçük model, 2-4 kat daha hızlı çıkarım, daha düşük güç tüketimi. Modern tekniklerle doğruluk kaybı genellikle %1'den azdır.
2.1 Eğitim Sonrası Niceleme (PTQ)
PyTorch ile INT8 nicelemesi
import torch
import torch.quantization as quant
from torch.ao.quantization import get_default_qconfig, prepare, convert
from torchvision import models
import copy
def quantize_model_ptq(
model: torch.nn.Module,
calibration_loader,
backend: str = 'x86' # 'x86' per CPU Intel, 'qnnpack' per ARM
) -> torch.nn.Module:
"""
Post-Training Quantization (PTQ): quantizza il modello senza retraining.
Richiede solo un piccolo calibration dataset (~100-1000 immagini).
Flusso:
1. Fuse operazioni (Conv+BN+ReLU -> singola op)
2. Insert observer per calibrazione
3. Esegui calibrazione (forward pass sul dataset di calibrazione)
4. Converti in modello quantizzato
"""
torch.backends.quantized.engine = backend
model_to_quantize = copy.deepcopy(model)
model_to_quantize.eval()
# Step 1: Fuse layer comuni per efficienza
# Esempio per ResNet: (Conv, BN, ReLU) -> singola operazione fused
model_to_quantize = torch.quantization.fuse_modules(
model_to_quantize,
[['conv1', 'bn1', 'relu']], # adatta ai nomi del tuo modello
inplace=True
)
# Step 2: Set qconfig e prepara per calibrazione
qconfig = get_default_qconfig(backend)
model_to_quantize.qconfig = qconfig
prepared_model = prepare(model_to_quantize, inplace=False)
# Step 3: Calibrazione con dati reali
print("Calibrazione quantizzazione...")
prepared_model.eval()
with torch.no_grad():
for i, (images, _) in enumerate(calibration_loader):
prepared_model(images)
if i >= 99: # 100 batch di calibrazione sufficienti
break
if i % 10 == 0:
print(f" Batch {i+1}/100")
# Step 4: Conversione al modello quantizzato
quantized_model = convert(prepared_model, inplace=False)
# Verifica dimensioni
def model_size_mb(m: torch.nn.Module) -> float:
param_size = sum(p.nelement() * p.element_size() for p in m.parameters())
buffer_size = sum(b.nelement() * b.element_size() for b in m.buffers())
return (param_size + buffer_size) / (1024 ** 2)
original_size = model_size_mb(model)
quantized_size = model_size_mb(quantized_model)
print(f"Dimensione originale: {original_size:.1f} MB")
print(f"Dimensione quantizzata: {quantized_size:.1f} MB")
print(f"Riduzione: {original_size / quantized_size:.1f}x")
return quantized_model
def compare_inference_speed(original_model, quantized_model,
input_tensor: torch.Tensor, n_runs: int = 100) -> dict:
"""Confronta velocità tra modello originale e quantizzato."""
import time
results = {}
for name, model in [('FP32', original_model), ('INT8', quantized_model)]:
model.eval()
# Warmup
with torch.no_grad():
for _ in range(10):
model(input_tensor)
# Benchmark
start = time.perf_counter()
with torch.no_grad():
for _ in range(n_runs):
model(input_tensor)
elapsed = time.perf_counter() - start
avg_ms = (elapsed / n_runs) * 1000
results[name] = avg_ms
print(f"{name}: {avg_ms:.2f}ms / inference")
speedup = results['FP32'] / results['INT8']
print(f"Speedup INT8: {speedup:.2f}x")
return results2.2 YOLO (Ultralytics) ile Kantitasyon
YOLO26: Edge için Niceleştirilmiş Dışa Aktarma
from ultralytics import YOLO
model = YOLO('yolo26n.pt') # nano per edge
# ---- TFLite INT8 per Raspberry Pi / Coral TPU ----
model.export(
format='tflite',
imgsz=320, # risoluzione ridotta per edge
int8=True, # quantizzazione INT8
data='coco.yaml' # dataset per calibrazione PTQ
)
# Output: yolo26n_int8.tflite
# ---- NCNN per CPU ARM (Raspberry Pi, Android) ----
model.export(
format='ncnn',
imgsz=320,
half=False # NCNN usa FP32 o INT8 nativo
)
# Output: yolo26n_ncnn_model/
# ---- TensorRT FP16 per Jetson ----
model.export(
format='engine',
imgsz=640,
half=True, # FP16
workspace=2, # GB workspace (ridotto per Jetson Nano)
device=0
)
# Output: yolo26n.engine
# ---- ONNX + ONNX Runtime per CPU/NPU ----
model.export(
format='onnx',
imgsz=320,
opset=17,
simplify=True,
dynamic=False # batch size fisso per deployment edge
)
print("Export completati per tutti i target edge")3. Raspberry Pi 5'te YOLO
Il Ahududu Pi 5 8GB RAM ve ARM Cortex-A76 işlemci ile uç yapay zeka için en erişilebilir giriş noktası. Doğru optimizasyonlarla (NCNN, çözünürlük azaltılmış, çıkarım sıklığını azaltmak için izleme) bir tespit sistemi elde edilebilir gerçek zamanlı olarak işlevseldir.
Raspberry Pi 5 Kurulumu ve Optimizasyonu
# ============================================
# SETUP RASPBERRY PI 5 per Computer Vision
# ============================================
# 1. Installazione dipendenze base
# sudo apt update && sudo apt install -y python3-pip libopencv-dev
# pip install ultralytics ncnn onnxruntime
# 2. Ottimizzazioni sistema per AI
# In /boot/firmware/config.txt:
# gpu_mem=256 # Aumenta memoria GPU (VideoCore VII)
# over_voltage=6 # Overclock lieve
# arm_freq=2800 # Frequenza CPU max (stock 2.4GHz)
# ============================================
# INFERENCE con NCNN su Raspberry Pi
# ============================================
import ncnn
import cv2
import numpy as np
import time
class YOLOncnn:
"""
YOLO inference con NCNN - ottimizzato per CPU ARM.
NCNN e sviluppato da Tencent ed e il runtime più veloce per ARM CPU.
"""
def __init__(self, param_path: str, bin_path: str,
num_threads: int = 4, input_size: int = 320):
self.net = ncnn.Net()
self.net.opt.num_threads = num_threads # usa tutti i core
self.net.opt.use_vulkan_compute = False # no GPU su RPi
self.net.load_param(param_path)
self.net.load_model(bin_path)
self.input_size = input_size
def predict(self, img_bgr: np.ndarray, conf_thresh: float = 0.4) -> list[dict]:
"""Inference NCNN su CPU ARM."""
h, w = img_bgr.shape[:2]
# Resize + normalizzazione per NCNN
img_resized = cv2.resize(img_bgr, (self.input_size, self.input_size))
img_rgb = cv2.cvtColor(img_resized, cv2.COLOR_BGR2RGB)
mat_in = ncnn.Mat.from_pixels(
img_rgb, ncnn.Mat.PixelType.PIXEL_RGB, self.input_size, self.input_size
)
mean_vals = [0.485 * 255, 0.456 * 255, 0.406 * 255]
norm_vals = [1/0.229/255, 1/0.224/255, 1/0.225/255]
mat_in.substract_mean_normalize(mean_vals, norm_vals)
ex = self.net.create_extractor()
ex.input("images", mat_in)
_, mat_out = ex.extract("output0")
return self._parse_output(mat_out, conf_thresh, w, h)
def _parse_output(self, mat_out, conf_thresh, orig_w, orig_h) -> list[dict]:
"""Parsing dell'output NCNN in formato detection."""
detections = []
for i in range(mat_out.h):
row = np.array(mat_out.row(i))
confidence = row[4]
if confidence < conf_thresh:
continue
class_scores = row[5:]
class_id = int(np.argmax(class_scores))
class_conf = confidence * class_scores[class_id]
if class_conf >= conf_thresh:
# Coordinate normalizzate -> pixel
cx, cy, bw, bh = row[:4]
x1 = int((cx - bw/2) * orig_w / self.input_size)
y1 = int((cy - bh/2) * orig_h / self.input_size)
x2 = int((cx + bw/2) * orig_w / self.input_size)
y2 = int((cy + bh/2) * orig_h / self.input_size)
detections.append({
'class_id': class_id,
'confidence': float(class_conf),
'bbox': (x1, y1, x2, y2)
})
return detections
def run_rpi_detection_loop(model_param: str, model_bin: str,
camera_id: int = 0) -> None:
"""Loop di detection real-time ottimizzato per Raspberry Pi."""
detector = YOLOncnn(model_param, model_bin, num_threads=4, input_size=320)
cap = cv2.VideoCapture(camera_id)
# Ottimizza acquisizione per RPi
cap.set(cv2.CAP_PROP_FRAME_WIDTH, 640)
cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 480)
cap.set(cv2.CAP_PROP_FPS, 30)
cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)
frame_skip = 2 # Processa 1 frame su 3 per risparmiare CPU
frame_count = 0
cached_dets = []
fps_history = []
while True:
ret, frame = cap.read()
if not ret:
break
t0 = time.perf_counter()
if frame_count % frame_skip == 0:
cached_dets = detector.predict(frame, conf_thresh=0.4)
elapsed = time.perf_counter() - t0
fps = 1.0 / elapsed if elapsed > 0 else 0
fps_history.append(fps)
# Visualizzazione
for det in cached_dets:
x1, y1, x2, y2 = det['bbox']
cv2.rectangle(frame, (x1, y1), (x2, y2), (0, 255, 0), 2)
cv2.putText(frame, f"{det['confidence']:.2f}",
(x1, y1-5), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0,255,0), 2)
avg_fps = sum(fps_history[-30:]) / min(len(fps_history), 30)
cv2.putText(frame, f"FPS: {avg_fps:.1f}", (10, 30),
cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)
cv2.imshow('RPi Detection', frame)
if cv2.waitKey(1) & 0xFF == ord('q'):
break
frame_count += 1
cap.release()
cv2.destroyAllWindows()
print(f"FPS medio: {sum(fps_history)/len(fps_history):.1f}")4. NVIDIA Jetson Orin: TensorRT ve DLA
Il Jetson Orin Nano (25W), 1024 CUDA çekirdeği ve DLA (Derin Öğrenme) sunar Hızlandırıcı) özel. TensorRT FP16 ve YOLO26n modeliyle bunların üstesinden kolaylıkla gelinebilir 640x640 videoda 100 FPS.
Jetson'da TensorRT: Kurulum ve Çıkarım
from ultralytics import YOLO
import cv2
import time
def setup_jetson_pipeline(model_path: str = 'yolo26n.pt') -> YOLO:
"""
Setup ottimale per Jetson Orin:
1. Esporta in TensorRT FP16
2. Configura jetson_clocks per prestazioni massime
3. Imposta modalità performance per la GPU
"""
import subprocess
# Massimizza performance Jetson (esegui una sola volta)
# subprocess.run(['sudo', 'jetson_clocks'], check=True)
# subprocess.run(['sudo', 'nvpmodel', '-m', '0'], check=True) # MAXN mode
model = YOLO(model_path)
print("Esportazione TensorRT FP16...")
model.export(
format='engine',
imgsz=640,
half=True, # FP16 - quasi la stessa accuratezza di FP32 ma 2x più veloce
workspace=2, # GB workspace GPU (Jetson Orin Nano ha 8GB shared)
device=0,
batch=1,
simplify=True
)
# Carica il modello TensorRT
trt_model = YOLO('yolo26n.engine')
print("Modello TensorRT pronto")
return trt_model
def run_jetson_pipeline(model: YOLO, source=0) -> None:
"""Pipeline real-time ottimizzata per Jetson con statistiche."""
cap = cv2.VideoCapture(source)
cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)
fps_list = []
frame_count = 0
try:
while True:
ret, frame = cap.read()
if not ret:
break
t0 = time.perf_counter()
results = model.predict(
frame, conf=0.35, iou=0.45,
verbose=False, half=True # FP16 inference
)
elapsed = time.perf_counter() - t0
fps = 1.0 / elapsed
fps_list.append(fps)
# Annotazione con informazioni performance
annotated = results[0].plot()
avg_fps = sum(fps_list[-30:]) / min(len(fps_list), 30)
info_text = [
f"FPS: {fps:.0f} (avg: {avg_fps:.0f})",
f"Detections: {len(results[0].boxes)}",
f"Inference: {elapsed*1000:.1f}ms"
]
for i, text in enumerate(info_text):
cv2.putText(annotated, text, (10, 30 + i * 30),
cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 255, 0), 2)
cv2.imshow('Jetson Pipeline', annotated)
if cv2.waitKey(1) & 0xFF == ord('q'):
break
frame_count += 1
finally:
cap.release()
cv2.destroyAllWindows()
if fps_list:
print(f"\n=== Stats Jetson ===")
print(f"Frame: {frame_count}")
print(f"FPS medio: {sum(fps_list)/len(fps_list):.1f}")
print(f"FPS massimo: {max(fps_list):.1f}")
print(f"Latenza minima: {1000/max(fps_list):.1f}ms")5. Budama ve Bilgi Damıtma
5.1 Yapılandırılmış Budama
PyTorch ile Yapılandırılmış Budama
import torch
import torch.nn as nn
import torch.nn.utils.prune as prune
def apply_structured_pruning(model: nn.Module,
amount: float = 0.3,
n: int = 2) -> nn.Module:
"""
Structured L2-norm pruning: rimuove interi filtri/neuroni.
Produce modelli più veloci in inferenza (a differenza del pruning non strutturato
che produce solo modelli più piccoli ma non necessariamente più veloci).
amount: percentuale di filtri da rimuovere (0.3 = 30%)
n: norma L_n usata per il ranking dei filtri
"""
for name, module in model.named_modules():
if isinstance(module, nn.Conv2d):
# Prune i filtri convoluzionali meno importanti
prune.ln_structured(
module,
name='weight',
amount=amount,
n=n,
dim=0 # dim=0 = prune filtri in output
)
elif isinstance(module, nn.Linear):
prune.ln_structured(
module,
name='weight',
amount=amount,
n=n,
dim=0
)
return model
def remove_pruning_masks(model: nn.Module) -> nn.Module:
"""
Rende permanente il pruning: rimuove le maschere e i parametri "orig",
lasciando solo i pesi pruned. Necessario prima dell'export.
"""
for name, module in model.named_modules():
if isinstance(module, (nn.Conv2d, nn.Linear)):
try:
prune.remove(module, 'weight')
except ValueError:
pass
return model
def prune_and_finetune(model: nn.Module, train_loader, val_loader,
prune_amount: float = 0.2, finetune_epochs: int = 5) -> nn.Module:
"""
Pipeline completa:
1. Prune il modello (rimuove il prune_amount% dei filtri)
2. Fine-tunes per recuperare l'accuratezza persa
3. Rimuove le maschere e finalizza
"""
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model.to(device)
print(f"Applying {prune_amount*100:.0f}% structured pruning...")
model = apply_structured_pruning(model, amount=prune_amount)
# Fine-tuning rapido per recupero accuratezza
criterion = nn.CrossEntropyLoss()
optimizer = torch.optim.AdamW(model.parameters(), lr=1e-4)
for epoch in range(finetune_epochs):
model.train()
total_loss = 0.0
for images, labels in train_loader:
images, labels = images.to(device), labels.to(device)
loss = criterion(model(images), labels)
optimizer.zero_grad(set_to_none=True)
loss.backward()
optimizer.step()
total_loss += loss.item()
model.eval()
correct = total = 0
with torch.no_grad():
for images, labels in val_loader:
images, labels = images.to(device), labels.to(device)
preds = model(images).argmax(1)
correct += preds.eq(labels).sum().item()
total += labels.size(0)
print(f" FT Epoch {epoch+1}/{finetune_epochs} | "
f"Loss: {total_loss/len(train_loader):.4f} | "
f"Acc: {100.*correct/total:.2f}%")
# Finalizza pruning
model = remove_pruning_masks(model)
print("Pruning completato e finalizzato")
return model6. Uç Modeller için Bilginin Damıtılması
Il Bilgi Damıtma (KD, Hinton ve diğerleri, 2015) "bilgiyi" aktarır büyük bir modelin (öğretmen) küçük bir modele (öğrenci) dönüştürülmesi. Öğrenci sadece öğrenmekle kalmıyor veri kümesinin sert etiketleri, ancak yumuşak tahminler öğretmenin dağılımları: veri alanının yapısı hakkında bilgi içeren olasılıklar (örn. "kedi", "araba"dan çok "kaplan"a benzer).
Bilginin Damıtılması: Öğretmen-Öğrenci Eğitimi
import torch
import torch.nn as nn
import torch.nn.functional as F
class DistillationLoss(nn.Module):
"""
Loss combinata per Knowledge Distillation.
L_total = alpha * L_hard + (1 - alpha) * L_soft
L_hard = CrossEntropyLoss(student_logits, true_labels)
L_soft = KLDivLoss(softmax(student/T), softmax(teacher/T)) * T^2
T (temperature): valori alti -> distribuzioni più soft -> più informazione strutturale
alpha: peso relativo tra label reali e distillazione dal teacher
"""
def __init__(self, temperature: float = 4.0, alpha: float = 0.7):
super().__init__()
self.T = temperature
self.alpha = alpha
self.hard_loss = nn.CrossEntropyLoss()
self.soft_loss = nn.KLDivLoss(reduction='batchmean')
def forward(self,
student_logits: torch.Tensor,
teacher_logits: torch.Tensor,
labels: torch.Tensor) -> torch.Tensor:
# Loss su label reali (hard labels)
hard = self.hard_loss(student_logits, labels)
# Loss su soft predictions del teacher (KL divergence)
student_soft = F.log_softmax(student_logits / self.T, dim=1)
teacher_soft = F.softmax(teacher_logits / self.T, dim=1)
soft = self.soft_loss(student_soft, teacher_soft) * (self.T ** 2)
return self.alpha * hard + (1 - self.alpha) * soft
def train_with_distillation(
teacher: nn.Module, # modello grande, già addestrato
student: nn.Module, # modello piccolo da addestrare
train_loader,
val_loader,
n_epochs: int = 50,
temperature: float = 4.0,
alpha: float = 0.7,
lr: float = 1e-3
) -> nn.Module:
"""
Training del modello student con KD.
Il teacher rimane frozen durante tutto il training.
Tipico risultato:
- MobileNetV3 senza KD su ImageNet: ~67% Top-1
- MobileNetV3 con KD da ResNet-50: ~72% Top-1
- ResNet-50 (teacher): ~76% Top-1
- Delta: +5% con 5x meno parametri!
"""
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
teacher.eval() # Teacher sempre in eval mode
student.to(device)
teacher.to(device)
criterion = DistillationLoss(temperature=temperature, alpha=alpha)
optimizer = torch.optim.AdamW(student.parameters(), lr=lr, weight_decay=0.01)
scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=n_epochs)
best_val_acc = 0.0
best_state = None
for epoch in range(n_epochs):
student.train()
total_loss = 0.0
for images, labels in train_loader:
images, labels = images.to(device), labels.to(device)
# Forward pass
student_logits = student(images)
with torch.no_grad(): # Teacher: nessun gradiente
teacher_logits = teacher(images)
# Loss combinata
loss = criterion(student_logits, teacher_logits, labels)
optimizer.zero_grad(set_to_none=True)
loss.backward()
torch.nn.utils.clip_grad_norm_(student.parameters(), 1.0)
optimizer.step()
total_loss += loss.item()
scheduler.step()
# Validation
student.eval()
correct = total = 0
with torch.no_grad():
for images, labels in val_loader:
images, labels = images.to(device), labels.to(device)
preds = student(images).argmax(1)
correct += preds.eq(labels).sum().item()
total += labels.size(0)
val_acc = 100.0 * correct / total
if val_acc > best_val_acc:
best_val_acc = val_acc
best_state = {k: v.cpu().clone() for k, v in student.state_dict().items()}
if (epoch + 1) % 10 == 0:
print(f"Epoch {epoch+1}/{n_epochs} | "
f"Loss: {total_loss/len(train_loader):.4f} | "
f"Val Acc: {val_acc:.2f}% | "
f"Best: {best_val_acc:.2f}%")
student.load_state_dict(best_state)
print(f"\nBest validation accuracy: {best_val_acc:.2f}%")
return studentEdge için Sıkıştırma Stratejileri Karşılaştırması
| Teknik | Param azaltma | Hızlanma | Acc. Kayıp | Yeniden Eğitim Gerektirir |
|---|---|---|---|---|
| Niceleme INT8 | 4x | 2-4x | <%1 | Hayır (PTQ) / Evet (QAT) |
| Yapılandırılmış budama %30 | 1,4x | 1,3-1,6x | %1-3 | Evet (ince ayar) |
| Bilgi Damıtma | 5-10x (model değişimi) | 5-10x | %3-8 | Evet (tam eğitim) |
| FP16 (TensorRT) | 2x | 1,5-2x | <%0,5 | No |
| Soru + Budama + KD | 10-20x | 8-15x | %2-5 | Si |
7. ONNX Çalışma Zamanı: Donanımlar Arasında Taşınabilirlik
ONNX (Açık Sinir Ağı Değişimi) ve taşınabilirlik için standart boyut derin öğrenme modellerinden ONNX'e aktarıldıktan sonra aynı model CPU, NVIDIA GPU, ARM NPU, Intel OpenVINO, Apple Neural Engine'de ONNX Runtime ile çalıştırın çıkarım kodunda değişiklik yapılmadan.
ONNX Çalışma Zamanı ile ONNX Dışa Aktarma ve Çıkarım
import torch
import onnx
import onnxruntime as ort
import numpy as np
import time
def export_to_onnx(model: torch.nn.Module,
input_shape: tuple = (1, 3, 640, 640),
output_path: str = 'model.onnx',
opset: int = 17) -> str:
"""
Esporta modello PyTorch in formato ONNX ottimizzato.
opset=17: versione del opset ONNX (più alta = più operatori supportati)
dynamic_axes: permette batch size variabile (utile per server, non per edge)
"""
model.eval()
dummy_input = torch.zeros(input_shape)
# Export con ottimizzazioni
torch.onnx.export(
model,
dummy_input,
output_path,
opset_version=opset,
input_names=['images'],
output_names=['output'],
dynamic_axes={
'images': {0: 'batch'},
'output': {0: 'batch'}
},
do_constant_folding=True, # ottimizza operazioni costanti
verbose=False
)
# Verifica il modello esportato
onnx_model = onnx.load(output_path)
onnx.checker.check_model(onnx_model)
print(f"Modello ONNX valido: {output_path}")
return output_path
class ONNXRuntimeInference:
"""
Inference ottimizzata con ONNX Runtime.
Supporta CPU, GPU CUDA, ARM (QNN), Intel OpenVINO come backend.
"""
def __init__(self, model_path: str, device: str = 'cpu'):
providers = self._get_providers(device)
sess_options = ort.SessionOptions()
sess_options.graph_optimization_level = (
ort.GraphOptimizationLevel.ORT_ENABLE_ALL
)
# Numero di thread per CPU inference
sess_options.intra_op_num_threads = 4
sess_options.inter_op_num_threads = 2
self.session = ort.InferenceSession(
model_path, sess_options, providers=providers
)
# Cache nomi input/output
self.input_name = self.session.get_inputs()[0].name
self.output_name = self.session.get_outputs()[0].name
print(f"ONNX Runtime caricato su: {providers[0]}")
def _get_providers(self, device: str) -> list:
if device == 'cuda':
return ['CUDAExecutionProvider', 'CPUExecutionProvider']
elif device == 'openvino':
return ['OpenVINOExecutionProvider', 'CPUExecutionProvider']
else:
return ['CPUExecutionProvider']
def predict(self, image: np.ndarray) -> np.ndarray:
"""Inference su immagine numpy preprocessata."""
# Assicura formato float32 [B, C, H, W]
if image.ndim == 3:
image = image[np.newaxis, ...]
image = image.astype(np.float32)
return self.session.run(
[self.output_name], {self.input_name: image}
)[0]
def benchmark(self, input_shape: tuple = (1, 3, 640, 640),
n_runs: int = 100) -> dict:
"""Misura latenza e throughput."""
dummy = np.random.rand(*input_shape).astype(np.float32)
# Warmup
for _ in range(10):
self.predict(dummy)
# Benchmark
start = time.perf_counter()
for _ in range(n_runs):
self.predict(dummy)
elapsed = time.perf_counter() - start
avg_ms = (elapsed / n_runs) * 1000
fps = 1000.0 / avg_ms
print(f"ONNX Runtime: {avg_ms:.2f}ms ({fps:.1f} FPS)")
return {'avg_ms': avg_ms, 'fps': fps}8. Uç Dağıtımı için En İyi Uygulamalar
Üretime Hazır Uç Dağıtımı için Kontrol Listesi
- Gereksinimleri karşılayan en küçük modeli seçin: RPi için YOLOv8n veya YOLO26n, Jetson Orin için YOLOv8m. Large veya XLarge modellerini kenarda kullanmayın. HER ZAMAN hedef donanımı ölçün.
- Giriş çözünürlüğünü azaltın: 640x640 yerine 320x320, orta düzeyde doğruluk kaybıyla çıkarım süresini %75 azaltır. Büyük parçalar için 320 yeterlidir.
- Akıllı çerçeve atlama: Nesneler yavaş hareket ediyorsa 3-5 kareden 1'ini işleyin. Atlanan karelerdeki konumları enterpolasyona tabi tutmak için bir izleyici (CSRT, ByteTrack) kullanın.
- Edinme hattınızı optimize edin: Gecikmeyi en aza indirmek için CAP_PROP_BUFFERSIZE=1 olarak ayarlayın. OpenCV'den daha az ek yük için V4L2'yi doğrudan Linux'ta kullanın.
- Jetson'da TensorRT: Her zaman. PyTorch ile TensorRT FP16 arasındaki fark 5-8x'tir. Jetson'da çıkarım üretimi için PyTorch'u kullanmanın hiçbir nedeni yoktur.
- Termal kısma: RPi ve Jetson'da aşırı ısınma kısılmaya neden olur. Soğutucu ekleyin, sıcaklığı kontrol edin
vcgencmd measure_temp(RPi) veyategrastats(Jetson). - Sadece hızı değil, enerjiyi de ölçün: FPS/watt, pilli cihazlar için önemli olan ölçümdür. 2 kat daha yavaş ancak 4 kat daha fazla enerji tasarrufu sağlayan ve sıklıkla tercih edilen bir model.
- Watchdog ve zarif yeniden başlatma: Üretim uç cihazlarında her zaman bir çökme veya donma durumunda çıkarım sürecini yeniden başlatan bir gözlemci uygulayın.
- Kenar dostu günlük kaydı: RPi'de olayları yerel olarak kaydetmek için uzak veritabanları yerine SQLite kullanın. Bağlantı mevcut olduğunda buluta toplu olarak senkronize edin.
Raspberry Pi'de Termal İzleme ve Watchdog
import subprocess
import threading
import time
import logging
class ThermalMonitor:
"""
Monitor termico per Raspberry Pi/Jetson.
Riduce automaticamente il carico di lavoro se la temperatura e troppo alta.
"""
TEMP_WARNING = 75.0 # Celsius: riduce frame rate
TEMP_CRITICAL = 85.0 # Celsius: ferma il processing
def __init__(self, platform: str = 'rpi',
check_interval: float = 5.0):
self.platform = platform
self.check_interval = check_interval
self.current_temp = 0.0
self.throttle_factor = 1.0 # 1.0 = nessun throttling
self._stop = threading.Event()
def get_temperature(self) -> float:
"""Legge la temperatura del SoC."""
try:
if self.platform == 'rpi':
result = subprocess.run(
['vcgencmd', 'measure_temp'],
capture_output=True, text=True
)
# Output: "temp=62.1'C"
temp_str = result.stdout.strip()
return float(temp_str.split('=')[1].replace("'C", ''))
elif self.platform == 'jetson':
# Legge da sysfs
with open('/sys/class/thermal/thermal_zone0/temp') as f:
return float(f.read().strip()) / 1000.0
except Exception as e:
logging.warning(f"Impossibile leggere temperatura: {e}")
return 0.0
def get_throttle_factor(self) -> float:
"""Restituisce il fattore di throttling (0.0-1.0)."""
temp = self.current_temp
if temp < self.TEMP_WARNING:
return 1.0
elif temp < self.TEMP_CRITICAL:
# Throttling lineare tra 75 e 85 gradi
factor = 1.0 - (temp - self.TEMP_WARNING) / (
self.TEMP_CRITICAL - self.TEMP_WARNING
)
return max(0.2, factor) # mai sotto il 20%
else:
return 0.0 # ferma il processing
def monitor_loop(self) -> None:
"""Thread di monitoraggio termico."""
while not self._stop.is_set():
self.current_temp = self.get_temperature()
self.throttle_factor = self.get_throttle_factor()
if self.current_temp >= self.TEMP_CRITICAL:
logging.critical(f"TEMP CRITICA: {self.current_temp:.1f}C - "
f"Processing fermato!")
elif self.current_temp >= self.TEMP_WARNING:
logging.warning(f"TEMP ALTA: {self.current_temp:.1f}C - "
f"Throttle: {self.throttle_factor:.2f}")
time.sleep(self.check_interval)
def start(self) -> None:
t = threading.Thread(target=self.monitor_loop, daemon=True)
t.start()
def stop(self) -> None:
self._stop.set()Sonuçlar
Bilgisayarla görme modellerini uç cihazlara dağıtmak bütünsel bir yaklaşım gerektirir donanım seçimini, model optimizasyonunu ve boru hattı mühendisliğini birleştiren bir teknolojidir. Bu mevcut değil benzersiz bir çözüm: en uygun kombinasyon, baskın kısıtlamaya (gecikme, enerji, doğruluk, maliyet). Bu makalede eksiksiz bir araç seti oluşturduk:
- Edge donanımı: Bütçe senaryoları için Raspberry Pi 5, gerçek zamanlı performans için Jetson Orin, ultra düşük güç için Coral TPU ve Hailo-8
- INT8 nicemleme: 4 kat boyut küçültme, 2-4 kat hızlandırma, PTQ ile <%1 doğruluk kaybı
- ARM CPU için NCNN, NVIDIA GPU için TensorRT, ultra düşük güç için TFLite + Coral TPU
- Yapılandırılmış budama + ince ayar: minimum doğruluk kaybıyla filtrelerin %20-30'unu kaldırın
- Bilginin Damıtma: Bilgiyi büyük modellerden gömülü modellere aktarın
- ONNX Çalışma Zamanı: farklı donanım platformları arasında model taşınabilirliği
- Termal izleme ve gözlemci: 7/24 uç üretim için sağlam sistemler
- Kare atlama + izleme: az hareketin olduğu sahnelerde hesaplamayı %70-80 azaltın
Seri Gezintisi
Seriler Arası Kaynaklar
- MLOps: Üretimde Hizmet Veren Model - Kubernetes ve Triton ile bulut dağıtımı
- Gelişmiş Derin Öğrenme: Niceleme ve Sıkıştırma