jhonny-editor/code/engine/testes/test_analise_visual_local.py

import unittest
from pathlib import Path
import json
import tempfile

from engine.dominio import Clipe, Faixa, IntervaloDeTempo, Timeline
from engine.integracoes.visual.contratos import (AnalisadorDeFrame, AnalisadorDeSequenciaDeQuadros,
                                                 DetectorDeIntervalosDeCena,
                                                 ExtratorDeQuadros)
from engine.integracoes.visual.modelos import QuadroDeVideo
from engine.integracoes.visual.extrator_open_cv import ExtratorDeQuadrosOpenCV
from engine.integracoes.visual.apple_vision import (AnalisadorAppleVisionNativo,
                                                     AnalisadorSequenciaAppleVisionNativo)
from engine.scanner.coordenacao import ContextoDeAnalise
from engine.scanner.modelos import Cena, EvidenciaVisual
from engine.scanner.visual import AnaliseVisualDaTimeline, DetectorDeCenas, ResultadoVisualDoClipe
from engine.scanner.relatorio_visual import gerar_relatorio_visual, gerar_resumo_visual_markdown
from engine.scanner.configuracao_visual import ConfiguracaoVisualDoScanner


class ExtratorSimulado(ExtratorDeQuadros):
    def extrair(self, clipe):
        return [
            QuadroDeVideo(0.0, 0, 1920, 1080, "frame-0"),
            QuadroDeVideo(1.0, 1, 1920, 1080, "frame-1"),
        ]


class AnalisadorSimulado(AnalisadorDeFrame):
    nome = "simulado"

    def analisar(self, quadro):
        return [EvidenciaVisual("qualidade", quadro.timestamp, quadro.timestamp,
                                {"nitidez_laplaciana": 42.0}, provider=self.nome)]


class DetectorSimulado(DetectorDeIntervalosDeCena):
    nome = "simulado"

    def detectar(self, clipe, quadros):
        return [Cena(0.0, 1.0, referencias=(0.0,)), Cena(1.0, 2.0, referencias=(1.0,))]


class AnalisadorTemporalSimulado(AnalisadorDeSequenciaDeQuadros):
    nome = "temporal_simulado"

    def analisar(self, quadros):
        return [EvidenciaVisual("continuidade", quadros[0].timestamp, quadros[-1].timestamp,
                                {"possivel_descontinuidade": False}, provider=self.nome)]


class ExecutorAppleSimulado:
    def executar(self, carga, timeout):
        self.carga, self.timeout = carga, timeout
        return {"evidencias": [{"tipo": "pessoa", "valor": {"ocupacao_do_quadro": 0.3},
                                 "confianca": 0.9, "modelo": "VNDetectHumanRectanglesRequest"}],
                "avisos": ["ocr: indisponível"]}


class ExecutorAppleSequenciaSimulado:
    def executar(self, carga, timeout):
        self.carga, self.timeout = carga, timeout
        return {"evidencias": [{"tipo": "similaridade_visual", "valor": {
                    "frame_inicial": 0, "frame_final": 1, "distancia_feature_print": 0.2,
                    "possivel_mudanca_de_cena": False,
                }, "modelo": "VNGenerateImageFeaturePrintRequest"}], "avisos": []}


