feat: versão inicial limpa

.env.example

@@ -0,0 +1,52 @@
+# Configurações da Aplicação
+DEBUG=False
+PORT=7860
+HOST=0.0.0.0
+
+# Configurações do Modelo
+MODEL_THRESHOLD=0.4
+TARGET_FPS=2
+
+# Configurações de Cache
+CACHE_DIR=/tmp/weapon_detection_cache
+MAX_CACHE_SIZE=1000000000  # 1GB em bytes
+RESULT_CACHE_SIZE=1000
+
+# Configurações de Upload
+MAX_CONTENT_LENGTH=100000000  # 100MB em bytes
+ALLOWED_EXTENSIONS=mp4,avi,mov,webm
+
+# Configurações de Processamento
+MAX_VIDEO_DURATION=300  # 5 minutos em segundos
+MIN_CONFIDENCE=0.3
+MAX_DETECTIONS_PER_FRAME=10
+
+# Configurações de E-mail
+EMAIL_APP_PASSWORD=sua_senha_de_app_aqui
+NOTIFICATION_EMAIL=""  # E-mail para envio de notificações
+
+# Configurações do Modelo
+HUGGING_FACE_TOKEN=""  # Token do Hugging Face para acesso aos modelos
+TOKENIZERS_PARALLELISM=false
+MODEL_CACHE_DIR=./.model_cache
+BATCH_SIZE=16
+MAX_WORKERS=2
+USE_HALF_PRECISION=true
+DETECTION_CONFIDENCE_THRESHOLD=0.5
+MODEL_CONFIDENCE_THRESHOLD=0.5
+MODEL_IOU_THRESHOLD=0.45
+
+# Configurações do Servidor
+SERVER_HOST=0.0.0.0
+SERVER_PORT=7860
+ENABLE_SHARING=true  # true para ambiente local, false para Hugging Face
+
+# Configurações de Vídeo
+DEFAULT_FPS=2
+DEFAULT_RESOLUTION=640
+
+# Configurações de GPU (opcional - apenas se tiver GPU)
+CUDA_VISIBLE_DEVICES=0
+TORCH_CUDA_ARCH_LIST="7.5"
+NVIDIA_VISIBLE_DEVICES=all
+NVIDIA_DRIVER_CAPABILITIES=compute,utility

.env.huggingface

@@ -0,0 +1,33 @@
+# Configurações do Modelo
+HUGGING_FACE_TOKEN=""  # Configure no Hugging Face Space
+TOKENIZERS_PARALLELISM=false
+MODEL_CACHE_DIR=./.model_cache
+BATCH_SIZE=16
+MAX_WORKERS=2
+USE_HALF_PRECISION=true
+DETECTION_CONFIDENCE_THRESHOLD=0.5
+MODEL_CONFIDENCE_THRESHOLD=0.5
+MODEL_IOU_THRESHOLD=0.45
+
+# Configurações de Cache
+CACHE_DIR=/code/.cache/weapon_detection_cache
+RESULT_CACHE_SIZE=1000
+
+# Configurações de E-mail
+NOTIFICATION_EMAIL=""  # Configure no Hugging Face Space
+SENDGRID_API_KEY=""  # Configure no Hugging Face Space
+
+# Configurações do Servidor
+SERVER_HOST=0.0.0.0
+SERVER_PORT=7860
+ENABLE_SHARING=false
+
+# Configurações de Vídeo
+DEFAULT_FPS=2
+DEFAULT_RESOLUTION=640
+
+# Configurações de GPU
+CUDA_VISIBLE_DEVICES=0
+TORCH_CUDA_ARCH_LIST="7.5"
+NVIDIA_VISIBLE_DEVICES=all
+NVIDIA_DRIVER_CAPABILITIES=compute,utility 

.gitignore

@@ -0,0 +1,62 @@
+# Ambiente virtual
+.venv/
+venv/
+ENV/
+env/
+.env/
+.python-version
+
+# Arquivos Python
+__pycache__/
+*.py[cod]
+*$py.class
+*.so
+build/
+dist/
+*.egg-info/
+
+# Arquivos de ambiente
+.env
+.env.*
+!.env.example
+!.env.huggingface
+
+# IDE
+.idea/
+.vscode/
+*.swp
+*.swo
+
+# Sistema
+.DS_Store
+Thumbs.db
+
+# Cache e logs
+*.log
+.cache/
+.pytest_cache/
+.model_cache/
+
+# Hugging Face
+.huggingface/
+.hf-cache/
+.transformers/
+.torch/
+
+# Arquivos grandes
+videos/
+*.mp4
+*.avi
+*.mov
+*.mkv
+*.webm
+*.pt
+*.pth
+*.onnx
+*.tflite
+*.h5
+*.model
+*.bin
+*.tar
+*.gz
+*.zip 

Dockerfile

