## Adding red box for failure
import os
import gradio as gr
import json
import time
import base64
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 PatientInfoExtractionEngine import PatientInfoExtractionEngine
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
# ---------- JSON to Markdown ----------
# --- MODIFICATION: Added 'status' parameter to control color scheme ---
def json_to_markdown(data, status='success') -> str:
try:
if isinstance(data, dict):
file_name = data.get("file_name", "Unknown Patient")
hcc_code = data.get("hcc_code", "N/A")
model_version = data.get("model_version", "N/A")
analyses = data.get("final_analysis", [])
# Demographics
patient_name = data.get("patient_name", "")
dob = data.get("dob", "")
age = data.get("age", "")
gender = data.get("gender", "")
address = data.get("address", "")
phone = data.get("phone", "")
patient_identifier = data.get("patient_identifier", "")
else:
return "โ ๏ธ Invalid data format for report.
"
# Dynamically set colors based on status
if status == 'error':
border_color, bg_color, header_color = "#f44336", "#ffeb ee", "#d32f2f" # Red theme
else:
border_color, bg_color, header_color = "#4CAF50", "#f9fdf9", "#2e7d32" # Green theme
md = f"""
๐ HCC Chart Validation Report
๐งพ File Name: {file_name}
๐ท๏ธ HCC Code: {hcc_code}
โ๏ธ Model Version: {model_version}
"""
# Add demographics
if any([patient_name, dob, age, gender, address, phone, patient_identifier]):
md += "
"
md += f"
๐ค Patient Demographics
" # Use dynamic color
if patient_name: md += f"
Name: {patient_name}
"
if patient_identifier: md += f"
Patient Identifier: {patient_identifier}
"
if dob: md += f"
Date of Birth: {dob}
"
if age: md += f"
Age: {age}
"
if gender: md += f"
Gender: {gender}
"
if address: md += f"
Address: {address}
"
if phone: md += f"
Phone: {phone}
"
md += "
"
# Render analyses if they exist
if analyses:
for idx, diag in enumerate(analyses, 1):
md += f"""
Diagnosis {idx}. {diag.get("diagnosis", "Unknown Diagnosis")}
ICD-10: {diag.get("icd10", "N/A")}
Reference: {diag.get("reference","")}
"""
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"
Explicit: {explicit_ans} โ {explicit_rat}
"
else:
md += f"
Implicit: {implicit_ans} โ {implicit_rat}
"
md += f"""
Clinical Status: {diag.get("clinical_status","N/A")}
Status Rationale: {diag.get("status_rationale","")}
"""
if "tests" in diag:
md += "
๐งช Tests & Procedures
"
tests = diag["tests"]
if "vitals" in tests:
md += "- Vitals:
"
for k, v in tests["vitals"].items(): md += f"- {k}: {v}
"
md += "
"
if "procedures" in tests:
md += "- Procedures:
"
for k, v in tests["procedures"].items(): md += f"- {k}: {v}
"
md += "
"
if "lab_test" in tests:
md += "- Lab Tests:
"
for k, v in tests["lab_test"].items(): md += f"- {k}: {v}
"
md += "
"
md += "
"
if "meat" in diag:
md += "
๐ MEAT Validation
"
for k, v in diag["meat"].items():
emoji = "โ
" if v else "โ"
md += f"- {k.capitalize()}: {emoji}
"
md += "
"
md += f"MEAT Rationale: {diag.get('meat_rationale','')}
"
md += ""
if "comorbidities" in diag and diag["comorbidities"]:
present_comorbidities = [c for c in diag["comorbidities"] if c.get("is_present")]
if present_comorbidities:
md += "
๐ฉบ Comorbidities
"
for c in present_comorbidities:
md += f"- โ
{c.get('condition')}
{c.get('rationale')} "
md += "
"
md += "
"
return md
except Exception as e:
return f"โ ๏ธ Error rendering report: {e}
"
# ---------- Processing Pipeline with Gradio Progress ----------
def process_pipeline(pdf_file, hcc_code, model_version, csv_path=CSV_PATH, output_folder="outputs"):
try:
start = time.time()
step = 0
total_steps = 8 # Total number of steps in the pipeline
def log(msg, current_step=0):
elapsed = time.time() - start
bar_html = ''
for i in range(1, total_steps + 1):
if i < current_step: color = "#1e40af"
elif i == current_step: color = "#3b82f6"
else: color = "#e5e7eb"
bar_html += f'
'
bar_html += '
{msg}{bar_html}โณ Elapsed: {elapsed:.1f} sec
"
if pdf_file is None:
yield log("โ ๏ธ Please upload a patient chart PDF.", 0)
return
hcc_code_str = str(hcc_code or "").strip()
if not hcc_code_str:
yield log("โ ๏ธ Please enter a valid HCC Code before running validation.", 0)
return
os.makedirs(output_folder, exist_ok=True)
pdf_path = pdf_file.name
file_name = os.path.splitext(os.path.basename(pdf_path))[0]
print(f"[PROCESSING] {file_name}")
# Step 1: Extract Demographics
step += 1
initial_progress_msg = log(f"๐ง Step {step}/{total_steps}: Extracting patient demographics...", step)
yield initial_progress_msg
demographics_engine = PatientInfoExtractionEngine(pdf_path)
demographics_info = demographics_engine.run()
print(f"[DEMOGRAPHICS] Extracted: {demographics_info}")
initial_report_data = {
"file_name": file_name,
"hcc_code": hcc_code_str,
"model_version": model_version,
"final_analysis": [],
"patient_name": demographics_info.get("name", ""),
"dob": demographics_info.get("dob", ""),
"age": demographics_info.get("age", ""),
"gender": demographics_info.get("gender", ""),
"address": demographics_info.get("address", ""),
"phone": demographics_info.get("phone", ""),
"patient_identifier": demographics_info.get("patient_identifier", "")
}
demographics_md = json_to_markdown(initial_report_data)
yield demographics_md
time.sleep(0.5)
# Step 2: Diagnoses
step += 1
yield demographics_md + log(f"๐ Step {step}/{total_steps}: Extracting possible HCC Diagnoses...", step)
diagnoses = HCCDiagnosisListEngine(hcc_code_str, model_version, csv_path).run()
if not diagnoses:
yield demographics_md + log(f"โ No diagnoses found for HCC {hcc_code_str}.", step)
return
# Step 3: Chart checking
step += 1
yield demographics_md + log(f"๐ Step {step}/{total_steps}: Checking diagnoses in patient chart...", step)
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"]
# --- MODIFICATION: Custom failure UI for this specific condition ---
if not confirmed_diagnoses:
error_report_md = json_to_markdown(initial_report_data, status='error')
error_message = f"No confirmed diagnoses found for HCC {hcc_code_str} in the patient chart."
error_box_html = f"""
โ Validation Failed: {error_message}
"""
yield error_report_md + error_box_html
return
# --- END OF MODIFICATION ---
# Step 4: Tests
step += 1
yield demographics_md + log(f"๐งช Step {step}/{total_steps}: Finding relevant tests...", step)
diagnoses_with_tests = TestFindingAgent(hcc_code=hcc_code_str, model_version=model_version).run(confirmed_diagnoses)
# Step 5: Clinical Status
step += 1
yield demographics_md + log(f"โ๏ธ Step {step}/{total_steps}: Determining clinical status...", step)
diagnoses_with_status = ClinicalStatusAgent().run(diagnoses_with_tests)
active_diagnoses = [d for d in diagnoses_with_status if d.get("clinical_status") == "ACTIVE"]
# Step 6: MEAT
step += 1
if active_diagnoses:
yield demographics_md + log(f"๐ Step {step}/{total_steps}: Validating MEAT...", step)
validated_meat_diagnoses = MEATValidatorAgent().run(active_diagnoses)
else:
validated_meat_diagnoses = []
yield demographics_md + log("โน๏ธ No ACTIVE diagnoses found. Skipping MEAT/Comorbidity.", step)
# Step 7: Comorbidities
step += 1
diagnoses_passed_meat = [d for d in validated_meat_diagnoses if any(d.get("meat", {}).values())]
if diagnoses_passed_meat:
yield demographics_md + log(f"๐ค Step {step}/{total_steps}: Checking comorbidities...", step)
comorbidity_results = ComorbidityCheckerAgent(pdf_path, hcc_code_str, model_version).run(diagnoses_passed_meat)
else:
comorbidity_results = []
# Step 8: Final Report
step += 1
yield demographics_md + log(f"โ
Step {step}/{total_steps}: Generating final report...", step)
# Merge results for final output
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 = {
"file_name": file_name,
"hcc_code": hcc_code_str,
"model_version": model_version,
"final_analysis": filtered_final_analysis,
"patient_name": demographics_info.get("name", ""),
"dob": demographics_info.get("dob", ""),
"age": demographics_info.get("age", ""),
"gender": demographics_info.get("gender", ""),
"address": demographics_info.get("address", ""),
"phone": demographics_info.get("phone", ""),
"patient_identifier": demographics_info.get("patient_identifier", "")
}
elapsed = time.time() - start
yield json_to_markdown(output_data) + f"
โ
Completed in {elapsed:.1f} sec
"
except Exception as e:
print(f"[ERROR] {e}")
yield f"โ ๏ธ Error: {e}
"
# ---------- Gradio Theme and Helpers ----------
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",
)
def load_sample_pdf():
if not os.path.exists(SAMPLE_PDF):
raise FileNotFoundError(f"Sample PDF not found at {SAMPLE_PDF}")
class PDFWrapper:
def __init__(self, path):
self.name = path
return PDFWrapper(SAMPLE_PDF)
def pdf_to_iframe(file):
if file is None: return "No PDF uploaded.
"
try:
with open(file.name, "rb") as f: pdf_bytes = f.read()
encoded = base64.b64encode(pdf_bytes).decode("utf-8")
return f''
except Exception as e:
return f"Failed to display PDF: {e}
"
def clear_outputs():
initial_md = "๐ Upload a PDF and click Run Validation to start.
"
initial_preview = "Upload a PDF to preview
"
return initial_md, initial_preview
# ---------- Gradio UI ----------
with gr.Blocks(theme=simple_theme, title=APP_TITLE) as interface:
gr.HTML(f"""
๐ฉ {APP_TITLE}
Upload a chart, set HCC + model version, and validate MEAT criteria.
""")
with gr.Row():
pdf_upload = gr.File(label="Upload Patient Chart (PDF)", file_types=[".pdf"], scale=1)
hcc_code = gr.Textbox(label="HCC Code (e.g., 12)", placeholder="Enter HCC code", scale=1)
model_version = gr.Dropdown(choices=["V24", "V28"], label="Model Version", value="V24", scale=1)
run_btn = gr.Button("๐ Run Validation", variant="primary", scale=1)
with gr.Row():
with gr.Column(scale=2):
pdf_preview = gr.HTML(label="๐ PDF Preview", value="Upload a PDF to preview
")
with gr.Column(scale=2):
output_md = gr.Markdown(
label="Validation Report",
value="๐ Upload a PDF and click Run Validation to start.
",
)
pdf_upload.change(fn=pdf_to_iframe, inputs=pdf_upload, outputs=pdf_preview)
pdf_upload.clear(fn=clear_outputs, inputs=[], outputs=[output_md, pdf_preview])
run_btn.click(
fn=process_pipeline,
inputs=[pdf_upload, hcc_code, model_version],
outputs=[output_md],
)
with gr.Row():
with gr.Column(scale=1):
gr.Examples(
examples=[[SAMPLE_PDF, "12", "V24"]],
inputs=[pdf_upload, hcc_code, model_version],
label="Click to load an example",
cache_examples=False
)
with gr.Column(scale=2):
pass
if __name__ == "__main__":
interface.queue().launch(
server_name="0.0.0.0",
server_port=int(os.environ.get("PORT", 7860))
)
# ## small samples
# import os
# import gradio as gr
# import json
# import time
# import base64
# 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 PatientInfoExtractionEngine import PatientInfoExtractionEngine
# 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
# # ---------- JSON to Markdown ----------
# def json_to_markdown(data) -> str:
# try:
# if isinstance(data, dict):
# file_name = data.get("file_name", "Unknown Patient")
# hcc_code = data.get("hcc_code", "N/A")
# model_version = data.get("model_version", "N/A")
# analyses = data.get("final_analysis", [])
# # Demographics
# patient_name = data.get("patient_name", "")
# dob = data.get("dob", "")
# age = data.get("age", "")
# gender = data.get("gender", "")
# address = data.get("address", "")
# phone = data.get("phone", "")
# patient_identifier = data.get("patient_identifier", "")
# else:
# return "โ ๏ธ Invalid data format for report.
"
# md = f"""
#
#
๐ HCC Chart Validation Report
#
๐งพ File Name: {file_name}
#
๐ท๏ธ HCC Code: {hcc_code}
#
โ๏ธ Model Version: {model_version}
# """
# # Add demographics
# if any([patient_name, dob, age, gender, address, phone, patient_identifier]):
# md += "
"
# md += "
๐ค Patient Demographics
"
# if patient_name: md += f"
Name: {patient_name}
"
# if patient_identifier: md += f"
Patient Identifier: {patient_identifier}
"
# if dob: md += f"
Date of Birth: {dob}
"
# if age: md += f"
Age: {age}
"
# if gender: md += f"
Gender: {gender}
"
# if address: md += f"
Address: {address}
"
# if phone: md += f"
Phone: {phone}
"
# md += "
"
# # Render analyses if they exist
# if analyses:
# for idx, diag in enumerate(analyses, 1):
# md += f"""
#
#
Diagnosis {idx}. {diag.get("diagnosis", "Unknown Diagnosis")}
#
ICD-10: {diag.get("icd10", "N/A")}
#
Reference: {diag.get("reference","")}
# """
# 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"
Explicit: {explicit_ans} โ {explicit_rat}
"
# else:
# md += f"
Implicit: {implicit_ans} โ {implicit_rat}
"
# md += f"""
#
Clinical Status: {diag.get("clinical_status","N/A")}
#
Status Rationale: {diag.get("status_rationale","")}
# """
# if "tests" in diag:
# md += "
๐งช Tests & Procedures
"
# tests = diag["tests"]
# if "vitals" in tests:
# md += "- Vitals:
"
# for k, v in tests["vitals"].items(): md += f"- {k}: {v}
"
# md += "
"
# if "procedures" in tests:
# md += "- Procedures:
"
# for k, v in tests["procedures"].items(): md += f"- {k}: {v}
"
# md += "
"
# if "lab_test" in tests:
# md += "- Lab Tests:
"
# for k, v in tests["lab_test"].items(): md += f"- {k}: {v}
"
# md += "
"
# md += "
"
# if "meat" in diag:
# md += "
๐ MEAT Validation
"
# for k, v in diag["meat"].items():
# emoji = "โ
" if v else "โ"
# md += f"- {k.capitalize()}: {emoji}
"
# md += "
"
# md += f"MEAT Rationale: {diag.get('meat_rationale','')}
"
# md += ""
# if "comorbidities" in diag and diag["comorbidities"]:
# present_comorbidities = [c for c in diag["comorbidities"] if c.get("is_present")]
# if present_comorbidities:
# md += "
๐ฉบ Comorbidities
"
# for c in present_comorbidities:
# md += f"- โ
{c.get('condition')}
{c.get('rationale')} "
# md += "
"
# md += "
โ ๏ธ Error rendering report: {e}
"
# # ---------- Processing Pipeline with Gradio Progress ----------
# def process_pipeline(pdf_file, hcc_code, model_version, csv_path=CSV_PATH, output_folder="outputs"):
# try:
# start = time.time()
# step = 0
# total_steps = 8 # Total number of steps in the pipeline
# def log(msg, current_step=0):
# elapsed = time.time() - start
# bar_html = ''
# for i in range(1, total_steps + 1):
# if i < current_step: color = "#1e40af"
# elif i == current_step: color = "#3b82f6"
# else: color = "#e5e7eb"
# bar_html += f'
'
# bar_html += '
{msg}{bar_html}โณ Elapsed: {elapsed:.1f} sec
"
# if pdf_file is None:
# yield log("โ ๏ธ Please upload a patient chart PDF.", 0)
# return
# hcc_code_str = str(hcc_code or "").strip()
# if not hcc_code_str:
# yield log("โ ๏ธ Please enter a valid HCC Code before running validation.", 0)
# return
# os.makedirs(output_folder, exist_ok=True)
# pdf_path = pdf_file.name
# file_name = os.path.splitext(os.path.basename(pdf_path))[0]
# print(f"[PROCESSING] {file_name}")
# # Step 1: Extract Demographics
# step += 1
# initial_progress_msg = log(f"๐ง Step {step}/{total_steps}: Extracting patient demographics...", step)
# yield initial_progress_msg
# demographics_engine = PatientInfoExtractionEngine(pdf_path)
# demographics_info = demographics_engine.run()
# print(f"[DEMOGRAPHICS] Extracted: {demographics_info}")
# initial_report_data = {
# "file_name": file_name,
# "hcc_code": hcc_code_str,
# "model_version": model_version,
# "final_analysis": [],
# "patient_name": demographics_info.get("name", ""),
# "dob": demographics_info.get("dob", ""),
# "age": demographics_info.get("age", ""),
# "gender": demographics_info.get("gender", ""),
# "address": demographics_info.get("address", ""),
# "phone": demographics_info.get("phone", ""),
# "patient_identifier": demographics_info.get("patient_identifier", "")
# }
# demographics_md = json_to_markdown(initial_report_data)
# yield demographics_md
# time.sleep(0.5)
# # Step 2: Diagnoses
# step += 1
# yield demographics_md + log(f"๐ Step {step}/{total_steps}: Extracting possible HCC Diagnoses...", step)
# diagnoses = HCCDiagnosisListEngine(hcc_code_str, model_version, csv_path).run()
# if not diagnoses:
# yield demographics_md + log(f"โ No diagnoses found for HCC {hcc_code_str}.", step)
# return
# # Step 3: Chart checking
# step += 1
# yield demographics_md + log(f"๐ Step {step}/{total_steps}: Checking diagnoses in patient chart...", step)
# 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:
# yield demographics_md + log(f"โ No confirmed diagnoses for HCC {hcc_code_str} in {file_name}.", step)
# return
# # Step 4: Tests
# step += 1
# yield demographics_md + log(f"๐งช Step {step}/{total_steps}: Finding relevant tests...", step)
# diagnoses_with_tests = TestFindingAgent(hcc_code=hcc_code_str, model_version=model_version).run(confirmed_diagnoses)
# # Step 5: Clinical Status
# step += 1
# yield demographics_md + log(f"โ๏ธ Step {step}/{total_steps}: Determining clinical status...", step)
# diagnoses_with_status = ClinicalStatusAgent().run(diagnoses_with_tests)
# active_diagnoses = [d for d in diagnoses_with_status if d.get("clinical_status") == "ACTIVE"]
# # Step 6: MEAT
# step += 1
# if active_diagnoses:
# yield demographics_md + log(f"๐ Step {step}/{total_steps}: Validating MEAT...", step)
# validated_meat_diagnoses = MEATValidatorAgent().run(active_diagnoses)
# else:
# validated_meat_diagnoses = []
# yield demographics_md + log("โน๏ธ No ACTIVE diagnoses found. Skipping MEAT/Comorbidity.", step)
# # Step 7: Comorbidities
# step += 1
# diagnoses_passed_meat = [d for d in validated_meat_diagnoses if any(d.get("meat", {}).values())]
# if diagnoses_passed_meat:
# yield demographics_md + log(f"๐ค Step {step}/{total_steps}: Checking comorbidities...", step)
# comorbidity_results = ComorbidityCheckerAgent(pdf_path, hcc_code_str, model_version).run(diagnoses_passed_meat)
# else:
# comorbidity_results = []
# # Step 8: Final Report
# step += 1
# yield demographics_md + log(f"โ
Step {step}/{total_steps}: Generating final report...", step)
# # Merge results for final output
# 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 = {
# "file_name": file_name,
# "hcc_code": hcc_code_str,
# "model_version": model_version,
# "final_analysis": filtered_final_analysis,
# "patient_name": demographics_info.get("name", ""),
# "dob": demographics_info.get("dob", ""),
# "age": demographics_info.get("age", ""),
# "gender": demographics_info.get("gender", ""),
# "address": demographics_info.get("address", ""),
# "phone": demographics_info.get("phone", ""),
# "patient_identifier": demographics_info.get("patient_identifier", "")
# }
# elapsed = time.time() - start
# yield json_to_markdown(output_data) + f"
โ
Completed in {elapsed:.1f} sec
"
# except Exception as e:
# print(f"[ERROR] {e}")
# yield f"โ ๏ธ Error: {e}
"
# # ---------- Gradio Theme and Helpers ----------
# 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",
# )
# def load_sample_pdf():
# if not os.path.exists(SAMPLE_PDF):
# raise FileNotFoundError(f"Sample PDF not found at {SAMPLE_PDF}")
# class PDFWrapper:
# def __init__(self, path):
# self.name = path
# return PDFWrapper(SAMPLE_PDF)
# def pdf_to_iframe(file):
# if file is None: return "No PDF uploaded.
"
# try:
# with open(file.name, "rb") as f: pdf_bytes = f.read()
# encoded = base64.b64encode(pdf_bytes).decode("utf-8")
# return f''
# except Exception as e:
# return f"Failed to display PDF: {e}
"
# def clear_outputs():
# initial_md = "๐ Upload a PDF and click Run Validation to start.
"
# initial_preview = "Upload a PDF to preview
"
# return initial_md, initial_preview
# # ---------- Gradio UI ----------
# with gr.Blocks(theme=simple_theme, title=APP_TITLE) as interface:
# gr.HTML(f"""
# ๐ฉ {APP_TITLE}
#
# Upload a chart, set HCC + model version, and validate MEAT criteria.
#
# """)
# with gr.Row():
# pdf_upload = gr.File(label="Upload Patient Chart (PDF)", file_types=[".pdf"], scale=1)
# hcc_code = gr.Textbox(label="HCC Code (e.g., 12)", placeholder="Enter HCC code", scale=1)
# model_version = gr.Dropdown(choices=["V24", "V28"], label="Model Version", value="V24", scale=1)
# run_btn = gr.Button("๐ Run Validation", variant="primary", scale=1)
# with gr.Row():
# with gr.Column(scale=2):
# pdf_preview = gr.HTML(label="๐ PDF Preview", value="Upload a PDF to preview
")
# with gr.Column(scale=2):
# output_md = gr.Markdown(
# label="Validation Report",
# value="๐ Upload a PDF and click Run Validation to start.
",
# )
# pdf_upload.change(fn=pdf_to_iframe, inputs=pdf_upload, outputs=pdf_preview)
# pdf_upload.clear(fn=clear_outputs, inputs=[], outputs=[output_md, pdf_preview])
# run_btn.click(
# fn=process_pipeline,
# inputs=[pdf_upload, hcc_code, model_version],
# outputs=[output_md],
# )
# # --- MODIFICATION: Adjusted column scaling to make the Examples section smaller ---
# with gr.Row():
# with gr.Column(scale=1): # This column will be smaller
# gr.Examples(
# examples=[[SAMPLE_PDF, "12", "V24"]],
# inputs=[pdf_upload, hcc_code, model_version],
# label="Click to load an example",
# cache_examples=False
# )
# with gr.Column(scale=2): # This empty column will take up the remaining space
# pass
# # --- END OF MODIFICATION ---
# if __name__ == "__main__":
# interface.queue().launch(
# server_name="0.0.0.0",
# server_port=int(os.environ.get("PORT", 7860))
# )
# import os
# import gradio as gr
# import json
# import time
# import base64
# 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 PatientInfoExtractionEngine import PatientInfoExtractionEngine
# 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
# # ---------- JSON to Markdown ----------
# def json_to_markdown(data) -> str:
# try:
# if isinstance(data, dict):
# file_name = data.get("file_name", "Unknown Patient")
# hcc_code = data.get("hcc_code", "N/A")
# model_version = data.get("model_version", "N/A")
# analyses = data.get("final_analysis", [])
# # Demographics
# patient_name = data.get("patient_name", "")
# dob = data.get("dob", "")
# age = data.get("age", "")
# gender = data.get("gender", "")
# address = data.get("address", "")
# phone = data.get("phone", "")
# patient_identifier = data.get("patient_identifier", "")
# else:
# return "โ ๏ธ Invalid data format for report.
"
# md = f"""
#
#
๐ HCC Chart Validation Report
#
๐งพ File Name: {file_name}
#
๐ท๏ธ HCC Code: {hcc_code}
#
โ๏ธ Model Version: {model_version}
# """
# # Add demographics
# if any([patient_name, dob, age, gender, address, phone, patient_identifier]):
# md += "
"
# md += "
๐ค Patient Demographics
"
# if patient_name: md += f"
Name: {patient_name}
"
# if patient_identifier: md += f"
Patient Identifier: {patient_identifier}
"
# if dob: md += f"
Date of Birth: {dob}
"
# if age: md += f"
Age: {age}
"
# if gender: md += f"
Gender: {gender}
"
# if address: md += f"
Address: {address}
"
# if phone: md += f"
Phone: {phone}
"
# md += "
"
# # Render analyses if they exist
# if analyses:
# for idx, diag in enumerate(analyses, 1):
# md += f"""
#
#
Diagnosis {idx}. {diag.get("diagnosis", "Unknown Diagnosis")}
#
ICD-10: {diag.get("icd10", "N/A")}
#
Reference: {diag.get("reference","")}
# """
# 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"
Explicit: {explicit_ans} โ {explicit_rat}
"
# else:
# md += f"
Implicit: {implicit_ans} โ {implicit_rat}
"
# md += f"""
#
Clinical Status: {diag.get("clinical_status","N/A")}
#
Status Rationale: {diag.get("status_rationale","")}
# """
# if "tests" in diag:
# md += "
๐งช Tests & Procedures
"
# tests = diag["tests"]
# if "vitals" in tests:
# md += "- Vitals:
"
# for k, v in tests["vitals"].items(): md += f"- {k}: {v}
"
# md += "
"
# if "procedures" in tests:
# md += "- Procedures:
"
# for k, v in tests["procedures"].items(): md += f"- {k}: {v}
"
# md += "
"
# if "lab_test" in tests:
# md += "- Lab Tests:
"
# for k, v in tests["lab_test"].items(): md += f"- {k}: {v}
"
# md += "
"
# md += "
"
# if "meat" in diag:
# md += "
๐ MEAT Validation
"
# for k, v in diag["meat"].items():
# emoji = "โ
" if v else "โ"
# md += f"- {k.capitalize()}: {emoji}
"
# md += "
"
# md += f"MEAT Rationale: {diag.get('meat_rationale','')}
"
# md += ""
# if "comorbidities" in diag and diag["comorbidities"]:
# present_comorbidities = [c for c in diag["comorbidities"] if c.get("is_present")]
# if present_comorbidities:
# md += "
๐ฉบ Comorbidities
"
# for c in present_comorbidities:
# md += f"- โ
{c.get('condition')}
{c.get('rationale')} "
# md += "
"
# md += "
โ ๏ธ Error rendering report: {e}
"
# # ---------- Processing Pipeline with Gradio Progress ----------
# def process_pipeline(pdf_file, hcc_code, model_version, csv_path=CSV_PATH, output_folder="outputs"):
# try:
# start = time.time()
# step = 0
# total_steps = 8 # Total number of steps in the pipeline
# def log(msg, current_step=0):
# elapsed = time.time() - start
# bar_html = ''
# for i in range(1, total_steps + 1):
# if i < current_step: color = "#1e40af"
# elif i == current_step: color = "#3b82f6"
# else: color = "#e5e7eb"
# bar_html += f'
'
# bar_html += '
{msg}{bar_html}โณ Elapsed: {elapsed:.1f} sec
"
# if pdf_file is None:
# yield log("โ ๏ธ Please upload a patient chart PDF.", 0)
# return
# hcc_code_str = str(hcc_code or "").strip()
# if not hcc_code_str:
# yield log("โ ๏ธ Please enter a valid HCC Code before running validation.", 0)
# return
# os.makedirs(output_folder, exist_ok=True)
# pdf_path = pdf_file.name
# file_name = os.path.splitext(os.path.basename(pdf_path))[0]
# print(f"[PROCESSING] {file_name}")
# # Step 1: Extract Demographics
# step += 1
# initial_progress_msg = log(f"๐ง Step {step}/{total_steps}: Extracting patient demographics...", step)
# yield initial_progress_msg
# demographics_engine = PatientInfoExtractionEngine(pdf_path)
# demographics_info = demographics_engine.run()
# print(f"[DEMOGRAPHICS] Extracted: {demographics_info}")
# # --- MODIFICATION: Using the elaborated method for demographics ---
# initial_report_data = {
# "file_name": file_name,
# "hcc_code": hcc_code_str,
# "model_version": model_version,
# "final_analysis": [],
# "patient_name": demographics_info.get("name", ""),
# "dob": demographics_info.get("dob", ""),
# "age": demographics_info.get("age", ""),
# "gender": demographics_info.get("gender", ""),
# "address": demographics_info.get("address", ""),
# "phone": demographics_info.get("phone", ""),
# "patient_identifier": demographics_info.get("patient_identifier", "")
# }
# # --- END OF MODIFICATION ---
# demographics_md = json_to_markdown(initial_report_data)
# yield demographics_md
# time.sleep(0.5)
# # Step 2: Diagnoses
# step += 1
# yield demographics_md + log(f"๐ Step {step}/{total_steps}: Extracting possible HCC Diagnoses...", step)
# diagnoses = HCCDiagnosisListEngine(hcc_code_str, model_version, csv_path).run()
# if not diagnoses:
# yield demographics_md + log(f"โ No diagnoses found for HCC {hcc_code_str}.", step)
# return
# # Step 3: Chart checking
# step += 1
# yield demographics_md + log(f"๐ Step {step}/{total_steps}: Checking diagnoses in patient chart...", step)
# 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:
# yield demographics_md + log(f"โ No confirmed diagnoses for HCC {hcc_code_str} in {file_name}.", step)
# return
# # Step 4: Tests
# step += 1
# yield demographics_md + log(f"๐งช Step {step}/{total_steps}: Finding relevant tests...", step)
# diagnoses_with_tests = TestFindingAgent(hcc_code=hcc_code_str, model_version=model_version).run(confirmed_diagnoses)
# # Step 5: Clinical Status
# step += 1
# yield demographics_md + log(f"โ๏ธ Step {step}/{total_steps}: Determining clinical status...", step)
# diagnoses_with_status = ClinicalStatusAgent().run(diagnoses_with_tests)
# active_diagnoses = [d for d in diagnoses_with_status if d.get("clinical_status") == "ACTIVE"]
# # Step 6: MEAT
# step += 1
# if active_diagnoses:
# yield demographics_md + log(f"๐ Step {step}/{total_steps}: Validating MEAT...", step)
# validated_meat_diagnoses = MEATValidatorAgent().run(active_diagnoses)
# else:
# validated_meat_diagnoses = []
# yield demographics_md + log("โน๏ธ No ACTIVE diagnoses found. Skipping MEAT/Comorbidity.", step)
# # Step 7: Comorbidities
# step += 1
# diagnoses_passed_meat = [d for d in validated_meat_diagnoses if any(d.get("meat", {}).values())]
# if diagnoses_passed_meat:
# yield demographics_md + log(f"๐ค Step {step}/{total_steps}: Checking comorbidities...", step)
# comorbidity_results = ComorbidityCheckerAgent(pdf_path, hcc_code_str, model_version).run(diagnoses_passed_meat)
# else:
# comorbidity_results = []
# # Step 8: Final Report
# step += 1
# yield demographics_md + log(f"โ
Step {step}/{total_steps}: Generating final report...", step)
# # Merge results for final output
# 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"]
# # --- MODIFICATION: Using the elaborated method for demographics ---
# output_data = {
# "file_name": file_name,
# "hcc_code": hcc_code_str,
# "model_version": model_version,
# "final_analysis": filtered_final_analysis,
# "patient_name": demographics_info.get("name", ""),
# "dob": demographics_info.get("dob", ""),
# "age": demographics_info.get("age", ""),
# "gender": demographics_info.get("gender", ""),
# "address": demographics_info.get("address", ""),
# "phone": demographics_info.get("phone", ""),
# "patient_identifier": demographics_info.get("patient_identifier", "")
# }
# # --- END OF MODIFICATION ---
# elapsed = time.time() - start
# yield json_to_markdown(output_data) + f"
โ
Completed in {elapsed:.1f} sec
"
# except Exception as e:
# print(f"[ERROR] {e}")
# yield f"โ ๏ธ Error: {e}
"
# # ---------- Gradio Theme and Helpers ----------
# 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",
# )
# def load_sample_pdf():
# if not os.path.exists(SAMPLE_PDF):
# raise FileNotFoundError(f"Sample PDF not found at {SAMPLE_PDF}")
# class PDFWrapper:
# def __init__(self, path):
# self.name = path
# return PDFWrapper(SAMPLE_PDF)
# def pdf_to_iframe(file):
# if file is None: return "No PDF uploaded.
"
# try:
# with open(file.name, "rb") as f: pdf_bytes = f.read()
# encoded = base64.b64encode(pdf_bytes).decode("utf-8")
# return f''
# except Exception as e:
# return f"Failed to display PDF: {e}
"
# def clear_outputs():
# initial_md = "๐ Upload a PDF and click Run Validation to start.
"
# initial_preview = "Upload a PDF to preview
"
# return initial_md, initial_preview
# # ---------- Gradio UI ----------
# with gr.Blocks(theme=simple_theme, title=APP_TITLE) as interface:
# gr.HTML(f"""
# ๐ฉ {APP_TITLE}
#
# Upload a chart, set HCC + model version, and validate MEAT criteria.
#
# """)
# with gr.Row():
# pdf_upload = gr.File(label="Upload Patient Chart (PDF)", file_types=[".pdf"], scale=1)
# hcc_code = gr.Textbox(label="HCC Code (e.g., 12)", placeholder="Enter HCC code", scale=1)
# model_version = gr.Dropdown(choices=["V24", "V28"], label="Model Version", value="V24", scale=1)
# run_btn = gr.Button("๐ Run Validation", variant="primary", scale=1)
# with gr.Row():
# with gr.Column(scale=2):
# pdf_preview = gr.HTML(label="๐ PDF Preview", value="Upload a PDF to preview
")
# with gr.Column(scale=2):
# output_md = gr.Markdown(
# label="Validation Report",
# value="๐ Upload a PDF and click Run Validation to start.
",
# )
# pdf_upload.change(fn=pdf_to_iframe, inputs=pdf_upload, outputs=pdf_preview)
# pdf_upload.clear(fn=clear_outputs, inputs=[], outputs=[output_md, pdf_preview])
# run_btn.click(
# fn=process_pipeline,
# inputs=[pdf_upload, hcc_code, model_version],
# outputs=[output_md],
# )
# gr.Examples(
# examples=[[SAMPLE_PDF, "12", "V24"]],
# inputs=[pdf_upload, hcc_code, model_version],
# cache_examples=False
# )
# if __name__ == "__main__":
# interface.queue().launch(
# server_name="0.0.0.0",
# server_port=int(os.environ.get("PORT", 7860))
# )
# import os
# import gradio as gr
# import json
# import time
# import base64
# 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 PatientInfoExtractionEngine import PatientInfoExtractionEngine
# 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
# # ---------- JSON to Markdown ----------
# def json_to_markdown(data) -> str:
# try:
# if isinstance(data, dict):
# file_name = data.get("file_name", "Unknown Patient")
# hcc_code = data.get("hcc_code", "N/A")
# model_version = data.get("model_version", "N/A")
# analyses = data.get("final_analysis", [])
# # Demographics
# patient_name = data.get("patient_name", "")
# dob = data.get("dob", "")
# age = data.get("age", "")
# gender = data.get("gender", "")
# address = data.get("address", "")
# phone = data.get("phone", "")
# patient_identifier = data.get("patient_identifier", "")
# else:
# return "โ ๏ธ Invalid data format for report.
"
# md = f"""
#
#
๐ HCC Chart Validation Report
#
๐งพ File Name: {file_name}
#
๐ท๏ธ HCC Code: {hcc_code}
#
โ๏ธ Model Version: {model_version}
# """
# # Add demographics
# if any([patient_name, dob, age, gender, address, phone, patient_identifier]):
# md += "
"
# md += "
๐ค Patient Demographics
"
# if patient_name: md += f"
Name: {patient_name}
"
# if patient_identifier: md += f"
Patient Identifier: {patient_identifier}
"
# if dob: md += f"
Date of Birth: {dob}
"
# if age: md += f"
Age: {age}
"
# if gender: md += f"
Gender: {gender}
"
# if address: md += f"
Address: {address}
"
# if phone: md += f"
Phone: {phone}
"
# md += "
"
# # Render analyses if they exist
# if analyses:
# for idx, diag in enumerate(analyses, 1):
# md += f"""
#
#
Diagnosis {idx}. {diag.get("diagnosis", "Unknown Diagnosis")}
#
ICD-10: {diag.get("icd10", "N/A")}
#
Reference: {diag.get("reference","")}
# """
# 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"
Explicit: {explicit_ans} โ {explicit_rat}
"
# else:
# md += f"
Implicit: {implicit_ans} โ {implicit_rat}
"
# md += f"""
#
Clinical Status: {diag.get("clinical_status","N/A")}
#
Status Rationale: {diag.get("status_rationale","")}
# """
# if "tests" in diag:
# md += "
๐งช Tests & Procedures
"
# tests = diag["tests"]
# if "vitals" in tests:
# md += "- Vitals:
"
# for k, v in tests["vitals"].items(): md += f"- {k}: {v}
"
# md += "
"
# if "procedures" in tests:
# md += "- Procedures:
"
# for k, v in tests["procedures"].items(): md += f"- {k}: {v}
"
# md += "
"
# if "lab_test" in tests:
# md += "- Lab Tests:
"
# for k, v in tests["lab_test"].items(): md += f"- {k}: {v}
"
# md += "
"
# md += "
"
# if "meat" in diag:
# md += "
๐ MEAT Validation
"
# for k, v in diag["meat"].items():
# emoji = "โ
" if v else "โ"
# md += f"- {k.capitalize()}: {emoji}
"
# md += "
"
# md += f"MEAT Rationale: {diag.get('meat_rationale','')}
"
# md += ""
# # --- MODIFICATION: Filter to only show present comorbidities ---
# if "comorbidities" in diag and diag["comorbidities"]:
# present_comorbidities = [c for c in diag["comorbidities"] if c.get("is_present")]
# if present_comorbidities:
# md += "
๐ฉบ Comorbidities
"
# for c in present_comorbidities:
# md += f"- โ
{c.get('condition')}
{c.get('rationale')} "
# md += "
"
# # --- END OF MODIFICATION ---
# md += "
โ ๏ธ Error rendering report: {e}
"
# # ---------- Processing Pipeline with Gradio Progress ----------
# def process_pipeline(pdf_file, hcc_code, model_version, csv_path=CSV_PATH, output_folder="outputs"):
# try:
# start = time.time()
# step = 0
# total_steps = 8 # Total number of steps in the pipeline
# def log(msg, current_step=0):
# elapsed = time.time() - start
# bar_html = ''
# for i in range(1, total_steps + 1):
# if i < current_step: color = "#1e40af"
# elif i == current_step: color = "#3b82f6"
# else: color = "#e5e7eb"
# bar_html += f'
'
# bar_html += '
{msg}{bar_html}โณ Elapsed: {elapsed:.1f} sec
"
# if pdf_file is None:
# yield log("โ ๏ธ Please upload a patient chart PDF.", 0)
# return
# hcc_code_str = str(hcc_code or "").strip()
# if not hcc_code_str:
# yield log("โ ๏ธ Please enter a valid HCC Code before running validation.", 0)
# return
# os.makedirs(output_folder, exist_ok=True)
# pdf_path = pdf_file.name
# file_name = os.path.splitext(os.path.basename(pdf_path))[0]
# print(f"[PROCESSING] {file_name}")
# # Step 1: Extract Demographics
# step += 1
# initial_progress_msg = log(f"๐ง Step {step}/{total_steps}: Extracting patient demographics...", step)
# yield initial_progress_msg
# demographics_engine = PatientInfoExtractionEngine(pdf_path)
# demographics_info = demographics_engine.run()
# print(f"[DEMOGRAPHICS] Extracted: {demographics_info}")
# initial_report_data = {
# "file_name": file_name,
# "hcc_code": hcc_code_str,
# "model_version": model_version,
# "final_analysis": [],
# **demographics_info
# }
# demographics_md = json_to_markdown(initial_report_data)
# yield demographics_md
# time.sleep(0.5)
# # Step 2: Diagnoses
# step += 1
# yield demographics_md + log(f"๐ Step {step}/{total_steps}: Extracting possible HCC Diagnoses...", step)
# diagnoses = HCCDiagnosisListEngine(hcc_code_str, model_version, csv_path).run()
# if not diagnoses:
# yield demographics_md + log(f"โ No diagnoses found for HCC {hcc_code_str}.", step)
# return
# # Step 3: Chart checking
# step += 1
# yield demographics_md + log(f"๐ Step {step}/{total_steps}: Checking diagnoses in patient chart...", step)
# 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:
# yield demographics_md + log(f"โ No confirmed diagnoses for HCC {hcc_code_str} in {file_name}.", step)
# return
# # Step 4: Tests
# step += 1
# yield demographics_md + log(f"๐งช Step {step}/{total_steps}: Finding relevant tests...", step)
# diagnoses_with_tests = TestFindingAgent(hcc_code=hcc_code_str, model_version=model_version).run(confirmed_diagnoses)
# # Step 5: Clinical Status
# step += 1
# yield demographics_md + log(f"โ๏ธ Step {step}/{total_steps}: Determining clinical status...", step)
# diagnoses_with_status = ClinicalStatusAgent().run(diagnoses_with_tests)
# active_diagnoses = [d for d in diagnoses_with_status if d.get("clinical_status") == "ACTIVE"]
# # Step 6: MEAT
# step += 1
# if active_diagnoses:
# yield demographics_md + log(f"๐ Step {step}/{total_steps}: Validating MEAT...", step)
# validated_meat_diagnoses = MEATValidatorAgent().run(active_diagnoses)
# else:
# validated_meat_diagnoses = []
# yield demographics_md + log("โน๏ธ No ACTIVE diagnoses found. Skipping MEAT/Comorbidity.", step)
# # Step 7: Comorbidities
# step += 1
# diagnoses_passed_meat = [d for d in validated_meat_diagnoses if any(d.get("meat", {}).values())]
# if diagnoses_passed_meat:
# yield demographics_md + log(f"๐ค Step {step}/{total_steps}: Checking comorbidities...", step)
# comorbidity_results = ComorbidityCheckerAgent(pdf_path, hcc_code_str, model_version).run(diagnoses_passed_meat)
# else:
# comorbidity_results = []
# # Step 8: Final Report
# step += 1
# yield demographics_md + log(f"โ
Step {step}/{total_steps}: Generating final report...", step)
# # Merge results for final output
# 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 = {
# "file_name": file_name,
# "hcc_code": hcc_code_str,
# "model_version": model_version,
# "final_analysis": filtered_final_analysis,
# **demographics_info
# }
# elapsed = time.time() - start
# yield json_to_markdown(output_data) + f"
โ
Completed in {elapsed:.1f} sec
"
# except Exception as e:
# print(f"[ERROR] {e}")
# yield f"โ ๏ธ Error: {e}
"
# # ---------- Gradio Theme and Helpers ----------
# 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",
# )
# def load_sample_pdf():
# if not os.path.exists(SAMPLE_PDF):
# raise FileNotFoundError(f"Sample PDF not found at {SAMPLE_PDF}")
# class PDFWrapper:
# def __init__(self, path):
# self.name = path
# return PDFWrapper(SAMPLE_PDF)
# def pdf_to_iframe(file):
# if file is None: return "No PDF uploaded.
"
# try:
# with open(file.name, "rb") as f: pdf_bytes = f.read()
# encoded = base64.b64encode(pdf_bytes).decode("utf-8")
# return f''
# except Exception as e:
# return f"Failed to display PDF: {e}
"
# # ---------- Gradio UI ----------
# with gr.Blocks(theme=simple_theme, title=APP_TITLE) as interface:
# gr.HTML(f"""
# ๐ฉ {APP_TITLE}
#
# Upload a chart, set HCC + model version, and validate MEAT criteria.
#
# """)
# with gr.Row():
# pdf_upload = gr.File(label="Upload Patient Chart (PDF)", file_types=[".pdf"], scale=1)
# hcc_code = gr.Textbox(label="HCC Code (e.g., 12)", placeholder="Enter HCC code", scale=1)
# model_version = gr.Dropdown(choices=["V24", "V28"], label="Model Version", value="V24", scale=1)
# run_btn = gr.Button("๐ Run Validation", variant="primary", scale=1)
# with gr.Row():
# with gr.Column(scale=2):
# pdf_preview = gr.HTML(label="๐ PDF Preview", value="Upload a PDF to preview
")
# with gr.Column(scale=2):
# output_md = gr.Markdown(
# label="Validation Report",
# value="๐ Upload a PDF and click Run Validation to start.
",
# )
# pdf_upload.change(fn=pdf_to_iframe, inputs=pdf_upload, outputs=pdf_preview)
# run_btn.click(
# fn=process_pipeline,
# inputs=[pdf_upload, hcc_code, model_version],
# outputs=[output_md],
# )
# gr.Examples(
# examples=[[SAMPLE_PDF]],
# inputs=[pdf_upload],
# outputs=[output_md],
# fn=lambda x: process_pipeline(load_sample_pdf(), hcc_code="12", model_version="V24"),
# cache_examples=False
# )
# if __name__ == "__main__":
# interface.queue().launch(
# server_name="0.0.0.0",
# server_port=int(os.environ.get("PORT", 7860))
# )
# import os
# import gradio as gr
# import json
# import time
# import base64
# 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 PatientInfoExtractionEngine import PatientInfoExtractionEngine
# 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
# # ---------- JSON to Markdown ----------
# def json_to_markdown(data) -> str:
# try:
# if isinstance(data, dict):
# file_name = data.get("file_name", "Unknown Patient")
# hcc_code = data.get("hcc_code", "N/A")
# model_version = data.get("model_version", "N/A")
# analyses = data.get("final_analysis", [])
# # Demographics
# patient_name = data.get("patient_name", "")
# dob = data.get("dob", "")
# age = data.get("age", "")
# gender = data.get("gender", "")
# address = data.get("address", "")
# phone = data.get("phone", "")
# patient_identifier = data.get("patient_identifier", "")
# else:
# return "โ ๏ธ Invalid data format for report.
"
# md = f"""
#
#
๐ HCC Chart Validation Report
#
๐งพ File Name: {file_name}
#
๐ท๏ธ HCC Code: {hcc_code}
#
โ๏ธ Model Version: {model_version}
# """
# # Add demographics
# if any([patient_name, dob, age, gender, address, phone, patient_identifier]):
# md += "
"
# md += "
๐ค Patient Demographics
"
# if patient_name: md += f"
Name: {patient_name}
"
# if patient_identifier: md += f"
Patient Identifier: {patient_identifier}
"
# if dob: md += f"
Date of Birth: {dob}
"
# if age: md += f"
Age: {age}
"
# if gender: md += f"
Gender: {gender}
"
# if address: md += f"
Address: {address}
"
# if phone: md += f"
Phone: {phone}
"
# md += "
"
# # Render analyses if they exist
# if analyses:
# for idx, diag in enumerate(analyses, 1):
# md += f"""
#
#
{idx}. {diag.get("diagnosis", "Unknown Diagnosis")}
#
ICD-10: {diag.get("icd10", "N/A")}
#
Reference: {diag.get("reference","")}
# """
# 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"
Explicit: {explicit_ans} โ {explicit_rat}
"
# else:
# md += f"
Implicit: {implicit_ans} โ {implicit_rat}
"
# md += f"""
#
Clinical Status: {diag.get("clinical_status","N/A")}
#
Status Rationale: {diag.get("status_rationale","")}
# """
# if "tests" in diag:
# md += "
๐งช Tests & Procedures
"
# tests = diag["tests"]
# if "vitals" in tests:
# md += "- Vitals:
"
# for k, v in tests["vitals"].items(): md += f"- {k}: {v}
"
# md += "
"
# if "procedures" in tests:
# md += "- Procedures:
"
# for k, v in tests["procedures"].items(): md += f"- {k}: {v}
"
# md += "
"
# if "lab_test" in tests:
# md += "- Lab Tests:
"
# for k, v in tests["lab_test"].items(): md += f"- {k}: {v}
"
# md += "
"
# md += "
"
# if "meat" in diag:
# md += "
๐ MEAT Validation
"
# for k, v in diag["meat"].items():
# emoji = "โ
" if v else "โ"
# md += f"- {k.capitalize()}: {emoji}
"
# md += "
"
# md += f"MEAT Rationale: {diag.get('meat_rationale','')}
"
# md += ""
# if "comorbidities" in diag and diag["comorbidities"]:
# md += "
๐ฉบ Comorbidities
"
# for c in diag["comorbidities"]:
# emoji = "โ
" if c.get("is_present") else "โ"
# md += f"- {emoji} {c.get('condition')}
{c.get('rationale')} "
# md += "
"
# md += "
โ ๏ธ Error rendering report: {e}
"
# # ---------- Processing Pipeline with Gradio Progress ----------
# def process_pipeline(pdf_file, hcc_code, model_version, csv_path=CSV_PATH, output_folder="outputs", progress=gr.Progress()):
# try:
# start = time.time()
# step = 0
# total_steps = 8 # Total number of steps in the pipeline
# def log(msg, current_step=0):
# elapsed = time.time() - start
# bar_html = ''
# for i in range(1, total_steps + 1):
# if i < current_step: color = "#1e40af"
# elif i == current_step: color = "#3b82f6"
# else: color = "#e5e7eb"
# bar_html += f'
'
# bar_html += '
โ
Completed in {elapsed:.1f} sec
"
# except Exception as e:
# print(f"[ERROR] {e}")
# yield f"โ ๏ธ Error: {e}
"
# # ---------- Gradio Theme and Helpers ----------
# 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",
# )
# def load_sample_pdf():
# if not os.path.exists(SAMPLE_PDF):
# raise FileNotFoundError(f"Sample PDF not found at {SAMPLE_PDF}")
# class PDFWrapper:
# def __init__(self, path):
# self.name = path
# return PDFWrapper(SAMPLE_PDF)
# def pdf_to_iframe(file):
# if file is None: return "No PDF uploaded.
"
# try:
# with open(file.name, "rb") as f: pdf_bytes = f.read()
# encoded = base64.b64encode(pdf_bytes).decode("utf-8")
# return f''
# except Exception as e:
# return f"Failed to display PDF: {e}
"
# # ---------- Gradio UI ----------
# with gr.Blocks(theme=simple_theme, title=APP_TITLE) as interface:
# gr.HTML(f"""
# ๐ฉ {APP_TITLE}
#
# Upload a chart, set HCC + model version, and validate MEAT criteria.
#
# """)
# with gr.Row():
# pdf_upload = gr.File(label="Upload Patient Chart (PDF)", file_types=[".pdf"], scale=1)
# hcc_code = gr.Textbox(label="HCC Code (e.g., 12)", placeholder="Enter HCC code", scale=1)
# model_version = gr.Dropdown(choices=["V24", "V28"], label="Model Version", value="V24", scale=1)
# run_btn = gr.Button("๐ Run Validation", variant="primary", scale=1)
# with gr.Row():
# with gr.Column(scale=2):
# pdf_preview = gr.HTML(label="๐ PDF Preview", value="Upload a PDF to preview
")
# with gr.Column(scale=2):
# output_md = gr.Markdown(
# label="Validation Report",
# value="๐ Upload a PDF and click Run Validation to start.
",
# )
# pdf_upload.change(fn=pdf_to_iframe, inputs=pdf_upload, outputs=pdf_preview)
# run_btn.click(
# fn=process_pipeline,
# inputs=[pdf_upload, hcc_code, model_version],
# outputs=[output_md],
# )
# gr.Examples(
# examples=[[SAMPLE_PDF]],
# inputs=[pdf_upload],
# outputs=[output_md],
# fn=lambda x: process_pipeline(load_sample_pdf(), hcc_code="12", model_version="V24"),
# cache_examples=False
# )
# if __name__ == "__main__":
# interface.queue().launch(
# server_name="0.0.0.0",
# server_port=int(os.environ.get("PORT", 7860))
# )
# import os
# import gradio as gr
# import json
# import time
# import base64
# 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 PatientInfoExtractionEngine import PatientInfoExtractionEngine
# 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
# # ---------- JSON to Markdown ----------
# def json_to_markdown(data) -> str:
# try:
# if isinstance(data, dict) and "final_analysis" in data:
# file_name = data.get("file_name", "Unknown Patient")
# hcc_code = data.get("hcc_code", "N/A")
# model_version = data.get("model_version", "N/A")
# analyses = data.get("final_analysis", [])
# # Demographics
# patient_name = data.get("patient_name", "")
# dob = data.get("dob", "")
# age = data.get("age", "")
# gender = data.get("gender", "")
# address = data.get("address", "")
# phone = data.get("phone", "")
# patient_identifier = data.get("patient_identifier", "")
# elif isinstance(data, list):
# file_name = "N/A"
# hcc_code = "N/A"
# model_version = "N/A"
# analyses = data
# patient_name = dob = age = gender = address = phone = patient_identifier = ""
# else:
# return "โ ๏ธ Invalid data format for report.
"
# md = f"""
#
#
๐ HCC Chart Validation Report
#
๐งพ Patient ID: {patient_id}
#
๐ท๏ธ HCC Code: {hcc_code}
#
โ๏ธ Model Version: {model_version}
# """
# # Add demographics
# if any([patient_name, dob, age, gender, address, phone, patient_identifier]):
# md += "
"
# md += "
๐ค Patient Demographics
"
# if patient_name: md += f"
Name: {patient_name}
"
# if patient_identifier: md += f"
Patient Identifier: {patient_identifier}
"
# if dob: md += f"
Date of Birth: {dob}
"
# if age: md += f"
Age: {age}
"
# if gender: md += f"
Gender: {gender}
"
# if address: md += f"
Address: {address}
"
# if phone: md += f"
Phone: {phone}
"
# md += "
"
# # Render analyses
# for idx, diag in enumerate(analyses, 1):
# md += f"""
#
#
{idx}. {diag.get("diagnosis", "Unknown Diagnosis")}
#
ICD-10: {diag.get("icd10", "N/A")}
#
Reference: {diag.get("reference","")}
# """
# 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"
Explicit: {explicit_ans} โ {explicit_rat}
"
# else:
# md += f"
Implicit: {implicit_ans} โ {implicit_rat}
"
# md += f"""
#
Clinical Status: {diag.get("clinical_status","N/A")}
#
Status Rationale: {diag.get("status_rationale","")}
# """
# if "tests" in diag:
# md += "
๐งช Tests & Procedures
"
# tests = diag["tests"]
# if "vitals" in tests:
# md += "- Vitals:
"
# for k, v in tests["vitals"].items():
# md += f"- {k}: {v}
"
# md += "
"
# if "procedures" in tests:
# md += "- Procedures:
"
# for k, v in tests["procedures"].items():
# md += f"- {k}: {v}
"
# md += "
"
# if "lab_test" in tests:
# md += "- Lab Tests:
"
# for k, v in tests["lab_test"].items():
# md += f"- {k}: {v}
"
# md += "
"
# md += "
"
# if "meat" in diag:
# md += "
๐ MEAT Validation
"
# for k, v in diag["meat"].items():
# emoji = "โ
" if v else "โ"
# md += f"- {k.capitalize()}: {emoji}
"
# md += "
"
# md += f"MEAT Rationale: {diag.get('meat_rationale','')}
"
# md += ""
# if "comorbidities" in diag and diag["comorbidities"]:
# md += "
๐ฉบ Comorbidities
"
# for c in diag["comorbidities"]:
# emoji = "โ
" if c.get("is_present") else "โ"
# md += f"- {emoji} {c.get('condition')}
{c.get('rationale')} "
# md += "
"
# md += "
โ ๏ธ Error rendering report: {e}
"
# # ---------- Processing Pipeline with Gradio Progress ----------
# def process_pipeline(pdf_file, hcc_code, model_version, csv_path=CSV_PATH, output_folder="outputs", progress=gr.Progress()):
# try:
# start = time.time()
# step = 0
# total_steps = 8 # Total number of steps in the pipeline
# def log(msg, current_step=0):
# elapsed = time.time() - start
# # Generate the segmented progress bar HTML
# bar_html = '' # Added top margin
# for i in range(1, total_steps + 1):
# if i < current_step:
# color = "#1e40af" # Completed (Dark Blue)
# elif i == current_step:
# color = "#3b82f6" # In Progress (Lighter Blue)
# else:
# color = "#e5e7eb" # Pending (Grey)
# bar_html += f'
'
# bar_html += '
โ
Completed in {elapsed:.1f} sec
"
# except Exception as e:
# print(f"[ERROR] {e}")
# yield f"โ ๏ธ Error: {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",
# )
# # ---------- Helper for Sample PDF ----------
# def load_sample_pdf():
# if not os.path.exists(SAMPLE_PDF):
# raise FileNotFoundError(f"Sample PDF not found at {SAMPLE_PDF}")
# class PDFWrapper:
# def __init__(self, path):
# self.name = path
# return PDFWrapper(SAMPLE_PDF)
# # ---------- Helper for PDF Preview ----------
# def pdf_to_iframe(file):
# if file is None:
# return "No PDF uploaded.
"
# try:
# with open(file.name, "rb") as f:
# pdf_bytes = f.read()
# encoded = base64.b64encode(pdf_bytes).decode("utf-8")
# return f"""
#
# """
# except Exception as e:
# return f"Failed to display PDF: {e}
"
# # ---------- Gradio UI ----------
# with gr.Blocks(theme=simple_theme, title=APP_TITLE) as interface:
# gr.HTML(f"""
# ๐ฉ {APP_TITLE}
#
# Upload a chart, set HCC + model version, and validate MEAT criteria.
#
# """)
# with gr.Row():
# pdf_upload = gr.File(label="Upload Patient Chart (PDF)", file_types=[".pdf"], scale=1)
# hcc_code = gr.Textbox(label="HCC Code (e.g., 12)", placeholder="Enter HCC code", scale=1)
# model_version = gr.Dropdown(choices=["V24", "V28"], label="Model Version", value="V24", scale=1)
# run_btn = gr.Button("๐ Run Validation", variant="primary", scale=1)
# with gr.Row():
# with gr.Column(scale=2):
# pdf_preview = gr.HTML(label="๐ PDF Preview", value="Upload a PDF to preview
")
# with gr.Column(scale=2):
# output_md = gr.Markdown(
# label="Validation Report",
# value="๐ Upload a PDF and click Run Validation to start.
",
# )
# pdf_upload.change(fn=pdf_to_iframe, inputs=pdf_upload, outputs=pdf_preview)
# run_btn.click(
# fn=process_pipeline,
# inputs=[pdf_upload, hcc_code, model_version],
# outputs=[output_md],
# )
# gr.Examples(
# examples=[[SAMPLE_PDF]],
# inputs=[pdf_upload],
# outputs=[output_md],
# fn=lambda x: process_pipeline(load_sample_pdf(), hcc_code="12", model_version="V24"),
# cache_examples=False
# )
# if __name__ == "__main__":
# interface.queue().launch(
# server_name="0.0.0.0",
# server_port=int(os.environ.get("PORT", 7860))
# )
# import os
# import gradio as gr
# import json
# import time
# import base64
# 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 PatientInfoExtractionEngine import PatientInfoExtractionEngine
# 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
# # ---------- JSON to Markdown ----------
# def json_to_markdown(data) -> str:
# try:
# 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", [])
# # Demographics
# patient_name = data.get("patient_name", "")
# dob = data.get("dob", "")
# age = data.get("age", "")
# gender = data.get("gender", "")
# address = data.get("address", "")
# phone = data.get("phone", "")
# patient_identifier = data.get("patient_identifier", "")
# elif isinstance(data, list):
# patient_id = "N/A"
# hcc_code = "N/A"
# model_version = "N/A"
# analyses = data
# patient_name = dob = age = gender = address = phone = patient_identifier = ""
# else:
# return "โ ๏ธ Invalid data format for report.
"
# md = f"""
#
#
๐ HCC Chart Validation Report
#
๐งพ Patient ID: {patient_id}
#
๐ท๏ธ HCC Code: {hcc_code}
#
โ๏ธ Model Version: {model_version}
# """
# # Add demographics
# if any([patient_name, dob, age, gender, address, phone, patient_identifier]):
# md += "
"
# md += "
๐ค Patient Demographics
"
# if patient_name: md += f"
Name: {patient_name}
"
# if patient_identifier: md += f"
Patient Identifier: {patient_identifier}
"
# if dob: md += f"
Date of Birth: {dob}
"
# if age: md += f"
Age: {age}
"
# if gender: md += f"
Gender: {gender}
"
# if address: md += f"
Address: {address}
"
# if phone: md += f"
Phone: {phone}
"
# md += "
"
# # Render analyses
# for idx, diag in enumerate(analyses, 1):
# md += f"""
#
#
{idx}. {diag.get("diagnosis", "Unknown Diagnosis")}
#
ICD-10: {diag.get("icd10", "N/A")}
#
Reference: {diag.get("reference","")}
# """
# 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"
Explicit: {explicit_ans} โ {explicit_rat}
"
# else:
# md += f"
Implicit: {implicit_ans} โ {implicit_rat}
"
# md += f"""
#
Clinical Status: {diag.get("clinical_status","N/A")}
#
Status Rationale: {diag.get("status_rationale","")}
# """
# if "tests" in diag:
# md += "
๐งช Tests & Procedures
"
# tests = diag["tests"]
# if "vitals" in tests:
# md += "- Vitals:
"
# for k, v in tests["vitals"].items():
# md += f"- {k}: {v}
"
# md += "
"
# if "procedures" in tests:
# md += "- Procedures:
"
# for k, v in tests["procedures"].items():
# md += f"- {k}: {v}
"
# md += "
"
# if "lab_test" in tests:
# md += "- Lab Tests:
"
# for k, v in tests["lab_test"].items():
# md += f"- {k}: {v}
"
# md += "
"
# md += "
"
# if "meat" in diag:
# md += "
๐ MEAT Validation
"
# for k, v in diag["meat"].items():
# emoji = "โ
" if v else "โ"
# md += f"- {k.capitalize()}: {emoji}
"
# md += "
"
# md += f"MEAT Rationale: {diag.get('meat_rationale','')}
"
# md += ""
# if "comorbidities" in diag and diag["comorbidities"]:
# md += "
๐ฉบ Comorbidities
"
# for c in diag["comorbidities"]:
# emoji = "โ
" if c.get("is_present") else "โ"
# md += f"- {emoji} {c.get('condition')}
{c.get('rationale')} "
# md += "
"
# md += "
โ ๏ธ Error rendering report: {e}
"
# # ---------- Processing Pipeline with Gradio Progress ----------
# def process_pipeline(pdf_file, hcc_code, model_version, csv_path=CSV_PATH, output_folder="outputs", progress=gr.Progress()):
# try:
# start = time.time()
# step = 0
# total_steps = 8 # Step 0 added for demographics
# def log(msg, step_progress=None):
# elapsed = time.time() - start
# progress_html = ""
# if step_progress is not None:
# progress_html = f"""
#
# """
# return f"{progress_html}{msg}
โณ Elapsed: {elapsed:.1f} sec"
# if pdf_file is None:
# yield log("โ ๏ธ Please upload a patient chart PDF.", 0)
# return
# hcc_code_str = str(hcc_code or "").strip()
# if not hcc_code_str:
# yield log("โ ๏ธ Please enter a valid HCC Code before running validation.", 0)
# 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"[PROCESSING] {patient_name}")
# # Step 0: Extract Demographics
# step += 1
# progress((step, total_steps), desc="Extracting demographics")
# yield log(f"๐ง Step {step}/{total_steps}: Extracting patient demographics...", step/total_steps)
# demographics_engine = PatientInfoExtractionEngine(pdf_path)
# demographics_info = demographics_engine.run()
# print(f"[DEMOGRAPHICS] Extracted: {demographics_info}")
# # Step 1: Diagnoses
# step += 1
# progress((step, total_steps), desc="Extracting diagnoses")
# yield log(f"๐ Step {step}/{total_steps}: Extracting possible HCC Diagnoses...", step/total_steps)
# diagnoses = HCCDiagnosisListEngine(hcc_code_str, model_version, csv_path).run()
# if not diagnoses:
# yield log(f"โ No diagnoses found for HCC {hcc_code_str}.", step/total_steps)
# 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...", step/total_steps)
# 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:
# yield log(f"โ No confirmed diagnoses for HCC {hcc_code_str} in {patient_name}.", step/total_steps)
# return
# # Step 3: Tests
# step += 1
# progress((step, total_steps), desc="Finding tests")
# yield log(f"๐งช Step {step}/{total_steps}: Finding relevant tests...", step/total_steps)
# 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...", step/total_steps)
# diagnoses_with_status = ClinicalStatusAgent().run(diagnoses_with_tests)
# active_diagnoses = [d for d in diagnoses_with_status if d.get("clinical_status") == "ACTIVE"]
# # Step 5: MEAT
# step += 1
# progress((step, total_steps), desc="Validating MEAT")
# if active_diagnoses:
# yield log(f"๐ Step {step}/{total_steps}: Validating MEAT...", step/total_steps)
# validated_meat_diagnoses = MEATValidatorAgent().run(active_diagnoses)
# else:
# validated_meat_diagnoses = []
# yield log("โน๏ธ No ACTIVE diagnoses found. Skipping MEAT/Comorbidity.", step/total_steps)
# # Step 6: Comorbidities
# step += 1
# progress((step, total_steps), desc="Checking comorbidities")
# diagnoses_passed_meat = [d for d in validated_meat_diagnoses if any(d.get("meat", {}).values())]
# if diagnoses_passed_meat:
# yield log(f"๐ค Step {step}/{total_steps}: Checking comorbidities...", step/total_steps)
# comorbidity_results = ComorbidityCheckerAgent(pdf_path, hcc_code_str, model_version).run(diagnoses_passed_meat)
# else:
# comorbidity_results = []
# # Step 7: Final Report
# step += 1
# progress((step, total_steps), desc="Generating report")
# yield log(f"โ
Step {step}/{total_steps}: Generating final report...", step/total_steps)
# # 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"]
# 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"
# ]
# # Include demographics in output
# output_data = {
# "patient_id": patient_name,
# "hcc_code": hcc_code_str,
# "model_version": model_version,
# "patient_name": demographics_info.get("name", ""),
# "dob": demographics_info.get("dob", ""),
# "age": demographics_info.get("age", ""),
# "gender": demographics_info.get("gender", ""),
# "address": demographics_info.get("address", ""),
# "phone": demographics_info.get("phone", ""),
# "patient_identifier": demographics_info.get("patient_identifier", ""),
# "final_analysis": filtered_final_analysis
# }
# elapsed = time.time() - start
# yield json_to_markdown(output_data) + f"
โ
Completed in {elapsed:.1f} sec
"
# except Exception as e:
# print(f"[ERROR] {e}")
# yield f"โ ๏ธ Error: {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",
# )
# # ---------- Helper for Sample PDF ----------
# def load_sample_pdf():
# if not os.path.exists(SAMPLE_PDF):
# raise FileNotFoundError(f"Sample PDF not found at {SAMPLE_PDF}")
# class PDFWrapper:
# def __init__(self, path):
# self.name = path
# return PDFWrapper(SAMPLE_PDF)
# # ---------- Helper for PDF Preview ----------
# def pdf_to_iframe(file):
# if file is None:
# return "No PDF uploaded.
"
# try:
# with open(file.name, "rb") as f:
# pdf_bytes = f.read()
# encoded = base64.b64encode(pdf_bytes).decode("utf-8")
# return f"""
#
# """
# except Exception as e:
# return f"Failed to display PDF: {e}
"
# # ---------- Gradio UI ----------
# with gr.Blocks(theme=simple_theme, title=APP_TITLE) as interface:
# gr.HTML(f"""
# ๐ฉ {APP_TITLE}
#
# Upload a chart, set HCC + model version, and validate MEAT criteria.
#
# """)
# with gr.Row():
# pdf_upload = gr.File(label="Upload Patient Chart (PDF)", file_types=[".pdf"], scale=1)
# hcc_code = gr.Textbox(label="HCC Code (e.g., 12)", placeholder="Enter HCC code", scale=1)
# model_version = gr.Dropdown(choices=["V24", "V28"], label="Model Version", value="V24", scale=1)
# run_btn = gr.Button("๐ Run Validation", variant="primary", scale=1)
# with gr.Row():
# with gr.Column(scale=2):
# pdf_preview = gr.HTML(label="๐ PDF Preview", value="Upload a PDF to preview
")
# with gr.Column(scale=2):
# output_md = gr.Markdown(
# label="Validation Report",
# value="๐ Upload a PDF and click Run Validation to start.
",
# )
# pdf_upload.change(fn=pdf_to_iframe, inputs=pdf_upload, outputs=pdf_preview)
# run_btn.click(
# fn=process_pipeline,
# inputs=[pdf_upload, hcc_code, model_version],
# outputs=[output_md],
# )
# gr.Examples(
# examples=[[SAMPLE_PDF]],
# inputs=[pdf_upload],
# outputs=[output_md],
# fn=lambda x: process_pipeline(load_sample_pdf(), hcc_code="12", model_version="V24"),
# cache_examples=False
# )
# if __name__ == "__main__":
# interface.queue().launch(
# server_name="0.0.0.0",
# server_port=int(os.environ.get("PORT", 7860))
# )
# import os
# import gradio as gr
# import json
# import time
# import base64
# 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 PatientInfoExtractionEngine import PatientInfoExtractionEngine
# 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
# # ---------- JSON to Markdown ----------
# def json_to_markdown(data) -> str:
# try:
# 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", [])
# elif isinstance(data, list):
# patient_id = "N/A"
# hcc_code = "N/A"
# model_version = "N/A"
# analyses = data
# else:
# return "โ ๏ธ Invalid data format for report.
"
# md = f"""
#
#
๐ HCC Chart Validation Report
#
๐งพ Patient ID: {patient_id}
#
๐ท๏ธ HCC Code: {hcc_code}
#
โ๏ธ Model Version: {model_version}
#
# """
# for idx, diag in enumerate(analyses, 1):
# md += f"""
#
#
{idx}. {diag.get("diagnosis", "Unknown Diagnosis")}
#
ICD-10: {diag.get("icd10", "N/A")}
#
Reference: {diag.get("reference","")}
# """
# 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"
Explicit: {explicit_ans} โ {explicit_rat}
"
# else:
# md += f"
Implicit: {implicit_ans} โ {implicit_rat}
"
# md += f"""
#
Clinical Status: {diag.get("clinical_status","N/A")}
#
Status Rationale: {diag.get("status_rationale","")}
# """
# if "tests" in diag:
# md += "
๐งช Tests & Procedures
"
# tests = diag["tests"]
# if "vitals" in tests:
# md += "- Vitals:
"
# for k, v in tests["vitals"].items():
# md += f"- {k}: {v}
"
# md += "
"
# if "procedures" in tests:
# md += "- Procedures:
"
# for k, v in tests["procedures"].items():
# md += f"- {k}: {v}
"
# md += "
"
# if "lab_test" in tests:
# md += "- Lab Tests:
"
# for k, v in tests["lab_test"].items():
# md += f"- {k}: {v}
"
# md += "
"
# md += "
"
# if "meat" in diag:
# md += "
๐ MEAT Validation
"
# for k, v in diag["meat"].items():
# emoji = "โ
" if v else "โ"
# md += f"- {k.capitalize()}: {emoji}
"
# md += "
"
# md += f"MEAT Rationale: {diag.get('meat_rationale','')}
"
# md += ""
# if "comorbidities" in diag and diag["comorbidities"]:
# md += "
๐ฉบ Comorbidities
"
# for c in diag["comorbidities"]:
# emoji = "โ
" if c.get("is_present") else "โ"
# md += f"- {emoji} {c.get('condition')}
{c.get('rationale')} "
# md += "
"
# md += "
โ ๏ธ Error rendering report: {e}
"
# # ---------- Processing Pipeline with Gradio Progress ----------
# def process_pipeline(pdf_file, hcc_code, model_version, csv_path=CSV_PATH, output_folder="outputs", progress=gr.Progress()):
# try:
# start = time.time()
# step = 0
# total_steps = 7
# def log(msg, step_progress=None):
# elapsed = time.time() - start
# progress_html = ""
# if step_progress is not None:
# progress_html = f"""
#
# """
# return f"{progress_html}{msg}
โณ Elapsed: {elapsed:.1f} sec"
# if pdf_file is None:
# yield log("โ ๏ธ Please upload a patient chart PDF.", 0)
# return
# hcc_code_str = str(hcc_code or "").strip()
# if not hcc_code_str:
# yield log("โ ๏ธ Please enter a valid HCC Code before running validation.", 0)
# 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"[PROCESSING] {patient_name}")
# # Step 1: Diagnoses
# step += 1
# progress((step, total_steps), desc="Extracting diagnoses")
# yield log(f"๐ Step {step}/{total_steps}: Extracting possible HCC Diagnoses...", step/total_steps)
# diagnoses = HCCDiagnosisListEngine(hcc_code_str, model_version, csv_path).run()
# if not diagnoses:
# yield log(f"โ No diagnoses found for HCC {hcc_code_str}.", step/total_steps)
# 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...", step/total_steps)
# 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:
# yield log(f"โ No confirmed diagnoses for HCC {hcc_code_str} in {patient_name}.", step/total_steps)
# return
# # Step 3: Tests
# step += 1
# progress((step, total_steps), desc="Finding tests")
# yield log(f"๐งช Step {step}/{total_steps}: Finding relevant tests...", step/total_steps)
# 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...", step/total_steps)
# diagnoses_with_status = ClinicalStatusAgent().run(diagnoses_with_tests)
# active_diagnoses = [d for d in diagnoses_with_status if d.get("clinical_status") == "ACTIVE"]
# # Step 5: MEAT
# step += 1
# progress((step, total_steps), desc="Validating MEAT")
# if active_diagnoses:
# yield log(f"๐ Step {step}/{total_steps}: Validating MEAT...", step/total_steps)
# validated_meat_diagnoses = MEATValidatorAgent().run(active_diagnoses)
# else:
# validated_meat_diagnoses = []
# yield log("โน๏ธ No ACTIVE diagnoses found. Skipping MEAT/Comorbidity.", step/total_steps)
# # Step 6: Comorbidities
# step += 1
# progress((step, total_steps), desc="Checking comorbidities")
# diagnoses_passed_meat = [d for d in validated_meat_diagnoses if any(d.get("meat", {}).values())]
# if diagnoses_passed_meat:
# yield log(f"๐ค Step {step}/{total_steps}: Checking comorbidities...", step/total_steps)
# comorbidity_results = ComorbidityCheckerAgent(pdf_path, hcc_code_str, model_version).run(diagnoses_passed_meat)
# else:
# comorbidity_results = []
# # Step 7: Final Report
# step += 1
# progress((step, total_steps), desc="Generating report")
# yield log(f"โ
Step {step}/{total_steps}: Generating final report...", step/total_steps)
# # 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"]
# 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
# yield json_to_markdown(output_data) + f"
โ
Completed in {elapsed:.1f} sec
"
# except Exception as e:
# print(f"[ERROR] {e}")
# yield f"โ ๏ธ Error: {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",
# )
# # ---------- Helper for Sample PDF ----------
# def load_sample_pdf():
# if not os.path.exists(SAMPLE_PDF):
# raise FileNotFoundError(f"Sample PDF not found at {SAMPLE_PDF}")
# class PDFWrapper:
# def __init__(self, path):
# self.name = path
# return PDFWrapper(SAMPLE_PDF)
# # ---------- Helper for PDF Preview ----------
# def pdf_to_iframe(file):
# if file is None:
# return "No PDF uploaded.
"
# try:
# with open(file.name, "rb") as f:
# pdf_bytes = f.read()
# encoded = base64.b64encode(pdf_bytes).decode("utf-8")
# return f"""
#
# """
# except Exception as e:
# return f"Failed to display PDF: {e}
"
# # ---------- Gradio UI ----------
# with gr.Blocks(theme=simple_theme, title=APP_TITLE) as interface:
# gr.HTML(f"""
# ๐ฉ {APP_TITLE}
#
# Upload a chart, set HCC + model version, and validate MEAT criteria.
#
# """)
# with gr.Row():
# pdf_upload = gr.File(label="Upload Patient Chart (PDF)", file_types=[".pdf"], scale=1)
# hcc_code = gr.Textbox(label="HCC Code (e.g., 12)", placeholder="Enter HCC code", scale=1)
# model_version = gr.Dropdown(choices=["V24", "V28"], label="Model Version", value="V24", scale=1)
# run_btn = gr.Button("๐ Run Validation", variant="primary", scale=1)
# with gr.Row():
# with gr.Column(scale=2):
# pdf_preview = gr.HTML(label="๐ PDF Preview", value="Upload a PDF to preview
")
# with gr.Column(scale=2):
# output_md = gr.Markdown(
# label="Validation Report",
# value="๐ Upload a PDF and click Run Validation to start.
",
# )
# # Connect PDF upload to preview
# pdf_upload.change(fn=pdf_to_iframe, inputs=pdf_upload, outputs=pdf_preview)
# # Connect run button with progress
# run_btn.click(
# fn=process_pipeline,
# inputs=[pdf_upload, hcc_code, model_version],
# outputs=[output_md],
# )
# gr.Examples(
# examples=[[SAMPLE_PDF]],
# inputs=[pdf_upload],
# outputs=[output_md],
# fn=lambda x: process_pipeline(load_sample_pdf(), hcc_code="12", model_version="V24"),
# cache_examples=False
# )
# if __name__ == "__main__":
# interface.queue().launch(
# server_name="0.0.0.0",
# server_port=int(os.environ.get("PORT", 7860))
# )