Combinar tarefas em um pipeline
Crie uma cascata de detecção seguida de classificação, classifique cada objeto rastreado apenas uma vez em vez de a cada quadro e transforme os pontos-chave de pose em séries temporais por pessoa.
Os sistemas mais úteis além da detecção geralmente usam dois modelos, não um. Um detector encontra cada trabalhador e um classificador decide se cada capacete é do tipo correto; um modelo de pose encontra as articulações e um rastreador as transforma em uma linha do tempo a partir da qual você pode contar repetições. O código é curto. O que costuma dar errado nesses sistemas é um conjunto de dados da segunda etapa que não corresponde aos recortes da primeira etapa e um custo que se multiplica por cada objeto em cada quadro — executado sem otimização, um fluxo de 30 FPS com dez pessoas custa 300 classificações por segundo.
Prático
Dois padrões que atendem à maioria dos produtos#
Ambos começam com o mesmo quadro e a mesma API da Ultralytics; diferem no que a segunda etapa consome. A seleção e o ajuste do rastreador, assim como o tratamento do fluxo para o rastreamento de objetos, fazem parte de Rastrear objetos com ByteTrack, BoT-SORT e TrackTrack; aqui, o foco é combinar as tarefas.
Detectar e, em seguida, classificar os 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)A integração está correta: os recortes vêm de orig_img na resolução original, nas coordenadas de pixels do próprio detector, e o classificador os processa todos em uma única chamada em lote. Os rótulos são absurdos de propósito.
O classificador ImageNet nunca foi treinado com figuras de rua muito recortadas, parcialmente cortadas, então ele faz suposições — observe que nenhuma confiança é superior a 0.50. O mesmo acontece com um classificador treinado com fotos nítidas de produtos que recebe recortes borrados do detector.
Criar o conjunto de dados da segunda etapa a partir dos seus próprios recortes#
A maneira confiável de obter dados correspondentes é deixar o detector gerá-los:
- Execute o detector em imagens representativas com
save_crop=True, por exemplo,yolo predict model=best.pt source=footage/ save_crop=True. Os recortes são salvos emruns/detect/predict/crops/<class>/. - Organize os recortes de cada classe nas pastas de classificação detalhada (
helmet_ok/,helmet_missing/) — essa já é a estrutura de classificação da lição anterior. - Treine o classificador nessas pastas e valide-o com recortes de outro dia ou outra câmera.
Como o detector extraiu os recortes, eles mantêm o enquadramento, o desfoque e as oclusões reais — exatamente o que o classificador verá em produção — e o ajuste fino neles é um exemplo comum de aprendizado por transferência.
Classificar uma vez por ID de rastreamento#
Um capacete não muda de um quadro para outro, então classificar cada caixa em cada quadro é trabalho desperdiçado. Rastreie as detecções e classifique cada novo ID de rastreamento uma única 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)}Três chamadas ao classificador, em vez de 33, em um clipe de 11 quadros; em um fluxo longo, a economia aumenta a cada quadro em que uma pessoa permanece visível. Os rótulos ainda são previsões do ImageNet — substitua-as pelo seu classificador ajustado e elas se tornarão as respostas para cada pessoa.
Classifique uma vez por trajetória quando o atributo não puder mudar enquanto o objeto estiver visível (tipo de capacete, marca do veículo, SKU do produto). Reclassifique a cada N quadros quando puder mudar (porta aberta, mão levantada). Quando as primeiras aparições tendem a ser pequenas e borradas, classifique no quadro em que a caixa da trajetória é maior.
Pose e rastreamento#
O modo Track atribui um result.boxes.id persistente a cada pessoa, transformando os pontos-chave em uma série temporal. Este script registra o ângulo do joelho esquerdo de cada pessoa ao longo do mesmo vídeo curto:
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 degO joelho dobrado da ID 2 corresponde à pessoa sentada no clipe; as outras duas estão de pé, com as pernas esticadas, perto de 180°. A partir daqui, um contador de repetições verifica se a série de cada ID cruza determinados limites — desce abaixo de um ângulo e volta a subir acima de outro —, algo que a solução de monitoramento de treinos oferece pronta para uso.
Calcular o custo do pipeline#
| Fator de custo | Detectar e depois classificar | Pose e rastreamento |
|---|---|---|
| Escala com | objetos por quadro | quadros |
| Principal alavanca | recorta em lote; classifica uma vez por ID de rastreamento e armazena em cache | reduz imgsz ou vid_stride quando o movimento é lento |
| Modo de falha | o classificador recebe recortes diferentes dos dados de treino | as trocas de ID dividem o histórico de uma pessoa em dois |
Sintoma: um contador de repetições zera no meio da série, ou uma pessoa aparece com dois IDs. Causa: o rastreador perdeu a pessoa (oclusão, uma virada rápida) e atribuiu um novo ID. Solução: ajuste o rastreador conforme o curso de produção e faça com que a lógica por ID tolere lacunas curtas — um rótulo de classificador em cache simplesmente é recalculado para o novo ID.
Mostrar solução
Adiciona save_crop=True à chamada do detector:
result = detector("https://ultralytics.com/images/bus.jpg", conf=0.5, save_crop=True)[0]Os recortes aparecem em runs/detect/predict*/crops/person/ e crops/bus/ (o nome da pasta da execução aumenta a cada execução de predição). O armazenamento em cache por ID de rastreamento fez 3 chamadas ao classificador para 33 caixas de pessoas, poupando 30 chamadas num clipe de 11 frames.
Para contar os ângulos do joelho, substitui o ciclo 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 degO ID 2, correspondente à pessoa sentada, está abaixo de 120° em todos os 11 frames; os IDs 1 e 3 nunca estão.