03 - YOLO26: Rilevamento Oggetti Real-Time

# Ogni detection e rappresentata da:
# [x_center, y_center, width, height, confidence] + [p_class1, p_class2, ..., p_classN]

# Esempio: detection di un gatto (classe 0) in un'immagine 640x640
detection = {
    'bbox': (0.45, 0.60, 0.30, 0.40),  # x_c, y_c, w, h (normalizzati 0-1)
    'confidence': 0.94,                 # confidence score del box
    'class_id': 0,                      # indice classe
    'class_name': 'gatto',
    'class_prob': 0.96                  # probabilità condizionale della classe
}

# Il "final score" e: confidence * class_prob = 0.94 * 0.96 = 0.90

# Coordinate in pixel (immagine 640x640):
x_c_px = 0.45 * 640   # = 288
y_c_px = 0.60 * 640   # = 384
w_px   = 0.30 * 640   # = 192
h_px   = 0.40 * 640   # = 256

# Conversione a [x1, y1, x2, y2]
x1 = x_c_px - w_px / 2   # = 192
y1 = y_c_px - h_px / 2   # = 256
x2 = x_c_px + w_px / 2   # = 384
y2 = y_c_px + h_px / 2   # = 512

import numpy as np

def compute_iou(box1: np.ndarray, box2: np.ndarray) -> float:
    """
    Calcola Intersection over Union tra due bounding boxes.
    Input: [x1, y1, x2, y2] per entrambi i box.
    Output: IoU in [0, 1]
    """
    # Coordinate dell'intersezione
    x_left   = max(box1[0], box2[0])
    y_top    = max(box1[1], box2[1])
    x_right  = min(box1[2], box2[2])
    y_bottom = min(box1[3], box2[3])

    if x_right < x_left or y_bottom < y_top:
        return 0.0  # Nessuna intersezione

    intersection = (x_right - x_left) * (y_bottom - y_top)
    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

def non_maximum_suppression(
    boxes: np.ndarray,
    scores: np.ndarray,
    iou_threshold: float = 0.45,
    score_threshold: float = 0.25
) -> list[int]:
    """
    Applica NMS per eliminare bounding box sovrapposti.

    Args:
        boxes: [N, 4] array di box [x1, y1, x2, y2]
        scores: [N] array di confidence scores
        iou_threshold: soglia IoU per considerare due box duplicati
        score_threshold: filtra box sotto questa confidenza

    Returns:
        Lista di indici dei box selezionati
    """
    # Filtra box sotto la soglia di confidenza
    valid_mask = scores >= score_threshold
    boxes = boxes[valid_mask]
    scores = scores[valid_mask]
    indices = np.where(valid_mask)[0]

    # Ordina per score decrescente
    order = np.argsort(scores)[::-1]
    selected_indices = []

    while len(order) > 0:
        # Prendi il box con score più alto
        best_idx = order[0]
        selected_indices.append(indices[best_idx])
        order = order[1:]

        if len(order) == 0:
            break

        # Calcola IoU del box selezionato con tutti i restanti
        ious = np.array([
            compute_iou(boxes[best_idx], boxes[i])
            for i in order
        ])

        # Mantieni solo i box con IoU basso (non sovrapposti)
        order = order[ious < iou_threshold]

    return selected_indices

# Test
boxes = np.array([
    [100, 100, 300, 300],  # box principale
    [110, 105, 310, 305],  # quasi identico - da eliminare
    [500, 200, 700, 400],  # box diverso - da mantenere
])
scores = np.array([0.92, 0.88, 0.75])

kept = non_maximum_suppression(boxes, scores, iou_threshold=0.5)
print(f"Box mantenuti: {kept}")  # [0, 2] - elimina il duplicato

Input Immagine (640x640x3)
        |
        v
+------------------+
|    BACKBONE      |  Estrazione feature multi-scala
|  (CSPDarkNet /   |  Output: feature maps a scale diverse
|   EfficientRep)  |  P3: 80x80  (oggetti piccoli)
|                  |  P4: 40x40  (oggetti medi)
|                  |  P5: 20x20  (oggetti grandi)
+------------------+
        |
        v
+------------------+
|      NECK        |  Aggregazione multi-scala
|   (PANet / BiFPN)|  Feature Pyramid Network
|                  |  Fonde informazioni semantiche (deep)
|                  |  con informazioni spaziali (shallow)
+------------------+
        |
        v
