Computer Vision per Controllo Qualità Alimentare con PyTorch YOLO

# augmentation_food.py
# Configurazione augmentation specifica per food quality inspection

import albumentations as A
import cv2
import numpy as np

def build_food_augmentation_pipeline(image_size: int = 640) -> A.Compose:
    """
    Pipeline di augmentation per dataset di difetti alimentari.
    Simula variazioni reali di illuminazione, orientamento e condizioni di linea.
    """
    return A.Compose([
        # Geometria: prodotti alimentari arrivano in orientamenti casuali
        A.RandomRotate90(p=0.5),
        A.HorizontalFlip(p=0.5),
        A.VerticalFlip(p=0.3),
        A.ShiftScaleRotate(
            shift_limit=0.1,
            scale_limit=0.2,
            rotate_limit=45,
            border_mode=cv2.BORDER_REFLECT,
            p=0.7
        ),

        # Illuminazione: variazioni reali in linea (vapore, sporco sulle lenti)
        A.OneOf([
            A.RandomBrightnessContrast(
                brightness_limit=0.3,
                contrast_limit=0.3,
                p=1.0
            ),
            A.RandomGamma(gamma_limit=(70, 130), p=1.0),
            A.CLAHE(clip_limit=4.0, tile_grid_size=(8, 8), p=1.0),
        ], p=0.8),

        # Colore: variazioni stagionali e di maturazione
        A.HueSaturationValue(
            hue_shift_limit=15,     # Piccolo: non cambiare il colore del prodotto
            sat_shift_limit=30,
            val_shift_limit=20,
            p=0.5
        ),

        # Rumore: sensore camera industriale, interferenze EMI
        A.OneOf([
            A.GaussNoise(var_limit=(10.0, 50.0), p=1.0),
            A.ISONoise(color_shift=(0.01, 0.05), intensity=(0.1, 0.5), p=1.0),
            A.MultiplicativeNoise(multiplier=(0.9, 1.1), p=1.0),
        ], p=0.4),

        # Blur: motion blur da velocità nastro, defocus per DOF limitata
        A.OneOf([
            A.MotionBlur(blur_limit=7, p=1.0),    # Più realistico per nastro
            A.GaussianBlur(blur_limit=(3, 7), p=1.0),
        ], p=0.3),

        # Riflessioni: superfici cerate, glazze, film di acqua
        A.RandomSunFlare(
            flare_roi=(0.0, 0.0, 1.0, 0.5),
            num_flare_circles_lower=1,
            num_flare_circles_upper=3,
            src_radius=100,
            p=0.1
        ),

        # Ritaglio e padding finale
        A.PadIfNeeded(
            min_height=image_size,
            min_width=image_size,
            border_mode=cv2.BORDER_REFLECT
        ),
        A.Resize(image_size, image_size),

        # Normalizzazione finale
        A.Normalize(
            mean=[0.485, 0.456, 0.406],
            std=[0.229, 0.224, 0.225]
        ),
    ], bbox_params=A.BboxParams(
        format='yolo',
        label_fields=['class_labels'],
        min_area=100,           # Ignora bbox troppo piccole dopo crop
        min_visibility=0.3      # Ignora bbox con meno del 30% visibile
    ))


def augment_rare_class(
    image: np.ndarray,
    bboxes: list,
    class_labels: list,
    n_augmentations: int = 10
) -> list:
    """
    Over-sampling per classi rare come foreign_object.
    Genera n varianti augmentate di un singolo campione.
    """
    pipeline = build_food_augmentation_pipeline()
    augmented_samples = []

    for _ in range(n_augmentations):
        result = pipeline(
            image=image,
            bboxes=bboxes,
            class_labels=class_labels
        )
        augmented_samples.append({
            'image': result['image'],
            'bboxes': result['bboxes'],
            'class_labels': result['class_labels']
        })

    return augmented_samples

# Requisiti: Python 3.11+, CUDA 12.1+, NVIDIA GPU con 8GB+ VRAM
# Consigliato: RTX 3080/4080 per training locale, A100 per training veloce

# 1. Installazione dipendenze
# pip install ultralytics==8.3.0 albumentations roboflow torch torchvision

# 2. Struttura directory dataset (formato YOLO)
# dataset/
# ├── images/
# │   ├── train/     (70% campioni)
# │   ├── val/       (20% campioni)
# │   └── test/      (10% campioni)
# ├── labels/
# │   ├── train/     (file .txt corrispondenti)
# │   ├── val/
# │   └── test/
# └── data.yaml      (configurazione dataset)

# data.yaml - Configurazione dataset
DATA_YAML_CONTENT = """
path: /data/food_quality_dataset
train: images/train
val: images/val
test: images/test

nc: 10  # Numero di classi

names:
  0: ok
  1: mold
  2: bruise
  3: burn
  4: crack
  5: foreign_object
  6: rot
  7: insect_damage
  8: size_defect
  9: color_defect
"""

import yaml
with open('/data/food_quality_dataset/data.yaml', 'w') as f:
    f.write(DATA_YAML_CONTENT)

# dataset_preparation.py
# Scarica dataset da Roboflow Universe o usa dataset locale

from roboflow import Roboflow
import os

def download_food_dataset(api_key: str, workspace: str, project: str, version: int) -> str:
    """
    Scarica dataset da Roboflow e prepara per training YOLO11.
    """
    rf = Roboflow(api_key=api_key)
    proj = rf.workspace(workspace).project(project)
    dataset = proj.version(version).download("yolov8")  # Formato YOLO11 compatibile

    dataset_path = dataset.location
    print(f"Dataset scaricato in: {dataset_path}")
    print(f"Training images: {len(os.listdir(os.path.join(dataset_path, 'train', 'images')))}")
    print(f"Validation images: {len(os.listdir(os.path.join(dataset_path, 'valid', 'images')))}")

    return dataset_path


