import json from pathlib import Path import subprocess import sys from typing import Any, Callable from ...scanner.modelos import EvidenciaVisual from .contratos import AnalisadorDeFrame, AnalisadorDeSequenciaDeQuadros from .modelos import QuadroDeVideo class AnalisadorDeQualidadeOpenCV(AnalisadorDeFrame): """Mede nitidez, brilho e contraste de cada frame usando OpenCV.""" nome = "opencv" def __init__(self, cv2_module: Any | None = None) -> None: self._cv2 = cv2_module @property def cv2(self) -> Any: if self._cv2 is None: try: import cv2 except ImportError as exc: raise RuntimeError("OpenCV não está instalado. Instale opencv-python.") from exc self._cv2 = cv2 return self._cv2 def analisar(self, quadro: QuadroDeVideo) -> list[EvidenciaVisual]: imagem = quadro.imagem cinza = self.cv2.cvtColor(imagem, self.cv2.COLOR_BGR2GRAY) if len(imagem.shape) == 3 else imagem media, desvio = self.cv2.meanStdDev(cinza) nitidez = float(self.cv2.Laplacian(cinza, self.cv2.CV_64F).var()) valor = { "largura": quadro.largura, "altura": quadro.altura, "brilho": float(media[0][0]), "contraste": float(desvio[0][0]), "nitidez_laplaciana": nitidez, } return [EvidenciaVisual("qualidade", quadro.timestamp, quadro.timestamp, valor, provider=self.nome)] class _AdapterOpenCV: """Implementação compartilhada para adapters OpenCV, com importação tardia.""" def __init__(self, cv2_module: Any | None = None) -> None: self._cv2 = cv2_module @property def cv2(self) -> Any: if self._cv2 is None: try: import cv2 except ImportError as exc: raise RuntimeError("OpenCV não está instalado. Instale opencv-python.") from exc self._cv2 = cv2 return self._cv2 class AnalisadorDeRostosOpenCV(_AdapterOpenCV, AnalisadorDeFrame): """Encontra rostos e olhos localmente; não infere identidade nem emoção.""" nome = "opencv_haar" def __init__(self, escala: float = 1.1, vizinhos_minimos: int = 5, diretorio_de_modelos: str | Path | None = None, tamanho_minimo_relativo: float = 0.04, cv2_module: Any | None = None) -> None: super().__init__(cv2_module) self.escala = escala self.vizinhos_minimos = vizinhos_minimos self.diretorio_de_modelos = Path(diretorio_de_modelos) if diretorio_de_modelos else None self.tamanho_minimo_relativo = tamanho_minimo_relativo self._detectores: tuple[Any, Any] | None = None def analisar(self, quadro: QuadroDeVideo) -> list[EvidenciaVisual]: detector_de_rostos, detector_de_olhos = self._obter_detectores() cinza = self.cv2.cvtColor(quadro.imagem, self.cv2.COLOR_BGR2GRAY) rostos = detector_de_rostos.detectMultiScale(cinza, scaleFactor=self.escala, minNeighbors=self.vizinhos_minimos) evidencias: list[EvidenciaVisual] = [] for x, y, largura, altura in rostos: caixa = _caixa_normalizada(x, y, largura, altura, quadro.largura, quadro.altura) if min(caixa["largura"], caixa["altura"]) < self.tamanho_minimo_relativo: continue rosto = EvidenciaVisual("rosto", quadro.timestamp, quadro.timestamp, {"bounding_box": caixa, "proximo_da_borda": _proximo_da_borda(caixa)}, provider=self.nome) olhos = detector_de_olhos.detectMultiScale(cinza[y:y + altura, x:x + largura], scaleFactor=1.1, minNeighbors=4) evidencias.extend((rosto, EvidenciaVisual("olhos_visiveis", quadro.timestamp, quadro.timestamp, {"quantidade": len(olhos), "rosto": caixa}, provider=self.nome))) return evidencias def _obter_detectores(self) -> tuple[Any, Any]: if self._detectores is None: base = self.diretorio_de_modelos or Path(self.cv2.data.haarcascades) rostos = self.cv2.CascadeClassifier(str(base / "haarcascade_frontalface_default.xml")) olhos = self.cv2.CascadeClassifier(str(base / "haarcascade_eye.xml")) if rostos.empty() or olhos.empty(): raise RuntimeError(f"Cascades Haar do OpenCV não estão disponíveis em: {base}") self._detectores = rostos, olhos return self._detectores class AnalisadorDeComposicaoOpenCV(_AdapterOpenCV, AnalisadorDeFrame): """Mede orientação, poluição visual e disponibilidade das zonas para legendas.""" nome = "opencv_composicao" def analisar(self, quadro: QuadroDeVideo) -> list[EvidenciaVisual]: cinza = self.cv2.cvtColor(quadro.imagem, self.cv2.COLOR_BGR2GRAY) bordas = self.cv2.Canny(cinza, 80, 160) densidade_de_bordas = float((bordas > 0).mean()) terco_inferior = cinza[int(quadro.altura * 2 / 3):, :] zona_de_legenda_livre = float((self.cv2.Canny(terco_inferior, 80, 160) > 0).mean()) < 0.08 valor = { "orientacao": "vertical" if quadro.altura > quadro.largura else "horizontal", "densidade_de_bordas": densidade_de_bordas, "fundo_visual_poluido": densidade_de_bordas > 0.16, "zona_inferior_livre_para_legenda": zona_de_legenda_livre, } return [EvidenciaVisual("composicao", quadro.timestamp, quadro.timestamp, valor, provider=self.nome)] class AnalisadorDeTremorOpenCV(_AdapterOpenCV, AnalisadorDeSequenciaDeQuadros): """Estima deslocamento global entre frames para sinalizar possível tremor de câmera.""" nome = "opencv_movimento" def analisar(self, quadros: list[QuadroDeVideo]) -> list[EvidenciaVisual]: if len(quadros) < 2: return [] deslocamentos = [_deslocamento_global(self.cv2, anterior.imagem, atual.imagem) for anterior, atual in zip(quadros, quadros[1:])] deslocamentos = [item for item in deslocamentos if item is not None] if not deslocamentos: return [] medio = sum(deslocamentos) / len(deslocamentos) variacao = sum(abs(item - medio) for item in deslocamentos) / len(deslocamentos) inicio, fim = quadros[0].timestamp, quadros[-1].timestamp return [EvidenciaVisual("movimento_de_camera", inicio, fim, { "deslocamento_medio_pixels": medio, "variacao_do_deslocamento": variacao, "possivel_tremor": variacao > 2.0 and medio > 1.0, "amostras": len(deslocamentos), }, provider=self.nome)] class AnalisadorDeContinuidadeOpenCV(_AdapterOpenCV, AnalisadorDeSequenciaDeQuadros): """Compara pares de frames e retorna indícios, não certezas, de descontinuidade ou congelamento.""" nome = "opencv_continuidade" def analisar(self, quadros: list[QuadroDeVideo]) -> list[EvidenciaVisual]: evidencias: list[EvidenciaVisual] = [] for anterior, atual in zip(quadros, quadros[1:]): cinza_anterior = self.cv2.cvtColor(anterior.imagem, self.cv2.COLOR_BGR2GRAY) cinza_atual = self.cv2.cvtColor(atual.imagem, self.cv2.COLOR_BGR2GRAY) diferenca = float(self.cv2.absdiff(cinza_anterior, cinza_atual).mean()) evidencias.append(EvidenciaVisual("continuidade", anterior.timestamp, atual.timestamp, { "diferenca_media_de_luminancia": diferenca, "possivel_frame_congelado": diferenca < 0.8, "possivel_descontinuidade": diferenca > 38.0, }, provider=self.nome)) return evidencias class AnalisadorDeObjetosONNX(AnalisadorDeFrame): """Adapter de modelo ONNX; pré e pós-processamento pertencem ao modelo escolhido.""" nome = "onnxruntime" def __init__(self, modelo: str | Path, preparar: Callable[[Any, list[str]], dict[str, Any]], decodificar: Callable[[list[Any], QuadroDeVideo], list[dict[str, Any]]], usar_coreml: bool = True, ort_module: Any | None = None) -> None: self.modelo = str(modelo) self.preparar = preparar self.decodificar = decodificar self._ort = ort_module ort = self.ort preferidos = ["CoreMLExecutionProvider", "CPUExecutionProvider"] if usar_coreml else ["CPUExecutionProvider"] disponiveis = set(ort.get_available_providers()) self.providers = [provider for provider in preferidos if provider in disponiveis] if not self.providers: raise RuntimeError("ONNX Runtime não possui provider compatível neste ambiente.") self.sessao = ort.InferenceSession(self.modelo, providers=self.providers) self.entradas = [entrada.name for entrada in self.sessao.get_inputs()] @property def ort(self) -> Any: if self._ort is None: try: import onnxruntime except ImportError as exc: raise RuntimeError("ONNX Runtime não está instalado. Instale onnxruntime.") from exc self._ort = onnxruntime return self._ort def analisar(self, quadro: QuadroDeVideo) -> list[EvidenciaVisual]: entradas = self.preparar(quadro.imagem, self.entradas) saidas = self.sessao.run(None, entradas) deteccoes = self.decodificar(saidas, quadro) return [EvidenciaVisual("objeto", quadro.timestamp, quadro.timestamp, deteccao, confianca=deteccao.get("confianca"), provider=self.nome, modelo=Path(self.modelo).name) for deteccao in deteccoes] class AnalisadorDePoseMediaPipe(AnalisadorDeFrame): """Adapter MediaPipe Tasks para pose e mãos em um frame.""" nome = "mediapipe" def __init__(self, modelo_de_pose: str | Path | None = None, modelo_de_maos: str | Path | None = None) -> None: self.modelo_de_pose = Path(modelo_de_pose) if modelo_de_pose else None self.modelo_de_maos = Path(modelo_de_maos) if modelo_de_maos else None if self.modelo_de_pose is None and self.modelo_de_maos is None: raise ValueError("Informe ao menos um modelo MediaPipe de pose ou mãos.") def analisar(self, quadro: QuadroDeVideo) -> list[EvidenciaVisual]: if quadro.caminho is None: raise ValueError("MediaPipe requer que o frame tenha caminho persistido em disco.") for modelo in (self.modelo_de_pose, self.modelo_de_maos): if modelo: self._validar_modelo(modelo) carga = { "frame": str(quadro.caminho), "pose": str(self.modelo_de_pose) if self.modelo_de_pose else None, "maos": str(self.modelo_de_maos) if self.modelo_de_maos else None, } processo = subprocess.run( [sys.executable, "-m", "engine.integracoes.visual.mediapipe_runner"], input=json.dumps(carga), capture_output=True, text=True, timeout=60, ) if processo.returncode != 0: detalhe = processo.stderr.strip().splitlines()[-1] if processo.stderr.strip() else "falha nativa sem diagnóstico" raise RuntimeError(f"MediaPipe falhou em processo isolado (código {processo.returncode}): {detalhe}") dados = json.loads(processo.stdout) return [EvidenciaVisual(item["tipo"], quadro.timestamp, quadro.timestamp, item["valor"], provider=self.nome) for item in dados] @staticmethod def _validar_modelo(caminho: Path) -> None: if not caminho.is_file(): raise FileNotFoundError(f"Modelo MediaPipe não encontrado: {caminho}") class AnalisadorAppleVision(AnalisadorDeFrame): """Usa PyObjC/Vision para OCR e detecção de faces em frames persistidos.""" nome = "apple_vision" def __init__(self, reconhecer_texto: bool = True, detectar_faces: bool = True) -> None: self.reconhecer_texto = reconhecer_texto self.detectar_faces = detectar_faces def analisar(self, quadro: QuadroDeVideo) -> list[EvidenciaVisual]: if quadro.caminho is None: raise ValueError("Apple Vision requer que o frame tenha caminho persistido em disco.") try: from Foundation import NSURL from Vision import VNDetectFaceRectanglesRequest, VNImageRequestHandler, VNRecognizeTextRequest except ImportError as exc: raise RuntimeError("Apple Vision requer PyObjC e macOS.") from exc requisicoes = [] texto = VNRecognizeTextRequest.alloc().initWithCompletionHandler_(None) if self.reconhecer_texto else None faces = VNDetectFaceRectanglesRequest.alloc().initWithCompletionHandler_(None) if self.detectar_faces else None if texto: requisicoes.append(texto) if faces: requisicoes.append(faces) handler = VNImageRequestHandler.alloc().initWithURL_options_(NSURL.fileURLWithPath_(str(quadro.caminho)), {}) sucesso, erro = handler.performRequests_error_(requisicoes, None) if not sucesso: detalhe = str(erro) if erro else "o macOS não retornou detalhe; verifique Vision/Neural Engine no host" raise RuntimeError(f"Apple Vision falhou: {detalhe}") evidencias: list[EvidenciaVisual] = [] if texto: for observacao in texto.results() or []: candidatos = observacao.topCandidates_(1) if candidatos: candidato = candidatos[0] evidencias.append(EvidenciaVisual("ocr", quadro.timestamp, quadro.timestamp, {"texto": str(candidato.string()), "bounding_box": _retangulo(observacao.boundingBox())}, confianca=float(candidato.confidence()), provider=self.nome)) if faces: for observacao in faces.results() or []: evidencias.append(EvidenciaVisual("rosto", quadro.timestamp, quadro.timestamp, {"bounding_box": _retangulo(observacao.boundingBox())}, provider=self.nome)) return evidencias def _retangulo(retangulo: Any) -> dict[str, float]: return {"x": float(retangulo.origin.x), "y": float(retangulo.origin.y), "largura": float(retangulo.size.width), "altura": float(retangulo.size.height)} def _caixa_normalizada(x: int, y: int, largura: int, altura: int, largura_do_frame: int, altura_do_frame: int) -> dict[str, float]: return {"x": float(x / largura_do_frame), "y": float(y / altura_do_frame), "largura": float(largura / largura_do_frame), "altura": float(altura / altura_do_frame)} def _proximo_da_borda(caixa: dict[str, float], margem: float = 0.04) -> bool: return (caixa["x"] < margem or caixa["y"] < margem or caixa["x"] + caixa["largura"] > 1 - margem or caixa["y"] + caixa["altura"] > 1 - margem) def _deslocamento_global(cv2: Any, imagem_anterior: Any, imagem_atual: Any) -> float | None: """Retorna o módulo do deslocamento de câmera, descartando pares sem pontos úteis.""" import math anterior = cv2.cvtColor(imagem_anterior, cv2.COLOR_BGR2GRAY) atual = cv2.cvtColor(imagem_atual, cv2.COLOR_BGR2GRAY) pontos = cv2.goodFeaturesToTrack(anterior, maxCorners=150, qualityLevel=0.01, minDistance=12, blockSize=7) if pontos is None or len(pontos) < 8: return None encontrados, status, _ = cv2.calcOpticalFlowPyrLK(anterior, atual, pontos, None) if encontrados is None or status is None: return None origem = pontos[status.ravel() == 1] destino = encontrados[status.ravel() == 1] if len(origem) < 8: return None matriz, _ = cv2.estimateAffinePartial2D(origem, destino, method=cv2.RANSAC) if matriz is None: return None return math.hypot(float(matriz[0, 2]), float(matriz[1, 2]))