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