def analyze_class_distribution(dataset_path: str) -> dict:
    """
    Analizza la distribuzione delle classi nel dataset.
    Identifica classi sbilanciate che richiedono over-sampling.
    """
    import glob
    from collections import Counter

    class_counts = Counter()
    label_files = glob.glob(os.path.join(dataset_path, 'train', 'labels', '*.txt'))

    for label_file in label_files:
        with open(label_file, 'r') as f:
            for line in f:
                class_id = int(line.split()[0])
                class_counts[class_id] += 1

    total = sum(class_counts.values())
    distribution = {
        class_id: {
            'count': count,
            'percentage': round(count / total * 100, 2),
            'needs_oversampling': count < total / len(class_counts) * 0.3  # < 30% della media
        }
        for class_id, count in sorted(class_counts.items())
    }

    return distribution


# Analisi distribuzione per identificare classi rare
if __name__ == "__main__":
    dataset_path = "/data/food_quality_dataset"
    distribution = analyze_class_distribution(dataset_path)

    print("\nDistribuzione classi nel dataset:")
    class_names = ['ok', 'mold', 'bruise', 'burn', 'crack',
                   'foreign_object', 'rot', 'insect_damage', 'size_defect', 'color_defect']

    for class_id, info in distribution.items():
        name = class_names[class_id] if class_id < len(class_names) else f"class_{class_id}"
        warning = " *** RICHIEDE OVERSAMPLING ***" if info['needs_oversampling'] else ""
        print(f"  {name}: {info['count']} campioni ({info['percentage']}%){warning}")

# train_yolo11_food.py
# Training pipeline completa per food quality inspection

from ultralytics import YOLO
import torch
import yaml
import os
from pathlib import Path

def train_food_quality_model(
    dataset_yaml: str,
    output_dir: str = "./runs/food_quality",
    epochs: int = 150,
    batch_size: int = 16,
    image_size: int = 640,
    model_variant: str = "yolo11m.pt"  # n/s/m/l/x
) -> dict:
    """
    Training YOLO11 per food quality inspection con ottimizzazioni specifiche.

    Args:
        dataset_yaml: Path al file data.yaml del dataset
        output_dir: Directory per salvare i risultati
        epochs: Numero di epoche (150 e un buon punto di partenza)
        batch_size: Dimensione batch (16 per 8GB VRAM, 32 per 16GB+)
        image_size: Dimensione immagine (640 standard, 1280 per difetti piccoli)
        model_variant: Variante YOLO11 da usare come punto di partenza

    Returns:
        dict con metriche finali del training
    """

    # Verifica GPU disponibile
    device = 'cuda' if torch.cuda.is_available() else 'cpu'
    if device == 'cpu':
        print("ATTENZIONE: Training su CPU, sarà molto lento. Usa una GPU.")

    print(f"Device: {device}")
    if device == 'cuda':
        print(f"GPU: {torch.cuda.get_device_name(0)}")
        print(f"VRAM: {torch.cuda.get_device_properties(0).total_memory / 1e9:.1f} GB")

    # Carica modello pre-addestrato su COCO (transfer learning)
    model = YOLO(model_variant)

    # Configurazione training
    training_args = {
        'data': dataset_yaml,
        'epochs': epochs,
        'batch': batch_size,
        'imgsz': image_size,
        'device': device,
        'project': output_dir,
        'name': 'food_quality_v1',

        # Ottimizzatore: AdamW e ottimo per fine-tuning
        'optimizer': 'AdamW',
        'lr0': 0.001,           # Learning rate iniziale
        'lrf': 0.01,            # Learning rate finale = lr0 * lrf
        'momentum': 0.937,
        'weight_decay': 0.0005,

        # Learning rate scheduling: cosine annealing
        'cos_lr': True,
        'warmup_epochs': 5,     # 5 epoche di warmup per stabilizzare

        # Augmentation integrata (albumentations)
        'hsv_h': 0.015,         # Hue shift (piccolo per preservare colore prodotto)
        'hsv_s': 0.7,           # Saturation
        'hsv_v': 0.4,           # Value (luminosita)
        'degrees': 30.0,        # Rotazione
        'translate': 0.1,       # Traslazione
        'scale': 0.5,           # Scaling
        'shear': 5.0,
        'perspective': 0.0001,
        'flipud': 0.3,          # Flip verticale (utile per frutta)
        'fliplr': 0.5,          # Flip orizzontale
        'mosaic': 1.0,          # Mosaic augmentation (4 immagini)
        'mixup': 0.1,           # MixUp augmentation
        'copy_paste': 0.1,      # Copy-paste per classi rare

        # Loss weights: aumenta peso classificazione per difetti rari
        'cls': 1.5,             # Classification loss weight (default 0.5)
        'box': 7.5,             # Box regression loss weight
        'dfl': 1.5,             # Distribution Focal Loss weight

        # Valutazione e salvataggio
        'val': True,
        'save': True,
        'save_period': 10,      # Salva checkpoint ogni 10 epoche
        'patience': 30,         # Early stopping se no miglioramento per 30 epoche

        # Performances
        'workers': 8,           # DataLoader workers
        'cache': True,          # Cache dataset in RAM per velocità
        'amp': True,            # Automatic Mixed Precision (FP16)

        # Metriche
        'plots': True,          # Genera grafici di training
        'verbose': True,
    }

    # Avvia training
    print(f"\nAvvio training YOLO11 per food quality inspection")
    print(f"Epoche: {epochs}, Batch: {batch_size}, Immagini: {image_size}x{image_size}")

    results = model.train(**training_args)

    # Estrai metriche finali
    metrics = {
        'mAP50': float(results.results_dict.get('metrics/mAP50(B)', 0)),
        'mAP50_95': float(results.results_dict.get('metrics/mAP50-95(B)', 0)),
        'precision': float(results.results_dict.get('metrics/precision(B)', 0)),
        'recall': float(results.results_dict.get('metrics/recall(B)', 0)),
        'best_model_path': str(Path(output_dir) / 'food_quality_v1' / 'weights' / 'best.pt')
    }

    print(f"\nTraining completato!")
    print(f"mAP@50: {metrics['mAP50']:.4f}")
    print(f"mAP@50-95: {metrics['mAP50_95']:.4f}")
    print(f"Precision: {metrics['precision']:.4f}")
    print(f"Recall: {metrics['recall']:.4f}")
    print(f"Modello salvato in: {metrics['best_model_path']}")

    return metrics


