Spaces:
Build error
Build error
| 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/<temp_stem> | |
| 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("<h1 style='text-align:center;color:#1e40af;'>🏥 ChartWise AI</h1>") | |
| gr.HTML("<p style='text-align:center;color:#64748b;'>Patient Chart Analysis & HEDIS Compliance</p>") | |
| with gr.Row(equal_height=True): | |
| # Left: Patient Summary | |
| with gr.Column(): | |
| gr.HTML("<h3 style='color:#1e40af;'>📋 Patient Chart Analysis</h3>") | |
| 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("<h3 style='color:#1e40af;'>🎯 HEDIS Measure Analysis</h3>") | |
| 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"))) | |