+------------------+
|      HEAD        |  Predizioni finali
|  (Decoupled      |  Per ogni cella della griglia:
|   Detection)     |  - Box regression: [x, y, w, h]
|                  |  - Objectness: p(oggetto)
|                  |  - Classification: [p_c1, ..., p_cN]
+------------------+
        |
        v
Output: [batch, num_predictions, 4 + 1 + num_classes]
        # Per YOLOv8 nano su 640x640: 8400 predizioni totali
        # (80x80 + 40x40 + 20x20 = 6400 + 1600 + 400 = 8400)

dataset/
├── images/
│   ├── train/          # Immagini di training
│   │   ├── img001.jpg
│   │   ├── img002.jpg
│   │   └── ...
│   ├── val/            # Immagini di validazione (20%)
│   │   └── ...
│   └── test/           # Immagini di test (opzionale)
│       └── ...
├── labels/
│   ├── train/          # Annotazioni training
│   │   ├── img001.txt  # Una riga per oggetto
│   │   ├── img002.txt
│   │   └── ...
│   ├── val/
│   │   └── ...
│   └── test/
│       └── ...
└── dataset.yaml        # Configurazione dataset

# Contenuto di img001.txt (due oggetti):
# class_id x_c  y_c   w    h
# 0        0.45 0.60 0.30 0.40   # gatto al centro
# 1        0.85 0.25 0.20 0.35   # cane in alto a destra

# Contenuto di dataset.yaml:
# path: /path/to/dataset
# train: images/train
# val: images/val
# test: images/test  # opzionale
# nc: 2              # numero classi
# names: ['gatto', 'cane']

import json
import os
from pathlib import Path

def convert_coco_to_yolo(coco_json_path: str, output_dir: str) -> None:
    """
    Converte annotazioni COCO JSON in formato YOLO TXT.
    Utile per dataset pubblici (COCO, Open Images, etc.)
    """
    with open(coco_json_path) as f:
        coco_data = json.load(f)

    # Mappa image_id -> info immagine
    images = {img['id']: img for img in coco_data['images']}

    # Mappa category_id -> indice YOLO (0-based)
    cat_id_to_yolo = {
        cat['id']: i
        for i, cat in enumerate(coco_data['categories'])
    }

    # Raggruppa annotazioni per immagine
    annotations_by_image: dict[int, list] = {}
    for ann in coco_data['annotations']:
        img_id = ann['image_id']
        if img_id not in annotations_by_image:
            annotations_by_image[img_id] = []
        annotations_by_image[img_id].append(ann)

    output_path = Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)

    for img_id, anns in annotations_by_image.items():
        img_info = images[img_id]
        img_w = img_info['width']
        img_h = img_info['height']
        img_name = Path(img_info['file_name']).stem

        txt_lines = []
        for ann in anns:
            # COCO: [x_top_left, y_top_left, width, height]
            x_tl, y_tl, w, h = ann['bbox']

            # Converti a formato YOLO (normalizzato)
            x_c = (x_tl + w / 2) / img_w
            y_c = (y_tl + h / 2) / img_h
            w_n = w / img_w
            h_n = h / img_h

            class_idx = cat_id_to_yolo[ann['category_id']]
            txt_lines.append(f"{class_idx} {x_c:.6f} {y_c:.6f} {w_n:.6f} {h_n:.6f}")

        with open(output_path / f"{img_name}.txt", 'w') as f:
            f.write('\n'.join(txt_lines))

    print(f"Convertite {len(annotations_by_image)} immagini in {output_dir}")

from ultralytics import YOLO
import yaml
from pathlib import Path

# ---- Configurazione dataset ----
dataset_config = {
    'path': '/path/to/dataset',
    'train': 'images/train',
    'val': 'images/val',
    'nc': 80,  # numero classi (es. COCO)
    'names': ['person', 'bicycle', 'car', '...']  # lista classi
}

config_path = 'dataset.yaml'
with open(config_path, 'w') as f:
    yaml.dump(dataset_config, f, allow_unicode=True)

# ---- Scelta del modello ----
# Varianti disponibili (velocità vs accuratezza):
# yolov8n.pt  - nano   (fastest, lowest accuracy)
# yolov8s.pt  - small
# yolov8m.pt  - medium
# yolov8l.pt  - large
# yolov8x.pt  - xlarge (slowest, highest accuracy)

model = YOLO('yolov8m.pt')  # carica pesi pre-addestrati su COCO

