import os import tempfile import hashlib import gradio as gr from embedding_manager import EmbeddingManager from summarizer import PatientChartSummarizer from hedis_engine import HedisComplianceEngine APP_TITLE = "ChartWise AI" DEFAULT_MEASURE_YEAR = 2024 # --- Simple in-process cache (per Space runtime) --- # Maps content_hash -> vectordb _VDB_CACHE = {} _LAST_HASH = None # optional: for quick reuse/debug def _pdf_hash(pdf_bytes: bytes) -> str: return hashlib.sha256(pdf_bytes).hexdigest() def _get_vectordb_from_bytes(pdf_bytes: bytes): """ Return a cached vectordb for this PDF content if available, otherwise build it once and cache it for subsequent calls. """ global _VDB_CACHE, _LAST_HASH content_hash = _pdf_hash(pdf_bytes) _LAST_HASH = content_hash if content_hash in _VDB_CACHE: print(f"[CACHE] Using cached vectordb for hash {content_hash[:12]}...") return _VDB_CACHE[content_hash] # Not cached yet: create temp file, build embeddings once, then cache with tempfile.NamedTemporaryFile(delete=False, suffix=".pdf") as tmp: tmp.write(pdf_bytes) pdf_path = tmp.name try: manager = EmbeddingManager(pdf_path) # persists under ./embeddings/ vectordb = manager.get_or_create_embeddings() _VDB_CACHE[content_hash] = vectordb print(f"[CACHE] Cached vectordb for hash {content_hash[:12]}.") return vectordb finally: # Remove only the temp PDF; embeddings are persisted separately by EmbeddingManager try: os.unlink(pdf_path) except Exception: pass # --- Handlers --- def generate_patient_summary(pdf_file): try: if pdf_file is None: return "⚠️ Please upload a PDF file first." # Reuse embeddings if already built for this file vectordb = _get_vectordb_from_bytes(pdf_file) summarizer = PatientChartSummarizer(vectordb) result = summarizer.summarize_chart() return result except Exception as e: return f"❌ Error processing chart: {e}" def generate_hedis_analysis(pdf_file, measure_name, measurement_year): try: if pdf_file is None: return "⚠️ Please upload a PDF file first." if not measure_name: return "⚠️ Please enter a HEDIS measure code (e.g., COL, BCS, CCS)." if not measurement_year: return "⚠️ Please enter a measurement year." # Reuse embeddings if already built for this file vectordb = _get_vectordb_from_bytes(pdf_file) hedis = HedisComplianceEngine(vectordb, measure_name, int(measurement_year)) result = hedis.run() return result except Exception as e: return f"❌ Error processing HEDIS analysis: {e}" # --- Gradio Theme --- simple_theme = gr.themes.Soft( primary_hue=gr.themes.colors.blue, secondary_hue=gr.themes.colors.slate, neutral_hue=gr.themes.colors.slate, ).set( button_primary_background_fill="#1e40af", button_primary_background_fill_hover="#1d4ed8", button_primary_text_color="white", background_fill_primary="white", background_fill_secondary="#f8fafc", ) # --- Interface --- with gr.Blocks(theme=simple_theme, title="ChartWise AI") as interface: gr.HTML("

🏥 ChartWise AI

") gr.HTML("

Patient Chart Analysis & HEDIS Compliance

") with gr.Row(equal_height=True): # Left: Patient Summary with gr.Column(): gr.HTML("

📋 Patient Chart Analysis

") pdf_upload = gr.File(label="Upload Patient Chart (PDF)", file_types=[".pdf"], type="binary") summary_btn = gr.Button("📊 Generate Patient Summary", variant="primary") summary_output = gr.Markdown( label="Patient Summary Results", value="Upload a patient chart and click 'Generate Patient Summary' to see results here.", height=400 ) # Right: HEDIS with gr.Column(): gr.HTML("

🎯 HEDIS Measure Analysis

") hedis_measure = gr.Textbox( label="HEDIS Measure Code", placeholder="e.g., COL, BCS, CCS, AAB", info="Measure code per NCQA HEDIS" ) measurement_year = gr.Number( label="Measurement Year", value=DEFAULT_MEASURE_YEAR, precision=0, info="e.g., 2024" ) hedis_btn = gr.Button("🎯 Run HEDIS Analysis", variant="secondary") hedis_output = gr.Markdown( label="HEDIS Analysis Results", value="Enter measure and year, then click 'Run HEDIS Analysis'.", height=400 ) # Wire events summary_btn.click(fn=generate_patient_summary, inputs=[pdf_upload], outputs=[summary_output]) hedis_btn.click(fn=generate_hedis_analysis, inputs=[pdf_upload, hedis_measure, measurement_year], outputs=[hedis_output]) # Spaces auto-runs the script; these hints are fine, too: if __name__ == "__main__": interface.queue().launch(server_name="0.0.0.0", server_port=int(os.environ.get("GRADIO_SERVER_PORT", "7860")))