if __name__ == "__main__":
    metrics = train_food_quality_model(
        dataset_yaml="/data/food_quality_dataset/data.yaml",
        output_dir="./runs/food_quality",
        epochs=150,
        batch_size=16,
        image_size=640,
        model_variant="yolo11m.pt"
    )

# hyperparameter_tuning.py
# Ricerca automatica degli hyperparametri ottimali con Ray Tune

from ultralytics import YOLO
from ray import tune
from ray.tune.schedulers import ASHAScheduler
import torch

def objective(config: dict) -> dict:
    """Funzione obiettivo per Ray Tune."""
    model = YOLO("yolo11m.pt")

    results = model.train(
        data="/data/food_quality_dataset/data.yaml",
        epochs=50,          # Epoche ridotte per tuning rapido
        batch=config['batch'],
        lr0=config['lr0'],
        lrf=config['lrf'],
        momentum=config['momentum'],
        weight_decay=config['weight_decay'],
        cls=config['cls'],
        hsv_s=config['hsv_s'],
        mixup=config['mixup'],
        amp=True,
        verbose=False,
        plots=False,
    )

    mAP50 = results.results_dict.get('metrics/mAP50(B)', 0)
    return {"mAP50": mAP50}


def run_hyperparameter_search(n_trials: int = 20) -> dict:
    """Esegue ricerca hyperparametri con ASHA scheduler."""

    search_space = {
        'batch': tune.choice([8, 16, 32]),
        'lr0': tune.loguniform(1e-4, 1e-2),
        'lrf': tune.uniform(0.001, 0.1),
        'momentum': tune.uniform(0.85, 0.98),
        'weight_decay': tune.loguniform(1e-5, 1e-3),
        'cls': tune.uniform(0.5, 2.0),
        'hsv_s': tune.uniform(0.4, 0.9),
        'mixup': tune.uniform(0.0, 0.3),
    }

    scheduler = ASHAScheduler(
        max_t=50,
        grace_period=10,
        reduction_factor=2
    )

    analysis = tune.run(
        objective,
        config=search_space,
        num_samples=n_trials,
        scheduler=scheduler,
        metric="mAP50",
        mode="max",
        resources_per_trial={"cpu": 4, "gpu": 1},
    )

    best_config = analysis.best_config
    print(f"Migliori hyperparametri trovati:")
    for key, value in best_config.items():
        print(f"  {key}: {value}")

    return best_config

# evaluate_model.py
# Valutazione completa del modello con metriche per food QC

from ultralytics import YOLO
import numpy as np
from sklearn.metrics import confusion_matrix
import matplotlib.pyplot as plt
import seaborn as sns
from pathlib import Path

