1. Introdução e Contexto Técnico

A detecção de máscaras em tempo real é uma aplicação crucial em sistemas de visão computacional, exigindo alto desempenho para processar fluxos de vídeo contínuos. Uma implementação monolítica que executa a leitura da imagem, o pré-processamento e a inferência de forma sequencial frequentemente não consegue atender aos requisitos de latência, especialmente com múltiplas câmeras ou resoluções elevadas.

Este artigo detalha a arquitetura de uma solução otimizada que segmenta o pipeline em estágios paralelos. A ideia central é isolar as operações intensivas em CPU (como redimensionamento e normalização) das operações na GPU (inferência do modelo neural), utilizando um paradigma de produtor-consumidor para coordenar o fluxo de dados entre elas. Isso permite que a GPU inicie a análise de uma imagem enquanto a CPU já está preparando a próxima, maximizando a utilização de ambos os recursos computacionais.

2. Fundamentos do Framework DAMO-YOLO

A base do sistema de detecção é o framework DAMO-YOLO, projetado para equilibrar precisão e eficiência em ambientes de produção. Suas principais características incluem:

• Precisão Aprimorada: Utiliza uma arquitetura com "large neck, small head" para uma fusão eficiente de características espaciais de baixo nível e semânticas de alto nível.
• Velocidade de Inferência: A estrutura de rede é otimizada para operar com baixa latência em hardware dedicado.
• Adaptabilidade Industrial: Suporte nativo a diversas plataformas e cenários de implantação.

Para explorar todo o potencial do framework, implementamos uma fila de processamento que desacopla completamente a aquisição, o pré-processamento e a inferência.

3. Configuração do Ambiente

Preparar o ambiente é o primeiro passo. Assegure-se de ter os pacotes essenciais instalados:

# Cria e ativa um ambiente virtual isolado
python -m venv deteccao_mascara_venv
source deteccao_mascara_venv/bin/activate

# Instala as bibliotecas centrais
pip install torch torchvision
pip install modelscope numpy opencv-python pillow

O modelo pré-treinado pode ser carregado diretamente a partir do repositório de modelos:

from modelscope import snapshot_download

caminho_modelo = snapshot_download('damo/cv_tinynas_object-detection_damoyolo_mask')

4. Construção do Pipelien Multithread

A implementação segue uma arquitetura de fila compartilhada para comunicação entre threads.

4.1 Gerenciamento de Dados com Filas

Definimos uma classe que encapsula as filas de comunicação e os controles de fluxo:

import threading
import queue
import cv2
import numpy as np
import torch

class ProcessadorVideo:
    def __init__(self, capacidade_fila=15):
        self.fila_quadros_brutos = queue.Queue(maxsize=capacidade_fila)
        self.fila_quadros_preparados = queue.Queue(maxsize=capacidade_fila)
        self.evento_parada = threading.Event()

    def capturar_quadros(self, fonte_video):
        """Thread dedicada à leitura do fluxo de vídeo."""
        captura = cv2.VideoCapture(fonte_video)
        while not self.evento_parada.is_set():
            sucesso, quadro = captura.read()
            if not sucesso:
                break
            try:
                # Enfileira o quadro para pré-processamento
                self.fila_quadros_brutos.put(quadro, block=True, timeout=2.0)
            except queue.Full:
                # Descarta o quadro se a fila estiver cheia
                continue
        captura.release()

4.2 Função de Pré-processamento para a CPU

Cada thread de pré-processamento consome quadros brutos e os transforma em tensores prontos para a GPU:

    def processar_para_tensor(self, identificador):
        """Thread do pool de pré-processamento."""
        while not self.evento_parada.is_set():
            try:
                quadro_bruto = self.fila_quadros_brutos.get(timeout=1.0)
                
                # Etapas de transformação da imagem
                quadro_redimensionado = cv2.resize(quadro_bruto, (640, 640), interpolation=cv2.INTER_LINEAR)
                quadro_float = (quadro_redimensionado.astype(np.float32)) / 255.0
                quadro_hwc_to_chw = np.ascontiguousarray(quadro_float.transpose(2, 0, 1))
                
                # Converte para tensor e move para a GPU (operação de cópia síncrona)
                tensor_saida = torch.from_numpy(quadro_hwc_to_chw).unsqueeze(0).cuda()
                
                # Enfileira o tensor processado para a inferência
                self.fila_quadros_preparados.put(tensor_saida, timeout=1.0)
                self.fila_quadros_brutos.task_done()
            except queue.Empty:
                continue