class TesteAnaliseVisualLocal(unittest.TestCase):
    def setUp(self):
        self.clipe = Clipe("clip-1", "camera", IntervaloDeTempo(0, 2), arquivo="/video.mp4")

    def test_detector_envia_cada_frame_aos_analisadores_e_consolida_cenas(self):
        detector = DetectorDeCenas(ExtratorSimulado(), [AnalisadorSimulado()], [DetectorSimulado()])

        resultado = detector.detectar(self.clipe)

        self.assertEqual(resultado.identificador_do_clipe, "clip-1")
        self.assertEqual([item.inicio for item in resultado.evidencias], [0.0, 1.0])
        self.assertEqual(len(resultado.cenas), 1)
        self.assertEqual((resultado.cenas[0].inicio, resultado.cenas[0].fim), (0.0, 2.0))

    def test_etapa_da_timeline_guarda_evidencias_por_clipe(self):
        detector = DetectorDeCenas(ExtratorSimulado(), [AnalisadorSimulado()], [DetectorSimulado()])
        timeline = Timeline("timeline-1", "Principal", faixas=[
            Faixa("video-1", "V1", "video", 0, [self.clipe]),
        ])
        contexto = ContextoDeAnalise(timeline=timeline)

        AnaliseVisualDaTimeline(detector).executar(contexto)

        self.assertEqual(len(contexto.caracteristicas_visuais["clip-1"]["evidencias"]), 2)
        self.assertEqual(len(contexto.cenas), 1)

    def test_detector_executa_analisadores_de_sequencia_uma_vez_por_clipe(self):
        detector = DetectorDeCenas(ExtratorSimulado(), analisadores_de_sequencia=[AnalisadorTemporalSimulado()])

        resultado = detector.detectar(self.clipe)

        self.assertEqual(len(resultado.evidencias), 1)
        self.assertEqual(resultado.evidencias[0].tipo, "continuidade")

    def test_fluxo_produz_cena_visual_com_observacoes_e_continuidade(self):
        detector = DetectorDeCenas(
            ExtratorSimulado(), [AnalisadorSimulado()],
            analisadores_de_sequencia=[AnalisadorTemporalSimulado()],
        )
        contexto = ContextoDeAnalise(timeline=Timeline(
            "timeline-1", "Principal", faixas=[Faixa("video-1", "V1", "video", 0, [self.clipe])]
        ))

        AnaliseVisualDaTimeline(detector).executar(contexto)

        cena = contexto.cenas_visuais[0]
        self.assertEqual(cena.identificador_do_clipe, "clip-1")
        self.assertEqual([item.tipo for item in cena.observacoes], ["qualidade", "qualidade", "continuidade"])
        self.assertTrue(contexto.caracteristicas_visuais["clip-1"]["disponivel"])

    def test_falha_de_adapter_e_registrada_sem_interromper_outro_clipe(self):
        class ExtratorQueFalha(ExtratorDeQuadros):
            def extrair(self, clipe):
                if clipe.identificador == "clip-1":
                    raise RuntimeError("OpenCV não instalado")
                return [QuadroDeVideo(0.0, 0, 1920, 1080, "frame")]

        segundo = Clipe("clip-2", "camera-2", IntervaloDeTempo(2, 3), arquivo="/video-2.mp4")
        timeline = Timeline("timeline-1", "Principal", faixas=[
            Faixa("video-1", "V1", "video", 0, [self.clipe, segundo]),
        ])
        contexto = ContextoDeAnalise(timeline=timeline)
        detector = DetectorDeCenas(ExtratorQueFalha(), [AnalisadorSimulado()])

        AnaliseVisualDaTimeline(detector).executar(contexto)

        self.assertFalse(contexto.caracteristicas_visuais["clip-1"]["disponivel"])
        self.assertTrue(contexto.caracteristicas_visuais["clip-2"]["disponivel"])
        self.assertEqual(len(contexto.avisos), 1)

    def test_adapter_apple_converte_fatos_e_preserva_avisos_por_recurso(self):
        executor = ExecutorAppleSimulado()
        quadro = QuadroDeVideo(0.0, 0, 1920, 1080, "frame", Path("/frame.jpg"))

        evidencias = AnalisadorAppleVisionNativo(executor=executor).analisar(quadro)

        self.assertEqual(evidencias[0].tipo, "pessoa")
        self.assertEqual(evidencias[0].modelo, "VNDetectHumanRectanglesRequest")
        self.assertEqual(evidencias[1].tipo, "diagnostico_apple_vision")
        self.assertTrue(executor.carga["resumir"])

    def test_adapter_apple_de_sequencia_converte_indices_em_intervalos(self):
        executor = ExecutorAppleSequenciaSimulado()
        quadros = [QuadroDeVideo(2.0, 0, 1920, 1080, "frame", Path("/frame-0.jpg")),
                   QuadroDeVideo(3.0, 1, 1920, 1080, "frame", Path("/frame-1.jpg"))]

        evidencias = AnalisadorSequenciaAppleVisionNativo(executor=executor).analisar(quadros)

        self.assertEqual(evidencias[0].tipo, "similaridade_visual")
        self.assertEqual((evidencias[0].inicio, evidencias[0].fim), (2.0, 3.0))
        self.assertEqual(evidencias[0].valor["distancia_feature_print"], 0.2)
        self.assertEqual(executor.carga["frames"], ["/frame-0.jpg", "/frame-1.jpg"])

    def test_evidencia_rejeita_intervalo_invalido(self):
        with self.assertRaises(ValueError):
            EvidenciaVisual("objeto", 2.0, 1.0, {})

    def test_relatorio_visual_separa_frames_de_evidencias_temporais(self):
        resultado = ResultadoVisualDoClipe("clip-1", [
            EvidenciaVisual("categoria", 2.0, 2.0, {"identificador": "drink"}),
            EvidenciaVisual("similaridade_visual", 2.0, 3.0, {"distancia": 0.2}),
        ])
        with tempfile.TemporaryDirectory() as pasta:
            caminho = gerar_relatorio_visual(self.clipe, resultado, Path(pasta) / "visual.json", 1.0)
            dados = json.loads(caminho.read_text(encoding="utf-8"))

        self.assertEqual(dados["clipe"]["identificador"], "clip-1")
        self.assertEqual(dados["observacoes_por_frame"][0]["timestamp_na_origem"], 2.0)
        self.assertEqual(dados["observacoes_por_frame"][0]["evidencias"][0]["tipo"], "categoria")
        self.assertEqual(dados["evidencias_temporais"][0]["tipo"], "similaridade_visual")

    def test_resumo_visual_markdown_lista_frames_e_transicoes(self):
        resultado = ResultadoVisualDoClipe("clip-1", [
            EvidenciaVisual("categoria", 2.0, 2.0, {"identificador": "drink"}),
            EvidenciaVisual("similaridade_visual", 2.0, 3.0, {"distancia": 0.2}),
        ])
        with tempfile.TemporaryDirectory() as pasta:
            caminho = gerar_resumo_visual_markdown(self.clipe, resultado, Path(pasta) / "visual.md", 1.0)
            conteudo = caminho.read_text(encoding="utf-8")

        self.assertIn("Origem 2.000s", conteudo)
        self.assertIn("similaridade_visual", conteudo)

    def test_configuracao_visual_valida_o_perfil_do_painel(self):
        perfil = ConfiguracaoVisualDoScanner.de_dict({
            "intervalo_em_segundos": 5,
            "faixas_de_video": [2, 0, 2],
            "recursos_vision": ["faces", "pessoas"],
            "analisar_continuidade": False,
        })

        self.assertEqual(perfil.faixas_de_video, (0, 2))
        self.assertEqual(perfil.para_dict()["recursos_vision"], ["faces", "pessoas"])
        self.assertFalse(perfil.analisar_continuidade)

    def test_extrator_respeita_intervalo_na_origem_e_limite_de_frames(self):
        clipe = Clipe("clip-1", "camera", IntervaloDeTempo(0, 2),
                      intervalo_na_origem=IntervaloDeTempo(2.3, 5.0), arquivo="/video.mp4")
        extrator = ExtratorDeQuadrosOpenCV(intervalo_em_segundos=1.0, maximo_de_quadros=1,
                                            cv2_module=object())

        inicio, fim = extrator._intervalo_de_origem(clipe, 10.0)

        self.assertEqual(extrator._timestamps(inicio, fim), [2.3])


if __name__ == "__main__":
    unittest.main()