def evaluate_food_model(
    model_path: str,
    test_dataset_yaml: str,
    class_names: list,
    confidence_thresholds: dict  # Soglie per classe
) -> dict:
    """
    Valutazione completa con metriche specifiche per food quality inspection.

    Args:
        model_path: Path al modello addestrato (best.pt)
        test_dataset_yaml: Path al data.yaml del test set
        class_names: Nomi delle classi
        confidence_thresholds: Soglie confidence per classe {"mold": 0.35, "ok": 0.6}
    """
    model = YOLO(model_path)

    # Validazione standard YOLO
    results = model.val(
        data=test_dataset_yaml,
        split='test',
        imgsz=640,
        conf=0.001,     # Basso per calcolare la curva completa
        iou=0.6,
        plots=True,
        verbose=False,
    )

    # Metriche per classe
    class_metrics = {}
    for i, name in enumerate(class_names):
        class_metrics[name] = {
            'precision': float(results.box.p[i]) if i < len(results.box.p) else 0,
            'recall': float(results.box.r[i]) if i < len(results.box.r) else 0,
            'mAP50': float(results.box.ap50[i]) if i < len(results.box.ap50) else 0,
            'mAP50_95': float(results.box.ap[i]) if i < len(results.box.ap) else 0,
        }

    # Soglie di accettabilita per l'industria alimentare
    acceptance_thresholds = {
        'mold': {'recall_min': 0.98, 'precision_min': 0.85},
        'foreign_object': {'recall_min': 0.995, 'precision_min': 0.80},
        'rot': {'recall_min': 0.97, 'precision_min': 0.87},
        'bruise': {'recall_min': 0.90, 'precision_min': 0.85},
        'burn': {'recall_min': 0.88, 'precision_min': 0.83},
        'crack': {'recall_min': 0.93, 'precision_min': 0.85},
        'insect_damage': {'recall_min': 0.92, 'precision_min': 0.84},
        'size_defect': {'recall_min': 0.85, 'precision_min': 0.88},
        'color_defect': {'recall_min': 0.85, 'precision_min': 0.88},
        'ok': {'recall_min': 0.95, 'precision_min': 0.92},
    }

    # Verifica accettabilita
    go_nogo_results = {}
    for class_name, metrics in class_metrics.items():
        thresholds = acceptance_thresholds.get(class_name, {})
        recall_ok = metrics['recall'] >= thresholds.get('recall_min', 0.0)
        precision_ok = metrics['precision'] >= thresholds.get('precision_min', 0.0)

        go_nogo_results[class_name] = {
            'go': recall_ok and precision_ok,
            'recall_ok': recall_ok,
            'precision_ok': precision_ok,
            'recall': metrics['recall'],
            'precision': metrics['precision'],
        }

    # Report finale
    print("\n=== VALUTAZIONE FOOD QUALITY MODEL ===\n")
    print(f"mAP@50 globale:    {float(results.box.map50):.4f}")
    print(f"mAP@50-95 globale: {float(results.box.map):.4f}")

    print("\nRisultati per classe:")
    print(f"{'Classe':-20} {'Recall':>8} {'Precision':>10} {'mAP50':>8} {'GO/NO-GO':>10}")
    print("-" * 60)

    for class_name, gng in go_nogo_results.items():
        status = "GO ✓" if gng['go'] else "NO-GO ✗"
        print(f"{class_name:-20} {gng['recall']:8.4f} {gng['precision']:10.4f} "
              f"{class_metrics[class_name]['mAP50']:8.4f} {status:>10}")

    overall_go = all(gng['go'] for gng in go_nogo_results.values())
    print(f"\nVALUTAZIONE FINALE: {'SISTEMA APPROVATO PER DEPLOY' if overall_go else 'NON APPROVATO - RICHIEDE MIGLIORAMENTI'}")

    return {
        'class_metrics': class_metrics,
        'go_nogo': go_nogo_results,
        'overall_go': overall_go,
        'global_mAP50': float(results.box.map50),
    }


# Esecuzione valutazione
if __name__ == "__main__":
    class_names = ['ok', 'mold', 'bruise', 'burn', 'crack',
                   'foreign_object', 'rot', 'insect_damage', 'size_defect', 'color_defect']

    confidence_thresholds = {
        'mold': 0.35,
        'foreign_object': 0.30,
        'rot': 0.40,
        'bruise': 0.50,
        'burn': 0.50,
        'crack': 0.45,
        'insect_damage': 0.45,
        'size_defect': 0.60,
        'color_defect': 0.60,
        'ok': 0.60,
    }

    results = evaluate_food_model(
        model_path="./runs/food_quality/food_quality_v1/weights/best.pt",
        test_dataset_yaml="/data/food_quality_dataset/data.yaml",
        class_names=class_names,
        confidence_thresholds=confidence_thresholds
    )

# camera_gige_acquisition.py
# Acquisizione immagini da camera GigE Vision con trigger hardware

import pypylon.pylon as pylon
import numpy as np
import cv2
import threading
import queue
import time
from dataclasses import dataclass
from typing import Optional

@dataclass
class AcquiredFrame:
    """Frame acquisito dalla camera con metadati."""
    image: np.ndarray
    timestamp_ns: int
    frame_id: int
    trigger_counter: int