# ---- Training ----
results = model.train(
    data=config_path,
    epochs=100,
    imgsz=640,           # dimensione input immagine
    batch=16,            # batch size (riduci se OOM)
    workers=8,           # CPU workers per data loading
    device='0',          # GPU index ('cpu' per CPU-only)

    # Ottimizzazione
    optimizer='AdamW',
    lr0=0.001,           # learning rate iniziale
    lrf=0.01,            # lr finale = lr0 * lrf
    momentum=0.937,
    weight_decay=0.0005,
    warmup_epochs=3,
    cos_lr=True,         # cosine LR schedule

    # Augmentation
    mosaic=1.0,          # mosaic augmentation (YOLOv5+ feature)
    mixup=0.1,           # mixup augmentation
    copy_paste=0.1,      # copy-paste augmentation
    degrees=10.0,        # rotation
    translate=0.1,       # translation
    scale=0.5,           # scaling
    fliplr=0.5,          # horizontal flip
    flipud=0.0,          # vertical flip

    # Regularization
    dropout=0.0,
    label_smoothing=0.0,

    # Checkpointing
    save=True,
    save_period=10,      # salva ogni N epoche
    project='runs/train',
    name='yolov8m_custom',

    # Early stopping
    patience=50,         # stop se mAP non migliora per N epoche

    # Logging
    plots=True,
    verbose=True
)

print(f"Best mAP@0.5: {results.results_dict['metrics/mAP50(B)']:.3f}")
print(f"Best model salvato in: runs/train/yolov8m_custom/weights/best.pt")

from ultralytics import YOLO
import cv2
import numpy as np
from pathlib import Path

class YOLOInferenceEngine:
    """
    Engine di inference per YOLOv8/YOLO26 con supporto
    per immagini, video e stream live.
    """

    def __init__(
        self,
        model_path: str = 'yolov8m.pt',
        conf_threshold: float = 0.25,
        iou_threshold: float = 0.45,
        device: str = 'auto'
    ):
        self.model = YOLO(model_path)
        self.conf = conf_threshold
        self.iou = iou_threshold

        # Palette colori per classi
        np.random.seed(42)
        self.colors = np.random.randint(0, 255, size=(100, 3), dtype=np.uint8)

    def predict_image(self, image_path: str, save_path: str | None = None) -> list[dict]:
        """Inference su singola immagine con visualizzazione opzionale."""
        results = self.model.predict(
            source=image_path,
            conf=self.conf,
            iou=self.iou,
            verbose=False
        )

        detections = []
        for r in results:
            for box in r.boxes:
                det = {
                    'bbox': box.xyxy[0].tolist(),     # [x1, y1, x2, y2]
                    'confidence': float(box.conf[0]),
                    'class_id': int(box.cls[0]),
                    'class_name': r.names[int(box.cls[0])]
                }
                detections.append(det)

            if save_path:
                annotated = r.plot()
                cv2.imwrite(save_path, annotated)

        return detections

    def process_video(self, video_path: str, output_path: str | None = None) -> None:
        """
        Processa un video file con detection frame per frame.
        Mostra FPS e statistiche in tempo reale.
        """
        cap = cv2.VideoCapture(video_path)

        if output_path:
            fps = cap.get(cv2.CAP_PROP_FPS)
            w   = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
            h   = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
            fourcc = cv2.VideoWriter_fourcc(*'mp4v')
            writer = cv2.VideoWriter(output_path, fourcc, fps, (w, h))

        frame_count = 0
        import time

        while cap.isOpened():
            ret, frame = cap.read()
            if not ret:
                break

            start = time.perf_counter()
            results = self.model(frame, conf=self.conf, iou=self.iou, verbose=False)
            elapsed = time.perf_counter() - start

            # Annota il frame
            annotated = results[0].plot()

            # Overlay FPS
            fps_text = f"FPS: {1/elapsed:.1f}"
            cv2.putText(annotated, fps_text, (10, 30),
                       cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)

            if output_path:
                writer.write(annotated)

            cv2.imshow('YOLO Detection', annotated)
            if cv2.waitKey(1) & 0xFF == ord('q'):
                break

            frame_count += 1

        cap.release()
        if output_path:
            writer.release()
        cv2.destroyAllWindows()
        print(f"Processati {frame_count} frame")

    def run_webcam(self, camera_id: int = 0) -> None:
        """Detection live da webcam."""
        print("Avvio webcam detection. Premi 'q' per uscire.")
        self.process_video(camera_id, output_path=None)


