import cv2 import time import threading import queue import os import torch from ultralytics import YOLO import yt_dlp from config import Config class DetectionService: def __init__(self, socketio, db, session_id): self.socketio = socketio self.db = db self.session_id = session_id self.model = None self.cap = None self.running = False self.thread = None self.frame_queue = queue.Queue(maxsize=1) self.logged_ids = set() self.detected_count = 0 self.current_target_url = Config.TARGET_URL self.load_model() def load_model(self): try: device = 'cuda' if torch.cuda.is_available() else 'cpu' if not os.path.exists(Config.MODEL_PATH): self.model = YOLO('yolov8n.pt').to(device) else: self.model = YOLO(Config.MODEL_PATH).to(device) print(f"Modelo carregado. Dispositivo: {device.upper()}") except Exception as e: print(f"Erro ao carregar modelo: {e}") def get_stream_url(self, target_url): """Extrai URL de stream, especialmente para YouTube, HLS, RTSP, etc.""" final_url = target_url # Detectar tipo de stream is_hls = target_url.endswith('.m3u8') or target_url.endswith('.ts') is_rtsp = target_url.startswith('rtsp://') is_rtmp = target_url.startswith('rtmp://') is_http_stream = target_url.startswith('http://') or target_url.startswith('https://') # URLs diretas (HLS, RTSP, RTMP, HTTP) - usar diretamente if is_hls or is_rtsp or is_rtmp or (is_http_stream and not ("youtube.com" in target_url or "youtu.be" in target_url)): print(f"URL direta detectada: {target_url[:80]}...") return target_url # YouTube if "youtube.com" in target_url or "youtu.be" in target_url: try: ydl_opts = { 'format': 'best[ext=mp4]/best', 'quiet': False, 'no_warnings': False, 'ignoreerrors': False } with yt_dlp.YoutubeDL(ydl_opts) as ydl: info = ydl.extract_info(target_url, download=False) # Verificar se é uma live stream if info.get('is_live'): formats = info.get('formats', []) # Tentar encontrar formato de vídeo direto url_found = None for f in formats: if f.get('ext') in ('mp4', 'webm') and f.get('protocol') in ('https', 'http'): if f.get('vcodec') != 'none': # Tem vídeo url_found = f['url'] break if not url_found: # Tentar URL direta url_found = info.get('url') if url_found: return url_found else: raise Exception("Não foi possível extrair URL da stream ao vivo") else: # Vídeo normal formats = info.get('formats', []) url_found = next( (f['url'] for f in formats if f.get('ext') == 'mp4' and f.get('protocol') in ('https', 'http')), info.get('url')) return url_found if url_found else target_url except Exception as e: error_msg = str(e) print(f"Erro yt-dlp: {error_msg}") # Mensagens de erro mais amigáveis if "will begin in a few moments" in error_msg or "is not currently live" in error_msg: raise Exception("A transmissão ao vivo ainda não começou. Aguarde alguns momentos e tente novamente.") elif "Private video" in error_msg or "Video unavailable" in error_msg: raise Exception("Vídeo privado ou indisponível. Verifique se a URL está correta e se o vídeo é público.") elif "Sign in to confirm your age" in error_msg: raise Exception("Este vídeo requer autenticação. Use um vídeo público.") else: raise Exception(f"Erro ao acessar YouTube: {error_msg}") # Arquivo local elif os.path.exists(target_url): return target_url # URL direta HTTP/HTTPS (fallback) elif target_url.startswith('http://') or target_url.startswith('https://'): print(f"Usando URL HTTP/HTTPS direta: {target_url[:80]}...") return target_url else: return final_url def start_monitoring(self, target_url=None): if self.running: raise Exception("Monitoramento já está em execução") if target_url: self.current_target_url = target_url try: stream_url = self.get_stream_url(self.current_target_url) except Exception as e: raise e # Re-raise para que o erro seja capturado no endpoint if not stream_url: raise Exception("Não foi possível obter URL da stream") # Configurar VideoCapture com FFMPEG para suportar mais formatos # Para HLS/HTTP streams, usar CAP_FFMPEG if stream_url.startswith('http://') or stream_url.startswith('https://'): self.cap = cv2.VideoCapture(stream_url, cv2.CAP_FFMPEG) elif stream_url.startswith('rtsp://'): self.cap = cv2.VideoCapture(stream_url, cv2.CAP_FFMPEG) elif stream_url.startswith('rtmp://'): self.cap = cv2.VideoCapture(stream_url, cv2.CAP_FFMPEG) else: # Arquivo local ou outros formatos self.cap = cv2.VideoCapture(stream_url) if not self.cap.isOpened(): if self.cap: self.cap.release() raise Exception(f"Não foi possível abrir a stream: {stream_url[:100]}. Verifique se a URL está correta e se a transmissão está ativa.") self.running = True self.logged_ids = set() self.detected_count = 0 self.thread = threading.Thread(target=self.video_loop, daemon=True) self.thread.start() return True def stop_monitoring(self): self.running = False if self.thread and self.thread.is_alive(): self.thread.join(timeout=1.0) if self.cap and self.cap.isOpened(): self.cap.release() self.cap = None def video_loop(self): frame_count = 0 last_results = None while self.running and self.cap and self.cap.isOpened(): ret, frame = self.cap.read() if not ret: self.socketio.emit('stream_lost', {'message': 'Sinal de vídeo perdido'}, room=self.session_id) break frame_count += 1 h, w = frame.shape[:2] scale = Config.IMG_SIZE / max(h, w) nh, nw = int(h * scale), int(w * scale) if frame_count % Config.FRAME_SKIP == 0 or last_results is None: frame_resized = cv2.resize(frame, (nw, nh), interpolation=cv2.INTER_LINEAR) if frame_count % Config.FRAME_SKIP == 0: try: results = self.model.track( frame_resized, persist=True, conf=0.25, classes=[0], verbose=False, tracker="bytetrack.yaml", half=True ) last_results = results for result in results: if result.boxes and result.boxes.id is not None: ids = result.boxes.id.cpu().numpy().astype(int) for track_id in ids: if track_id not in self.logged_ids: self.logged_ids.add(track_id) self.detected_count += 1 # Salvar no banco from models import DetectionEvent event = DetectionEvent( track_id=int(track_id), session_id=self.session_id, detail="Detecção de Pessoa" ) self.db.session.add(event) self.db.session.commit() # Emitir evento via WebSocket self.socketio.emit('detection', { 'track_id': int(track_id), 'count': self.detected_count, 'timestamp': time.strftime('%H:%M:%S') }, room=self.session_id) # Emitir alarme self.socketio.emit('alarm', {'message': 'Nova detecção!'}, room=self.session_id) except Exception as e: print(f"Erro Inferência: {e}") # Renderizar frame if last_results: self.model.names[0] = 'Pessoa' res_plotted = last_results[0].plot(img=frame_resized.copy()) else: res_plotted = cv2.resize(frame, (nw, nh), interpolation=cv2.INTER_LINEAR) # Converter para base64 e enviar import base64 _, buffer = cv2.imencode('.jpg', res_plotted, [cv2.IMWRITE_JPEG_QUALITY, 85]) frame_base64 = base64.b64encode(buffer).decode('utf-8') self.socketio.emit('frame', { 'image': f'data:image/jpeg;base64,{frame_base64}', 'count': self.detected_count, 'timestamp': time.strftime('%d/%m/%Y %H:%M:%S') }, room=self.session_id) time.sleep(0.033) # ~30 FPS self.running = False