class GigEVisionCamera:
    """
    Wrapper per camera GigE Vision via Basler pylibpylon.
    Gestisce trigger hardware, acquisizione e buffer.
    """

    def __init__(
        self,
        device_index: int = 0,
        trigger_mode: str = "Line1",  # Trigger hardware su Line1
        exposure_us: int = 500,        # 500 microsec: congela il moto a 2m/s
        gain_db: float = 6.0,
        buffer_count: int = 10,
    ) -> None:
        self._device_index = device_index
        self._trigger_mode = trigger_mode
        self._exposure_us = exposure_us
        self._gain_db = gain_db
        self._buffer_count = buffer_count
        self._camera: Optional[pylon.InstantCamera] = None
        self._frame_queue: queue.Queue = queue.Queue(maxsize=100)
        self._acquiring = False
        self._frame_counter = 0

    def connect(self) -> None:
        """Connette e configura la camera GigE Vision."""
        transport_factory = pylon.TlFactory.GetInstance()
        devices = transport_factory.EnumerateDevices()

        if len(devices) == 0:
            raise RuntimeError("Nessuna camera GigE Vision trovata")
        if self._device_index >= len(devices):
            raise RuntimeError(f"Camera index {self._device_index} non disponibile")

        self._camera = pylon.InstantCamera(
            transport_factory.CreateDevice(devices[self._device_index])
        )
        self._camera.Open()

        # Configurazione trigger hardware
        self._camera.TriggerMode.SetValue("On")
        self._camera.TriggerSource.SetValue(self._trigger_mode)
        self._camera.TriggerActivation.SetValue("RisingEdge")

        # Configurazione esposizione
        self._camera.ExposureTime.SetValue(self._exposure_us)
        self._camera.Gain.SetValue(self._gain_db)

        # Pixel format: Mono8 per velocità massima, BayerRG8 per colore
        self._camera.PixelFormat.SetValue("BayerRG8")

        # Buffer pool
        self._camera.MaxNumBuffer.SetValue(self._buffer_count)

        print(f"Camera connessa: {self._camera.DeviceModelName.GetValue()}")
        print(f"Risoluzione: {self._camera.Width.GetValue()}x{self._camera.Height.GetValue()}")
        print(f"Frame rate max: {self._camera.AcquisitionFrameRate.GetValue():.1f} FPS")

    def start_acquisition(self) -> None:
        """Avvia acquisizione continua in thread separato."""
        if self._camera is None:
            raise RuntimeError("Camera non connessa")

        self._acquiring = True
        self._camera.StartGrabbing(pylon.GrabStrategy_LatestImageOnly)

        acquisition_thread = threading.Thread(
            target=self._acquisition_loop,
            daemon=True
        )
        acquisition_thread.start()
        print("Acquisizione avviata in modalità trigger hardware")

    def _acquisition_loop(self) -> None:
        """Loop di acquisizione continua."""
        while self._acquiring and self._camera.IsGrabbing():
            try:
                grab_result = self._camera.RetrieveResult(
                    5000,  # Timeout 5 secondi
                    pylon.TimeoutHandling_ThrowException
                )

                if grab_result.GrabSucceeded():
                    # Conversione immagine
                    converter = pylon.ImageFormatConverter()
                    converter.OutputPixelFormat = pylon.PixelType_BGR8packed
                    converted = converter.Convert(grab_result)
                    image = converted.GetArray()

                    frame = AcquiredFrame(
                        image=image.copy(),
                        timestamp_ns=grab_result.TimeStamp,
                        frame_id=grab_result.ImageNumber,
                        trigger_counter=self._frame_counter
                    )

                    # Non-blocking: scarta se coda piena (priorità ai frame più recenti)
                    try:
                        self._frame_queue.put_nowait(frame)
                    except queue.Full:
                        # Scarta frame vecchio, inserisci nuovo
                        try:
                            self._frame_queue.get_nowait()
                        except queue.Empty:
                            pass
                        self._frame_queue.put_nowait(frame)

                    self._frame_counter += 1

                grab_result.Release()

            except pylon.TimeoutException:
                pass  # Nessun trigger ricevuto nel timeout, normale
            except Exception as e:
                print(f"Errore acquisizione: {e}")

    def get_frame(self, timeout: float = 1.0) -> Optional[AcquiredFrame]:
        """Recupera il prossimo frame dalla coda."""
        try:
            return self._frame_queue.get(timeout=timeout)
        except queue.Empty:
            return None

    def stop(self) -> None:
        """Ferma acquisizione e disconnette la camera."""
        self._acquiring = False
        if self._camera and self._camera.IsGrabbing():
            self._camera.StopGrabbing()
        if self._camera:
            self._camera.Close()
        print("Camera disconnessa")

# sorting_system.py
# Sistema di sorting integrato con PLC via OPC-UA

import asyncio
import asyncua
from ultralytics import YOLO
import numpy as np
import time
from dataclasses import dataclass
from typing import Optional
from enum import IntEnum


class SortingDecision(IntEnum):
    """Decisioni di sorting basate sulla criticita del difetto."""
    ACCEPT = 0          # Prodotto conforme: procede
    REJECT_DEFECT = 1   # Difetto standard: canale scarto generico
    REJECT_CRITICAL = 2 # Difetto critico (corpo estraneo, muffa): canale separato
    REINSPECT = 3       # Bassa confidence: re-ispezione manuale


@dataclass
class InspectionResult:
    """Risultato ispezione con decisione di sorting."""
    frame_id: int
    timestamp_ms: float
    decision: SortingDecision
    primary_defect: Optional[str]
    confidence: float
    defect_count: int
    inference_time_ms: float


