#--- START OF FILE app.py --- import gradio as gr import cv2 import numpy as np from skimage.metrics import structural_similarity as ssim import matplotlib.pyplot as plt from PIL import Image import imagehash import torch import collections # Adicionado para a nova lógica de visualização import os # --- MÓDULO DE INICIALIZAÇÃO --- # Tenta carregar os modelos e módulos, definindo flags de disponibilidade. CLIP_AVAILABLE, SALIENCY_AVAILABLE = False, False try: from transformers import CLIPProcessor, CLIPModel MODEL_ID = "openai/clip-vit-base-patch32" clip_model = CLIPModel.from_pretrained(MODEL_ID) clip_processor = CLIPProcessor.from_pretrained(MODEL_ID) CLIP_AVAILABLE = True print("Modelo CLIP carregado com sucesso.") except Exception as e: print(f"AVISO: Modelo CLIP não carregado. Teste de Inteligência desabilitado. Erro: {e}") try: saliency_detector = cv2.saliency.StaticSaliencySpectralResidual_create() SALIENCY_AVAILABLE = True print("Módulo de Saliência carregado com sucesso.") except AttributeError: print("AVISO: Módulo de Saliência não encontrado. Análise de Foco Móvel desabilitada.") print("Certifique-se de que 'opencv-contrib-python-headless' está no requirements.txt") except Exception as e: print(f"AVISO: Módulo de Saliência não carregado. Erro: {e}") # --- FUNÇÕES DE ANÁLISE --- def analisar_fidelidade(video_path, start_frame=0, end_frame=0): """Lê um trecho do vídeo e calcula as métricas de fidelidade (SSIM, pHash).""" cap = cv2.VideoCapture(video_path) if not cap.isOpened(): raise gr.Error("Não foi possível abrir o arquivo de vídeo.") frames, ssim_scores, phash_distances = [], [], [] fps = cap.get(cv2.CAP_PROP_FPS) or 30 total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) # Valida e ajusta os frames de início e fim. Se inválido, analisa o vídeo todo. process_full_video = (start_frame == 0 and end_frame == 0) or (start_frame >= end_frame) if process_full_video: start_frame = 0 end_frame = total_frames else: start_frame = max(0, start_frame) end_frame = min(total_frames, end_frame) cap.set(cv2.CAP_PROP_POS_FRAMES, start_frame) current_frame_pos = start_frame while current_frame_pos < end_frame: ret, frame = cap.read() if not ret: break frames.append(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)) current_frame_pos += 1 cap.release() if len(frames) < 2: gr.Warning("O trecho selecionado é muito curto para análise (menos de 2 frames).") return frames, fps, [], [] for i in range(len(frames) - 1): gray1 = cv2.cvtColor(frames[i], cv2.COLOR_RGB2GRAY) gray2 = cv2.cvtColor(frames[i+1], cv2.COLOR_RGB2GRAY) ssim_val, _ = ssim(gray1, gray2, full=True, data_range=gray1.max() - gray1.min()) pil_img1 = Image.fromarray(frames[i]) pil_img2 = Image.fromarray(frames[i+1]) phash_dist = imagehash.phash(pil_img1) - imagehash.phash(pil_img2) ssim_scores.append(ssim_val) phash_distances.append(phash_dist) return frames, fps, ssim_scores, phash_distances def analisar_cor_iluminacao(frames): lum_corr_scores, color_corr_scores = [], [] if len(frames) < 2: return [], [] for i in range(len(frames) - 1): frame1, frame2 = frames[i], frames[i+1] gray1, gray2 = cv2.cvtColor(frame1, cv2.COLOR_RGB2GRAY), cv2.cvtColor(frame2, cv2.COLOR_RGB2GRAY) hist1_lum = cv2.calcHist([gray1], [0], None, [256], [0,256]) hist2_lum = cv2.calcHist([gray2], [0], None, [256], [0,256]) cv2.normalize(hist1_lum, hist1_lum, alpha=0, beta=1, norm_type=cv2.NORM_MINMAX) cv2.normalize(hist2_lum, hist2_lum, alpha=0, beta=1, norm_type=cv2.NORM_MINMAX) lum_corr = cv2.compareHist(hist1_lum, hist2_lum, cv2.HISTCMP_CORREL) lum_corr_scores.append(lum_corr) corrs = [] for chan in range(3): hist1 = cv2.calcHist([frame1],[chan],None,[256],[0,256]) hist2 = cv2.calcHist([frame2],[chan],None,[256],[0,256]) cv2.normalize(hist1, hist1, alpha=0, beta=1, norm_type=cv2.NORM_MINMAX) cv2.normalize(hist2, hist2, alpha=0, beta=1, norm_type=cv2.NORM_MINMAX) corrs.append(cv2.compareHist(hist1, hist2, cv2.HISTCMP_CORREL)) color_corr_scores.append(np.mean(corrs)) return lum_corr_scores, color_corr_scores def analisar_anomalias_movimento(frames): magnitude_scores, orientation_variance_scores = [], [] if len(frames) < 2: return [], [] prev_gray = cv2.cvtColor(frames[0], cv2.COLOR_RGB2GRAY) for i in range(1, len(frames)): current_gray = cv2.cvtColor(frames[i], cv2.COLOR_RGB2GRAY) flow = cv2.calcOpticalFlowFarneback(prev_gray, current_gray, None, 0.5, 3, 15, 3, 5, 1.2, 0) magnitude, angle = cv2.cartToPolar(flow[...,0], flow[...,1], angleInDegrees=True) magnitude_scores.append(np.mean(magnitude)) orientation_variance_scores.append(np.var(angle)) prev_gray = current_gray return magnitude_scores, orientation_variance_scores def analisar_estabilidade_foco(frames): ssim_foco_scores, jitter_foco_scores = [], [] if len(frames) < 2: return [], [] last_roi_center = None for i in range(len(frames) - 1): frame1_cv, frame2_cv = cv2.cvtColor(frames[i], cv2.COLOR_RGB2BGR), cv2.cvtColor(frames[i+1], cv2.COLOR_RGB2BGR) try: _, saliencyMap1 = saliency_detector.computeSaliency(frame1_cv) _, saliencyMap2 = saliency_detector.computeSaliency(frame2_cv) saliencyMap1_8bit, saliencyMap2_8bit = (saliencyMap1 * 255).astype("uint8"), (saliencyMap2 * 255).astype("uint8") _, thresh1 = cv2.threshold(saliencyMap1_8bit, 0, 255, cv2.THRESH_BINARY | cv2.THRESH_OTSU) _, thresh2 = cv2.threshold(saliencyMap2_8bit, 0, 255, cv2.THRESH_BINARY | cv2.THRESH_OTSU) contours1, _ = cv2.findContours(thresh1, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE) contours2, _ = cv2.findContours(thresh2, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE) if contours1 and contours2: x1, y1, w1, h1 = cv2.boundingRect(max(contours1, key=cv2.contourArea)) x2, y2, w2, h2 = cv2.boundingRect(max(contours2, key=cv2.contourArea)) roi1_gray, roi2_gray = cv2.cvtColor(frames[i][y1:y1+h1, x1:x1+w1], cv2.COLOR_RGB2GRAY), cv2.cvtColor(frames[i+1][y2:y2+h2, x2:x2+w2], cv2.COLOR_RGB2GRAY) if roi1_gray.size == 0 or roi2_gray.size == 0: raise ValueError("ROI vazia") roi2_gray_resized = cv2.resize(roi2_gray, (roi1_gray.shape[1], roi1_gray.shape[0])) ssim_foco, _ = ssim(roi1_gray, roi2_gray_resized, full=True, data_range=255) if min(roi1_gray.shape) > 7 else (0, None) ssim_foco_scores.append(ssim_foco) center = (x1 + w1/2, y1 + h1/2) jitter_foco_scores.append(np.linalg.norm(np.array(center) - np.array(last_roi_center)) if last_roi_center else 0) last_roi_center = center else: ssim_foco_scores.append(0); jitter_foco_scores.append(0) except Exception: ssim_foco_scores.append(0); jitter_foco_scores.append(0) return ssim_foco_scores, jitter_foco_scores def executar_teste_semantico(phash_distances, descriptions_text): # (Função como definida anteriormente) return None, "Função ainda não implementada completamente no template" def extrair_e_visualizar_frames(video_path, start_frame, end_frame): """Extrai frames de um trecho e os organiza em uma grade alinhada a múltiplos de 8.""" if not video_path or start_frame >= end_frame: return None cap = cv2.VideoCapture(video_path) if not cap.isOpened(): return None total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) start_frame = max(0, start_frame) end_frame = min(total_frames, end_frame) cap.set(cv2.CAP_PROP_POS_FRAMES, start_frame) extracted_frames, frame_shape = {}, None for i in range(start_frame, end_frame): ret, frame = cap.read() if not ret: break if frame_shape is None: frame_shape = frame.shape frame_num_text = f"Frame: {i}" cv2.putText(frame, frame_num_text, (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 0, 255), 2, cv2.LINE_AA) extracted_frames[i] = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) cap.release() if not extracted_frames: return None # Cria um frame preto para usar como preenchimento h, w, c = frame_shape black_frame = np.zeros((h, w, c), dtype=np.uint8) cv2.putText(black_frame, "VAZIO", (w//2 - 50, h//2), cv2.FONT_HERSHEY_SIMPLEX, 1, (100, 100, 100), 2) # Determina o início e o fim da grade, alinhados a múltiplos de 8 grid_start = (start_frame // 8) * 8 grid_end = ((end_frame - 1) // 8 + 1) * 8 # Preenche um dicionário com os frames reais ou de preenchimento rows_dict = collections.defaultdict(list) for i in range(grid_start, grid_end): row_index = i // 8 rows_dict[row_index].append(extracted_frames.get(i, black_frame)) # Monta as linhas horizontalmente e depois as empilha verticalmente image_rows = [np.hstack(rows_dict[row_index]) for row_index in sorted(rows_dict.keys())] combined_image = np.vstack(image_rows) # Salva e retorna o caminho da imagem final pil_img = Image.fromarray(combined_image) path = "frames_extraidos_grid.png" pil_img.save(path) return path # --- FUNÇÕES DE PLOTAGEM --- def plot_to_file(fig, filename): path = f"{filename}.png" fig.savefig(path) plt.close(fig) return path def gerar_grafico_fidelidade(ssim, phash, num_frames, fps): if not ssim: return None fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(16, 10), sharex=True) x_axis = [i/fps for i in range(len(ssim))] ax1.plot(x_axis, ssim, label='SSIM') ax2.plot(x_axis, phash, label='pHash Distance', color='red') ax1.set_title('Fidelidade Estrutural (SSIM)'); ax2.set_title('Mudança Perceptual (pHash)') return plot_to_file(fig, "fidelidade") def gerar_grafico_cor(lum, color, num_frames, fps): if not lum: return None fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(16, 10), sharex=True) x_axis = [i/fps for i in range(len(lum))] ax1.plot(x_axis, lum, label='Luminância', color='gold') ax2.plot(x_axis, color, label='Cor (RGB)', color='magenta') ax1.set_title('Consistência de Iluminação'); ax2.set_title('Consistência de Cor') return plot_to_file(fig, "cor") def gerar_grafico_anomalias(mag, var, num_frames, fps): if not mag: return None fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(16, 10), sharex=True) x_axis = [i/fps for i in range(len(mag))] ax1.plot(x_axis, mag, label='Magnitude') ax2.plot(x_axis, var, label='Variância de Orientação', color='red') ax1.set_title('Magnitude do Movimento'); ax2.set_title('Incoerência de Movimento (Glitches)') return plot_to_file(fig, "anomalias") def gerar_grafico_foco(ssim_global, ssim_foco, jitter, num_frames, fps): if not ssim_global: return None fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(16, 10), sharex=True) x_axis = [i/fps for i in range(len(ssim_global))] ax1.plot(x_axis, ssim_global, label='Global') ax1.plot(x_axis, ssim_foco, label='Foco') ax2.plot(x_axis, jitter, label='Jitter', color='coral') ax1.set_title('Estabilidade de Foco (SSIM)'); ax2.set_title('Jitter de Foco') return plot_to_file(fig, "foco") # --- FUNÇÃO DE CALLBACK PRINCIPAL --- def run_full_analysis(video_path, descriptions_text, start_frame, end_frame, progress=gr.Progress()): if video_path is None: raise gr.Error("Por favor, faça o upload de um vídeo.") start_frame, end_frame = int(start_frame), int(end_frame) progress(0, desc="Extraindo e visualizando frames...") frames_imagem_path = extrair_e_visualizar_frames(video_path, start_frame, end_frame) progress(0.1, desc="Analisando fidelidade do trecho...") frames, fps, ssim_scores, phash_distances = analisar_fidelidade(video_path, start_frame, end_frame) progress(0.2, desc="Gerando gráfico de fidelidade...") fidelidade_plot_path = gerar_grafico_fidelidade(ssim_scores, phash_distances, len(frames), fps) progress(0.3, desc="Analisando cor e iluminação...") lum_scores, color_scores = analisar_cor_iluminacao(frames) cor_plot_path = gerar_grafico_cor(lum_scores, color_scores, len(frames), fps) progress(0.4, desc="Analisando glitches de movimento...") mag_scores, var_scores = analisar_anomalias_movimento(frames) anomalias_plot_path = gerar_grafico_anomalias(mag_scores, var_scores, len(frames), fps) foco_plot_path = None if SALIENCY_AVAILABLE: progress(0.6, desc="Analisando foco móvel...") ssim_foco, jitter_foco = analisar_estabilidade_foco(frames) foco_plot_path = gerar_grafico_foco(ssim_scores, ssim_foco, jitter_foco, len(frames), fps) semantico_path = None if CLIP_AVAILABLE and descriptions_text.strip(): progress(0.8, desc="Executando teste semântico...") semantico_path, error_msg = executar_teste_semantico(phash_distances, descriptions_text) if error_msg: gr.Warning(error_msg) progress(1.0, desc="Análise completa!") return frames_imagem_path, fidelidade_plot_path, cor_plot_path, foco_plot_path, semantico_path, anomalias_plot_path # --- INTERFACE GRADIO --- with gr.Blocks(theme=gr.themes.Soft()) as demo: gr.Markdown("# Suíte de Validação Completa para Geração de Vídeo (ADUC-SDR)") with gr.Row(): with gr.Column(scale=1): video_input = gr.Video(label="1. Upload do vídeo") descriptions_input = gr.Textbox(lines=5, label="3. Descrições (Opcional)", placeholder="Uma descrição por cena para o teste de inteligência...") with gr.Row(): start_frame_input = gr.Number(label="Frame Inicial", value=0, precision=0, info="Deixe 0 e 0 para analisar o vídeo inteiro.") end_frame_input = gr.Number(label="Frame Final", value=0, precision=0, info="Ex: 88 e 94 para analisar esse trecho.") analyze_button = gr.Button("4. Executar Análise Completa", variant="primary") with gr.Tabs(): with gr.TabItem("Frames Extraídos"): extracted_frames_img = gr.Image(label="Visualização dos Frames em Grade") with gr.TabItem("1. Fidelidade e Coerência"): plot_fidelidade = gr.Image(label="Gráfico de Análise de Fidelidade (SSIM e pHash)") with gr.TabItem("2. Cor e Iluminação"): plot_cor = gr.Image(label="Gráfico de Análise de Cor e Luminância") with gr.TabItem("3. Foco (Vídeo Móvel)"): plot_foco = gr.Image(label="Gráfico de Análise de Foco e Jitter") with gr.TabItem("4. Glitches de Movimento"): plot_anomalias = gr.Image(label="Gráfico do Detector de Anomalias de Movimento") with gr.TabItem("5. Inteligência Adaptativa"): plot_semantico = gr.Image(label="Gráfico de Estresse Semântico") analyze_button.click( fn=run_full_analysis, inputs=[video_input, descriptions_input, start_frame_input, end_frame_input], outputs=[extracted_frames_img, plot_fidelidade, plot_cor, plot_foco, plot_semantico, plot_anomalias] # <-- LINHA CORRIGIDA ) if __name__ == "__main__": demo.queue().launch()