Combina tareas en un único flujo de trabajo
Crea una cascada de detección y clasificación, clasifica cada objeto rastreado una sola vez en lugar de en cada fotograma y convierte los puntos clave de pose en series temporales por persona.
Los sistemas más útiles que van más allá de la detección suelen combinar dos modelos, no uno. Un detector localiza a cada trabajador y un clasificador determina si cada casco es del tipo correcto; un modelo de pose localiza las articulaciones y un rastreador las convierte en una cronología que permite contar repeticiones. El código es breve. Lo que falla en estos sistemas es que el conjunto de datos de la segunda etapa no coincide con los recortes de la primera y que el coste se multiplica por cada objeto de cada fotograma: si se ejecuta sin optimizar, un flujo de 30 FPS con diez personas requiere 300 clasificaciones por segundo.
Práctico
Dos patrones que cubren la mayoría de los productos#
Ambos parten del mismo fotograma y de la misma API de Ultralytics; se diferencian en lo que consume la segunda etapa. La selección y el ajuste del rastreador, así como la gestión de flujos para el seguimiento de objetos, se tratan en Sigue objetos con ByteTrack, BoT-SORT y TrackTrack; aquí nos centramos en combinar las tareas.
Detecta y luego clasifica los recortes#
from ultralytics import YOLO
detector = YOLO("yolo26n.pt")
classifier = YOLO("yolo26n-cls.pt") # swap in your own fine-tuned classifier
result = detector("https://ultralytics.com/images/bus.jpg", conf=0.5)[0]
crops, labels = [], []
for box in result.boxes:
x1, y1, x2, y2 = box.xyxy[0].int().tolist()
crops.append(result.orig_img[y1:y2, x1:x2]) # BGR numpy crop, same as the detector saw
labels.append(result.names[int(box.cls)])
for label, crop_result in zip(labels, classifier(crops, verbose=False)): # one batched call for all crops
probs = crop_result.probs
print(f"{label:6s} -> {crop_result.names[probs.top1]} ({float(probs.top1conf):.2f})")bus -> bullet_train (0.31)
person -> sweatshirt (0.24)
person -> abaya (0.50)
person -> loudspeaker (0.49)
person -> frilled_lizard (0.24)La integración es correcta: los recortes proceden de orig_img a resolución completa y en las propias coordenadas de píxel del detector; el clasificador los procesa todos en una única llamada por lotes. Las etiquetas son deliberadamente absurdas.
El clasificador de ImageNet nunca se entrenó con figuras urbanas recortadas y encuadradas de cerca, así que hace conjeturas: fíjate en que ninguna confianza supera 0.50. Lo mismo ocurre con un clasificador entrenado con fotos de productos nítidas al recibir recortes borrosos del detector.
Crea el conjunto de datos de la segunda etapa a partir de tus propios recortes#
La forma fiable de obtener datos que coincidan consiste en dejar que el detector los genere:
- Ejecuta el detector sobre grabaciones representativas con
save_crop=True, por ejemplo,yolo predict model=best.pt source=footage/ save_crop=True. Los recortes se guardan enruns/detect/predict/crops/<class>/. - Ordena los recortes de cada clase en tus carpetas de categorías específicas (
helmet_ok/,helmet_missing/): esa es ya la estructura de clasificación de la lección anterior. - Entrena el clasificador con esas carpetas y valídalo con recortes de otro día o cámara.
Como el detector ha generado los recortes, estos conservan el encuadre, el desenfoque y las oclusiones reales: exactamente lo que verá el clasificador en producción; además, el ajuste fino es un caso habitual de aprendizaje por transferencia.
Clasifica una sola vez por ID de seguimiento#
Un casco no cambia de un fotograma a otro, así que clasificar cada caja en cada fotograma es trabajo desperdiciado. Rastrea las detecciones y clasifica cada ID de seguimiento nuevo una sola vez:
from ultralytics import YOLO
detector = YOLO("yolo26n.pt")
classifier = YOLO("yolo26n-cls.pt") # swap in your own fine-tuned classifier
source = "https://github.com/ultralytics/assets/releases/download/v0.0.0/decelera_portrait_min.mov"
labels = {} # track id -> (label, confidence), filled once per person
frames = boxes_seen = 0
for result in detector.track(source, classes=[0], stream=True, persist=True, verbose=False): # people only
frames += 1
if result.boxes.id is None:
continue
for track_id, (x1, y1, x2, y2) in zip(result.boxes.id.int().tolist(), result.boxes.xyxy.int().tolist()):
boxes_seen += 1
if track_id not in labels: # first sighting: classify once, then reuse
probs = classifier(result.orig_img[y1:y2, x1:x2], verbose=False)[0].probs
labels[track_id] = (classifier.names[probs.top1], round(float(probs.top1conf), 2))
print(f"{frames} frames, {boxes_seen} person boxes, {len(labels)} classifier calls")
print(labels)11 frames, 33 person boxes, 3 classifier calls
{1: ('basketball', 0.51), 2: ('maillot_(tights)', 0.14), 3: ('bulletproof_vest', 0.23)}Tres llamadas al clasificador en lugar de 33 en un clip de 11 fotogramas; en un flujo largo, el ahorro aumenta con cada fotograma que una persona permanece a la vista. Las etiquetas siguen siendo estimaciones de ImageNet: sustitúyelo por tu clasificador ajustado y obtendrás la respuesta por persona.
Clasifica una vez por track cuando el atributo no pueda cambiar mientras el objeto esté a la vista (tipo de casco, marca del vehículo, SKU del producto). Vuelve a clasificar cada N fotogramas cuando sí pueda cambiar (puerta abierta, mano levantada). Si las primeras detecciones suelen ser pequeñas y borrosas, clasifica el fotograma en el que la caja del track sea más grande.
Pose y seguimiento#
El modo de seguimiento asigna un result.boxes.id persistente a cada persona, por lo que los puntos clave se convierten en una serie temporal. Este script registra el ángulo de la rodilla izquierda de cada persona a lo largo del mismo vídeo corto:
from collections import defaultdict
import numpy as np
from ultralytics import YOLO
def angle(a, b, c):
"""Angle at b, in degrees, formed by points a-b-c."""
ba, bc = a - b, c - b
return np.degrees(np.arccos(np.dot(ba, bc) / (np.linalg.norm(ba) * np.linalg.norm(bc))))
model = YOLO("yolo26n-pose.pt")
source = "https://github.com/ultralytics/assets/releases/download/v0.0.0/decelera_portrait_min.mov"
HIP, KNEE, ANKLE = 11, 13, 15 # left leg, COCO keypoint order
knee_angles = defaultdict(list) # track id -> knee angle per frame
for result in model.track(source, stream=True, persist=True, verbose=False):
if result.boxes.id is None: # no confirmed tracks in this frame
continue
kpts = result.keypoints.cpu().numpy()
for track_id, xy, conf in zip(result.boxes.id.int().tolist(), kpts.xy, kpts.conf):
if conf[[HIP, KNEE, ANKLE]].min() > 0.5: # skip occluded legs
knee_angles[track_id].append(angle(xy[HIP], xy[KNEE], xy[ANKLE]))
for track_id, angles in sorted(knee_angles.items()):
print(f"id {track_id}: {len(angles)} frames, left knee {min(angles):.0f}-{max(angles):.0f} deg")id 1: 11 frames, left knee 170-176 deg
id 2: 11 frames, left knee 96-118 deg
id 3: 11 frames, left knee 177-180 degLa rodilla flexionada del ID 2 corresponde a la persona sentada del clip; las otras dos personas están de pie con las piernas rectas, cerca de los 180°. A partir de aquí, un contador de repeticiones detecta un cruce de umbral en la serie de cada ID: por debajo de un ángulo y luego por encima de otro; eso es lo que te ofrece la solución de monitorización de entrenamientos.
Calcula el coste de la cadena#
| Factor de coste | Detectar y luego clasificar | Pose y seguimiento |
|---|---|---|
| Aumenta con | objetos por fotograma | fotogramas |
| Palanca principal | recorta los lotes; clasifica una vez por ID de seguimiento y almacena el resultado en caché | reduce imgsz o vid_stride cuando el movimiento sea lento |
| Modo de fallo | el clasificador recibe recortes distintos de los datos de entrenamiento | los cambios de ID dividen el historial de una persona en dos |
Síntoma: el contador de repeticiones se reinicia a mitad de una serie o una persona aparece con dos ID. Causa: el rastreador perdió a la persona (por oclusión o un giro rápido) y le asignó un ID nuevo. Solución: ajusta el rastreador como se explica en el curso de producción y haz que la lógica por ID tolere interrupciones breves: la etiqueta del clasificador almacenada en caché simplemente se vuelve a calcular para el ID nuevo.
Mostrar solución
Añade save_crop=True a la llamada al detector:
result = detector("https://ultralytics.com/images/bus.jpg", conf=0.5, save_crop=True)[0]Los recortes aparecen en runs/detect/predict*/crops/person/ y crops/bus/ (el nombre de la carpeta de ejecución aumenta con cada predicción). La caché por seguimiento realizó 3 llamadas al clasificador para 33 cajas de personas, lo que supone un ahorro de 30 llamadas en un clip de 11 fotogramas.
Para contar los ángulos de rodilla, sustituye el bucle final de pose_track.py:
for track_id, angles in sorted(knee_angles.items()):
bent = sum(a < 120 for a in angles)
print(f"id {track_id}: {len(angles)} frames, {bent} with knee below 120 deg")id 1: 11 frames, 0 with knee below 120 deg
id 2: 11 frames, 11 with knee below 120 deg
id 3: 11 frames, 0 with knee below 120 degEl ID 2, correspondiente a la persona sentada, tiene un ángulo inferior a 120° en los 11 fotogramas; los ID 1 y 3 nunca lo tienen.