Update app.py

app.py

@@ -402,6 +402,410 @@ def process_pipeline(
 
 
 
+# # app.py
+import os
+import gradio as gr
+import json
+import time
+from dotenv import load_dotenv
+from ClinicalStatusAgent import ClinicalStatusAgent
+from TestFindingAgent import TestFindingAgent
+from ComorbidityCheckerAgent import ComorbidityCheckerAgent
+from HCCDiagnosisListEngine import HCCDiagnosisListEngine
+from chartdiagnosischecker import ChartDiagnosisChecker
+from MeatValidatorAgent import MEATValidatorAgent
+from typing import Optional
+
+load_dotenv()
+APP_TITLE = "Risk Adjustment (HCC Chart Validation)"
+CSV_PATH = "hcc_mapping.csv"
+SAMPLE_PDF = "sample_patient_chart.pdf"  # Place a sample PDF in the same folder
+
+
+def json_to_markdown(data) -> str:
+    """
+    Convert JSON (dict or list) into a clean, styled Markdown/HTML string.
+    Handles both:
+      - Full dict with patient_id, hcc_code, model_version, final_analysis
+      - Just a list of diagnoses
+    Priorities:
+      - Explicit = yes takes priority over implicit
+      - MEAT, Tests, Comorbidities are expandable
+    """
+    try:
+        # Case 1: full dict
+        if isinstance(data, dict) and "final_analysis" in data:
+            patient_id = data.get("patient_id", "Unknown Patient")
+            hcc_code = data.get("hcc_code", "N/A")
+            model_version = data.get("model_version", "N/A")
+            analyses = data.get("final_analysis", [])
+        # Case 2: list of diagnoses
+        elif isinstance(data, list):
+            patient_id = "N/A"
+            hcc_code = "N/A"
+            model_version = "N/A"
+            analyses = data
+        else:
+            return "<div style='color:red; font-weight:bold;'>⚠️ Invalid data format for report.</div>"
+
+        # Header
+        md = f"""
+<div style="border:2px solid #4CAF50; padding:15px; border-radius:10px; background:#f9fdf9;">
+  <h2 style="color:#2e7d32;">📋 HCC Chart Validation Report </h2>
+  <p><b>🧾 Patient ID:</b> {patient_id}</p>
+  <p><b>🏷️ HCC Code:</b> {hcc_code}</p>
+  <p><b>⚙️ Model Version:</b> {model_version}</p>
+</div>
+<br/>
+"""
+
+        # Each diagnosis
+        for idx, diag in enumerate(analyses, 1):
+            md += f"""
+<div style="border:1px solid #ccc; padding:12px; border-radius:8px; margin-bottom:12px;">
+  <h3 style="color:#1565c0;">{idx}. {diag.get("diagnosis", "Unknown Diagnosis")}</h3>
+  <p><b>ICD-10:</b> {diag.get("icd10", "N/A")}</p>
+  <p><b>Reference:</b> <a href="{diag.get("reference","")}" target="_blank">{diag.get("reference","")}</a></p>
+"""
+
+            # --- Explicit / Implicit Priority ---
+            explicit_ans = diag.get("answer_explicit", "N/A")
+            explicit_rat = diag.get("rationale_explicit", "")
+            implicit_ans = diag.get("answer_implicit", "N/A")
+            implicit_rat = diag.get("rationale_implicit", "")
+
+            if explicit_ans.lower() == "yes":
+                md += f"<p><b>Explicit:</b> {explicit_ans} — {explicit_rat}</p>"
+            else:
+                md += f"<p><b>Implicit:</b> {implicit_ans} — {implicit_rat}</p>"
+
+            md += f"""
+  <p><b>Clinical Status:</b> {diag.get("clinical_status","N/A")}</p>
+  <p><b>Status Rationale:</b> {diag.get("status_rationale","")}</p>
+"""
+
+            # --- Tests ---
+            if "tests" in diag:
+                md += "<details><summary><b>🧪 Tests & Procedures</b></summary><ul>"
+                tests = diag["tests"]
+
+                if "vitals" in tests:
+                    md += "<li><b>Vitals:</b><ul>"
+                    for k, v in tests["vitals"].items():
+                        md += f"<li>{k}: {v}</li>"
+                    md += "</ul></li>"
+
+                if "procedures" in tests:
+                    md += "<li><b>Procedures:</b><ul>"
+                    for k, v in tests["procedures"].items():
+                        md += f"<li>{k}: {v}</li>"
+                    md += "</ul></li>"
+
+                if "lab_test" in tests:
+                    md += "<li><b>Lab Tests:</b><ul>"
+                    for k, v in tests["lab_test"].items():
+                        md += f"<li>{k}: {v}</li>"
+                    md += "</ul></li>"
+
+                md += "</ul></details>"
+
+            # --- MEAT ---
+            if "meat" in diag:
+                md += "<details><summary><b>🍖 MEAT Validation</b></summary><ul>"
+                for k, v in diag["meat"].items():
+                    emoji = "✅" if v else "❌"
+                    md += f"<li>{k.capitalize()}: {emoji}</li>"
+                md += "</ul>"
+                md += f"<p><b>MEAT Rationale:</b> {diag.get('meat_rationale','')}</p>"
+                md += "</details>"
+
+            # --- Comorbidities ---
+            if "comorbidities" in diag and diag["comorbidities"]:
+                md += "<details><summary><b>🩺 Comorbidities</b></summary><ul>"
+                for c in diag["comorbidities"]:
+                    emoji = "✅" if c.get("is_present") else "❌"
+                    md += f"<li>{emoji} <b>{c.get('condition')}</b><br/><i>{c.get('rationale')}</i></li>"
+                md += "</ul></details>"
+
+            md += "</div>"
+
+        return md
+
+    except Exception as e:
+        return f"<div style='color:red; font-weight:bold;'>⚠️ Error rendering report: {e}</div>"
+
+# def process_pipeline(pdf_file, hcc_code, model_version, csv_path=CSV_PATH, output_folder="outputs"):
+#     try:
+#         start = time.time()
+
+#         if pdf_file is None:
+#             yield "<div style='color:orange; font-weight:bold;'>⚠️ Please upload a patient chart PDF.</div>"
+#             return
+
+#         hcc_code_str = str(hcc_code or "").strip()
+#         if not hcc_code_str:
+#             yield "<div style='color:orange; font-weight:bold;'>⚠️ Please enter a valid HCC Code before running validation.</div>"
+#             return
+
+#         os.makedirs(output_folder, exist_ok=True)
+#         pdf_path = pdf_file.name
+#         patient_name = os.path.splitext(os.path.basename(pdf_path))[0]
+#         print(f"\n[PROCESSING] {patient_name}")
+
+#         # Step 1: Get all possible HCC Diagnoses
+#         yield "🔍 Step 1: Extracting possible HCC Diagnoses..."
+#         diagnoses = HCCDiagnosisListEngine(hcc_code_str, model_version, csv_path).run()
+#         yield f"⏳ Elapsed: {time.time() - start:.1f} sec"
+
+#         if not diagnoses:
+#             msg = f"<b>❌ No diagnoses found for HCC {hcc_code_str}.</b>"
+#             yield msg
+#             return
+
+#         # Step 2: Check patient chart
+#         yield "📄 Step 2: Checking diagnoses in patient chart..."
+#         all_checked_results = ChartDiagnosisChecker(pdf_path).run(diagnoses)
+#         yield f"⏳ Elapsed: {time.time() - start:.1f} sec"
+
+#         confirmed_diagnoses = [
+#             d for d in all_checked_results
+#             if d.get("answer_explicit", "").lower() == "yes"
+#             or d.get("answer_implicit", "").lower() == "yes"
+#         ]
+
+#         if not confirmed_diagnoses:
+#             msg = f"<b>❌ No confirmed diagnoses for HCC {hcc_code_str} in {patient_name}. Report saved.</b>"
+#             output_data = {
+#                 "patient_id": patient_name,
+#                 "hcc_code": hcc_code_str,
+#                 "model_version": model_version,
+#                 "final_analysis": all_checked_results
+#             }
+#             outpath = os.path.join(output_folder, f"{patient_name}.json")
+#             with open(outpath, "w", encoding="utf-8") as f:
+#                 json.dump(output_data, f, indent=2, ensure_ascii=False)
+#             print(f"[DONE] Saved basic chart check report to {outpath}")
+#             yield msg
+#             return
+
+#         # Step 3: Tests
+#         yield "🧪 Step 3: Finding relevant tests..."
+#         diagnoses_with_tests = TestFindingAgent(hcc_code=hcc_code_str, model_version=model_version).run(confirmed_diagnoses)
+#         yield f"⏳ Elapsed: {time.time() - start:.1f} sec"
+
+#         # Step 4: Clinical Status
+#         yield "⚕️ Step 4: Determining clinical status..."
+#         diagnoses_with_status = ClinicalStatusAgent().run(diagnoses_with_tests)
+#         yield f"⏳ Elapsed: {time.time() - start:.1f} sec"
+
+#         active_diagnoses = [d for d in diagnoses_with_status if d.get("clinical_status") == "ACTIVE"]
+
+#         validated_meat_diagnoses = []
+#         if active_diagnoses:
+#             yield "🍖 Step 5: Validating MEAT..."
+#             validated_meat_diagnoses = MEATValidatorAgent().run(active_diagnoses)
+#             yield f"⏳ Elapsed: {time.time() - start:.1f} sec"
+#         else:
+#             yield "ℹ️ No ACTIVE diagnoses found. Skipping MEAT/Comorbidity."
+
+#         diagnoses_passed_meat = [d for d in validated_meat_diagnoses if any(d.get("meat", {}).values())]
+
+#         comorbidity_results = []
+#         if diagnoses_passed_meat:
+#             yield "🤝 Step 6: Checking comorbidities..."
+#             comorbidity_results = ComorbidityCheckerAgent(pdf_path, hcc_code_str, model_version).run(diagnoses_passed_meat)
+#             yield f"⏳ Elapsed: {time.time() - start:.1f} sec"
+
+#         # Merge results
+#         status_map = {d["diagnosis"]: d for d in diagnoses_with_status}
+#         meat_map = {d["diagnosis"]: d for d in validated_meat_diagnoses}
+#         comorbidity_map = {d["diagnosis"]: d for d in comorbidity_results}
+
+#         final_analysis = []
+#         for entry in all_checked_results:
+#             diag_name = entry["diagnosis"]
+#             if diag_name in status_map:
+#                 updated_entry = status_map[diag_name]
+#                 if diag_name in meat_map:
+#                     updated_entry.update(meat_map[diag_name])
+#                 if diag_name in comorbidity_map:
+#                     updated_entry.update(comorbidity_map[diag_name])
+#                 final_analysis.append(updated_entry)
+#             else:
+#                 final_analysis.append(entry)
+
+#         filtered_final_analysis = [
+#             e for e in final_analysis
+#             if e.get("answer_explicit", "").lower() == "yes"
+#             or e.get("answer_implicit", "").lower() == "yes"
+#         ]
+
+#         output_data = {
+#             "patient_id": patient_name,
+#             "hcc_code": hcc_code_str,
+#             "model_version": model_version,
+#             "final_analysis": filtered_final_analysis
+#         }
+
+#         # Final Report
+#         yield "✅ Step 7: Generating final report..."
+#         elapsed = time.time() - start
+#         final_report = json_to_markdown(output_data) + f"<br/><div style='color:green;'>✅ Completed in {elapsed:.1f} sec</div>"
+#         yield final_report
+
+#     except Exception as e:
+#         print(f"[ERROR] {e}")
+#         yield f"<div style='color:red; font-weight:bold;'>⚠️ Error: {e}</div>"
+
+
+import time
+import gradio as gr
+
+def process_pipeline(
+    pdf_file,
+    hcc_code,
+    model_version,
+    csv_path=CSV_PATH,
+    output_folder="outputs",
+    progress=gr.Progress()
+):
+    try:
+        start = time.time()
+
+        if pdf_file is None:
+            yield "<div style='color:orange; font-weight:bold;'>⚠️ Please upload a patient chart PDF.</div>"
+            return
+
+        hcc_code_str = str(hcc_code or "").strip()
+        if not hcc_code_str:
+            yield "<div style='color:orange; font-weight:bold;'>⚠️ Please enter a valid HCC Code before running validation.</div>"
+            return
+
+        os.makedirs(output_folder, exist_ok=True)
+        pdf_path = pdf_file.name
+        patient_name = os.path.splitext(os.path.basename(pdf_path))[0]
+        print(f"\n[PROCESSING] {patient_name}")
+
+        # total steps for progress bar
+        total_steps = 7
+        step = 0
+
+        def log(msg):
+            """Helper to include elapsed time like tqdm"""
+            elapsed = time.time() - start
+            return f"{msg}<br/>⏳ Elapsed: {elapsed:.1f} sec"
+
+        # ----------------------
+        # Step 1: Diagnoses
+        step += 1
+        progress((step, total_steps), desc="Extracting diagnoses")
+        yield log(f"🔍 Step {step}/{total_steps}: Extracting possible HCC Diagnoses...")
+        diagnoses = HCCDiagnosisListEngine(hcc_code_str, model_version, csv_path).run()
+
+        if not diagnoses:
+            yield f"<b>❌ No diagnoses found for HCC {hcc_code_str}.</b>"
+            return
+
+        # ----------------------
+        # Step 2: Chart checking
+        step += 1
+        progress((step, total_steps), desc="Checking chart")
+        yield log(f"📄 Step {step}/{total_steps}: Checking diagnoses in patient chart...")
+        all_checked_results = ChartDiagnosisChecker(pdf_path).run(diagnoses)
+
+        confirmed_diagnoses = [
+            d for d in all_checked_results
+            if d.get("answer_explicit", "").lower() == "yes"
+            or d.get("answer_implicit", "").lower() == "yes"
+        ]
+        if not confirmed_diagnoses:
+            msg = f"<b>❌ No confirmed diagnoses for HCC {hcc_code_str} in {patient_name}. Report saved.</b>"
+            yield log(msg)
+            return
+
+        # ----------------------
+        # Step 3: Tests
+        step += 1
+        progress((step, total_steps), desc="Finding tests")
+        yield log(f"🧪 Step {step}/{total_steps}: Finding relevant tests...")
+        diagnoses_with_tests = TestFindingAgent(hcc_code=hcc_code_str, model_version=model_version).run(confirmed_diagnoses)
+
+        # ----------------------
+        # Step 4: Clinical Status
+        step += 1
+        progress((step, total_steps), desc="Determining clinical status")
+        yield log(f"⚕️ Step {step}/{total_steps}: Determining clinical status...")
+        diagnoses_with_status = ClinicalStatusAgent().run(diagnoses_with_tests)
+
+        active_diagnoses = [d for d in diagnoses_with_status if d.get("clinical_status") == "ACTIVE"]
+
+        validated_meat_diagnoses = []
+        if active_diagnoses:
+            # ----------------------
+            # Step 5: MEAT
+            step += 1
+            progress((step, total_steps), desc="Validating MEAT")
+            yield log(f"🍖 Step {step}/{total_steps}: Validating MEAT...")
+            validated_meat_diagnoses = MEATValidatorAgent().run(active_diagnoses)
+        else:
+            yield log("ℹ️ No ACTIVE diagnoses found. Skipping MEAT/Comorbidity.")
+
+        # ----------------------
+        # Step 6: Comorbidities
+        diagnoses_passed_meat = [d for d in validated_meat_diagnoses if any(d.get("meat", {}).values())]
+        comorbidity_results = []
+        if diagnoses_passed_meat:
+            step += 1
+            progress((step, total_steps), desc="Checking comorbidities")
+            yield log(f"🤝 Step {step}/{total_steps}: Checking comorbidities...")
+            comorbidity_results = ComorbidityCheckerAgent(pdf_path, hcc_code_str, model_version).run(diagnoses_passed_meat)
+
+        # ----------------------
+        # Step 7: Final Report
+        step += 1
+        progress((step, total_steps), desc="Generating report")
+        yield log(f"✅ Step {step}/{total_steps}: Generating final report...")
+
+        # merge & filter results
+        status_map = {d["diagnosis"]: d for d in diagnoses_with_status}
+        meat_map = {d["diagnosis"]: d for d in validated_meat_diagnoses}
+        comorbidity_map = {d["diagnosis"]: d for d in comorbidity_results}
+
+        final_analysis = []
+        for entry in all_checked_results:
+            diag_name = entry["diagnosis"]
+            updated_entry = entry.copy()
+            if diag_name in status_map:
+                updated_entry.update(status_map[diag_name])
+            if diag_name in meat_map:
+                updated_entry.update(meat_map[diag_name])
+            if diag_name in comorbidity_map:
+                updated_entry.update(comorbidity_map[diag_name])
+            final_analysis.append(updated_entry)
+
+        filtered_final_analysis = [
+            e for e in final_analysis
+            if e.get("answer_explicit", "").lower() == "yes"
+            or e.get("answer_implicit", "").lower() == "yes"
+        ]
+
+        output_data = {
+            "patient_id": patient_name,
+            "hcc_code": hcc_code_str,
+            "model_version": model_version,
+            "final_analysis": filtered_final_analysis
+        }
+
+        elapsed = time.time() - start
+        final_report = json_to_markdown(output_data) + f"<br/><div style='color:green;'>✅ Completed in {elapsed:.1f} sec</div>"
+        yield final_report
+
+    except Exception as e:
+        print(f"[ERROR] {e}")
+        yield f"<div style='color:red; font-weight:bold;'>⚠️ Error: {e}</div>"
+
+
+
 # ---------- Gradio Theme ----------
 simple_theme = gr.themes.Soft(
     primary_hue=gr.themes.colors.blue,
@@ -470,3 +874,5 @@ if __name__ == "__main__":
     interface.queue().launch(server_name="0.0.0.0", server_port=int(os.environ.get("PORT", 7860)))
 
 
+
+