@@ -0,0 +1,64 @@
+FROM nvidia/cuda:12.1.0-cudnn8-runtime-ubuntu22.04
+
+# Configurar ambiente não interativo
+ENV DEBIAN_FRONTEND=noninteractive
+
+WORKDIR /code
+
+# Instalar Python e dependências do sistema
+RUN apt-get update && apt-get install -y \
+    python3.10 \
+    python3-pip \
+    python3.10-venv \
+    libgl1-mesa-glx \
+    libglib2.0-0 \
+    ffmpeg \
+    git \
+    git-lfs \
+    && apt-get clean \
+    && rm -rf /var/lib/apt/lists/*
+
+# Inicializar Git LFS
+RUN git lfs install
+
+# Copiar requirements primeiro para aproveitar cache
+COPY requirements.txt .
+
+# Configurar ambiente Python
+ENV VIRTUAL_ENV=/opt/venv
+RUN python3.10 -m venv $VIRTUAL_ENV
+ENV PATH="$VIRTUAL_ENV/bin:$PATH"
+
+# Instalar dependências Python otimizadas para GPU
+RUN pip install --no-cache-dir -U pip && \
+    pip install --no-cache-dir torch torchvision --extra-index-url https://download.pytorch.org/whl/cu121 && \
+    pip install --no-cache-dir -r requirements.txt
+
+# Criar diretório de vídeos e cache
+RUN mkdir -p /code/videos /code/.cache/huggingface /code/.cache/torch && \
+    chmod -R 777 /code/.cache /code/videos
+
+# Configurar variáveis de ambiente
+ENV HOST=0.0.0.0 \
+    PORT=7860 \
+    PYTHONUNBUFFERED=1 \
+    PYTHONPATH=/code \
+    TRANSFORMERS_CACHE=/code/.cache/huggingface \
+    TORCH_HOME=/code/.cache/torch \
+    GRADIO_SERVER_NAME=0.0.0.0 \
+    GRADIO_SERVER_PORT=7860 \
+    SYSTEM=spaces \
+    CUDA_VISIBLE_DEVICES=0 \
+    HUGGINGFACE_HUB_CACHE=/code/.cache/huggingface \
+    HF_HOME=/code/.cache/huggingface \
+    TORCH_CUDA_ARCH_LIST="7.5" \
+    MAX_WORKERS=2
+
+# Copiar arquivos do projeto
+COPY . .
+
+# Expor porta
+EXPOSE 7860
+
+# Comando para iniciar a aplicação
+CMD ["python3", "app.py"]

README.md

@@ -0,0 +1,111 @@
+---
+title: Weapon Detection App
+emoji: 🚨
+colorFrom: red
+colorTo: yellow
+sdk: gradio
+sdk_version: 5.15.0
+app_file: app.py
+pinned: false
+license: mit
+hardware: true
+resources:
+  accelerator: T4
+  gpu: true
+---
+
+# Sistema de Detecção de Riscos em Vídeo
+
+Este projeto implementa um sistema de detecção de riscos em vídeo utilizando YOLOv8 e Clean Architecture.
+
+## Pré-requisitos
+
+- Python 3.9 ou superior
+- pip (gerenciador de pacotes Python)
+- Ambiente virtual Python (recomendado)
+
+## Configuração do Ambiente
+
+1. Clone o repositório:
+```bash
+git clone [URL_DO_REPOSITORIO]
+cd [NOME_DO_DIRETORIO]
+```
+
+2. Crie e ative um ambiente virtual:
+```bash
+python -m venv .venv
+source .venv/bin/activate  # Linux/Mac
+# OU
+.venv\Scripts\activate  # Windows
+```
+
+3. Instale as dependências:
+```bash
+pip install -r requirements.txt
+```
+
+4. Configure as variáveis de ambiente:
+Crie um arquivo `.env` na raiz do projeto com as seguintes variáveis:
+```
+NOTIFICATION_API_KEY=sua_chave_api
+```
+
+## Executando o Projeto
+
+1. Ative o ambiente virtual (se ainda não estiver ativo)
+
+2. Execute o aplicativo:
+```bash
+python src/main.py
+```
+
+3. Acesse a interface web através do navegador no endereço mostrado no terminal (geralmente http://localhost:7860)
+
+## Funcionalidades
+
+- Upload de vídeos para análise
+- Detecção de objetos em tempo real
+- Configuração de parâmetros de detecção
+- Sistema de notificações
+- Monitoramento de recursos do sistema
+
+## Estrutura do Projeto
+
+O projeto segue os princípios da Clean Architecture:
+
+- `domain/`: Regras de negócio e entidades
+- `application/`: Casos de uso e interfaces
+- `infrastructure/`: Implementações concretas
+- `presentation/`: Interface com usuário (Gradio)
+
+## Contribuindo
+
+1. Fork o projeto
+2. Crie uma branch para sua feature (`git checkout -b feature/AmazingFeature`)
+3. Commit suas mudanças (`git commit -m 'Add some AmazingFeature'`)
+4. Push para a branch (`git push origin feature/AmazingFeature`)
+5. Abra um Pull Request
+
+## Tecnologias
+
+- Python 3.8+
+- PyTorch com CUDA
+- OWL-ViT
+- Gradio
+- FFmpeg
+
+## Requisitos de Hardware
+
+- GPU NVIDIA T4 (fornecida pelo Hugging Face)
+- 16GB de RAM
+- Armazenamento para cache de modelos
+
+## Limitações
+
+- Processamento pode ser lento em CPUs menos potentes
+- Requer GPU para melhor performance
+- Alguns falsos positivos em condições de baixa luz
+
+---
+Desenvolvido com ❤️ para o Hackathon FIAP

app.py

@@ -0,0 +1,31 @@
+import os
+import logging
+from src.main import main
+from dotenv import load_dotenv
+
+# Configurar logging
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+)
+
+logger = logging.getLogger(__name__)
+
+if __name__ == "__main__":
+    try:
+        # Verificar se está rodando no Hugging Face
+        IS_HUGGINGFACE = os.getenv('SPACE_ID') is not None
+        
+        # Carregar configurações do ambiente apropriado
+        if IS_HUGGINGFACE:
+            load_dotenv('.env.huggingface')
+            logger.info("Ambiente HuggingFace detectado")
+        else:
+            load_dotenv('.env')
+            logger.info("Ambiente local detectado")
+            
+        # Iniciar aplicação
+        main()
+    except Exception as e:
+        logger.error(f"Erro ao iniciar aplicação: {str(e)}")
+        raise 

pytest.ini

@@ -0,0 +1,19 @@
+[pytest]
+python_files = test_*.py
+python_classes = Test*
+python_functions = test_*
+
+addopts = -v -s --strict-markers
+
+markers =
+    slow: marks tests as slow (deselect with '-m "not slow"')
+    integration: marks tests as integration tests
+    unit: marks tests as unit tests
+
+testpaths =
+    tests
+
+log_cli = true
+log_cli_level = INFO
+log_cli_format = %(asctime)s [%(levelname)8s] %(message)s (%(filename)s:%(lineno)s)
+log_cli_date_format = %Y-%m-%d %H:%M:%S 

requirements.txt

@@ -0,0 +1,77 @@
+accelerate==1.3.0
+aiofiles==23.2.1
+annotated-types==0.7.0
+anyio==4.8.0
+blinker==1.9.0
+certifi==2024.12.14
+charset-normalizer==3.4.1
+click==8.1.8
+fastapi==0.115.7
+ffmpeg-python==0.2.0
+ffmpy==0.5.0
+filelock==3.17.0
+Flask==3.1.0
+fsspec==2024.12.0
+future==1.0.0
+gradio==5.15.0
+gradio_client==1.7.0
+h11==0.14.0
+httpcore==1.0.7
+httpx==0.28.1
+huggingface-hub==0.28.1
+idna==3.10
+itsdangerous==2.2.0
+Jinja2==3.1.5
+markdown-it-py==3.0.0
+MarkupSafe==2.1.5
+mdurl==0.1.2
+mpmath==1.3.0
+networkx==3.4.2
+numpy==2.2.2
+nvidia-ml-py3==7.352.0
+opencv-python==4.11.0.86
+opencv-python-headless==4.11.0.86
+orjson==3.10.15
+packaging==24.2
+pandas==2.2.3
+pillow==11.1.0
+psutil==5.9.5
+pydantic==2.10.6
+pydantic_core==2.27.2
+pydub==0.25.1
+Pygments==2.19.1
+python-dateutil==2.9.0.post0
+python-dotenv==1.0.1
+python-http-client==3.3.7
+python-multipart==0.0.20
+pytz==2024.2
+PyYAML==6.0.2
+regex==2024.11.6
+requests==2.32.3
+rich==13.9.4
+ruff==0.9.3
+safehttpx==0.1.6
+safetensors==0.5.2
+scipy==1.15.1
+semantic-version==2.10.0
+sendgrid==6.11.0
+setuptools==75.8.0
+shellingham==1.5.4
+six==1.17.0
+sniffio==1.3.1
+starkbank-ecdsa==2.2.0
+starlette==0.45.2
+sympy==1.13.1
+tokenizers==0.18.0
+tomlkit==0.13.2
+torch==2.5.1
+torchvision==0.20.1
+tqdm==4.67.1
+transformers==4.36.2
+typer==0.15.1
+typing_extensions==4.12.2
+tzdata==2025.1
+urllib3==2.3.0
+uvicorn==0.34.0
+websockets==14.2
+Werkzeug==3.1.3

src/application/use_cases/process_video.py

@@ -0,0 +1,168 @@
+from dataclasses import dataclass
+from typing import Optional, Dict, Any
+from ...domain.interfaces.detector import DetectorInterface
+from ...domain.interfaces.notification import NotificationFactory
+from ...domain.entities.detection import DetectionResult
+import logging
+
+logger = logging.getLogger(__name__)
+
+@dataclass
+class ProcessVideoRequest:
+    """DTO para requisição de processamento de vídeo."""
+    video_path: str
+    threshold: float = 0.5
+    fps: Optional[int] = None
+    resolution: Optional[int] = None
+    notification_type: Optional[str] = None
+    notification_target: Optional[str] = None
+
+@dataclass
+class ProcessVideoResponse:
+    """DTO para resposta do processamento de vídeo."""
+    status_message: str
+    detection_result: DetectionResult
+    memory_info: str
+    device_info: str
+    cache_stats: Optional[Dict[str, Any]] = None
+
+class ProcessVideoUseCase:
+    """Caso de uso para processamento de vídeo e notificação."""
+    
+    def __init__(
+        self,
+        detector: DetectorInterface,
+        notification_factory: NotificationFactory,
+        default_fps: int,
+        default_resolution: int
+    ):
+        self.detector = detector
+        self.notification_factory = notification_factory
+        self.default_fps = default_fps
+        self.default_resolution = default_resolution
+    
+    def execute(self, request: ProcessVideoRequest) -> ProcessVideoResponse:
+        """Executa o processamento do vídeo e envia notificações se necessário."""
+        try:
+            # Usar valores padrão se não especificados
+            fps = request.fps or self.default_fps
+            resolution = request.resolution or self.default_resolution
+            
+            # Processar vídeo
+            output_path, result = self.detector.process_video(
+                request.video_path,
+                fps=fps,
+                threshold=request.threshold,
+                resolution=resolution
+            )
+            
+            # Enviar notificação se houver detecções e destino configurado
+            if result.detections and request.notification_type and request.notification_target:
+                notification_service = self.notification_factory.create_service(request.notification_type)
+                if notification_service:
+                    detection_data = {
+                        'detections': [
+                            {
+                                'label': det.label,
+                                'confidence': det.confidence,
+                                'box': det.box,
+                                'timestamp': det.timestamp
+                            } for det in result.detections
+                        ],
+                        'technical': {
+                            'threshold': request.threshold,
+                            'fps': fps,
+                            'resolution': resolution
+                        }
+                    }
+                    notification_service.send_notification(detection_data, request.notification_target)
+            
+            # Formatar mensagem de status
+            status_msg = self._format_status_message(result)
+            
+            # Obter informações do sistema de forma segura
+            try:
+                device_info = self.detector.get_device_info() if hasattr(self.detector, 'get_device_info') else {}
+            except Exception as e:
+                logger.error(f"Erro ao obter informações do dispositivo: {str(e)}")
+                device_info = {}
+                
+            try:
+                cache_stats = self.detector.get_cache_stats() if hasattr(self.detector, 'get_cache_stats') else {}
+            except Exception as e:
+                logger.error(f"Erro ao obter estatísticas do cache: {str(e)}")
+                cache_stats = {}
+            
+            # Limpar memória
+            try:
+                self.detector.clean_memory()
+            except Exception as e:
+                logger.error(f"Erro ao limpar memória: {str(e)}")
+            
+            return ProcessVideoResponse(
+                status_message=status_msg,
+                detection_result=result,
+                memory_info=self._format_memory_info(device_info),
+                device_info=self._format_device_info(device_info),
+                cache_stats=cache_stats
+            )
+            
+        except Exception as e:
+            logger.error(f"Erro ao executar caso de uso: {str(e)}")
+            # Criar um resultado vazio em caso de erro
+            empty_result = DetectionResult(
+                video_path=request.video_path,
+                detections=[],
+                frames_analyzed=0,
+                total_time=0.0,
+                device_type="Unknown",
+                frame_extraction_time=0.0,
+                analysis_time=0.0
+            )
+            return ProcessVideoResponse(
+                status_message="Erro ao processar o vídeo. Por favor, tente novamente.",
+                detection_result=empty_result,
+                memory_info="N/A",
+                device_info="N/A",
+                cache_stats={}
+            )
+    
+    def _format_status_message(self, result: DetectionResult) -> str:
+        """Formata a mensagem de status do processamento."""
+        try:
+            status = "⚠️ RISCO DETECTADO" if result.detections else "✅ SEGURO"
+            
+            message = f"""Processamento concluído! ({result.device_type})
+            
+Status: {status}
+Detecções: {len(result.detections)}
+Frames analisados: {result.frames_analyzed}
+Tempo total: {result.total_time:.2f}s
+Tempo de extração: {result.frame_extraction_time:.2f}s
+Tempo de análise: {result.analysis_time:.2f}s"""
+
+            # Adicionar detalhes das detecções se houver
+            if result.detections:
+                message += "\n\nDetecções encontradas:"
+                for i, det in enumerate(result.detections[:3], 1):  # Mostrar até 3 detecções
+                    message += f"\n{i}. {det.label} (Confiança: {det.confidence:.1%}, Frame: {det.frame})"
+                if len(result.detections) > 3:
+                    message += f"\n... e mais {len(result.detections) - 3} detecção(ões)"
+            
+            return message
+            
+        except Exception as e:
+            logger.error(f"Erro ao formatar mensagem de status: {str(e)}")
+            return "Erro ao processar o vídeo. Por favor, tente novamente."
+    
+    def _format_memory_info(self, device_info: Dict[str, Any]) -> str:
+        if device_info.get('type') == 'GPU':
+            return f"GPU: {device_info.get('memory_used', 0) / 1024**2:.1f}MB / {device_info.get('memory_total', 0) / 1024**2:.1f}MB"
+        else:
+            return f"RAM: {device_info.get('memory_used', 0) / 1024**2:.1f}MB / {device_info.get('memory_total', 0) / 1024**2:.1f}MB"
+    
+    def _format_device_info(self, device_info: Dict[str, Any]) -> str:
+        if device_info.get('type') == 'GPU':
+            return f"GPU: {device_info.get('name', 'Unknown')}"
+        else:
+            return f"CPU Threads: {device_info.get('threads', 'N/A')}" 

src/domain/detectors/base.py

@@ -0,0 +1,200 @@
+from abc import ABC, abstractmethod
+import gc
+import torch
+import logging
+from typing import Dict, Any, Optional, List, Tuple
+import os
+import cv2
+from PIL import Image
+import time
+
+logger = logging.getLogger(__name__)
+
+class BaseCache:
+    """Cache base para armazenar resultados de detecção."""
+    def __init__(self, max_size: int = 1000):
+        self.cache = {}
+        self.max_size = max_size
+        self.hits = 0
+        self.misses = 0
+        self.last_access = {}
+
+    def get(self, key: str) -> Optional[Dict]:
+        try:
+            if key in self.cache:
+                self.hits += 1
+                self.last_access[key] = time.time()
+                return self.cache[key]
+            self.misses += 1
+            return None
+        except Exception as e:
+            logger.error(f"Erro ao recuperar do cache: {str(e)}")
+            return None
+
+    def put(self, key: str, results: Dict):
+        try:
+            if len(self.cache) >= self.max_size:
+                oldest_key = min(self.last_access.items(), key=lambda x: x[1])[0]
+                del self.cache[oldest_key]
+                del self.last_access[oldest_key]
+            self.cache[key] = results
+            self.last_access[key] = time.time()
+        except Exception as e:
+            logger.error(f"Erro ao armazenar no cache: {str(e)}")
+
+    def clear(self):
+        """Limpa o cache e libera memória."""
+        self.cache.clear()
+        self.last_access.clear()
+        gc.collect()
+
+    def get_stats(self) -> dict:
+        total = self.hits + self.misses
+        hit_rate = (self.hits / total) * 100 if total > 0 else 0
+        return {
+            "cache_size": len(self.cache),
+            "max_size": self.max_size,
+            "hits": self.hits,
+            "misses": self.misses,
+            "hit_rate": f"{hit_rate:.2f}%",
+            "memory_usage": sum(sys.getsizeof(v) for v in self.cache.values())
+        }
+
+class BaseDetector(ABC):
+    """Classe base abstrata para detectores de objetos perigosos."""
+    def __init__(self):
+        self._initialized = False
+        self.device = None
+        self.owlv2_model = None
+        self.owlv2_processor = None
+        self.text_queries = None
+        self.processed_text = None
+        self.threshold = 0.3
+        self.result_cache = None
+        
+    @abstractmethod
+    def _initialize(self):
+        """Inicializa o modelo e o processador."""
+        pass
+        
+    @abstractmethod
+    def _get_best_device(self):
+        """Retorna o melhor dispositivo disponível."""
+        pass
+        
+    def initialize(self):
+        """Inicializa o detector se ainda não estiver inicializado."""
+        if not self._initialized:
+            self._initialize()
+            
+    def extract_frames(self, video_path: str, fps: int = None, resolution: int = 640) -> List:
+        """Extrai frames do vídeo com taxa e resolução especificadas."""
+        try:
+            if not os.path.exists(video_path):
+                logger.error(f"Arquivo de vídeo não encontrado: {video_path}")
+                return []
+                
+            cap = cv2.VideoCapture(video_path)
+            if not cap.isOpened():
+                logger.error("Erro ao abrir o vídeo")
+                return []
+                
+            original_fps = cap.get(cv2.CAP_PROP_FPS)
+            target_fps = fps if fps else min(2, original_fps)
+            frame_interval = int(original_fps / target_fps)
+            
+            frames = []
+            frame_count = 0
+            
+            while True:
+                ret, frame = cap.read()
+                if not ret:
+                    break
+                    
+                if frame_count % frame_interval == 0:
+                    if resolution:
+                        height, width = frame.shape[:2]
+                        scale = resolution / max(height, width)
+                        if scale < 1:
+                            new_width = int(width * scale)
+                            new_height = int(height * scale)
+                            frame = cv2.resize(frame, (new_width, new_height))
+                    frames.append(frame)
+                    
+                frame_count += 1
+                
+            cap.release()
+            return frames
+            
+        except Exception as e:
+            logger.error(f"Erro ao extrair frames: {str(e)}")
+            return []
+            
+    @abstractmethod
+    def detect_objects(self, image: Image.Image, threshold: float = 0.3) -> List[Dict]:
+        """Detecta objetos em uma imagem."""
+        pass
+        
+    @abstractmethod
+    def process_video(self, video_path: str, fps: int = None, threshold: float = 0.3, resolution: int = 640) -> Tuple[str, Dict]:
+        """Processa um vídeo para detecção de objetos."""
+        pass
+
+    def clean_memory(self):
+        """Limpa memória não utilizada."""
+        try:
+            if torch.cuda.is_available():
+                torch.cuda.empty_cache()
+                logger.debug("Cache GPU limpo")
+            gc.collect()
+            logger.debug("Garbage collector executado")
+        except Exception as e:
+            logger.error(f"Erro ao limpar memória: {str(e)}")
+
+    def _get_detection_queries(self) -> List[str]:
+        """Retorna as queries otimizadas para detecção de objetos perigosos."""
+        firearms = ["handgun", "rifle", "shotgun", "machine gun", "firearm"]
+        edged_weapons = ["knife", "dagger", "machete", "box cutter", "sword"]
+        ranged_weapons = ["crossbow", "bow"]
+        sharp_objects = ["blade", "razor", "glass shard", "screwdriver", "metallic pointed object"]
+        
+        firearm_contexts = ["close-up", "clear view", "detailed"]
+        edged_contexts = ["close-up", "clear view", "detailed", "metallic", "sharp"]
+        ranged_contexts = ["close-up", "clear view", "detailed"]
+        sharp_contexts = ["close-up", "clear view", "detailed", "sharp"]
+        
+        queries = []
+        
+        for weapon in firearms:
+            queries.append(f"a photo of a {weapon}")
+            for context in firearm_contexts:
+                queries.append(f"a photo of a {context} {weapon}")
+        
+        for weapon in edged_weapons:
+            queries.append(f"a photo of a {weapon}")
+            for context in edged_contexts:
+                queries.append(f"a photo of a {context} {weapon}")
+        
+        for weapon in ranged_weapons:
+            queries.append(f"a photo of a {weapon}")
+            for context in ranged_contexts:
+                queries.append(f"a photo of a {context} {weapon}")
+        
+        for weapon in sharp_objects:
+            queries.append(f"a photo of a {weapon}")
+            for context in sharp_contexts:
+                queries.append(f"a photo of a {context} {weapon}")
+        
+        queries = sorted(list(set(queries)))
+        logger.info(f"Total de queries otimizadas geradas: {len(queries)}")
+        return queries
+
+    @abstractmethod
+    def _apply_nms(self, detections: List[Dict], iou_threshold: float = 0.5) -> List[Dict]:
+        """Aplica Non-Maximum Suppression nas detecções."""
+        pass
+
+    @abstractmethod
+    def _preprocess_image(self, image: Any) -> Any:
+        """Pré-processa a imagem para o formato adequado."""
+        pass 

src/domain/detectors/cpu.py

@@ -0,0 +1,587 @@
+import torch
+from transformers import Owlv2Processor, Owlv2ForObjectDetection
+from PIL import Image
+import numpy as np
+import cv2
+import time
+from typing import List, Dict, Tuple, Optional, Union
+import os
+from tqdm import tqdm
+import json
+from pathlib import Path
+from contextlib import nullcontext
+import threading
+import hashlib
+import pickle
+from datetime import datetime
+from dotenv import load_dotenv
+import tempfile
+import subprocess
+import shutil
+import traceback
+import psutil
+import logging
+import gc
+import sys
+from concurrent.futures import ThreadPoolExecutor
+import torch.nn.functional as F
+from .base import BaseDetector, BaseCache
+
+
+# Configurar logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# Carregar variáveis de ambiente
+load_dotenv()
+
+class CPUCache(BaseCache):
+    """Cache otimizado para CPU."""
+    def __init__(self, max_size: int = 1000):
+        super().__init__(max_size)
+        self.device = torch.device('cpu')
+
+class WeaponDetectorCPU(BaseDetector):
+    """Implementação CPU do detector de armas."""
+    def __init__(self):
+        """Inicializa variáveis básicas."""
+        super().__init__()
+        self.default_resolution = 640
+        self.device = torch.device('cpu')
+
+    def _get_best_device(self):
+        return torch.device('cpu')
+
+    def _initialize(self):
+        """Inicializa o modelo e o processador para execução em CPU."""
+        try:
+            # Configurações otimizadas para CPU
+            torch.set_num_threads(min(8, os.cpu_count()))
+            torch.set_num_interop_threads(min(8, os.cpu_count()))
+            
+            # Carregar modelo com configurações otimizadas
+            cache_dir = os.getenv('CACHE_DIR', '/tmp/weapon_detection_cache')
+            os.makedirs(cache_dir, exist_ok=True)
+            
+            model_name = "google/owlv2-base-patch16"
+            logger.info("Carregando modelo e processador...")
+            
+            self.owlv2_processor = Owlv2Processor.from_pretrained(
+                model_name,
+                cache_dir=cache_dir
+            )
+
+            self.owlv2_model = Owlv2ForObjectDetection.from_pretrained(
+                model_name,
+                cache_dir=cache_dir,
+                torch_dtype=torch.float32,
+                low_cpu_mem_usage=True
+            ).to(self.device)
+
+            self.owlv2_model.eval()
+
+            # Usar queries do método base
+            self.text_queries = self._get_detection_queries()
+            logger.info(f"Total de queries carregadas: {len(self.text_queries)}")
+            
+            # Processar queries uma única vez
+            logger.info("Processando queries...")
+            self.processed_text = self.owlv2_processor(
+                text=self.text_queries,
+                return_tensors="pt",
+                padding=True
+            ).to(self.device)
+
+            # Inicializar cache
+            cache_size = int(os.getenv('RESULT_CACHE_SIZE', '1000'))
+            self.result_cache = CPUCache(max_size=cache_size)
+
+            logger.info("Inicialização CPU completa!")
+            self._initialized = True
+
+        except Exception as e:
+            logger.error(f"Erro na inicialização CPU: {str(e)}")
+            raise
+
+    def _apply_nms(self, detections: list, iou_threshold: float = 0.5) -> list:
+        """Aplica NMS usando operações em CPU."""
+        try:
+            if not detections:
+                return []
+
+            boxes = torch.tensor([[d["box"][0], d["box"][1], d["box"][2], d["box"][3]] for d in detections])
+            scores = torch.tensor([d["confidence"] for d in detections])
+            labels = [d["label"] for d in detections]
+
+            area = (boxes[:, 2] - boxes[:, 0]) * (boxes[:, 3] - boxes[:, 1])
+            _, order = scores.sort(descending=True)
+
+            keep = []
+            while order.numel() > 0:
+                if order.numel() == 1:
+                    keep.append(order.item())
+                    break
+                i = order[0]
+                keep.append(i.item())
+
+                xx1 = torch.max(boxes[i, 0], boxes[order[1:], 0])
+                yy1 = torch.max(boxes[i, 1], boxes[order[1:], 1])
+                xx2 = torch.min(boxes[i, 2], boxes[order[1:], 2])
+                yy2 = torch.min(boxes[i, 3], boxes[order[1:], 3])
+
+                w = torch.clamp(xx2 - xx1, min=0)
+                h = torch.clamp(yy2 - yy1, min=0)
+                inter = w * h
+
+                ovr = inter / (area[i] + area[order[1:]] - inter)
+                ids = (ovr <= iou_threshold).nonzero().squeeze()
+                if ids.numel() == 0:
+                    break
+                order = order[ids + 1]
+
+            filtered_detections = []
+            for idx in keep:
+                filtered_detections.append({
+                    "confidence": scores[idx].item(),
+                    "box": boxes[idx].tolist(),
+                    "label": labels[idx]
+                })
+            return filtered_detections
+
+        except Exception as e:
+            logger.error(f"Erro ao aplicar NMS: {str(e)}")
+            return []
+
+    def _preprocess_image(self, image: Image.Image) -> Image.Image:
+        """Pré-processa a imagem para o tamanho 640x640 e garante RGB."""
+        try:
+            target_size = (640, 640)
+            if image.mode != 'RGB':
+                image = image.convert('RGB')
+            if image.size != target_size:
+                ratio = min(target_size[0] / image.size[0], target_size[1] / image.size[1])
+                new_size = tuple(int(dim * ratio) for dim in image.size)
+                image = image.resize(new_size, Image.LANCZOS)
+                if new_size != target_size:
+                    new_image = Image.new('RGB', target_size, (0, 0, 0))
+                    paste_x = (target_size[0] - new_size[0]) // 2
+                    paste_y = (target_size[1] - new_size[1]) // 2
+                    new_image.paste(image, (paste_x, paste_y))
+                    image = new_image
+            return image
+        except Exception as e:
+            logger.error(f"Erro no pré-processamento: {str(e)}")
+            return image
+
+    def detect_objects(self, image: Image.Image, threshold: float = 0.3) -> list:
+        """Detecta objetos em uma imagem utilizando CPU."""
+        try:
+            image = self._preprocess_image(image)
+            with torch.no_grad():
+                image_inputs = self.owlv2_processor(
+                    images=image,
+                    return_tensors="pt"
+                ).to(self.device)
+                inputs = {**image_inputs, **self.processed_text}
+                outputs = self.owlv2_model(**inputs)
+
+                target_sizes = torch.tensor([image.size[::-1]])
+                results = self.owlv2_processor.post_process_grounded_object_detection(
+                    outputs=outputs,
+                    target_sizes=target_sizes,
+                    threshold=threshold
+                )[0]
+
+                detections = []
+                for score, box, label in zip(results["scores"], results["boxes"], results["labels"]):
+                    x1, y1, x2, y2 = box.tolist()
+                    detections.append({
+                        "confidence": score.item(),
+                        "box": [int(x1), int(y1), int(x2), int(y2)],
+                        "label": self.text_queries[label]
+                    })
+                return self._apply_nms(detections)
+
+        except Exception as e:
+            logger.error(f"Erro em detect_objects: {str(e)}")
+            return []
+
+    def process_video(self, video_path: str, fps: int = None, threshold: float = 0.3, resolution: int = 640) -> tuple:
+        """Processa um vídeo utilizando CPU. Para na primeira detecção encontrada."""
+        try:
+            metrics = {
+                "total_time": 0,
+                "frame_extraction_time": 0,
+                "analysis_time": 0,
+                "frames_analyzed": 0,
+                "video_duration": 0,
+                "device_type": self.device.type,
+                "detections": [],
+                "technical": {
+                    "model": "owlv2-base-patch16-ensemble",
+                    "input_size": f"{resolution}x{resolution}",
+                    "nms_threshold": 0.5,
+                    "preprocessing": "basic",
+                    "early_stop": True
+                },
+            }
+
+            start_time = time.time()
+            t0 = time.time()
+            frames = self.extract_frames(video_path, fps, resolution)
+            metrics["frame_extraction_time"] = time.time() - t0
+            metrics["frames_analyzed"] = len(frames)
+
+            if not frames:
+                logger.warning("Nenhum frame extraído do vídeo")
+                return video_path, metrics
+
+            metrics["video_duration"] = len(frames) / (fps or 2)
+            t0 = time.time()
+            detections = []
+            frames_processed = 0
+
+            # Processar um frame por vez para otimizar memória e permitir parada precoce
+            for frame_idx, frame in enumerate(frames):
+                frames_processed += 1
+                
+                # Converter frame para RGB e pré-processar
+                frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
+                image = Image.fromarray(frame_rgb)
+                image = self._preprocess_image(image)
+
+                # Detectar objetos com threshold direto
+                with torch.no_grad():
+                    image_inputs = self.owlv2_processor(
+                        images=image,
+                        return_tensors="pt"
+                    ).to(self.device)
+                    inputs = {**image_inputs, **self.processed_text}
+                    outputs = self.owlv2_model(**inputs)
+
+                    target_sizes = torch.tensor([image.size[::-1]])
+                    results = self.owlv2_processor.post_process_grounded_object_detection(
+                        outputs=outputs,
+                        target_sizes=target_sizes,
+                        threshold=threshold  # Aplicar threshold diretamente
+                    )[0]
+
+                    # Se encontrou alguma detecção acima do threshold
+                    if len(results["scores"]) > 0:
+                        # Pegar a detecção com maior confiança
+                        max_score_idx = torch.argmax(results["scores"])
+                        score = results["scores"][max_score_idx].item()
+                        box = results["boxes"][max_score_idx].tolist()
+                        label = results["labels"][max_score_idx].item()
+
+                        detections.append({
+                            "frame": frame_idx,
+                            "confidence": score,
+                            "box": [int(x) for x in box],
+                            "label": self.text_queries[label]
+                        })
+
+                        # Atualizar métricas e parar o processamento
+                        metrics["frames_processed_until_detection"] = frames_processed
+                        metrics["analysis_time"] = time.time() - t0
+                        metrics["total_time"] = time.time() - start_time
+                        metrics["detections"] = detections
+                        logger.info(f"Detecção encontrada após processar {frames_processed} frames")
+                        return video_path, metrics
+
+                # Liberar memória a cada 10 frames
+                if frames_processed % 10 == 0:
+                    gc.collect()
+
+            # Se chegou aqui, não encontrou nenhuma detecção
+            metrics["analysis_time"] = time.time() - t0
+            metrics["total_time"] = time.time() - start_time
+            metrics["frames_processed_until_detection"] = frames_processed
+            metrics["detections"] = detections
+            return video_path, metrics
+
+        except Exception as e:
+            logger.error(f"Erro ao processar vídeo: {str(e)}")
+            return video_path, {}
+
+    def extract_frames(self, video_path: str, fps: int = 2, resolution: int = 480) -> list:
+        """Extrai frames de um vídeo utilizando ffmpeg."""
+        frames = []
+        temp_dir = Path(tempfile.mkdtemp())
+        try:
+            threads = min(os.cpu_count(), 4)  # Menor número de threads para CPU
+            cmd = [
+                'ffmpeg', '-i', video_path,
+                '-threads', str(threads),
+                '-vf', (f'fps={fps},'
+                        f'scale={resolution}:{resolution}:force_original_aspect_ratio=decrease:flags=lanczos,'
+                        f'pad={resolution}:{resolution}:(ow-iw)/2:(oh-ih)/2'),
+                '-frame_pts', '1',
+                f'{temp_dir}/%d.jpg'
+            ]
+            subprocess.run(cmd, check=True, capture_output=True)
+            frame_files = sorted(temp_dir.glob('*.jpg'), key=lambda x: int(x.stem))
+            chunk_size = 50  # Menor chunk size para CPU
+            with ThreadPoolExecutor(max_workers=threads) as executor:
+                for i in range(0, len(frame_files), chunk_size):
+                    chunk = frame_files[i:i + chunk_size]
+                    chunk_frames = list(tqdm(
+                        executor.map(lambda f: cv2.imread(str(f)), chunk),
+                        desc=f"Carregando frames {i+1}-{min(i+chunk_size, len(frame_files))}",
+                        total=len(chunk)
+                    ))
+                    frames.extend(chunk_frames)
+                    if i % (chunk_size * 5) == 0:
+                        gc.collect()
+        finally:
+            shutil.rmtree(temp_dir)
+        return frames
+
+    def clear_cache(self):
+        """Limpa o cache de resultados e libera memória."""
+        try:
+            if hasattr(self, 'result_cache'):
+                self.result_cache.clear()
+            gc.collect()
+            logger.info("Cache CPU limpo com sucesso")
+        except Exception as e:
+            logger.error(f"Erro ao limpar cache CPU: {str(e)}")
+
+    def _apply_nms(self, detections: list, iou_threshold: float = 0.5) -> list:
+        """Aplica NMS usando operações em CPU."""
+        try:
+            if not detections:
+                return []
+
+            boxes = torch.tensor([[d["box"][0], d["box"][1], d["box"][2], d["box"][3]] for d in detections])
+            scores = torch.tensor([d["confidence"] for d in detections])
+            labels = [d["label"] for d in detections]
+
+            area = (boxes[:, 2] - boxes[:, 0]) * (boxes[:, 3] - boxes[:, 1])
+            _, order = scores.sort(descending=True)
+
+            keep = []
+            while order.numel() > 0:
+                if order.numel() == 1:
+                    keep.append(order.item())
+                    break
+                i = order[0]
+                keep.append(i.item())
+
+                xx1 = torch.max(boxes[i, 0], boxes[order[1:], 0])
+                yy1 = torch.max(boxes[i, 1], boxes[order[1:], 1])
+                xx2 = torch.min(boxes[i, 2], boxes[order[1:], 2])
+                yy2 = torch.min(boxes[i, 3], boxes[order[1:], 3])
+
+                w = torch.clamp(xx2 - xx1, min=0)
+                h = torch.clamp(yy2 - yy1, min=0)
+                inter = w * h
+
+                ovr = inter / (area[i] + area[order[1:]] - inter)
+                ids = (ovr <= iou_threshold).nonzero().squeeze()
+                if ids.numel() == 0:
+                    break
+                order = order[ids + 1]
+
+            filtered_detections = []
+            for idx in keep:
+                filtered_detections.append({
+                    "confidence": scores[idx].item(),
+                    "box": boxes[idx].tolist(),
+                    "label": labels[idx]
+                })
+            return filtered_detections
+
+        except Exception as e:
+            logger.error(f"Erro ao aplicar NMS: {str(e)}")
+            return []
+
+    def _preprocess_image(self, image: Image.Image) -> Image.Image:
+        """Pré-processa a imagem para o tamanho 640x640 e garante RGB."""
+        try:
+            target_size = (640, 640)
+            if image.mode != 'RGB':
+                image = image.convert('RGB')
+            if image.size != target_size:
+                ratio = min(target_size[0] / image.size[0], target_size[1] / image.size[1])
+                new_size = tuple(int(dim * ratio) for dim in image.size)
+                image = image.resize(new_size, Image.LANCZOS)
+                if new_size != target_size:
+                    new_image = Image.new('RGB', target_size, (0, 0, 0))
+                    paste_x = (target_size[0] - new_size[0]) // 2
+                    paste_y = (target_size[1] - new_size[1]) // 2
+                    new_image.paste(image, (paste_x, paste_y))
+                    image = new_image
+            return image
+        except Exception as e:
+            logger.error(f"Erro no pré-processamento: {str(e)}")
+            return image
+
+    def detect_objects(self, image: Image.Image, threshold: float = 0.3) -> list:
+        """Detecta objetos em uma imagem utilizando CPU."""
+        try:
+            image = self._preprocess_image(image)
+            with torch.no_grad():
+                image_inputs = self.owlv2_processor(
+                    images=image,
+                    return_tensors="pt"
+                ).to(self.device)
+                inputs = {**image_inputs, **self.processed_text}
+                outputs = self.owlv2_model(**inputs)
+
+                target_sizes = torch.tensor([image.size[::-1]])
+                results = self.owlv2_processor.post_process_grounded_object_detection(
+                    outputs=outputs,
+                    target_sizes=target_sizes,
+                    threshold=threshold
+                )[0]
+
+                detections = []
+                for score, box, label in zip(results["scores"], results["boxes"], results["labels"]):
+                    x1, y1, x2, y2 = box.tolist()
+                    detections.append({
+                        "confidence": score.item(),
+                        "box": [int(x1), int(y1), int(x2), int(y2)],
+                        "label": self.text_queries[label]
+                    })
+                return self._apply_nms(detections)
+
+        except Exception as e:
+            logger.error(f"Erro em detect_objects: {str(e)}")
+            return []
+
+    def process_video(self, video_path: str, fps: int = None, threshold: float = 0.3, resolution: int = 640) -> tuple:
+        """Processa um vídeo utilizando CPU. Para na primeira detecção encontrada."""
+        try:
+            metrics = {
+                "total_time": 0,
+                "frame_extraction_time": 0,
+                "analysis_time": 0,
+                "frames_analyzed": 0,
+                "video_duration": 0,
+                "device_type": self.device.type,
+                "detections": [],
+                "technical": {
+                    "model": "owlv2-base-patch16-ensemble",
+                    "input_size": f"{resolution}x{resolution}",
+                    "nms_threshold": 0.5,
+                    "preprocessing": "basic",
+                    "early_stop": True
+                },
+            }
+
+            start_time = time.time()
+            t0 = time.time()
+            frames = self.extract_frames(video_path, fps, resolution)
+            metrics["frame_extraction_time"] = time.time() - t0
+            metrics["frames_analyzed"] = len(frames)
+
+            if not frames:
+                logger.warning("Nenhum frame extraído do vídeo")
+                return video_path, metrics
+
+            metrics["video_duration"] = len(frames) / (fps or 2)
+            t0 = time.time()
+            detections = []
+            frames_processed = 0
+
+            # Processar um frame por vez para otimizar memória e permitir parada precoce
+            for frame_idx, frame in enumerate(frames):
+                frames_processed += 1
+                
+                # Converter frame para RGB e pré-processar
+                frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
+                image = Image.fromarray(frame_rgb)
+                image = self._preprocess_image(image)
+
+                # Detectar objetos com threshold direto
+                with torch.no_grad():
+                    image_inputs = self.owlv2_processor(
+                        images=image,
+                        return_tensors="pt"
+                    ).to(self.device)
+                    inputs = {**image_inputs, **self.processed_text}
+                    outputs = self.owlv2_model(**inputs)
+
+                    target_sizes = torch.tensor([image.size[::-1]])
+                    results = self.owlv2_processor.post_process_grounded_object_detection(
+                        outputs=outputs,
+                        target_sizes=target_sizes,
+                        threshold=threshold  # Aplicar threshold diretamente
+                    )[0]
+
+                    # Se encontrou alguma detecção acima do threshold
+                    if len(results["scores"]) > 0:
+                        # Pegar a detecção com maior confiança
+                        max_score_idx = torch.argmax(results["scores"])
+                        score = results["scores"][max_score_idx].item()
+                        box = results["boxes"][max_score_idx].tolist()
+                        label = results["labels"][max_score_idx].item()
+
+                        detections.append({
+                            "frame": frame_idx,
+                            "confidence": score,
+                            "box": [int(x) for x in box],
+                            "label": self.text_queries[label]
+                        })
+
+                        # Atualizar métricas e parar o processamento
+                        metrics["frames_processed_until_detection"] = frames_processed
+                        metrics["analysis_time"] = time.time() - t0
+                        metrics["total_time"] = time.time() - start_time
+                        metrics["detections"] = detections
+                        logger.info(f"Detecção encontrada após processar {frames_processed} frames")
+                        return video_path, metrics
+
+                # Liberar memória a cada 10 frames
+                if frames_processed % 10 == 0:
+                    gc.collect()
+
+            # Se chegou aqui, não encontrou nenhuma detecção
+            metrics["analysis_time"] = time.time() - t0
+            metrics["total_time"] = time.time() - start_time
+            metrics["frames_processed_until_detection"] = frames_processed
+            metrics["detections"] = detections
+            return video_path, metrics
+
+        except Exception as e:
+            logger.error(f"Erro ao processar vídeo: {str(e)}")
+            return video_path, {}
+
+    def extract_frames(self, video_path: str, fps: int = 2, resolution: int = 480) -> list:
+        """Extrai frames de um vídeo utilizando ffmpeg."""
+        frames = []
+        temp_dir = Path(tempfile.mkdtemp())
+        try:
+            threads = min(os.cpu_count(), 4)  # Menor número de threads para CPU
+            cmd = [
+                'ffmpeg', '-i', video_path,
+                '-threads', str(threads),
+                '-vf', (f'fps={fps},'
+                        f'scale={resolution}:{resolution}:force_original_aspect_ratio=decrease:flags=lanczos,'
+                        f'pad={resolution}:{resolution}:(ow-iw)/2:(oh-ih)/2'),
+                '-frame_pts', '1',
+                f'{temp_dir}/%d.jpg'
+            ]
+            subprocess.run(cmd, check=True, capture_output=True)
+            frame_files = sorted(temp_dir.glob('*.jpg'), key=lambda x: int(x.stem))
+            chunk_size = 50  # Menor chunk size para CPU
+            with ThreadPoolExecutor(max_workers=threads) as executor:
+                for i in range(0, len(frame_files), chunk_size):
+                    chunk = frame_files[i:i + chunk_size]
+                    chunk_frames = list(tqdm(
+                        executor.map(lambda f: cv2.imread(str(f)), chunk),
+                        desc=f"Carregando frames {i+1}-{min(i+chunk_size, len(frame_files))}",
+                        total=len(chunk)
+                    ))
+                    frames.extend(chunk_frames)
+                    if i % (chunk_size * 5) == 0:
+                        gc.collect()
+        finally:
+            shutil.rmtree(temp_dir)
+        return frames
+
+    def clear_cache(self):
+        """Limpa cache e libera memória."""
+        self.result_cache.clear()
+        gc.collect() 

src/domain/detectors/gpu.py

@@ -0,0 +1,396 @@
+import torch
+import torch.nn.functional as F
+import torch._dynamo
+import logging
+import os
+import time
+import gc
+import numpy as np
+import cv2
+from PIL import Image
+from transformers import Owlv2Processor, Owlv2ForObjectDetection
+from .base import BaseDetector, BaseCache
+
+logger = logging.getLogger(__name__)
+
+# Configurações globais do PyTorch para otimização em GPU
+torch.backends.cuda.matmul.allow_tf32 = True
+torch.backends.cudnn.allow_tf32 = True
+torch.backends.cudnn.benchmark = True
+torch.backends.cuda.matmul.allow_fp16_reduced_precision_reduction = True
+torch._dynamo.config.suppress_errors = True
+
+
+class GPUCache(BaseCache):
+    """Cache otimizado para GPU."""
+    def __init__(self, max_size: int = 1000):
+        super().__init__(max_size)
+        self.device = torch.device('cuda')
+
+
+class WeaponDetectorGPU(BaseDetector):
+    """Implementação GPU do detector de armas com otimizações para a última versão do OWLv2."""
+    
+    def __init__(self):
+        """Inicializa variáveis básicas."""
+        super().__init__()
+        self.default_resolution = 640
+        self.amp_dtype = torch.float16
+        self.preprocess_stream = torch.cuda.Stream()
+        self.max_batch_size = 16  # Aumentado para 16
+        self.current_batch_size = 8  # Aumentado para 8
+        self.min_batch_size = 2
+    
+    def _initialize(self):
+        """Inicializa o modelo e o processador para execução exclusiva em GPU."""
+        try:
+            # Configurar device
+            self.device = self._get_best_device()
+            
+            # Diretório de cache para o modelo
+            cache_dir = os.getenv('CACHE_DIR', '/tmp/weapon_detection_cache')
+            os.makedirs(cache_dir, exist_ok=True)
+            
+            # Limpar memória GPU
+            self._clear_gpu_memory()
+            
+            logger.info("Carregando modelo e processador...")
+            
+            # Carregar processador e modelo com otimizações
+            model_name = "google/owlv2-base-patch16"
+            self.owlv2_processor = Owlv2Processor.from_pretrained(
+                model_name,
+                cache_dir=cache_dir
+            )
+            
+            # Configurações otimizadas para T4
+            self.owlv2_model = Owlv2ForObjectDetection.from_pretrained(
+                model_name,
+                cache_dir=cache_dir,
+                torch_dtype=self.amp_dtype,
+                device_map="auto",
+                low_cpu_mem_usage=True
+            ).to(self.device)
+            
+            # Otimizar modelo para inferência
+            self.owlv2_model.eval()
+            torch.compile(self.owlv2_model)  # Usar torch.compile para otimização
+            
+            # Usar queries do método base
+            self.text_queries = self._get_detection_queries()
+            logger.info(f"Total de queries carregadas: {len(self.text_queries)}")
+            
+            # Processar queries uma única vez com otimização de memória
+            with torch.cuda.amp.autocast(dtype=self.amp_dtype):
+                self.processed_text = self.owlv2_processor(
+                    text=self.text_queries,
+                    return_tensors="pt",
+                    padding=True
+                )
+                
+                self.processed_text = {
+                    key: val.to(self.device, non_blocking=True) 
+                    for key, val in self.processed_text.items()
+                }
+            
+            # Ajustar batch size baseado na memória disponível
+            self._adjust_batch_size()
+            
+            logger.info(f"Inicialização GPU completa! Batch size inicial: {self.current_batch_size}")
+            self._initialized = True
+            
+        except Exception as e:
+            logger.error(f"Erro na inicialização GPU: {str(e)}")
+            raise
+
+    def _adjust_batch_size(self):
+        """Ajusta o batch size baseado na memória disponível."""
+        try:
+            gpu_mem = torch.cuda.get_device_properties(0).total_memory
+            free_mem = torch.cuda.memory_reserved() - torch.cuda.memory_allocated()
+            mem_ratio = free_mem / gpu_mem
+            
+            if mem_ratio < 0.2:  # Menos de 20% livre
+                self.current_batch_size = max(self.min_batch_size, self.current_batch_size // 2)
+            elif mem_ratio > 0.4:  # Mais de 40% livre
+                self.current_batch_size = min(self.max_batch_size, self.current_batch_size * 2)
+                
+            logger.debug(f"Batch size ajustado para {self.current_batch_size} (Memória livre: {mem_ratio:.1%})")
+        except Exception as e:
+            logger.warning(f"Erro ao ajustar batch size: {str(e)}")
+            self.current_batch_size = self.min_batch_size
+
+    def detect_objects(self, image: Image.Image, threshold: float = 0.3) -> list:
+        """Detecta objetos em uma imagem utilizando a última versão do OWLv2."""
+        try:
+            self.threshold = threshold
+            
+            # Pré-processar imagem
+            if image.mode != 'RGB':
+                image = image.convert('RGB')
+            
+            # Processar imagem
+            image_inputs = self.owlv2_processor(
+                images=image,
+                return_tensors="pt"
+            )
+            
+            image_inputs = {
+                key: val.to(self.device) 
+                for key, val in image_inputs.items()
+            }
+            
+            # Inferência
+            with torch.no_grad():
+                inputs = {**image_inputs, **self.processed_text}
+                outputs = self.owlv2_model(**inputs)
+                
+                target_sizes = torch.tensor([image.size[::-1]], device=self.device)
+                results = self.owlv2_processor.post_process_grounded_object_detection(
+                    outputs=outputs,
+                    target_sizes=target_sizes,
+                    threshold=threshold
+                )[0]
+            
+            # Processar detecções
+            detections = []
+            if len(results["scores"]) > 0:
+                scores = results["scores"]
+                boxes = results["boxes"]
+                labels = results["labels"]
+                
+                for score, box, label in zip(scores, boxes, labels):
+                    if score.item() >= threshold:
+                        detections.append({
+                            "confidence": score.item(),
+                            "box": [int(x) for x in box.tolist()],
+                            "label": self.text_queries[label]
+                        })
+            
+            return detections
+            
+        except Exception as e:
+            logger.error(f"Erro em detect_objects: {str(e)}")
+            return []
+
+    def process_video(self, video_path: str, fps: int = None, threshold: float = 0.3, resolution: int = 640) -> tuple:
+        """Processa um vídeo utilizando GPU com processamento em lote e otimizações para T4."""
+        try:
+            metrics = {
+                "total_time": 0,
+                "frame_extraction_time": 0,
+                "analysis_time": 0,
+                "frames_analyzed": 0,
+                "video_duration": 0,
+                "device_type": self.device.type,
+                "detections": [],
+                "technical": {
+                    "model": "owlv2-base-patch16",
+                    "input_size": f"{resolution}x{resolution}",
+                    "threshold": threshold,
+                    "batch_size": self.current_batch_size,
+                    "gpu_memory": f"{torch.cuda.get_device_properties(0).total_memory / 1024**3:.1f}GB"
+                }
+            }
+            
+            start_time = time.time()
+            frames = self.extract_frames(video_path, fps, resolution)
+            metrics["frame_extraction_time"] = time.time() - start_time
+            metrics["frames_analyzed"] = len(frames)
+            
+            if not frames:
+                logger.warning("Nenhum frame extraído do vídeo")
+                return video_path, metrics
+            
+            metrics["video_duration"] = len(frames) / (fps or 2)
+            analysis_start = time.time()
+            
+            # Processar frames em lotes com ajuste dinâmico de batch size
+            for i in range(0, len(frames), self.current_batch_size):
+                try:
+                    batch_frames = frames[i:i + self.current_batch_size]
+                    
+                    # Pré-processamento assíncrono
+                    with torch.cuda.stream(self.preprocess_stream):
+                        batch_images = [
+                            Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
+                            for frame in batch_frames
+                        ]
+                        
+                        batch_inputs = self.owlv2_processor(
+                            images=batch_images,
+                            return_tensors="pt"
+                        )
+                        
+                        batch_inputs = {
+                            key: val.to(self.device, non_blocking=True)
+                            for key, val in batch_inputs.items()
+                        }
+                    
+                    # Expandir texto processado para o batch
+                    batch_text = {
+                        key: val.repeat(len(batch_images), 1) 
+                        for key, val in self.processed_text.items()
+                    }
+                    
+                    inputs = {**batch_inputs, **batch_text}
+                    
+                    # Inferência com mixed precision
+                    with torch.cuda.amp.autocast(dtype=self.amp_dtype):
+                        with torch.no_grad():
+                            outputs = self.owlv2_model(**inputs)
+                    
+                    # Processar resultados
+                    target_sizes = torch.tensor([[img.size[::-1] for img in batch_images]], device=self.device)
+                    results = self.owlv2_processor.post_process_grounded_object_detection(
+                        outputs=outputs,
+                        target_sizes=target_sizes[0],
+                        threshold=threshold
+                    )
+                    
+                    # Verificar detecções
+                    for batch_idx, result in enumerate(results):
+                        if len(result["scores"]) > 0:
+                            frame_idx = i + batch_idx
+                            max_score_idx = torch.argmax(result["scores"])
+                            score = result["scores"][max_score_idx]
+                            
+                            if score.item() >= threshold:
+                                detection = {
+                                    "frame": frame_idx,
+                                    "confidence": score.item(),
+                                    "box": [int(x) for x in result["boxes"][max_score_idx].tolist()],
+                                    "label": self.text_queries[result["labels"][max_score_idx]]
+                                }
+                                metrics["detections"].append(detection)
+                                metrics["analysis_time"] = time.time() - analysis_start
+                                metrics["total_time"] = time.time() - start_time
+                                return video_path, metrics
+                    
+                    # Limpar memória e ajustar batch size periodicamente
+                    if (i // self.current_batch_size) % 5 == 0:
+                        self._clear_gpu_memory()
+                        self._adjust_batch_size()
+                        
+                except RuntimeError as e:
+                    if "out of memory" in str(e):
+                        logger.warning("OOM detectado, reduzindo batch size")
+                        self._clear_gpu_memory()
+                        self.current_batch_size = max(self.min_batch_size, self.current_batch_size // 2)
+                        continue
+                    raise
+            
+            metrics["analysis_time"] = time.time() - analysis_start
+            metrics["total_time"] = time.time() - start_time
+            return video_path, metrics
+            
+        except Exception as e:
+            logger.error(f"Erro ao processar vídeo: {str(e)}")
+            return video_path, metrics
+
+    def _clear_gpu_memory(self):
+        """Limpa memória GPU de forma agressiva."""
+        try:
+            torch.cuda.empty_cache()
+            torch.cuda.synchronize()
+            gc.collect()
+        except Exception as e:
+            logger.error(f"Erro ao limpar memória GPU: {str(e)}")
+
+    def _get_best_device(self):
+        if not torch.cuda.is_available():
+            raise RuntimeError("CUDA não está disponível!")
+        return torch.device('cuda')
+
+    def _preprocess_image(self, image: Image.Image) -> Image.Image:
+        """Pré-processa a imagem com otimizações para GPU."""
+        try:
+            target_size = (self.default_resolution, self.default_resolution)
+            if image.mode != 'RGB':
+                image = image.convert('RGB')
+
+            if image.size != target_size:
+                ratio = min(target_size[0] / image.size[0], target_size[1] / image.size[1])
+                new_size = (int(image.size[0] * ratio), int(image.size[1] * ratio))
+                
+                with torch.cuda.stream(self.preprocess_stream), torch.amp.autocast(device_type='cuda', dtype=self.amp_dtype):
+                    img_tensor = torch.from_numpy(np.array(image)).permute(2, 0, 1).unsqueeze(0)
+                    img_tensor = img_tensor.to(self.device, dtype=self.amp_dtype, non_blocking=True)
+                    img_tensor = img_tensor / 255.0
+                    
+                    mode = 'bilinear' if ratio < 1 else 'nearest'
+                    img_tensor = F.interpolate(
+                        img_tensor,
+                        size=new_size,
+                        mode=mode,
+                        align_corners=False if mode == 'bilinear' else None
+                    )
+
+                    if new_size != target_size:
+                        final_tensor = torch.zeros(
+                            (1, 3, target_size[1], target_size[0]),
+                            device=self.device,
+                            dtype=self.amp_dtype
+                        )
+                        pad_left = (target_size[0] - new_size[0]) // 2
+                        pad_top = (target_size[1] - new_size[1]) // 2
+                        final_tensor[
+                            :,
+                            :,
+                            pad_top:pad_top + new_size[1],
+                            pad_left:pad_left + new_size[0]
+                        ] = img_tensor
+
+                        img_tensor = final_tensor
+
+                    img_tensor = img_tensor.squeeze(0).permute(1, 2, 0).cpu()
+                    image = Image.fromarray((img_tensor.numpy() * 255).astype(np.uint8))
+
+            return image
+
+        except Exception as e:
+            logger.error(f"Erro no pré-processamento: {str(e)}")
+            return image
+
+    def _get_memory_usage(self):
+        """Retorna o uso atual de memória GPU em porcentagem."""
+        try:
+            allocated = torch.cuda.memory_allocated()
+            reserved = torch.cuda.memory_reserved()
+            total = torch.cuda.get_device_properties(0).total_memory
+            return (allocated + reserved) / total * 100
+        except Exception as e:
+            logger.error(f"Erro ao obter uso de memória GPU: {str(e)}")
+            return 0
+
+    def _should_clear_cache(self):
+        """Determina se o cache deve ser limpo baseado no uso de memória."""
+        try:
+            memory_usage = self._get_memory_usage()
+            if memory_usage > 90:
+                return True
+            if memory_usage > 75 and not hasattr(self, '_last_cache_clear'):
+                return True
+            if hasattr(self, '_last_cache_clear'):
+                time_since_last_clear = time.time() - self._last_cache_clear
+                if memory_usage > 80 and time_since_last_clear > 300:
+                    return True
+            return False
+        except Exception as e:
+            logger.error(f"Erro ao verificar necessidade de limpeza: {str(e)}")
+            return False
+
+    def clear_cache(self):
+        """Limpa o cache de resultados e libera memória quando necessário."""
+        try:
+            if self._should_clear_cache():
+                if hasattr(self, 'result_cache'):
+                    self.result_cache.clear()
+                torch.cuda.empty_cache()
+                gc.collect()
+                self._last_cache_clear = time.time()
+                logger.info(f"Cache GPU limpo com sucesso. Uso de memória: {self._get_memory_usage():.1f}%")
+            else:
+                logger.debug("Limpeza de cache não necessária no momento")
+        except Exception as e:
+            logger.error(f"Erro ao limpar cache GPU: {str(e)}")

src/domain/entities/detection.py

@@ -0,0 +1,22 @@
+from dataclasses import dataclass
+from typing import List, Tuple, Optional
+
+@dataclass
+class Detection:
+    """Representa uma detecção de objeto perigoso."""
+    frame: int
+    confidence: float
+    label: str
+    box: List[int]  # [x1, y1, x2, y2]
+    timestamp: float = 0.0
+
+@dataclass
+class DetectionResult:
+    """Resultado completo do processamento de vídeo."""
+    video_path: str
+    detections: List[Detection]
+    frames_analyzed: int
+    total_time: float
+    device_type: str
+    frame_extraction_time: float
+    analysis_time: float

src/domain/factories/detector_factory.py

@@ -0,0 +1,296 @@
+import os
+import time
+import gc
+import cv2
+import json
+import torch
+import psutil
+import shutil
+import pickle
+import hashlib
+import tempfile
+import logging
+import subprocess
+import numpy as np
+import sys
+from tqdm import tqdm
+from pathlib import Path
+from datetime import datetime
+from concurrent.futures import ThreadPoolExecutor
+from PIL import Image
+from dotenv import load_dotenv
+from transformers import Owlv2Processor, Owlv2ForObjectDetection
+from typing import Optional
+from src.domain.detectors.base import BaseDetector, BaseCache
+from src.domain.detectors.gpu import WeaponDetectorGPU
+from src.domain.detectors.cpu import WeaponDetectorCPU
+
+# Carregar variáveis de ambiente
+load_dotenv()
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+class BaseCache:
+    """Cache base para armazenar resultados de detecção."""
+    def __init__(self, max_size: int = 1000):
+        self.cache = {}
+        self.max_size = max_size
+        self.hits = 0
+        self.misses = 0
+        self.last_access = {}
+
+    def get(self, image: np.ndarray) -> list:
+        try:
+            key = hashlib.blake2b(image.tobytes(), digest_size=16).hexdigest()
+            if key in self.cache:
+                self.hits += 1
+                self.last_access[key] = time.time()
+                return self.cache[key]
+            self.misses += 1
+            return None
+        except Exception as e:
+            logger.error(f"Erro ao recuperar do cache: {str(e)}")
+            return None
+
+    def put(self, image: np.ndarray, results: list):
+        try:
+            key = hashlib.blake2b(image.tobytes(), digest_size=16).hexdigest()
+            if len(self.cache) >= self.max_size:
+                oldest_key = min(self.last_access.items(), key=lambda x: x[1])[0]
+                del self.cache[oldest_key]
+                del self.last_access[oldest_key]
+            self.cache[key] = results
+            self.last_access[key] = time.time()
+        except Exception as e:
+            logger.error(f"Erro ao armazenar no cache: {str(e)}")
+
+    def clear(self):
+        """Limpa o cache e libera memória."""
+        self.cache.clear()
+        self.last_access.clear()
+        gc.collect()
+
+    def get_stats(self) -> dict:
+        total = self.hits + self.misses
+        hit_rate = (self.hits / total) * 100 if total > 0 else 0
+        return {
+            "cache_size": len(self.cache),
+            "max_size": self.max_size,
+            "hits": self.hits,
+            "misses": self.misses,
+            "hit_rate": f"{hit_rate:.2f}%",
+            "memory_usage": sum(sys.getsizeof(v) for v in self.cache.values())
+        }
+
+class BaseWeaponDetector:
+    """Classe base abstrata para detecção de armas."""
+    def __init__(self):
+        """Inicialização básica comum a todos os detectores."""
+        self._initialized = False
+        self.device = self._get_best_device()
+        self._initialize()
+
+    def _check_initialized(self):
+        """Verifica se o detector está inicializado."""
+        if not self._initialized:
+            raise RuntimeError("Detector não está inicializado")
+
+    def clean_memory(self):
+        """Limpa memória não utilizada."""
+        try:
+            if torch.cuda.is_available():
+                torch.cuda.empty_cache()
+                logger.debug("Cache GPU limpo")
+            gc.collect()
+            logger.debug("Garbage collector executado")
+        except Exception as e:
+            logger.error(f"Erro ao limpar memória: {str(e)}")
+
+    def _get_best_device(self):
+        """Deve ser implementado nas classes filhas."""
+        raise NotImplementedError
+
+    def _initialize(self):
+        """Deve ser implementado nas classes filhas."""
+        raise NotImplementedError
+
+    def detect_objects(self, image, threshold=0.3):
+        """Deve ser implementado nas classes filhas."""
+        raise NotImplementedError
+
+    def process_video(self, video_path, fps=None, threshold=0.3, resolution=640):
+        """Deve ser implementado nas classes filhas."""
+        raise NotImplementedError
+
+    def _apply_nms(self, detections: list, iou_threshold: float = 0.5) -> list:
+        """Deve ser implementado nas classes filhas."""
+        raise NotImplementedError
+
+    def extract_frames(self, video_path: str, fps: int = 2, resolution: int = 640) -> list:
+        """Extrai frames de um vídeo utilizando ffmpeg."""
+        frames = []
+        temp_dir = Path(tempfile.mkdtemp())
+        try:
+            threads = min(os.cpu_count(), 8)
+            cmd = [
+                'ffmpeg', '-i', video_path,
+                '-threads', str(threads),
+                '-vf', (f'fps={fps},'
+                        f'scale={resolution}:{resolution}:force_original_aspect_ratio=decrease:flags=lanczos,'
+                        f'pad={resolution}:{resolution}:(ow-iw)/2:(oh-ih)/2'),
+                '-frame_pts', '1',
+                f'{temp_dir}/%d.jpg'
+            ]
+            subprocess.run(cmd, check=True, capture_output=True)
+            frame_files = sorted(temp_dir.glob('*.jpg'), key=lambda x: int(x.stem))
+            chunk_size = 100
+            with ThreadPoolExecutor(max_workers=threads) as executor:
+                for i in range(0, len(frame_files), chunk_size):
+                    chunk = frame_files[i:i + chunk_size]
+                    chunk_frames = list(tqdm(
+                        executor.map(lambda f: cv2.imread(str(f)), chunk),
+                        desc=f"Carregando frames {i+1}-{min(i+chunk_size, len(frame_files))}",
+                        total=len(chunk)
+                    ))
+                    frames.extend(chunk_frames)
+                    if i % (chunk_size * 5) == 0:
+                        gc.collect()
+        finally:
+            shutil.rmtree(temp_dir)
+        return frames
+
+    def _preprocess_image(self, image: Image.Image) -> Image.Image:
+        """Deve ser implementado nas classes filhas."""
+        raise NotImplementedError
+
+    def _update_frame_metrics(self, detections: list, frame_idx: int, metrics: dict):
+        """Atualiza as métricas para um conjunto de detecções em um frame."""
+        try:
+            for detection in detections:
+                self._update_detection_metrics(detection, metrics)
+                if isinstance(detection, dict):
+                    metrics.setdefault("detections", []).append({
+                        "frame": frame_idx,
+                        "box": detection.get("box", []),
+                        "confidence": detection.get("confidence", 0),
+                        "label": detection.get("label", "unknown")
+                    })
+        except Exception as e:
+            logger.error(f"Erro ao atualizar métricas do frame: {str(e)}")
+
+    def _update_detection_metrics(self, detection: dict, metrics: dict):
+        """Atualiza as métricas de detecção."""
+        try:
+            if not isinstance(detection, dict):
+                logger.warning(f"Detection não é um dicionário: {detection}")
+                return
+            confidence = detection.get("confidence", 0)
+            if not confidence:
+                return
+            if "detection_stats" not in metrics:
+                metrics["detection_stats"] = {
+                    "total_detections": 0,
+                    "avg_confidence": 0,
+                    "confidence_distribution": {
+                        "low": 0,
+                        "medium": 0,
+                        "high": 0
+                    }
+                }
+            stats = metrics["detection_stats"]
+            stats["total_detections"] += 1
+            if confidence < 0.5:
+                stats["confidence_distribution"]["low"] += 1
+            elif confidence < 0.7:
+                stats["confidence_distribution"]["medium"] += 1
+            else:
+                stats["confidence_distribution"]["high"] += 1
+            n = stats["total_detections"]
+            old_avg = stats["avg_confidence"]
+            stats["avg_confidence"] = (old_avg * (n - 1) + confidence) / n
+        except Exception as e:
+            logger.error(f"Error updating metrics: {str(e)}")
+
+    def clear_cache(self):
+        """Deve ser implementado nas classes filhas."""
+        raise NotImplementedError
+
+class ResultCache(BaseCache):
+    """
+    Cache otimizado para armazenar resultados de detecção.
+    """
+    def __init__(self, max_size: int = 1000):
+        super().__init__(max_size)
+
+class WeaponDetector:
+    """Implementação do Factory Pattern para criar a instância apropriada do detector."""
+    _instance = None
+
+    def __new__(cls):
+        try:
+            if cls._instance is None:
+                if torch.cuda.is_available():
+                    cls._instance = WeaponDetectorGPU()
+                    logger.info("Detector GPU criado")
+                else:
+                    cls._instance = WeaponDetectorCPU()
+                    logger.info("Detector CPU criado")
+                
+                # Garantir que o detector foi inicializado corretamente
+                if not cls._instance:
+                    raise RuntimeError("Falha ao criar instância do detector")
+                    
+                # Inicializar o detector
+                if hasattr(cls._instance, 'initialize'):
+                    cls._instance.initialize()
+                    
+                # Verificar se os métodos necessários existem
+                required_methods = ['process_video', 'clean_memory', 'detect_objects']
+                for method in required_methods:
+                    if not hasattr(cls._instance, method):
+                        raise RuntimeError(f"Detector não possui método obrigatório: {method}")
+                
+            return cls._instance
+            
+        except Exception as e:
+            logger.error(f"Erro ao criar detector: {str(e)}")
+            raise
+
+    @classmethod
+    def get_instance(cls):
+        """Retorna a instância existente ou cria uma nova."""
+        if cls._instance is None:
+            return cls()
+        return cls._instance
+
+    def detect_objects(self, image: Image.Image, threshold: float = 0.3) -> list:
+        """Detecta objetos em uma imagem."""
+        if not self._instance:
+            raise RuntimeError("Detector não inicializado")
+        return self._instance.detect_objects(image, threshold)
+
+    def extract_frames(self, video_path: str, fps: int = 2, resolution: int = 640) -> list:
+        """Extrai frames de um vídeo."""
+        if not self._instance:
+            raise RuntimeError("Detector não inicializado")
+        return self._instance.extract_frames(video_path, fps, resolution)
+
+    def process_video(self, video_path: str, fps: int = None, threshold: float = 0.3, resolution: int = 640) -> tuple:
+        """Processa o vídeo e retorna os detalhes técnicos e as detecções."""
+        if not self._instance:
+            raise RuntimeError("Detector não inicializado")
+        return self._instance.process_video(video_path, fps, threshold, resolution)
+
+    def clean_memory(self):
+        """Limpa todo o cache do sistema."""
+        if not self._instance:
+            return
+        if hasattr(self._instance, 'clear_cache'):
+            self._instance.clear_cache()
+        if hasattr(self._instance, 'clean_memory'):
+            self._instance.clean_memory()
+        # Forçar limpeza de memória
+        gc.collect()
+        if torch.cuda.is_available():
+            torch.cuda.empty_cache()

src/domain/interfaces/detector.py

@@ -0,0 +1,26 @@
+from abc import ABC, abstractmethod
+from typing import List, Tuple, Dict, Any
+from ..entities.detection import DetectionResult
+
+class DetectorInterface(ABC):
+    """Interface base para detectores de objetos perigosos."""
+    
+    @abstractmethod
+    def process_video(self, video_path: str, fps: int, threshold: float, resolution: int) -> Tuple[str, DetectionResult]:
+        """Processa um vídeo e retorna as detecções encontradas."""
+        pass
+    
+    @abstractmethod
+    def clean_memory(self) -> None:
+        """Limpa a memória utilizada pelo detector."""
+        pass
+    
+    @abstractmethod
+    def get_device_info(self) -> Dict[str, Any]:
+        """Retorna informações sobre o dispositivo em uso."""
+        pass
+    
+    @abstractmethod
+    def get_cache_stats(self) -> Dict[str, Any]:
+        """Retorna estatísticas do cache."""
+        pass 

src/domain/interfaces/notification.py

@@ -0,0 +1,23 @@
+from abc import ABC, abstractmethod
+from typing import Dict, Any, List
+
+class NotificationService(ABC):
+    """Interface base para serviços de notificação."""
+    
+    @abstractmethod
+    def send_notification(self, detection_data: Dict[str, Any], recipient: str) -> bool:
+        """Envia notificação usando o serviço específico."""
+        pass
+
+class NotificationFactory(ABC):
+    """Interface para fábrica de serviços de notificação."""
+    
+    @abstractmethod
+    def create_service(self, service_type: str) -> NotificationService:
+        """Cria uma instância do serviço de notificação especificado."""
+        pass
+    
+    @abstractmethod
+    def get_available_services(self) -> List[str]:
+        """Retorna lista de serviços de notificação disponíveis."""
+        pass 

src/infrastructure/services/notification_services.py

@@ -0,0 +1,116 @@
+import logging
+import os
+from typing import Dict, Any
+from sendgrid import SendGridAPIClient
+from sendgrid.helpers.mail import Mail, Email, To, Content
+from src.domain.interfaces.notification import NotificationService, NotificationFactory
+
+logger = logging.getLogger(__name__)
+
+class EmailNotification(NotificationService):
+    def send_notification(self, detection_data: Dict[str, Any], recipient: str) -> bool:
+        try:
+            # Verificar se há detecções
+            if not detection_data.get("detections"):
+                logger.info("Nenhuma detecção para notificar")
+                return True  # Retorna True pois não é um erro
+                
+            sender_email = os.getenv('NOTIFICATION_EMAIL')
+            sendgrid_api_key = os.getenv('SENDGRID_API_KEY')
+            
+            if not sender_email:
+                logger.error("NOTIFICATION_EMAIL não configurado")
+                return False
+                
+            if not sendgrid_api_key:
+                logger.error("SENDGRID_API_KEY não configurada")
+                return False
+                
+            if not recipient:
+                logger.error("Destinatário de e-mail não fornecido")
+                return False
+                
+            body = self._format_email_body(detection_data)
+            
+            message = Mail(
+                from_email=sender_email,
+                to_emails=recipient,
+                subject='🚨 ALERTA DE SEGURANÇA - Detecção de Risco',
+                html_content=f'<pre style="font-family: monospace;">{body}</pre>'
+            )
+            
+            try:
+                sg = SendGridAPIClient(sendgrid_api_key)
+                response = sg.send(message)
+                success = response.status_code == 202
+                
+                if success:
+                    logger.info(f"E-mail enviado com sucesso para {recipient}")
+                    logger.debug(f"Status: {response.status_code}")
+                    logger.debug(f"Body: {response.body}")
+                    logger.debug(f"Headers: {response.headers}")
+                else:
+                    logger.error(f"Erro ao enviar e-mail. Status code: {response.status_code}")
+                
+                return success
+                
+            except Exception as e:
+                logger.error(f"Erro ao enviar e-mail via SendGrid: {str(e)}")
+                return False
+                
+        except Exception as e:
+            logger.error(f"Erro no serviço de e-mail: {str(e)}")
+            return False
+            
+    def _format_email_body(self, detection_data: Dict[str, Any]) -> str:
+        """Formata o corpo do e-mail com os dados da detecção."""
+        try:
+            detections = detection_data.get("detections", [])
+            if not detections:
+                return "Nenhuma detecção encontrada no vídeo."
+                
+            body = """
+⚠️ ALERTA DE SEGURANÇA ⚠️
+
+Uma detecção de risco foi identificada:
+
+"""
+            # Adicionar informações da primeira detecção
+            first_detection = detections[0]
+            body += f"""📹 Detecção:
+- Objeto: {first_detection.get('label', 'Desconhecido')}
+- Confiança: {first_detection.get('confidence', 0):.2%}
+- Timestamp: {first_detection.get('timestamp', 0):.2f}s
+
+"""
+            
+            # Adicionar informações técnicas
+            if "technical" in detection_data:
+                tech = detection_data["technical"]
+                body += f"""Informações Técnicas:
+- Threshold: {tech.get('threshold', 'N/A')}
+- FPS: {tech.get('fps', 'N/A')}
+- Resolução: {tech.get('resolution', 'N/A')}
+"""
+            
+            body += """
+--
+Este é um e-mail automático enviado pelo Sistema de Detecção de Riscos.
+Não responda este e-mail.
+"""
+            
+            return body
+            
+        except Exception as e:
+            logger.error(f"Erro ao formatar e-mail: {str(e)}")
+            return "Erro ao formatar dados da detecção."
+
+class NotificationServiceFactory(NotificationFactory):
+    def __init__(self):
+        self._services = {'email': EmailNotification()}
+    
+    def create_service(self, service_type: str) -> NotificationService:
+        return self._services.get(service_type)
+    
+    def get_available_services(self) -> list:
+        return list(self._services.keys())

src/infrastructure/services/weapon_detector.py

@@ -0,0 +1,190 @@
+import torch
+from typing import Tuple
+from src.domain.interfaces.detector import DetectorInterface
+from src.domain.entities.detection import Detection, DetectionResult
+from src.domain.factories.detector_factory import WeaponDetector
+from src.domain.detectors.gpu import WeaponDetectorGPU
+from src.domain.detectors.cpu import WeaponDetectorCPU
+import logging
+import gc
+
+logger = logging.getLogger(__name__)
+
+class WeaponDetectorService(DetectorInterface):
+    """Adaptador que conecta os detectores do domínio com a infraestrutura externa."""
+    
+    def __init__(self):
+        try:
+            # Usar o Factory Pattern do domínio para criar o detector apropriado
+            self.detector = WeaponDetector.get_instance()  # Usar get_instance ao invés do construtor direto
+            if not self.detector:
+                raise RuntimeError("Falha ao criar o detector")
+                
+            self.device_type = "GPU" if torch.cuda.is_available() else "CPU"
+            logger.info(f"Detector inicializado em modo {self.device_type}")
+            
+            # Manter referência à implementação específica para otimizações
+            if hasattr(self.detector, '_instance') and self.detector._instance is not None:
+                self._specific_detector = self.detector._instance
+            else:
+                self._specific_detector = self.detector
+                
+            # Verificar se o detector foi inicializado corretamente
+            if not hasattr(self._specific_detector, 'process_video'):
+                raise RuntimeError("Detector não possui método process_video")
+                
+            # Garantir que o detector está inicializado
+            if hasattr(self._specific_detector, 'initialize'):
+                self._specific_detector.initialize()
+                
+        except Exception as e:
+            logger.error(f"Erro ao inicializar WeaponDetectorService: {str(e)}")
+            raise RuntimeError(f"Falha na inicialização do detector: {str(e)}")
+    
+    def process_video(
+        self,
+        video_path: str,
+        fps: int,
+        threshold: float,
+        resolution: int
+    ) -> Tuple[str, DetectionResult]:
+        """Processa o vídeo usando o detector apropriado."""
+        try:
+            if not self._specific_detector:
+                raise RuntimeError("Detector não inicializado")
+                
+            # Garantir que o detector está inicializado
+            if hasattr(self._specific_detector, 'initialize'):
+                self._specific_detector.initialize()
+                
+            output_path, metrics = self._specific_detector.process_video(
+                video_path,
+                fps=fps,
+                threshold=threshold,
+                resolution=resolution
+            )
+            
+            if not metrics:
+                logger.warning("Nenhuma métrica retornada pelo detector")
+                metrics = {}
+            
+            # Converter detecções para entidades do domínio
+            detections = []
+            for d in metrics.get('detections', []):
+                try:
+                    detections.append(Detection(
+                        frame=d.get('frame', 0),
+                        confidence=d.get('confidence', 0.0),
+                        label=d.get('label', 'unknown'),
+                        box=d.get('box', [0, 0, 0, 0]),
+                        timestamp=d.get('frame', 0) / fps if fps else 0
+                    ))
+                except Exception as e:
+                    logger.error(f"Erro ao processar detecção: {str(e)}")
+            
+            result = DetectionResult(
+                video_path=output_path or video_path,
+                detections=detections,
+                frames_analyzed=metrics.get('frames_analyzed', 0),
+                total_time=metrics.get('total_time', 0.0),
+                device_type=self.device_type,
+                frame_extraction_time=metrics.get('frame_extraction_time', 0.0),
+                analysis_time=metrics.get('analysis_time', 0.0)
+            )
+            
+            return output_path or video_path, result
+            
+        except Exception as e:
+            logger.error(f"Erro ao processar vídeo: {str(e)}")
+            empty_result = DetectionResult(
+                video_path=video_path,
+                detections=[],
+                frames_analyzed=0,
+                total_time=0.0,
+                device_type=self.device_type,
+                frame_extraction_time=0.0,
+                analysis_time=0.0
+            )
+            return video_path, empty_result
+    
+    def clean_memory(self) -> None:
+        """Limpa a memória do detector."""
+        try:
+            if not self._specific_detector:
+                logger.warning("Nenhum detector específico para limpar memória")
+                return
+                
+            if hasattr(self._specific_detector, 'clear_cache'):
+                self._specific_detector.clear_cache()
+                
+            if hasattr(self._specific_detector, 'clean_memory'):
+                self._specific_detector.clean_memory()
+                
+            # Forçar coleta de lixo
+            gc.collect()
+            if torch.cuda.is_available():
+                torch.cuda.empty_cache()
+                
+        except Exception as e:
+            logger.error(f"Erro ao limpar memória: {str(e)}")
+    
+    def get_device_info(self) -> dict:
+        """Retorna informações detalhadas sobre o dispositivo em uso."""
+        try:
+            if not self._specific_detector:
+                return {
+                    "type": self.device_type,
+                    "memory_total": 0,
+                    "memory_used": 0
+                }
+                
+            if isinstance(self._specific_detector, WeaponDetectorGPU):
+                return {
+                    "type": "GPU",
+                    "name": torch.cuda.get_device_name(0),
+                    "memory_total": torch.cuda.get_device_properties(0).total_memory,
+                    "memory_used": torch.cuda.memory_allocated(),
+                    "memory_cached": torch.cuda.memory_reserved()
+                }
+            else:
+                import psutil
+                return {
+                    "type": "CPU",
+                    "threads": psutil.cpu_count(),
+                    "memory_total": psutil.virtual_memory().total,
+                    "memory_used": psutil.virtual_memory().used
+                }
+        except Exception as e:
+            logger.error(f"Erro ao obter informações do dispositivo: {str(e)}")
+            return {
+                "type": self.device_type,
+                "memory_total": 0,
+                "memory_used": 0
+            }
+    
+    def get_cache_stats(self) -> dict:
+        """Retorna estatísticas do cache se disponível."""
+        try:
+            if not self._specific_detector:
+                return self._get_empty_cache_stats()
+                
+            if (hasattr(self._specific_detector, 'result_cache') and 
+                self._specific_detector.result_cache is not None):
+                return self._specific_detector.result_cache.get_stats()
+                
+            return self._get_empty_cache_stats()
+            
+        except Exception as e:
+            logger.error(f"Erro ao obter estatísticas do cache: {str(e)}")
+            return self._get_empty_cache_stats()
+            
+    def _get_empty_cache_stats(self) -> dict:
+        """Retorna estatísticas vazias do cache."""
+        return {
+            "cache_size": 0,
+            "max_size": 0,
+            "hits": 0,
+            "misses": 0,
+            "hit_rate": "0.00%",
+            "memory_usage": 0
+        } 

src/main.py

@@ -0,0 +1,69 @@
+import os
+from dotenv import load_dotenv
+from src.presentation.web.gradio_interface import GradioInterface
+import logging
+import torch
+
+# Configurar logging
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+)
+logger = logging.getLogger(__name__)
+
+def main():
+    """Função principal que inicia a aplicação."""
+    try:
+        # Verificar se está rodando no Hugging Face
+        IS_HUGGINGFACE = os.getenv('SPACE_ID') is not None
+        
+        # Carregar configurações do ambiente apropriado
+        if IS_HUGGINGFACE:
+            load_dotenv('.env.huggingface')
+            logger.info("Ambiente HuggingFace detectado")
+        else:
+            load_dotenv('.env')
+            logger.info("Ambiente local detectado")
+        
+        # Criar e configurar interface
+        interface = GradioInterface()
+        demo = interface.create_interface()
+        
+        if IS_HUGGINGFACE:
+            # Calcular número ideal de workers baseado na GPU
+            if torch.cuda.is_available():
+                gpu_mem = torch.cuda.get_device_properties(0).total_memory / (1024**3)  # em GB
+                max_concurrent = min(2, int(gpu_mem / 8))  # 8GB por worker
+                logger.info(f"GPU Memory: {gpu_mem:.1f}GB, Max Concurrent: {max_concurrent}")
+            else:
+                max_concurrent = 1
+            
+            # Primeiro configurar a fila
+            demo = demo.queue(
+                max_size=16,  # Aumentado para corresponder ao max_batch_size
+                concurrency_count=max_concurrent,  # Baseado na memória GPU
+                status_update_rate=10,  # Atualizações mais frequentes
+                api_open=False,
+                max_batch_size=16  # Aumentado para corresponder ao detector
+            )
+            # Depois fazer o launch
+            demo.launch(
+                server_name="0.0.0.0",
+                server_port=7860,
+                share=False,
+                max_threads=4  # Limitar threads da CPU
+            )
+        else:
+            # Ambiente local - apenas launch direto
+            demo.launch(
+                server_name="0.0.0.0",
+                server_port=7860,
+                share=True
+            )
+            
+    except Exception as e:
+        logger.error(f"Erro ao iniciar aplicação: {str(e)}")
+        raise
+
+if __name__ == "__main__":
+    main() 

src/presentation/web/gradio_interface.py

@@ -0,0 +1,235 @@
+import gradio as gr
+import os
+from typing import Tuple, Any
+from pathlib import Path
+from src.application.use_cases.process_video import ProcessVideoUseCase, ProcessVideoRequest
+from src.infrastructure.services.weapon_detector import WeaponDetectorService
+from src.infrastructure.services.notification_services import NotificationServiceFactory
+
+class GradioInterface:
+    """Interface Gradio usando Clean Architecture."""
+    
+    def __init__(self):
+        self.detector = WeaponDetectorService()
+        self.notification_factory = NotificationServiceFactory()
+        self.default_fps = 2 if self.detector.device_type == "GPU" else 1
+        self.default_resolution = "640" if self.detector.device_type == "GPU" else "480"
+        
+        self.use_case = ProcessVideoUseCase(
+            detector=self.detector,
+            notification_factory=self.notification_factory,
+            default_fps=self.default_fps,
+            default_resolution=int(self.default_resolution)
+        )
+    
+    def list_sample_videos(self) -> list:
+        """Lista os vídeos de exemplo na pasta videos."""
+        video_dir = Path("videos")
+        if not video_dir.exists():
+            os.makedirs(video_dir)
+            return []
+        
+        video_extensions = ['.mp4', '.avi', '.mov', '.mkv']
+        videos = []
+        
+        # Procurar em subdiretórios específicos
+        for status_dir in ['seguro', 'risco_detectado']:
+            dir_path = video_dir / status_dir
+            if dir_path.exists():
+                for ext in video_extensions:
+                    for video_path in dir_path.glob(f'*{ext}'):
+                        videos.append({
+                            'path': str(video_path),
+                            'name': video_path.name,
+                            'ground_truth': '✅ SEGURO (Ground Truth)' if status_dir == 'seguro' else '⚠️ RISCO DETECTADO (Ground Truth)'
+                        })
+        
+        return videos
+    
+    def load_sample_video(self, video_path: str) -> str:
+        """Carrega um vídeo de exemplo."""
+        try:
+            return video_path
+        except Exception as e:
+            return None
+    
+    def create_interface(self) -> gr.Blocks:
+        """Cria a interface Gradio."""
+        title = "Detector de Riscos em Vídeos"
+        sample_videos = self.list_sample_videos()
+        
+        with gr.Blocks(
+            title=title,
+            theme=gr.themes.Ocean(),
+            css="footer {display: none !important}"
+        ) as demo:
+            gr.Markdown(f"""# 🚨 {title}
+            
+            Faça upload de um vídeo para detectar objetos perigosos.
+            Opcionalmente, configure notificações para receber alertas em caso de detecções.
+
+            **Importante para melhor performance:**
+            - Vídeos de até 60 segundos
+            - FPS entre 1-2 para análise com maior performance
+            - FPS maior que 2 para análise com maior precisão
+            """)
+            with gr.Group():
+                gr.Markdown("""### Configuração de Processamento""")
+                with gr.Row():
+                    threshold = gr.Slider(
+                        minimum=0.1,
+                        maximum=1.0,
+                        value=0.5,
+                        step=0.1,
+                        label="Limiar de Detecção",
+                    )
+                    fps = gr.Slider(
+                        minimum=1,
+                        maximum=5,
+                        value=self.default_fps,
+                        step=1,
+                        label="Frames por Segundo",
+                    )
+                    resolution = gr.Radio(
+                        choices=["480", "640", "768"],
+                        value=self.default_resolution,
+                        label="Resolução de Processamento",
+                )
+            with gr.Group():
+                gr.Markdown("""### Configuração de Notificações de Detecção (Opcional)""")
+                with gr.Row():
+                    notification_type = gr.Radio(
+                        choices=self.notification_factory.get_available_services(),
+                        value="email",
+                        label="Tipo de Notificação",
+                        interactive=True,
+                    )
+                    notification_target = gr.Textbox(
+                        label="Destino da Notificação (E-mail)",
+                        placeholder="[email protected]",
+                    )
+            with gr.Row():
+                with gr.Column(scale=2):
+                    input_video = gr.Video(
+                        label="Vídeo de Entrada",
+                        format="mp4",
+                        interactive=True,
+                        height=400
+                    )
+                
+                    submit_btn = gr.Button(
+                        "Analisar Vídeo",
+                        variant="primary",
+                        scale=2
+                    )
+                with gr.Column(scale=1):
+                    status = gr.Textbox(
+                        label="Status da Detecção",
+                        lines=4,
+                        show_copy_button=True
+                    )
+                    with gr.Accordion("Detalhes Técnicos", open=False):
+                        json_output = gr.JSON(
+                            label="Detalhes Técnicos",
+                        )
+                    
+                    # Informações adicionais
+                    with gr.Accordion("Informações Adicionais", open=False):
+                        gr.Markdown("""
+                        ### Sobre o Detector
+                        Este sistema utiliza um modelo de IA avançado para detectar objetos perigosos em vídeos.
+                        
+                        ### Tipos de Objetos Detectados
+                        - Armas de fogo (pistolas, rifles, etc.)
+                        - Armas brancas (facas, canivetes, etc.)
+                        - Objetos perigosos (bastões, objetos pontiagudos, etc.)
+                        
+                        ### Recomendações
+                        - Use vídeos com boa iluminação
+                        - Evite vídeos muito longos
+                        - Mantenha os objetos visíveis e em foco
+                        """)
+            # Vídeos de exemplo
+            if sample_videos:
+                with gr.Group():
+                    gr.Markdown("### Vídeos de Exemplo")
+                    with gr.Row():
+                        with gr.Column(scale=3):
+                            gr.Markdown("#### Vídeo")
+                        with gr.Column(scale=2):
+                            gr.Markdown("#### Status Real")
+                        with gr.Column(scale=1):
+                            gr.Markdown("#### Ação")
+                    
+                    for video in sample_videos:
+                        with gr.Row():
+                            with gr.Column(scale=3):
+                                gr.Video(
+                                    value=video['path'],
+                                    format="mp4",
+                                    height=150,
+                                    interactive=False,
+                                    show_label=False
+                                )
+                            with gr.Column(scale=2, min_width=200):
+                                gr.Markdown(video['ground_truth'])
+                            with gr.Column(scale=1, min_width=100):
+                                gr.Button(
+                                    "📥 Carregar",
+                                    size="sm"
+                                ).click(
+                                    fn=self.load_sample_video,
+                                    inputs=[gr.State(video['path'])],
+                                    outputs=[input_video]
+                                )
+                
+                
+            
+            # Configurar callback do botão
+            submit_btn.click(
+                fn=lambda *args: self._process_video(*args),
+                inputs=[
+                    input_video,
+                    threshold,
+                    fps,
+                    resolution,
+                    notification_type,
+                    notification_target
+                ],
+                outputs=[status, json_output]
+            )
+        
+        return demo
+    
+    def _process_video(
+        self,
+        video_path: str,
+        threshold: float = 0.5,
+        fps: int = None,
+        resolution: str = None,
+        notification_type: str = None,
+        notification_target: str = None
+    ) -> Tuple[str, Any]:
+        """Processa o vídeo usando o caso de uso."""
+        if not video_path:
+            return "Erro: Nenhum vídeo fornecido", {}
+            
+        # Usar valores padrão se não especificados
+        fps = fps or self.default_fps
+        resolution = resolution or self.default_resolution
+        
+        request = ProcessVideoRequest(
+            video_path=video_path,
+            threshold=threshold,
+            fps=fps,
+            resolution=int(resolution),
+            notification_type=notification_type,
+            notification_target=notification_target
+        )
+        
+        response = self.use_case.execute(request)
+        
+        return (
+            response.status_message,
+            response.detection_result.__dict__
+        ) 

tests/__init__.py

@@ -0,0 +1 @@
+ 

tests/conftest.py

@@ -0,0 +1,49 @@
+import pytest
+import os
+import sys
+from pathlib import Path
+
+# Adiciona o diretório src ao PYTHONPATH
+sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
+
+@pytest.fixture
+def sample_video_path():
+    """Retorna o caminho para um vídeo de teste"""
+    return str(Path(__file__).parent / "fixtures" / "sample_video.mp4")
+
+@pytest.fixture
+def mock_weapon_detector_service():
+    """Mock do serviço de detecção de armas"""
+    class MockWeaponDetectorService:
+        def detect(self, video_path, threshold=0.5):
+            return {
+                "detections": [
+                    {"label": "weapon", "confidence": 0.8, "bbox": [10, 10, 100, 100]},
+                ],
+                "frame_count": 30,
+                "processing_time": 1.5
+            }
+    
+    return MockWeaponDetectorService()
+
+@pytest.fixture
+def mock_notification_service():
+    """Mock do serviço de notificação"""
+    class MockNotificationService:
+        def send_notification(self, message, level="info"):
+            return {"status": "success", "message": message}
+    
+    return MockNotificationService()
+
+@pytest.fixture
+def mock_system_monitor():
+    """Mock do monitor de sistema"""
+    class MockSystemMonitor:
+        def get_system_info(self):
+            return {
+                "cpu_percent": 50.0,
+                "memory_percent": 60.0,
+                "gpu_info": {"name": "Test GPU", "memory_used": 1000}
+            }
+    
+    return MockSystemMonitor() 

tests/integration/__init__.py

@@ -0,0 +1 @@
+ 

tests/integration/test_gradio_interface.py

@@ -0,0 +1 @@
+ 

tests/unit/__init__.py

@@ -0,0 +1 @@
+ 

tests/unit/test_weapon_detector_service.py

@@ -0,0 +1 @@
+