class FoodInspectionEngine:
    """
    Engine principale di ispezione: integra vision AI e logica di sorting.
    """

    # Classificazione criticita difetti
    CRITICAL_DEFECTS = {'foreign_object', 'mold', 'rot'}
    STANDARD_DEFECTS = {'bruise', 'burn', 'crack', 'insect_damage'}
    QUALITY_DEFECTS = {'size_defect', 'color_defect'}

    # Soglie confidence per classe
    CLASS_THRESHOLDS = {
        'foreign_object': 0.30,
        'mold': 0.35,
        'rot': 0.38,
        'crack': 0.45,
        'bruise': 0.50,
        'insect_damage': 0.48,
        'burn': 0.50,
        'size_defect': 0.60,
        'color_defect': 0.60,
        'ok': 0.60,
    }

    def __init__(self, model_path: str) -> None:
        self._model = YOLO(model_path)
        self._class_names = self._model.names
        self._inspection_count = 0
        self._defect_count = 0
        self._reject_count = 0

    def inspect(self, image: np.ndarray, frame_id: int) -> InspectionResult:
        """
        Esegue l'ispezione AI su un frame e restituisce la decisione di sorting.
        """
        start_time = time.perf_counter()

        # Inference YOLO11
        results = self._model(
            image,
            conf=0.25,      # Soglia bassa: filtraggio per classe dopo
            iou=0.45,
            verbose=False,
            half=True,      # FP16 per velocità
        )

        inference_time_ms = (time.perf_counter() - start_time) * 1000

        # Analisi detections
        detections = []
        for result in results:
            if result.boxes is None:
                continue
            for box in result.boxes:
                class_id = int(box.cls[0])
                class_name = self._class_names[class_id]
                confidence = float(box.conf[0])

                # Applica soglia per classe
                class_threshold = self.CLASS_THRESHOLDS.get(class_name, 0.5)
                if confidence >= class_threshold and class_name != 'ok':
                    detections.append({
                        'class': class_name,
                        'confidence': confidence,
                        'bbox': box.xyxy[0].tolist(),
                    })

        # Logica di decisione sorting
        decision, primary_defect, max_confidence = self._make_sorting_decision(detections)

        self._inspection_count += 1
        if decision != SortingDecision.ACCEPT:
            self._reject_count += 1

        return InspectionResult(
            frame_id=frame_id,
            timestamp_ms=time.time() * 1000,
            decision=decision,
            primary_defect=primary_defect,
            confidence=max_confidence,
            defect_count=len(detections),
            inference_time_ms=inference_time_ms,
        )

    def _make_sorting_decision(
        self,
        detections: list
    ) -> tuple[SortingDecision, Optional[str], float]:
        """
        Logica di decisione sorting basata sui difetti rilevati.
        Priorità: REJECT_CRITICAL > REJECT_DEFECT > REINSPECT > ACCEPT
        """
        if not detections:
            return SortingDecision.ACCEPT, None, 0.0

        # Verifica presenza difetti critici (priorità assoluta)
        critical_detections = [
            d for d in detections
            if d['class'] in self.CRITICAL_DEFECTS
        ]
        if critical_detections:
            primary = max(critical_detections, key=lambda x: x['confidence'])
            return SortingDecision.REJECT_CRITICAL, primary['class'], primary['confidence']

        # Verifica difetti standard
        standard_detections = [
            d for d in detections
            if d['class'] in self.STANDARD_DEFECTS
        ]
        if standard_detections:
            primary = max(standard_detections, key=lambda x: x['confidence'])
            # Se confidence bassa (0.45-0.55) su difetto standard: reinspect
            if primary['confidence'] < 0.55:
                return SortingDecision.REINSPECT, primary['class'], primary['confidence']
            return SortingDecision.REJECT_DEFECT, primary['class'], primary['confidence']

        # Solo difetti qualità
        quality_detections = [
            d for d in detections
            if d['class'] in self.QUALITY_DEFECTS
        ]
        if quality_detections:
            primary = max(quality_detections, key=lambda x: x['confidence'])
            return SortingDecision.REJECT_DEFECT, primary['class'], primary['confidence']

        return SortingDecision.ACCEPT, None, 0.0

    def get_statistics(self) -> dict:
        """Statistiche di ispezione in tempo reale."""
        reject_rate = self._reject_count / max(self._inspection_count, 1) * 100
        return {
            'total_inspected': self._inspection_count,
            'total_rejected': self._reject_count,
            'reject_rate_pct': round(reject_rate, 2),
            'throughput': self._inspection_count,  # Da normalizzare nel tempo
        }


class PLCInterface:
    """
    Interfaccia OPC-UA verso PLC Siemens/Beckhoff per controllo attuatori.
    """

    # Indirizzi OPC-UA nodi PLC (configurati nel TIA Portal o TwinCAT)
    SORT_COMMAND_NODE = "ns=2;s=FoodLine.Sort.Command"
    SORT_DELAY_MS_NODE = "ns=2;s=FoodLine.Sort.DelayMs"
    CONVEYOR_SPEED_NODE = "ns=2;s=FoodLine.Conveyor.SpeedMps"
    INSPECTION_TO_EJECTOR_MM = 450.0  # Distanza fisica stazione ispezione -> espulsore

    def __init__(self, plc_url: str = "opc.tcp://192.168.1.100:4840") -> None:
        self._plc_url = plc_url
        self._client: Optional[asyncua.Client] = None

    async def connect(self) -> None:
        """Connette al PLC via OPC-UA."""
        self._client = asyncua.Client(url=self._plc_url)
        await self._client.connect()
        print(f"Connesso al PLC: {self._plc_url}")

    async def get_conveyor_speed(self) -> float:
        """Legge velocità nastro in m/s dal PLC."""
        node = self._client.get_node(self.CONVEYOR_SPEED_NODE)
        return await node.read_value()

    async def send_sort_command(
        self,
        decision: SortingDecision,
        conveyor_speed_mps: float
    ) -> None:
        """
        Invia comando di sorting al PLC con delay calcolato.

        Il delay compensa il ritardo di trasporto dalla stazione di ispezione
        all'attuatore di espulsione.
        """
        if decision == SortingDecision.ACCEPT:
            return  # Nessuna azione necessaria

        # Calcola delay: distanza / velocità - anticipo attuatore
        transport_delay_ms = (self.INSPECTION_TO_EJECTOR_MM / 1000) / conveyor_speed_mps * 1000
        actuator_lead_ms = 60  # Il valvola pneumatica richiede ~60ms per aprirsi
        total_delay_ms = max(0, int(transport_delay_ms - actuator_lead_ms))

        # Codice comando per PLC
        plc_command = 1 if decision in [SortingDecision.REJECT_DEFECT, SortingDecision.REJECT_CRITICAL] else 0

        # Scrivi delay e comando
        delay_node = self._client.get_node(self.SORT_DELAY_MS_NODE)
        command_node = self._client.get_node(self.SORT_COMMAND_NODE)

        await delay_node.write_value(total_delay_ms)
        await command_node.write_value(plc_command)

    async def disconnect(self) -> None:
        """Disconnette dal PLC."""
        if self._client:
            await self._client.disconnect()

# deploy_optimization.py
# Export e ottimizzazione modello per deploy industriale

from ultralytics import YOLO
import torch
import time
import numpy as np

