Riconoscimento degli oggetti in uno stream video
Riconoscere gli oggetti presenti in un'immagine statica è ormai un problema risolto per la maggior parte dei casi d'uso comuni: basta un modello pre-addestrato e poche righe di codice. Le cose si complicano quando l'immagine non è più una sola ma diventa un flusso continuo di fotogrammi proveniente da una webcam, da una telecamera IP o da un file video. In questo scenario non conta soltanto la precisione del modello, ma anche la latenza, la stabilità della connessione con la sorgente, la coerenza dei risultati tra un frame e il successivo e la capacità di trasformare le singole rilevazioni in informazioni utili per l'applicazione.
In questo articolo costruiremo passo dopo passo una pipeline completa in Python che legge uno stream video, rileva gli oggetti con un modello della famiglia YOLO, li traccia nel tempo assegnando a ciascuno un identificativo stabile e produce eventi applicativi, come il conteggio delle persone o dei veicoli che attraversano una linea virtuale. Lungo il percorso vedremo i problemi tipici dei flussi in tempo reale e le tecniche per affrontarli.
Classificazione, rilevamento e tracciamento
Prima di scrivere codice conviene chiarire tre concetti che spesso vengono confusi.
- Classificazione: il modello riceve un'immagine e restituisce un'etichetta per l'immagine intera, ad esempio "gatto". Non dice dove si trova il gatto né quanti gatti ci sono.
- Rilevamento (object detection): il modello restituisce un elenco di oggetti, ciascuno con una classe, un punteggio di confidenza e un riquadro di delimitazione (bounding box) espresso in coordinate pixel.
- Tracciamento (object tracking): partendo dalle rilevazioni dei singoli frame, un algoritmo associa tra loro gli oggetti dei fotogrammi consecutivi e assegna a ciascuno un identificativo che resta lo stesso finché l'oggetto rimane in scena.
Su uno stream video il rilevamento da solo non basta quasi mai. Se vogliamo sapere quante auto sono passate davanti a un cancello, contare le rilevazioni per frame darebbe un numero enorme e privo di significato, perché la stessa auto viene rilevata in decine di fotogrammi consecutivi. Serve il tracciamento per capire che si tratta sempre dello stesso oggetto.
L'architettura della pipeline
Una pipeline di riconoscimento su video si può scomporre in cinque fasi:
- Acquisizione: lettura dei frame dalla sorgente, gestione delle disconnessioni e del buffering.
- Pre-elaborazione: ridimensionamento, conversione dello spazio colore, eventuale ritaglio di una regione di interesse.
- Inferenza: esecuzione del modello sul frame.
- Post-elaborazione: filtraggio per classe e confidenza, tracciamento, logica applicativa.
- Output: visualizzazione, registrazione su file, invio di eventi a un altro sistema.
Il punto critico è che acquisizione e inferenza procedono a velocità diverse. Una telecamera produce tipicamente 25 o 30 frame al secondo, mentre il modello, a seconda dell'hardware, può elaborarne di più o di meno. Se il modello è più lento della sorgente e leggiamo i frame in modo sequenziale, i fotogrammi si accumulano nel buffer del decoder e il ritardo rispetto alla scena reale cresce senza limiti. Vedremo più avanti come risolvere questo problema separando le due fasi.
Preparare l'ambiente
Useremo due librerie: OpenCV per l'acquisizione e la visualizzazione dei frame e Ultralytics per i modelli YOLO. Creiamo un ambiente virtuale e installiamo le dipendenze.
python3 -m venv .venv
source .venv/bin/activate
pip install -U ultralytics opencv-pythonIl pacchetto ultralytics installa anche PyTorch. Se la macchina dispone di una GPU NVIDIA conviene verificare che la versione di PyTorch installata supporti CUDA, perché la differenza di prestazioni rispetto alla CPU è di uno o due ordini di grandezza.
python -c "import torch; print(torch.__version__, torch.cuda.is_available())"Come modello useremo YOLO26, l'ultima generazione della famiglia YOLO di Ultralytics, nella variante nano (yolo26n.pt). È il modello più piccolo e veloce, adatto a girare anche su CPU o su dispositivi edge, ed è addestrato sul dataset COCO, che comprende 80 classi di oggetti comuni come persone, biciclette, automobili, autobus, animali domestici e oggetti d'uso quotidiano. Una caratteristica rilevante di YOLO26 è che l'inferenza è end-to-end e non richiede il passaggio di Non-Maximum Suppression (NMS) per eliminare i riquadri duplicati, il che semplifica l'esportazione e riduce la latenza. Il file dei pesi viene scaricato automaticamente al primo utilizzo.
Leggere uno stream con OpenCV
La classe cv2.VideoCapture di OpenCV accetta come sorgente un indice numerico (per le webcam collegate alla macchina), il percorso di un file video oppure un URL di rete, ad esempio uno stream RTSP di una telecamera IP. Iniziamo con un semplice programma che apre la sorgente e mostra i frame a video.
import sys
import cv2
def parse_source(value: str):
# Un argomento numerico indica l'indice di una webcam locale
return int(value) if value.isdigit() else value
def main() -> None:
source = parse_source(sys.argv[1] if len(sys.argv) > 1 else "0")
capture = cv2.VideoCapture(source)
if not capture.isOpened():
raise SystemExit(f"Impossibile aprire la sorgente: {source}")
width = int(capture.get(cv2.CAP_PROP_FRAME_WIDTH))
height = int(capture.get(cv2.CAP_PROP_FRAME_HEIGHT))
fps = capture.get(cv2.CAP_PROP_FPS)
print(f"Sorgente aperta: {width}x{height} a {fps:.1f} fps")
while True:
ok, frame = capture.read()
if not ok:
# Fine del file o interruzione dello stream
break
cv2.imshow("stream", frame)
# Premere q per uscire
if cv2.waitKey(1) & 0xFF == ord("q"):
break
capture.release()
cv2.destroyAllWindows()
if __name__ == "__main__":
main()Il programma si avvia con python viewer.py 0 per la webcam predefinita, con python viewer.py video.mp4 per un file o con un URL del tipo rtsp://utente:password@192.168.1.50:554/stream1 per una telecamera IP. Il frame restituito da read() è un array NumPy di forma (altezza, larghezza, 3) con i canali nell'ordine BGR, che è la convenzione storica di OpenCV.
La chiamata cv2.waitKey(1) è indispensabile quando si usa imshow: oltre a leggere la tastiera, consente alla finestra di aggiornarsi. Su un server senza interfaccia grafica, invece, va eliminata insieme a imshow, come vedremo nella versione finale.
Il primo rilevamento con YOLO26
Aggiungere il rilevamento richiede pochissimo codice. Il modello si carica una sola volta all'avvio e poi si invoca su ogni frame.
import cv2
from ultralytics import YOLO
model = YOLO("yolo26n.pt")
capture = cv2.VideoCapture(0)
while True:
ok, frame = capture.read()
if not ok:
break
# Inferenza sul singolo frame: verbose=False evita il log a ogni chiamata
results = model(frame, conf=0.4, verbose=False)
# plot() restituisce una copia del frame con riquadri ed etichette
annotated = results[0].plot()
cv2.imshow("detections", annotated)
if cv2.waitKey(1) & 0xFF == ord("q"):
break
capture.release()
cv2.destroyAllWindows()Il parametro conf imposta la soglia minima di confidenza: le rilevazioni con un punteggio inferiore a 0.4 vengono scartate. Una soglia troppo bassa produce falsi positivi, una troppo alta fa perdere oggetti piccoli o parzialmente nascosti. Il valore giusto dipende dalla scena e va trovato sperimentalmente.
Il metodo plot() è comodo per una prima verifica, ma in un'applicazione reale vogliamo accedere ai dati grezzi delle rilevazioni. L'oggetto restituito per ogni frame espone l'attributo boxes, che contiene tre tensori allineati: le coordinate dei riquadri, le confidenze e gli indici di classe.
import cv2
import numpy as np
def extract_detections(result, names: dict) -> list[dict]:
# Converte le rilevazioni in una lista di dizionari indipendente da PyTorch
boxes = result.boxes
xyxy = boxes.xyxy.cpu().numpy()
confidences = boxes.conf.cpu().numpy()
classes = boxes.cls.cpu().numpy().astype(int)
detections = []
for (x1, y1, x2, y2), confidence, class_id in zip(xyxy, confidences, classes):
detections.append({
"label": names[class_id],
"confidence": float(confidence),
"box": (int(x1), int(y1), int(x2), int(y2)),
})
return detections
def draw_detections(frame: np.ndarray, detections: list[dict]) -> np.ndarray:
output = frame.copy()
for detection in detections:
x1, y1, x2, y2 = detection["box"]
text = f'{detection["label"]} {detection["confidence"]:.2f}'
cv2.rectangle(output, (x1, y1), (x2, y2), (0, 200, 0), 2)
# Sfondo pieno sotto l'etichetta per renderla leggibile
(text_width, text_height), baseline = cv2.getTextSize(
text, cv2.FONT_HERSHEY_SIMPLEX, 0.5, 1
)
top = max(y1 - text_height - baseline - 4, 0)
cv2.rectangle(output, (x1, top), (x1 + text_width + 4, y1), (0, 200, 0), -1)
cv2.putText(
output, text, (x1 + 2, y1 - baseline - 2),
cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 0, 0), 1, cv2.LINE_AA,
)
return outputLe coordinate in formato xyxy indicano l'angolo superiore sinistro e quello inferiore destro del riquadro, in pixel rispetto al frame originale: Ultralytics si occupa di riscalarle automaticamente anche se internamente il modello lavora su un'immagine ridimensionata. Il dizionario model.names associa a ogni indice di classe il nome corrispondente, ad esempio 0 a person e 2 a car.
Se ci interessano solo alcune classi, è più efficiente chiederlo direttamente al modello con il parametro classes invece di filtrare dopo: model(frame, classes=[0, 2]) restituisce soltanto persone e automobili.
Misurare le prestazioni
Prima di ottimizzare bisogna misurare. Il dato più significativo è il numero di frame elaborati al secondo dall'intera pipeline. Poiché il tempo di elaborazione varia leggermente da un frame all'altro, conviene calcolare una media mobile esponenziale, che produce un valore stabile senza dover conservare uno storico.
import time
class FpsMeter:
def __init__(self, smoothing: float = 0.9) -> None:
self.smoothing = smoothing
self.value = 0.0
self._last = time.perf_counter()
def tick(self) -> float:
now = time.perf_counter()
elapsed = now - self._last
self._last = now
if elapsed > 0:
instant = 1.0 / elapsed
# Media mobile esponenziale: il primo campione inizializza il valore
self.value = instant if self.value == 0 else (
self.smoothing * self.value + (1 - self.smoothing) * instant
)
return self.valueOltre agli FPS, ogni oggetto risultato di Ultralytics contiene l'attributo speed, un dizionario con i millisecondi spesi in pre-elaborazione, inferenza e post-elaborazione. È utile per capire dove si concentra il costo: se la pre-elaborazione pesa quanto l'inferenza, probabilmente il frame è molto più grande del necessario.
Il problema della latenza accumulata
Proviamo ora a usare lo script precedente con una telecamera IP via RTSP su una macchina senza GPU. Se il modello elabora 12 frame al secondo e la telecamera ne invia 25, dopo un minuto l'immagine mostrata a video sarà in ritardo di diversi secondi rispetto alla realtà, e il ritardo continuerà a crescere. Il motivo è che capture.read() restituisce il frame successivo nella coda del decoder, non quello più recente: i fotogrammi che non facciamo in tempo a leggere restano in attesa.
Per un file video questo comportamento è corretto, perché vogliamo analizzare ogni frame. Per uno stream dal vivo è invece un difetto grave: un sistema che segnala un intruso con trenta secondi di ritardo non serve a nulla. La soluzione è separare acquisizione e inferenza in due thread. Il thread di acquisizione legge continuamente dalla sorgente e conserva soltanto l'ultimo frame; il ciclo principale, quando è pronto, preleva quel frame e lo elabora. I fotogrammi intermedi vengono scartati, ma l'analisi lavora sempre sulla scena attuale.
Già che ci siamo, il thread di acquisizione si occupa anche delle riconnessioni: le telecamere di rete si disconnettono per i motivi più vari, dai riavvii del dispositivo ai problemi di rete, e un sistema che deve restare in funzione per mesi non può fermarsi alla prima interruzione.
import threading
import time
import cv2
import numpy as np
class FrameGrabber:
"""Legge i frame in un thread separato e conserva soltanto il più recente."""
def __init__(self, source, reconnect_delay: float = 2.0) -> None:
self.source = source
self.reconnect_delay = reconnect_delay
self._capture: cv2.VideoCapture | None = None
self._frame: np.ndarray | None = None
self._frame_id = 0
self._lock = threading.Lock()
self._stopped = threading.Event()
self._thread = threading.Thread(target=self._run, daemon=True)
def start(self) -> "FrameGrabber":
self._thread.start()
return self
def _open(self) -> cv2.VideoCapture:
capture = cv2.VideoCapture(self.source)
# Riduce il buffer interno dove il backend lo supporta
capture.set(cv2.CAP_PROP_BUFFERSIZE, 1)
return capture
def _reset(self) -> None:
if self._capture is not None:
self._capture.release()
self._capture = None
time.sleep(self.reconnect_delay)
def _run(self) -> None:
while not self._stopped.is_set():
if self._capture is None or not self._capture.isOpened():
self._capture = self._open()
if not self._capture.isOpened():
print(f"Sorgente non disponibile, nuovo tentativo tra {self.reconnect_delay}s")
self._reset()
continue
ok, frame = self._capture.read()
if not ok:
print("Stream interrotto, riconnessione in corso")
self._reset()
continue
with self._lock:
self._frame = frame
self._frame_id += 1
def read(self) -> tuple[int, np.ndarray | None]:
# Restituisce l'identificativo e una copia dell'ultimo frame disponibile
with self._lock:
if self._frame is None:
return 0, None
return self._frame_id, self._frame.copy()
def stop(self) -> None:
self._stopped.set()
self._thread.join(timeout=2.0)
if self._capture is not None:
self._capture.release()Il metodo read() restituisce anche un contatore progressivo. Il ciclo principale lo confronta con quello dell'iterazione precedente per evitare di elaborare due volte lo stesso frame quando il modello è più veloce della sorgente. La copia del frame all'interno del lock garantisce che il thread di acquisizione non sovrascriva l'array mentre il modello lo sta usando.
Questa classe è pensata per le sorgenti dal vivo. Su un file video il thread leggerebbe i frame il più velocemente possibile e ne scarterebbe la maggior parte; per l'analisi offline di un file va quindi usata la lettura sequenziale vista in precedenza.
Dal rilevamento al tracciamento
Ultralytics integra due algoritmi di tracciamento, ByteTrack e BoT-SORT, utilizzabili tramite il metodo track() al posto della normale chiamata al modello. Entrambi seguono lo schema tracking-by-detection: a ogni frame il detector produce i riquadri, poi l'algoritmo li associa alle tracce esistenti usando una previsione del movimento basata su un filtro di Kalman e la sovrapposizione tra il riquadro previsto e quello rilevato. ByteTrack ha la particolarità di recuperare anche le rilevazioni a bassa confidenza quando corrispondono a una traccia già nota, il che riduce le interruzioni quando un oggetto viene parzialmente coperto.
results = model.track(
frame,
persist=True, # mantiene lo stato del tracker tra le chiamate
tracker="bytetrack.yaml", # in alternativa botsort.yaml
conf=0.4,
classes=[0, 2], # persone e automobili
verbose=False,
)
result = results[0]
if result.boxes.id is not None:
track_ids = result.boxes.id.int().cpu().tolist()
centers = result.boxes.xywh.cpu().numpy()[:, :2]
for track_id, (center_x, center_y) in zip(track_ids, centers):
print(track_id, round(center_x), round(center_y))Il parametro persist=True è fondamentale quando si chiama track() un frame alla volta: senza di esso il tracker verrebbe reinizializzato a ogni invocazione e gli identificativi non sarebbero mai stabili. L'attributo boxes.id vale None quando nel frame non ci sono tracce confermate, quindi va sempre controllato prima dell'uso. Il formato xywh esprime ogni riquadro come centro, larghezza e altezza, comodo quando ci interessa la posizione dell'oggetto più che i suoi bordi.
Un dettaglio importante: il tracker dipende dalla continuità temporale. Se scartiamo molti frame perché il modello è lento, gli oggetti si spostano parecchio tra due fotogrammi elaborati e le associazioni diventano meno affidabili. È un motivo in più per scegliere un modello abbastanza leggero da tenere il passo con la scena.
Dalla traccia all'evento: contare gli attraversamenti
Le tracce sono il materiale grezzo; l'applicazione ha bisogno di eventi. Un caso d'uso classico è il conteggio degli oggetti che attraversano una linea virtuale, ad esempio le persone che entrano ed escono da un negozio o i veicoli che transitano su una strada. L'idea è semplice: per ogni traccia ricordiamo da quale lato della linea si trovava il suo centro nel frame precedente; quando il lato cambia, registriamo un attraversamento nella direzione corrispondente.
Bisogna anche ripulire periodicamente lo stato delle tracce che non si vedono più da un po', altrimenti in un processo che gira per settimane il dizionario crescerebbe senza limiti.
from dataclasses import dataclass, field
@dataclass
class TrackState:
is_below: bool
last_seen: int
@dataclass
class LineCounter:
line_y: int
max_age: int = 90
inbound: int = 0
outbound: int = 0
_tracks: dict[int, TrackState] = field(default_factory=dict)
def update(self, track_id: int, center_y: float, frame_index: int) -> str | None:
is_below = center_y >= self.line_y
state = self._tracks.get(track_id)
self._tracks[track_id] = TrackState(is_below=is_below, last_seen=frame_index)
# Prima osservazione o nessun cambio di lato: nessun evento
if state is None or state.is_below == is_below:
return None
if is_below:
self.inbound += 1
return "in"
self.outbound += 1
return "out"
def prune(self, frame_index: int) -> None:
# Elimina le tracce non più osservate da max_age frame
expired = [
track_id for track_id, state in self._tracks.items()
if frame_index - state.last_seen > self.max_age
]
for track_id in expired:
del self._tracks[track_id]La convenzione scelta considera "in ingresso" il passaggio dall'alto verso il basso dell'immagine. In una installazione reale la direzione va concordata in base alla posizione della telecamera. Per scene in cui la linea non è orizzontale si può generalizzare il confronto usando il segno del prodotto vettoriale tra il segmento della linea e il vettore che va da un suo estremo al centro dell'oggetto: il principio resta identico.
La pipeline completa
Mettiamo insieme tutti i pezzi in un unico programma configurabile da riga di comando. Il programma legge lo stream con FrameGrabber, esegue il tracciamento, aggiorna il contatore, scrive ogni evento come riga JSON sullo standard output e, se richiesto, mostra a video il frame annotato. Le classi FrameGrabber, FpsMeter e LineCounter sono quelle definite nelle sezioni precedenti e si suppone che stiano in un modulo pipeline.py.
import argparse
import json
import sys
import time
from datetime import datetime, timezone
import cv2
from ultralytics import YOLO
from pipeline import FpsMeter, FrameGrabber, LineCounter
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Rilevamento e conteggio oggetti su stream video")
parser.add_argument("source", help="indice webcam, file video o URL RTSP")
parser.add_argument("--model", default="yolo26n.pt")
parser.add_argument("--classes", type=int, nargs="*", default=[0])
parser.add_argument("--conf", type=float, default=0.4)
parser.add_argument("--imgsz", type=int, default=640)
parser.add_argument("--line", type=float, default=0.5, help="posizione della linea (0-1)")
parser.add_argument("--show", action="store_true", help="mostra la finestra di anteprima")
return parser.parse_args()
def emit(event: dict) -> None:
# Una riga JSON per evento: facile da inoltrare a un log collector o a una coda
event["timestamp"] = datetime.now(timezone.utc).isoformat()
sys.stdout.write(json.dumps(event) + "\n")
sys.stdout.flush()
def main() -> None:
args = parse_args()
source = int(args.source) if args.source.isdigit() else args.source
model = YOLO(args.model)
grabber = FrameGrabber(source).start()
fps_meter = FpsMeter()
counter: LineCounter | None = None
last_frame_id = 0
frame_index = 0
try:
while True:
frame_id, frame = grabber.read()
if frame is None or frame_id == last_frame_id:
# Nessun frame nuovo: breve attesa per non occupare la CPU
time.sleep(0.005)
continue
last_frame_id = frame_id
frame_index += 1
if counter is None:
# La linea si calcola sulla prima immagine ricevuta
counter = LineCounter(line_y=int(frame.shape[0] * args.line))
result = model.track(
frame,
persist=True,
tracker="bytetrack.yaml",
conf=args.conf,
classes=args.classes,
imgsz=args.imgsz,
verbose=False,
)[0]
if result.boxes.id is not None:
track_ids = result.boxes.id.int().cpu().tolist()
centers = result.boxes.xywh.cpu().numpy()[:, :2]
labels = result.boxes.cls.int().cpu().tolist()
for track_id, (_, center_y), class_id in zip(track_ids, centers, labels):
direction = counter.update(track_id, float(center_y), frame_index)
if direction is not None:
emit({
"event": "line_crossed",
"direction": direction,
"track_id": track_id,
"label": model.names[class_id],
"inbound_total": counter.inbound,
"outbound_total": counter.outbound,
})
counter.prune(frame_index)
fps = fps_meter.tick()
if args.show:
annotated = result.plot()
width = annotated.shape[1]
cv2.line(annotated, (0, counter.line_y), (width, counter.line_y), (0, 0, 255), 2)
summary = f"IN {counter.inbound} OUT {counter.outbound} FPS {fps:.1f}"
cv2.putText(annotated, summary, (10, 30), cv2.FONT_HERSHEY_SIMPLEX,
0.8, (255, 255, 255), 2, cv2.LINE_AA)
cv2.imshow("tracking", annotated)
if cv2.waitKey(1) & 0xFF == ord("q"):
break
except KeyboardInterrupt:
pass
finally:
grabber.stop()
if args.show:
cv2.destroyAllWindows()
if __name__ == "__main__":
main()Il programma si avvia ad esempio così, per contare persone e automobili su una telecamera di rete con la linea a due terzi dell'altezza:
python counter.py "rtsp://camera.local:554/stream1" --classes 0 2 --line 0.66 --showL'output è una sequenza di righe JSON, una per ogni attraversamento, che si presta a essere redirezionata su file, inviata a un sistema di raccolta log oppure pubblicata su una coda di messaggi come MQTT o Kafka. Separare la produzione degli eventi dalla loro destinazione rende il componente facile da integrare: chi lo usa non deve sapere nulla di visione artificiale, ma solo leggere eventi strutturati.
{"event": "line_crossed", "direction": "in", "track_id": 17, "label": "person", "inbound_total": 4, "outbound_total": 2, "timestamp": "2026-10-09T18:12:44.381205+00:00"}Ottimizzare le prestazioni
Quando la pipeline funziona, il passo successivo è renderla abbastanza veloce per l'hardware a disposizione. Le leve principali sono queste.
Risoluzione di inferenza
Il parametro imgsz stabilisce la dimensione a cui il frame viene ridimensionato prima di entrare nel modello. Il valore predefinito è 640 pixel sul lato lungo. Scendere a 480 o 416 riduce sensibilmente il tempo di calcolo, al prezzo di una peggiore capacità di rilevare oggetti piccoli o lontani. Salire a 960 o 1280 fa l'opposto. Non serve mai inviare al modello un frame 4K intero: la telecamera può trasmettere ad alta risoluzione per la registrazione, ma l'analisi lavora quasi sempre meglio su un sotto-stream a risoluzione ridotta, che molte telecamere IP offrono nativamente.
Regione di interesse
Se gli eventi rilevanti avvengono solo in una parte dell'inquadratura, ad esempio una porta o una corsia, conviene ritagliare quella porzione prima dell'inferenza. Il ritaglio con NumPy è praticamente gratuito e il modello lavora su meno pixel, con il vantaggio aggiuntivo che gli oggetti, a parità di imgsz, risultano più grandi.
# Ritaglio della regione di interesse: righe prima, colonne poi
roi_x, roi_y, roi_width, roi_height = 200, 120, 800, 500
roi = frame[roi_y:roi_y + roi_height, roi_x:roi_x + roi_width]
result = model.track(roi, persist=True, verbose=False)[0]
# Le coordinate sono relative al ritaglio: per disegnarle sul frame intero
# va aggiunto l'offset della regioneGPU e mezza precisione
Su GPU NVIDIA il parametro device=0 seleziona la prima scheda e half=True abilita il calcolo in virgola mobile a 16 bit, che dimezza l'uso di memoria e accelera l'inferenza con una perdita di precisione trascurabile nella maggior parte dei casi.
Esportazione del modello
Il formato PyTorch è comodo in sviluppo, ma per la produzione Ultralytics permette di esportare il modello in formati ottimizzati per il runtime di destinazione: ONNX per un'esecuzione portabile su CPU, TensorRT per le GPU NVIDIA e i dispositivi Jetson, OpenVINO per l'hardware Intel, CoreML per i dispositivi Apple. Il modello esportato si carica con la stessa classe YOLO e il resto del codice non cambia. L'architettura senza NMS di YOLO26 rende queste esportazioni particolarmente lineari, perché non c'è un passaggio di post-elaborazione da reimplementare fuori dal grafo del modello.
from ultralytics import YOLO
model = YOLO("yolo26n.pt")
# Esportazione per CPU generiche
model.export(format="onnx", imgsz=640)
# Esportazione per GPU NVIDIA in mezza precisione (richiede TensorRT installato)
model.export(format="engine", imgsz=640, half=True)
# Il modello esportato si usa esattamente come l'originale
onnx_model = YOLO("yolo26n.onnx")
results = onnx_model("frame.jpg")Più telecamere
Con più sorgenti la tentazione è avviare un processo per telecamera, ciascuno con il proprio modello caricato in memoria. Su GPU è più efficiente un solo processo che raccoglie l'ultimo frame di ogni telecamera e li passa al modello in un unico batch, perché il costo di un'inferenza su più immagini è molto inferiore alla somma dei costi delle inferenze singole. In questo caso il tracciamento va gestito con un tracker separato per ciascuna sorgente, altrimenti gli identificativi di telecamere diverse finirebbero mescolati.
Eseguire la pipeline su un server
In produzione la pipeline gira tipicamente su una macchina senza schermo, come servizio di sistema o in un container. In questo caso va avviata senza l'opzione --show e, per evitare dipendenze grafiche inutili, si può installare la variante opencv-python-headless al posto di opencv-python. Se occorre conservare il video annotato per una verifica successiva, la classe cv2.VideoWriter permette di scrivere i frame su file, ma va usata con giudizio: la codifica video ha un costo di CPU non trascurabile e i file crescono rapidamente.
fourcc = cv2.VideoWriter_fourcc(*"mp4v")
writer = cv2.VideoWriter("output.mp4", fourcc, 15.0, (frame.shape[1], frame.shape[0]))
# All'interno del ciclo principale
writer.write(annotated)
# Alla chiusura
writer.release()Per la supervisione conviene esporre almeno due metriche: gli FPS effettivi della pipeline e il tempo trascorso dall'ultimo frame ricevuto. Un valore di FPS che crolla segnala un problema di risorse; un frame vecchio di minuti segnala che la telecamera è irraggiungibile anche se il processo è ancora in vita.
Privacy e aspetti normativi
Un sistema che analizza immagini di persone tratta dati personali, e in Europa ricade nel campo di applicazione del GDPR. Prima di metterlo in produzione occorre valutarne la base giuridica, informare gli interessati con un'apposita segnaletica e, nei casi previsti, svolgere una valutazione d'impatto. Dal punto di vista tecnico, il principio di minimizzazione suggerisce alcune scelte concrete: elaborare i frame in memoria senza salvarli, conservare solo gli eventi aggregati come i conteggi, limitare la regione di interesse alla zona strettamente necessaria e, se le immagini devono essere archiviate, oscurare automaticamente i volti. In ambito lavorativo valgono inoltre le norme specifiche sui controlli a distanza dei lavoratori. Sono valutazioni da fare con chi si occupa della conformità, ma è bene che lo sviluppatore le conosca, perché incidono direttamente sull'architettura.
Conclusione
Abbiamo visto che il riconoscimento degli oggetti in uno stream video va ben oltre l'invocazione di un modello su ogni frame. Una pipeline affidabile separa l'acquisizione dall'inferenza per non accumulare latenza, gestisce le disconnessioni della sorgente, usa il tracciamento per trasformare rilevazioni isolate in oggetti con un'identità stabile e produce eventi strutturati che il resto del sistema può consumare senza conoscere i dettagli della visione artificiale. Su queste fondamenta si possono costruire applicazioni molto diverse: conteggio dei passaggi, rilevamento di intrusioni in aree vietate, misurazione dei tempi di permanenza, monitoraggio di linee produttive. Il passo naturale successivo è addestrare il modello su un dataset proprio, quando le 80 classi di COCO non bastano a descrivere gli oggetti che interessano al dominio applicativo.