# Uso pratico
engine = YOLOInferenceEngine('yolov8m.pt', conf_threshold=0.4)

# Inference su immagine
dets = engine.predict_image('test.jpg', save_path='result.jpg')
for d in dets:
    print(f"{d['class_name']}: {d['confidence']:.2f} @ {[int(c) for c in d['bbox']]}")

# Inference su video
engine.process_video('traffic.mp4', 'traffic_detected.mp4')

# Webcam live
engine.run_webcam(camera_id=0)

from ultralytics import YOLO

model = YOLO('runs/train/yolov8m_custom/weights/best.pt')

# Export ONNX (più portabile)
model.export(format='onnx', imgsz=640, opset=17, simplify=True)

# Export TensorRT (massima velocità su GPU NVIDIA)
# Richiede: NVIDIA GPU + CUDA + TensorRT installato
model.export(
    format='engine',      # TensorRT Engine
    imgsz=640,
    half=True,            # FP16 per maggiore velocità
    workspace=4,          # GB di workspace GPU
    batch=1,              # batch size fisso per TensorRT
    device=0
)

# Export OpenVINO (CPU Intel ottimizzata)
model.export(format='openvino', imgsz=640, half=False)

# Export TFLite (mobile/edge)
model.export(format='tflite', imgsz=640, int8=False)

# Carica e usa il modello esportato
# ONNX Runtime (CPU/GPU, no PyTorch)
model_onnx = YOLO('best.onnx')
results = model_onnx.predict('image.jpg', conf=0.25)

# TensorRT (GPU NVIDIA)
model_trt = YOLO('best.engine')
results = model_trt.predict('image.jpg', conf=0.25)
print(f"Inferenza TensorRT completata: {len(results[0].boxes)} oggetti rilevati")

# YOLO26 richiede ultralytics >= 8.3.0 (gennaio 2026)
# pip install ultralytics --upgrade

from ultralytics import YOLO

# Carica modello YOLO26
model = YOLO('yolo26n.pt')   # nano - massima velocità
model = YOLO('yolo26s.pt')   # small
model = YOLO('yolo26m.pt')   # medium - bilanciato
model = YOLO('yolo26l.pt')   # large
model = YOLO('yolo26x.pt')   # xlarge - massima accuratezza

# Training su dataset personalizzato (stessa API di YOLOv8)
results = model.train(
    data='dataset.yaml',
    epochs=100,
    imgsz=640,
    batch=16,
    device='0',
    # YOLO26 aggiunge: self-calibrating augmentation
    auto_augment='yolo26',   # NEW: augmentation policy ottimizzata
    # Dynamic NMS threshold scheduling
    nms_schedule=True        # NEW: NMS adattivo durante training
)

# Inference (identica a YOLOv8)
results = model.predict('image.jpg', conf=0.25, iou=0.45)

# Validazione sul validation set
metrics = model.val(data='dataset.yaml')
print(f"mAP@0.5: {metrics.box.map50:.3f}")
print(f"mAP@0.5:0.95: {metrics.box.map:.3f}")

from ultralytics import YOLO
from ultralytics.utils.callbacks import on_train_epoch_end
import torch
import yaml
from pathlib import Path

# ---- Dataset YAML con class weights per bilanciamento ----
dataset_config = {
    'path': './datasets/custom',
    'train': 'images/train',
    'val': 'images/val',
    'nc': 5,
    'names': ['person', 'car', 'bicycle', 'dog', 'cat'],

    # Pesi per classe: compensa sbilanciamento (persona 5x più frequente)
    'cls_weights': [1.0, 1.0, 2.0, 2.5, 2.5]
}

with open('dataset.yaml', 'w') as f:
    yaml.dump(dataset_config, f)


# ---- Training con hyperparameters ottimizzati ----
model = YOLO('yolo26m.pt')  # Pre-trained YOLO26 medium