def export_optimized_model(
    model_path: str,
    target_hardware: str = "tensorrt",  # tensorrt, openvino, onnx, hailo
    image_size: int = 640,
    batch_size: int = 1,
) -> str:
    """
    Esporta il modello YOLO11 nel formato ottimizzato per l'hardware target.

    Formati supportati per applicazioni industriali:
    - TensorRT (NVIDIA GPU): massima velocità su CUDA hardware
    - OpenVINO (Intel CPU/iGPU): ottimale per sistemi embedded Intel
    - ONNX (universale): compatibile con ONNX Runtime su qualsiasi hardware
    - Hailo: formato proprietario per chip Hailo-8/Hailo-15
    """
    model = YOLO(model_path)

    export_args = {
        'format': target_hardware,
        'imgsz': image_size,
        'batch': batch_size,
        'half': True,          # FP16: dimezza memoria, +30% velocità
        'int8': False,         # INT8 richiede calibration dataset
        'simplify': True,      # Semplifica grafo ONNX
        'dynamic': False,      # Batch size fisso per latenza deterministica
        'verbose': False,
    }

    if target_hardware == "tensorrt":
        export_args.update({
            'workspace': 4,     # GB di workspace TensorRT
            'device': '0',      # GPU 0
        })

    exported_path = model.export(**export_args)
    print(f"Modello esportato: {exported_path}")

    return str(exported_path)


def benchmark_inference(model_path: str, n_iterations: int = 1000) -> dict:
    """
    Benchmark di latenza per confronto tra formati export.
    """
    model = YOLO(model_path)

    # Immagine di test casuale (simula frame camera)
    dummy_image = np.random.randint(0, 255, (640, 640, 3), dtype=np.uint8)

    # Warmup
    for _ in range(50):
        model(dummy_image, verbose=False)

    # Benchmark
    latencies = []
    for _ in range(n_iterations):
        start = time.perf_counter()
        model(dummy_image, verbose=False, half=True)
        latencies.append((time.perf_counter() - start) * 1000)

    latencies = np.array(latencies)

    results = {
        'mean_ms': float(np.mean(latencies)),
        'median_ms': float(np.median(latencies)),
        'p95_ms': float(np.percentile(latencies, 95)),
        'p99_ms': float(np.percentile(latencies, 99)),
        'max_fps': round(1000 / np.mean(latencies), 1),
    }

    print(f"\nBenchmark {n_iterations} iterazioni:")
    print(f"  Latenza media:   {results['mean_ms']:.2f} ms")
    print(f"  Latenza P95:     {results['p95_ms']:.2f} ms")
    print(f"  Latenza P99:     {results['p99_ms']:.2f} ms")
    print(f"  FPS massimo:     {results['max_fps']:.1f} FPS")

    return results


# Esempio: confronto latenze per hardware diversi
# Risultati tipici su linea produttiva reale:
# PyTorch FP32:    12.4 ms -> 80 FPS
# PyTorch FP16:     7.1 ms -> 140 FPS
# TensorRT FP16:    2.8 ms -> 357 FPS
# TensorRT INT8:    1.9 ms -> 526 FPS
# ONNX Runtime:     5.2 ms -> 192 FPS
# OpenVINO:         3.8 ms -> 263 FPS

# validation_protocol.py
# Protocollo di validazione OQ/PQ per sistema vision alimentare

import json
import datetime
from dataclasses import dataclass, field, asdict
from typing import Optional
from ultralytics import YOLO
import numpy as np


@dataclass
class ChallengeTestResult:
    """Risultato di un singolo challenge test."""
    defect_class: str
    defect_severity: str       # 'mild', 'moderate', 'severe'
    n_samples: int
    detected_correctly: int
    false_positives_on_ok: int
    recall: float
    precision: float
    pass_fail: str


@dataclass
class ValidationReport:
    """Report di validazione OQ/PQ del sistema vision."""
    system_id: str
    product_type: str
    validation_date: str
    model_version: str
    hardware_config: dict
    challenge_results: list[ChallengeTestResult] = field(default_factory=list)
    overall_result: str = "PENDING"
    validated_by: str = ""
    notes: str = ""

    def to_json(self, output_path: str) -> None:
        """Esporta il report in formato JSON per documentazione normativa."""
        report_dict = asdict(self)
        with open(output_path, 'w', encoding='utf-8') as f:
            json.dump(report_dict, f, indent=2, ensure_ascii=False)
        print(f"Report salvato: {output_path}")


def run_challenge_test(
    model: YOLO,
    challenge_images_dir: str,
    defect_class: str,
    severity: str,
    n_ok_images: int = 50,
    acceptance_recall: float = 0.95,
    acceptance_precision: float = 0.85,
) -> ChallengeTestResult:
    """
    Esegue un challenge test su campioni noti.

    I challenge samples sono immagini acquisite dalla linea reale,
    etichettate da ispettori esperti, con difetti di gravita nota.
    Sono conservati fisicamente (campioni di riferimento) e fotografati
    in condizioni di linea controllate.
    """
    import glob
    import os

    # Immagini con difetto della classe specificata
    defect_images = glob.glob(
        os.path.join(challenge_images_dir, defect_class, severity, "*.jpg")
    )
    # Immagini ok di riferimento
    ok_images = glob.glob(
        os.path.join(challenge_images_dir, "ok", "*.jpg")
    )[:n_ok_images]

    detected = 0
    total_defect = len(defect_images)

    for img_path in defect_images:
        result = model(img_path, verbose=False)
        detections = [
            r for r in result[0].boxes
            if model.names[int(r.cls[0])] == defect_class
            and float(r.conf[0]) >= 0.35  # Soglia del sistema
        ]
        if len(detections) > 0:
            detected += 1

    # Falsi positivi su immagini ok
    fp_count = 0
    for img_path in ok_images:
        result = model(img_path, verbose=False)
        fp_detections = [
            r for r in result[0].boxes
            if model.names[int(r.cls[0])] == defect_class
            and float(r.conf[0]) >= 0.35
        ]
        if len(fp_detections) > 0:
            fp_count += 1

    recall = detected / max(total_defect, 1)
    precision = detected / max(detected + fp_count, 1)
    passed = recall >= acceptance_recall and precision >= acceptance_precision

    return ChallengeTestResult(
        defect_class=defect_class,
        defect_severity=severity,
        n_samples=total_defect,
        detected_correctly=detected,
        false_positives_on_ok=fp_count,
        recall=round(recall, 4),
        precision=round(precision, 4),
        pass_fail="PASS" if passed else "FAIL"
    )

