jhonny-editor/code/engine/integracoes/visual/analisadores.py

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]))