results = model.train(
    data='dataset.yaml',
    epochs=300,
    imgsz=640,
    batch=16,
    device='0',

    # ---- Optimizer ----
    optimizer='AdamW',          # AdamW migliore di SGD per fine-tuning
    lr0=0.001,                  # Learning rate iniziale
    lrf=0.01,                   # LR finale = lr0 * lrf (1/100 dell'iniziale)
    momentum=0.937,             # Momentum per SGD (ignorato per AdamW)
    weight_decay=0.0005,        # L2 regularization

    # ---- Learning Rate Scheduling ----
    cos_lr=True,                # Cosine annealing invece di step decay
    warmup_epochs=3,            # Warmup lineare da lr0/10 a lr0
    warmup_momentum=0.8,        # Momentum durante warmup

    # ---- Augmentation ----
    auto_augment='yolo26',      # Policy self-calibrating di YOLO26
    mosaic=1.0,                 # Mosaic augmentation (0.0 per disabilitare)
    mixup=0.2,                  # MixUp probability
    copy_paste=0.1,             # Copy-Paste probability
    degrees=10.0,               # Rotazione max +/- gradi
    translate=0.2,              # Traslazione max (fraction of image size)
    scale=0.9,                  # Scale augmentation range [1-scale, 1+scale]
    flipud=0.0,                 # Flip verticale (0 se non sensato)
    fliplr=0.5,                 # Flip orizzontale

    # ---- Loss weights ----
    box=7.5,                    # Peso perdita bounding box
    cls=0.5,                    # Peso perdita classificazione
    dfl=1.5,                    # Peso Distribution Focal Loss

    # ---- Early stopping e checkpointing ----
    patience=50,                # Ferma se non migliora per 50 epoch
    save_period=25,             # Salva checkpoint ogni 25 epoch
    val=True,                   # Valida ad ogni epoch

    # ---- Output ----
    project='runs/train',
    name='yolo26m_custom',
    exist_ok=True,
    plots=True,                 # Genera grafici training
    verbose=True
)

print(f"Best mAP@0.5: {results.results_dict['metrics/mAP50(B)']:.4f}")
print(f"Best mAP@0.5:0.95: {results.results_dict['metrics/mAP50-95(B)']:.4f}")


# ---- Hyperparameter Auto-Tuning con Ray Tune ----
def tune_hyperparameters(model_path: str, data_path: str) -> None:
    """
    Usa Ray Tune per ottimizzare automaticamente gli hyperparameter.
    Richiede: pip install ray[tune]
    Esplora learning rate, augmentation intensity, loss weights.
    """
    model = YOLO(model_path)

    # Spazio di ricerca degli hyperparameters
    space = {
        'lr0': (1e-5, 1e-1),           # log-uniform
        'lrf': (0.01, 1.0),
        'momentum': (0.6, 0.98),
        'weight_decay': (0.0, 0.001),
        'warmup_epochs': (0, 5),
        'warmup_momentum': (0.0, 0.95),
        'box': (0.02, 0.2),
        'cls': (0.2, 4.0),
        'mosaic': (0.0, 1.0),
        'mixup': (0.0, 0.5),
        'copy_paste': (0.0, 0.3),
    }

    result = model.tune(
        data=data_path,
        space=space,
        epochs=50,           # Epoch per ogni trial
        iterations=100,      # Numero di configurazioni da provare
        optimizer='AdamW',
        plots=True,
        save=True
    )

    print("Migliori hyperparameter trovati:")
    for k, v in result.items():
        print(f"  {k}: {v}")


# ---- Custom callback per monitoraggio avanzato ----
class YOLOTrainingMonitor:
    """
    Callback per monitoring avanzato durante il training.
    Traccia metriche custom e genera alert se necessario.
    """

    def __init__(self, alert_threshold: float = 0.3):
        self.best_map = 0.0
        self.no_improve_count = 0
        self.alert_threshold = alert_threshold
        self.history = []

    def on_train_epoch_end(self, trainer) -> None:
        """Chiamato alla fine di ogni epoch di training."""
        metrics = trainer.metrics
        current_map = metrics.get('metrics/mAP50(B)', 0.0)

        self.history.append({
            'epoch': trainer.epoch,
            'map50': current_map,
            'loss': trainer.loss.item()
        })

        if current_map > self.best_map:
            self.best_map = current_map
            self.no_improve_count = 0
        else:
            self.no_improve_count += 1

        # Alert se il modello non migliora dopo 30 epoch
        if self.no_improve_count == 30:
            print(f"[WARN] Nessun miglioramento per 30 epoch. Best mAP: {self.best_map:.4f}")


# Utilizzo del monitor
model = YOLO('yolo26m.pt')
monitor = YOLOTrainingMonitor()
model.add_callback('on_train_epoch_end', monitor.on_train_epoch_end)
# model.train(data='dataset.yaml', epochs=200)