# production_monitor.py
# Monitoring continuo del sistema vision in produzione

import sqlite3
import time
import json
from collections import deque
from typing import Optional
from dataclasses import dataclass


@dataclass
class ProductionStats:
    """Statistiche di produzione per monitoraggio."""
    timestamp: str
    window_minutes: int
    total_inspected: int
    reject_rate_pct: float
    defect_distribution: dict
    avg_inference_ms: float
    p99_inference_ms: float
    alert_triggered: bool
    alert_reason: Optional[str]


class ProductionMonitor:
    """
    Monitor continuo per sistema vision alimentare.
    Rileva anomalie statistiche che indicano degradazione del modello.
    """

    # Soglie di alert
    MAX_REJECT_RATE_PCT = 15.0      # >15% scarto e anomalo
    MIN_REJECT_RATE_PCT = 0.1       # <0.1% potrebbe indicare modello non funzionante
    MAX_INFERENCE_P99_MS = 15.0     # Latenza P99 non deve superare 15ms
    MAX_CONSECUTIVE_ACCEPTS = 500   # 500 ok di fila = modello probabilmente bloccato

    def __init__(self, db_path: str = "/data/production_log.db") -> None:
        self._db_path = db_path
        self._inspection_buffer: deque = deque(maxlen=10000)
        self._consecutive_accepts = 0
        self._init_db()

    def _init_db(self) -> None:
        """Inizializza database SQLite per log produzioni."""
        with sqlite3.connect(self._db_path) as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS inspections (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    timestamp REAL,
                    frame_id INTEGER,
                    decision INTEGER,
                    primary_defect TEXT,
                    confidence REAL,
                    inference_ms REAL,
                    line_speed_mps REAL
                )
            """)
            conn.execute("""
                CREATE TABLE IF NOT EXISTS alerts (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    timestamp REAL,
                    alert_type TEXT,
                    details TEXT,
                    acknowledged INTEGER DEFAULT 0
                )
            """)

    def log_inspection(self, result: dict) -> Optional[str]:
        """
        Logga un'ispezione e verifica anomalie.
        Ritorna il motivo dell'alert se presente, None altrimenti.
        """
        self._inspection_buffer.append(result)

        # Tracking accept consecutivi
        if result['decision'] == 0:  # ACCEPT
            self._consecutive_accepts += 1
        else:
            self._consecutive_accepts = 0

        # Alert: troppi accept consecutivi (modello potrebbe essere bloccato)
        if self._consecutive_accepts >= self.MAX_CONSECUTIVE_ACCEPTS:
            alert_msg = (
                f"Alert: {self._consecutive_accepts} accept consecutivi. "
                f"Verificare funzionamento sistema vision."
            )
            self._log_alert("CONSECUTIVE_ACCEPTS", alert_msg)
            self._consecutive_accepts = 0  # Reset dopo alert
            return alert_msg

        # Controlla reject rate su finestra recente (ultimi 100 pezzi)
        if len(self._inspection_buffer) >= 100:
            recent = list(self._inspection_buffer)[-100:]
            reject_count = sum(1 for r in recent if r['decision'] != 0)
            reject_rate = reject_count / len(recent) * 100

            if reject_rate > self.MAX_REJECT_RATE_PCT:
                alert_msg = f"Alert: tasso scarto {reject_rate:.1f}% (soglia: {self.MAX_REJECT_RATE_PCT}%)"
                self._log_alert("HIGH_REJECT_RATE", alert_msg)
                return alert_msg

        # Controlla latenza
        if result.get('inference_ms', 0) > self.MAX_INFERENCE_P99_MS:
            alert_msg = f"Alert: latenza inference {result['inference_ms']:.1f}ms supera soglia"
            self._log_alert("HIGH_LATENCY", alert_msg)
            return alert_msg

        return None

    def _log_alert(self, alert_type: str, details: str) -> None:
        """Logga alert nel database."""
        with sqlite3.connect(self._db_path) as conn:
            conn.execute(
                "INSERT INTO alerts (timestamp, alert_type, details) VALUES (?, ?, ?)",
                (time.time(), alert_type, details)
            )
        print(f"[ALERT] {alert_type}: {details}")

    def get_hourly_report(self) -> dict:
        """Genera report orario delle prestazioni di produzione."""
        recent = list(self._inspection_buffer)
        if not recent:
            return {}

        total = len(recent)
        rejects = sum(1 for r in recent if r['decision'] != 0)
        latencies = [r.get('inference_ms', 0) for r in recent]

        defect_dist = {}
        for r in recent:
            defect = r.get('primary_defect') or 'ok'
            defect_dist[defect] = defect_dist.get(defect, 0) + 1

        return {
            'total_inspected': total,
            'reject_rate_pct': round(rejects / total * 100, 2),
            'defect_distribution': defect_dist,
            'avg_inference_ms': round(sum(latencies) / len(latencies), 2),
            'p99_inference_ms': round(sorted(latencies)[int(len(latencies) * 0.99)], 2),
        }