4.3 Execução da Inferência na GPU

Uma thread específica irá consumir os tensores preparados e executar o modelo:

    def executar_inferencia(self, modelo_carga):
        """Thread única de inferência na GPU."""
        modelo_carga.eval()
        while not self.evento_parada.is_set():
            try:
                tensor_entrada = self.fila_quadros_preparados.get(timeout=1.0)
                
                with torch.no_grad():
                    saida_modelo = modelo_carga(tensor_entrada)
                
                # Pós-processamento (ex: NMS) e uso dos resultados...
                resultados_finais = self._pos_processar(saida_modelo)
                self._exibir_ou_salvar(resultados_finais)
                
                self.fila_quadros_preparados.task_done()
            except queue.Empty:
                continue

    def _pos_processar(self, saida_bruta):
        # Lógica específica para converter a saída do modelo em bounding boxes, etc.
        # Esta parte depende do formato de saída do DAMO-YOLO.
        pass

5. Estratégias para Maximizar o Desempenho

Para extrair o máximo desempenho, são necessárias algumas otimizações adicionais.

5.1 Ajuste Dinâmico de Recursos

A quantidade ideal de threads para pré-processamento varia conforme o hardware:

import os

def definir_configuracao_threads(self):
    nucleos_cpu = os.cpu_count()
    gpus_disponiveis = torch.cuda.device_count()
    
    # Reserva um núcleo para a thread principal e de captura
    self.num_threads_preprocessamento = max(1, nucleos_cpu - 2)
    self.num_threads_inferencia = gpus_disponiveis
    
    print(f"Configuração: {self.num_threads_preprocessamento} threads de CPU, {self.num_threads_inferencia} thread(s) de GPU.")

5.2 Monitoramento e Balanceamento

Um monitor periódico pode prevenir gargalos, ajustando o tamanho das filas ou o número de threads ativas:

    def monitorar_filas(self):
        while not self.evento_parada.is_set():
            tamanho_bruto = self.fila_quadros_brutos.qsize()
            tamanho_preparado = self.fila_quadros_preparados.qsize()
            
            # Exemplo de lógica: se a fila de processamento estiver muito cheia, pode-se sinalizar para a captura reduzir a taxa
            if tamanho_preparado > self.fila_quadros_preparados.maxsize * 0.8:
                print("Aviso: Fila de inferência saturada. Possível atraso.")
            
            time.sleep(3.0)

6. Resultados e Métricas de Desempenho

A comparação entre uma abordagem sequencial e o pipeline paralelo revela melhorias significativas, especialmente em cenários complexos:

O ganho mais expressivo ocorre com múltiplas entradas, pois o pipeline consegue manter tanto a CPU quanto a GPU continuamente ocupadas. A utilização média da GPU subiu de ~45% para ~90% com a implementação paralela.

7. Tratamento de Recursos e Encerramento

Uma finalização correta é essencial para evitar vazamentos de memória ou travamentos:

    def encerrar_sistema(self):
        """Procedimento seguro de encerramento."""
        self.evento_parada.set()
        
        # Espera as threads esvaziarem as filas de forma controlada
        self.fila_quadros_brutos.join()
        self.fila_quadros_preparados.join()
        
        # Libera memória da GPU explicitamente
        torch.cuda.empty_cache()
        print("Pipeline encerrado com sucesso.")

8. Conclusão e Extensões

A arquitetura de pipeline com paralelismo CPU-GPU demonstrada é uma técnica poderosa e generalizável para otimizar sistemas de visão computacional em tempo real. O princípio de desacoplar operações com perfis de recursos distintos e coordená-las através de filas seguras resulta em um ganho de desempenho substancial.

Extensões naturais deste trabalho incluem a integração com aceleradores como TensorRT para otimizar ainda mais o estágio de inferência, ou a implementação de um sistema de escalonamento dinâmico que varie o número de threads de pré-processamento em tempo real com base na carga das filas, garantindo eficiência mesmo sob demanda variável.