# app.py import tempfile import requests import streamlit as st from services.pdf_processor import extract_text_from_pdf from services.config import ORQUESTRATOR_URL import tempfile import cv2 import time import base64 from dotenv import load_dotenv load_dotenv() # ========================= # 🔧 UTIL # ========================= def encode_file(path): with open(path, "rb") as f: return base64.b64encode(f.read()).decode() # ========================= # 🧠 UI # ========================= st.title("Análise Multimodal com Suporte Clínico") # ========================= # ⚙️ CONFIGURAÇÕES # ========================= MAX_PDF_CHARS = 2000 # Caracteres do PDF MAX_TEXT_CHARS = 300 # Caracteres do Text Area MAX_VIDEO_DURATION = 150 # Duração máxima (segundos) MAX_VIDEO_FPS = 300 # FPS limite # ========================= # 📥 Uploads # ========================= video_file = st.file_uploader("Upload de vídeo", type=["mp4"]) pdf_file = st.file_uploader("Upload de laudo PDF (opcional)", type=["pdf"]) manual_text = st.text_area( f"Informações clínicas adicionais (máx {MAX_TEXT_CHARS} caracteres)", max_chars=MAX_TEXT_CHARS ) # ========================= # ▶️ BOTÃO # ========================= if st.button("Analisar"): if video_file is None: st.error("Envie um vídeo") st.stop() # ========================= # 🎥 VALIDAR VÍDEO # ========================= with tempfile.NamedTemporaryFile(delete=False, suffix=".mp4") as tmp_video: tmp_video.write(video_file.read()) video_path = tmp_video.name cap = cv2.VideoCapture(video_path) fps = cap.get(cv2.CAP_PROP_FPS) frame_count = cap.get(cv2.CAP_PROP_FRAME_COUNT) duration = frame_count / fps if fps > 0 else 0 cap.release() if duration > MAX_VIDEO_DURATION: st.error(f"Vídeo muito longo ({duration:.1f}s). Máx permitido: {MAX_VIDEO_DURATION}s") st.stop() if fps > MAX_VIDEO_FPS: st.error(f"FPS muito alto ({fps:.1f}). Máx permitido: {MAX_VIDEO_FPS}") st.stop() # ========================= # 📄 PROCESSAR PDF # ========================= medical_text = "" if pdf_file: extracted = extract_text_from_pdf(pdf_file) if len(extracted) > MAX_PDF_CHARS: st.warning(f"PDF truncado para {MAX_PDF_CHARS} caracteres") extracted = extracted[:MAX_PDF_CHARS] medical_text += extracted if manual_text: medical_text += "\n\nInformações clínicas adicionais:\n" + manual_text # ========================= # 📡 ENVIAR JOB # ========================= video_base64 = encode_file(video_path) data = { "video_base64": video_base64, "audio_base64": None, "medical_text": medical_text, } with st.spinner("Enviando para processamento..."): try: response = requests.post( ORQUESTRATOR_URL + "/analyze", json=data, timeout=180 ) except Exception as e: st.error(f"Erro ao conectar com backend: {e}") st.stop() if response.status_code != 200: st.error(f"Erro no backend: {response.status_code}") st.text(response.text) st.stop() job_id = response.json().get("job_id") if not job_id: st.error("Job ID não retornado") st.stop() st.success(f"Job criado: {job_id}") # ========================= # 🔄 POLLING RESULTADO # ========================= progress = st.progress(0) status_text = st.empty() start_time = time.time() timeout = 300 # 5 minutos while True: if time.time() - start_time > timeout: st.error("Timeout aguardando resultado") break try: result = requests.get( ORQUESTRATOR_URL + f"/result/{job_id}", timeout=10 ).json() except Exception: st.warning("Erro ao consultar resultado") time.sleep(2) continue status = result.get("status") if status == "queued": status_text.info("Na fila...") progress.progress(20) elif status == "processing": status_text.info("Processando...") progress.progress(60) elif status == "done": progress.progress(100) status_text.success("Processamento concluído!") final = result.get("result") if not final: st.error("Resultado vazio") break # ========================= # 🧠 RESULTADO COMPLETO # ========================= llm = final.get("llm_analysis", {}) ml = final.get("ml_decision", {}) emotion = final.get("emotion", {}) audio = final.get("audio", {}) video = final.get("video", {}) # ========================= # 🚨 LLM (CAMADA CLÍNICA) # ========================= st.header("🚨 Análise Clínica (LLM)") col1, col2 = st.columns(2) with col1: st.subheader("Nível de Risco") risk = llm.get("risk_level", "desconhecido") if risk.lower() in ["alto", "high"]: st.error(risk) elif risk.lower() in ["médio", "medio"]: st.warning(risk) else: st.success(risk) with col2: st.subheader("Revisão Humana") if llm.get("requires_human_review"): st.error("⚠️ Necessária") else: st.success("Não necessária") st.subheader("Alerta") st.warning(llm.get("alert")) st.subheader("Explicação") st.write(llm.get("explanation")) st.subheader("Recomendação") st.info(llm.get("recommendation")) st.caption(llm.get("disclaimer")) st.caption(f"Fonte: {llm.get('source')}") # ========================= # 🤖 DECISÃO ML # ========================= st.header("🤖 Decisão do Modelo (ML)") col1, col2 = st.columns(2) with col1: st.subheader("Classificação") label = ml.get("label") if "AGRESSAO" in label: st.error(label) elif "ESTRESSE" in label: st.warning(label) elif "APATIA" in label: st.info(label) elif "INQUIETACAO" in label: st.warning(label) elif "AGITACAO" in label: st.warning(label) else: st.success(label) with col2: st.metric("Confiança", f"{ml.get('confidence', 0):.2f}") st.subheader("Evidências") evidence = ml.get("evidence", {}) st.subheader("📊 Indicadores comportamentais") col1, col2, col3 = st.columns(3) with col1: st.metric("Head Movement", f"{evidence.get('head_movement', 0):.3f}") with col2: st.metric("Arm Movement", f"{evidence.get('arm_movement', 0):.3f}") with col3: st.metric("Leg Movement", f"{evidence.get('leg_movement', 0):.3f}") st.subheader("🧠 Interpretação do comportamento") if evidence.get("leg_movement", 0) > evidence.get("arm_movement", 0) * 2: st.info("Movimento predominante nas pernas (possível inquietação)") if evidence.get("arm_movement", 0) > 0.04: st.info("Movimentação elevada de braços (possível agitação)") if evidence.get("head_movement", 0) > 0.02: st.info("Movimento de cabeça elevado (possível desconforto)") st.json(evidence) # ========================= # 😀 EMOÇÃO FACIAL # ========================= st.header("😀 Emoção Facial") st.write(f"Emoção detectada: **{emotion.get('emotion')}**") # ========================= # 🎤 ÁUDIO # ========================= st.header("🎤 Análise de Áudio") st.subheader("Transcrição") st.write(audio.get("transcription")) st.subheader("Tradução") st.write(audio.get("translation")) st.subheader("Sentimento") sentiment = audio.get("sentiment", {}) st.write(f"Emoção: {sentiment.get('emotion')}") st.write(f"Confiança: {sentiment.get('confidence')}") # ========================= # 🎥 VÍDEO # ========================= st.header("🎥 Análise de Vídeo") video_result = video.get("result", {}) motion = video_result.get("motion", {}) col1, col2, col3 = st.columns(3) with col1: st.metric("Movimento total", f"{motion.get('motion', 0):.3f}") st.metric("Movimento corporal", f"{motion.get('body_movement', 0):.3f}") with col2: st.metric("Mov. cabeça", f"{motion.get('head_movement', 0):.3f}") st.metric("Inclinação cabeça", f"{motion.get('head_tilt', 0):.3f}") with col3: st.metric("Mov. braços", f"{motion.get('arm_movement', 0):.3f}") st.metric("Mov. pernas", f"{motion.get('leg_movement', 0):.3f}") st.write(f"Pessoa detectada: {motion.get('person_detected')}") # ========================= # 📊 DEBUG COMPLETO # ========================= with st.expander("🔍 JSON completo"): st.json(final) break else: status_text.warning("Status desconhecido") time.sleep(2)