big changes to the application flow

app.py

@@ -1,53 +1,28 @@
-from flask import Flask, render_template, request, send_file, jsonify, Response, redirect, url_for
+from flask import Flask, render_template, request, send_file, redirect, url_for
 import os
-import json
 import logging
-from utils.callbackmanager import CallbackManager
-from utils.generators import (generate_pdf_from_form, generate_pdf_from_meldrx,
-                              analyze_dicom_file_with_ai, analyze_hl7_file_with_ai,
-                              analyze_cda_xml_file_with_ai, analyze_pdf_file_with_ai,
-                              analyze_csv_file_with_ai)
-from utils.oneclick import generate_discharge_paper_one_click
 from utils.meldrx import MeldRxAPI
+from utils.oneclick import generate_discharge_paper_one_click
 
-# Set up logging
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
 app = Flask(__name__)
 
-# Set SPACE_URL from environment variable or default to the correct Space URL
+# Configuration from environment variables
+CLIENT_ID = os.getenv("MELDRX_CLIENT_ID", "04bdc9f9a23d488a868b93d594ee5a4a")
+CLIENT_SECRET = os.getenv("MELDRX_CLIENT_SECRET", None)
+WORKSPACE_ID = os.getenv("MELDRX_WORKSPACE_ID", "09ed4f76-b5ac-42bf-92d5-496933203dbe")
 SPACE_URL = os.getenv("SPACE_URL", "https://multitransformer-tonic-discharge-guard.hf.space")
 REDIRECT_URI = f"{SPACE_URL}/auth/callback"
-CALLBACK_MANAGER = CallbackManager(
-    redirect_uri=REDIRECT_URI,
-    client_secret=None,
-)
 
-# Mock display_form function for HTML rendering
-def display_form(first_name, last_name, middle_initial, dob, age, sex, address, city, state, zip_code,
-                 doctor_first_name, doctor_last_name, doctor_middle_initial, hospital_name, doctor_address,
-                 doctor_city, doctor_state, doctor_zip, admission_date, referral_source, admission_method,
-                 discharge_date, discharge_reason, date_of_death, diagnosis, procedures, medications,
-                 preparer_name, preparer_job_title):
-    return f"""
-    <div style='color:#00FFFF; font-family: monospace;'>
-    <strong>Patient Discharge Form</strong><br>
-    - Name: {first_name} {middle_initial} {last_name}<br>
-    - Date of Birth: {dob}, Age: {age}, Sex: {sex}<br>
-    - Address: {address}, {city}, {state}, {zip_code}<br>
-    - Doctor: {doctor_first_name} {doctor_middle_initial} {doctor_last_name}<br>
-    - Hospital/Clinic: {hospital_name}<br>
-    - Doctor Address: {doctor_address}, {doctor_city}, {doctor_state}, {doctor_zip}<br>
-    - Admission Date: {admission_date}, Source: {referral_source}, Method: {admission_method}<br>
-    - Discharge Date: {discharge_date}, Reason: {discharge_reason}<br>
-    - Date of Death: {date_of_death}<br>
-    - Diagnosis: {diagnosis}<br>
-    - Procedures: {procedures}<br>
-    - Medications: {medications}<br>
-    - Prepared By: {preparer_name}, {preparer_job_title}
-    </div>
-    """
+# Initialize MeldRx API
+meldrx_api = MeldRxAPI(
+    client_id=CLIENT_ID,
+    client_secret=CLIENT_SECRET,
+    workspace_id=WORKSPACE_ID,
+    redirect_uri=REDIRECT_URI
+)
 
 @app.route('/')
 def index():
@@ -55,110 +30,34 @@ def index():
 
 @app.route('/auth', methods=['GET', 'POST'])
 def auth():
-    auth_url = CALLBACK_MANAGER.get_auth_url()
     if request.method == 'POST':
-        redirected_url = request.form.get('redirected_url')
-        if redirected_url:
-            auth_code, error = CALLBACK_MANAGER.handle_callback(redirected_url)
-            result = CALLBACK_MANAGER.set_auth_code(auth_code) if auth_code else f"<span style='color:#FF4500;'>{error}</span>"
-            return render_template('auth.html', auth_url=auth_url, auth_result=result)
         auth_code = request.form.get('auth_code')
         if auth_code:
-            result = CALLBACK_MANAGER.set_auth_code(auth_code)
-            return render_template('auth.html', auth_url=auth_url, auth_result=result)
-    return render_template('auth.html', auth_url=auth_url)
+            if meldrx_api.authenticate_with_code(auth_code):
+                return redirect(url_for('dashboard'))
+            return render_template('auth.html', auth_url=meldrx_api.get_authorization_url(), auth_result="Authentication failed")
+    return render_template('auth.html', auth_url=meldrx_api.get_authorization_url())
 
 @app.route('/auth/callback', methods=['GET'])
 def auth_callback():
-    redirected_url = request.url
-    result = CALLBACK_MANAGER.handle_callback(redirected_url)
-    if "Authentication successful" in result:
+    auth_code = request.args.get('code')
+    if auth_code and meldrx_api.authenticate_with_code(auth_code):
         return redirect(url_for('dashboard'))
-    return render_template('auth.html', auth_url=CALLBACK_MANAGER.get_auth_url(), auth_result=result)
+    return render_template('auth.html', auth_url=meldrx_api.get_authorization_url(), auth_result="Callback failed")
 
 @app.route('/dashboard', methods=['GET'])
 def dashboard():
-    if not CALLBACK_MANAGER.access_token:
-        return render_template('auth.html', auth_url=CALLBACK_MANAGER.get_auth_url(), auth_result="<span style='color:#FF8C00;'>Please authenticate first.</span>")
-    
-    data = CALLBACK_MANAGER.get_patient_data()
-    if data.startswith('<span'):
-        return render_template('dashboard.html', error=data)
-    patients_data = json.loads(data)
-    patients = [entry['resource'] for entry in patients_data.get('entry', []) if entry['resource'].get('resourceType') == 'Patient']
+    if not meldrx_api.access_token:
+        return redirect(url_for('auth'))
+    patients_data = meldrx_api.get_patients()
+    if not patients_data or "entry" not in patients_data:
+        return render_template('dashboard.html', error="Failed to fetch patient data")
+    patients = [entry['resource'] for entry in patients_data.get('entry', [])]
     return render_template('dashboard.html', patients=patients, authenticated=True)
 
-@app.route('/auth/patient-data', methods=['GET'])
-def patient_data():
-    data = CALLBACK_MANAGER.get_patient_data()
-    return jsonify(json.loads(data) if data and not data.startswith('<span') else {"error": data})
-
-@app.route('/auth/pdf', methods=['GET'])
-def generate_meldrx_pdf():
-    patient_data = CALLBACK_MANAGER.get_patient_data()
-    pdf_path = generate_pdf_from_meldrx(patient_data)
-    return send_file(pdf_path, as_attachment=True, download_name="meldrx_patient_data.pdf")
-
-# @app.route('/dashboard', methods=['GET'])
-# def dashboard():
-#     data = CALLBACK_MANAGER.get_patient_data()
-#     if data.startswith('<span'):  # Indicates an error or unauthenticated state
-#         return render_template('dashboard.html', error=data)
-#     patients_data = json.loads(data)
-#     patients = [entry['resource'] for entry in patients_data.get('entry', []) if entry['resource'].get('resourceType') == 'Patient']
-#     return render_template('dashboard.html', patients=patients, authenticated=True)
-
-@app.route('/form', methods=['GET', 'POST'])
-def discharge_form():
-    if request.method == 'POST':
-        form_data = request.form.to_dict()
-        if 'display' in request.form:
-            html_form = display_form(**form_data)
-            return render_template('form.html', form_output=html_form)
-        elif 'generate_pdf' in request.form:
-            pdf_path = generate_pdf_from_form(**form_data)
-            return send_file(pdf_path, as_attachment=True, download_name="discharge_form.pdf")
-    return render_template('form.html')
-
-@app.route('/analysis', methods=['GET', 'POST'])
-def file_analysis():
-    if request.method == 'POST':
-        file = request.files.get('file')
-        file_type = request.form.get('file_type')
-        if file:
-            file_path = os.path.join('/tmp', file.filename)
-            file.save(file_path)
-            if file_type == 'dicom':
-                result = analyze_dicom_file_with_ai(file_path)
-            elif file_type == 'hl7':
-                result = analyze_hl7_file_with_ai(file_path)
-            elif file_type == 'xml' or file_type == 'ccda' or file_type == 'ccd':
-                result = analyze_cda_xml_file_with_ai(file_path)
-            elif file_type == 'pdf':
-                result = analyze_pdf_file_with_ai(file_path)
-            elif file_type == 'csv':
-                result = analyze_csv_file_with_ai(file_path)
-            else:
-                result = "Unsupported file type"
-            os.remove(file_path)
-            return render_template('analysis.html', result=result, file_type=file_type)
-    return render_template('analysis.html')
-
-# Configuration from environment variables
-CLIENT_ID = os.getenv("MELDRX_CLIENT_ID")
-CLIENT_SECRET = None
-WORKSPACE_ID = os.getenv("WORKSPACE_URL")
-
-# # Initialize MeldRx API
-# meldrx_api = MeldRxAPI(
-#     client_id=CLIENT_ID,
-#     client_secret=CLIENT_SECRET,
-#     workspace_id=WORKSPACE_ID,
-#     redirect_uri=REDIRECT_URI
-# )
 @app.route('/oneclick', methods=['GET', 'POST'])
 def one_click():
-    if not CALLBACK_MANAGER.access_token:
+    if not meldrx_api.access_token:
         return redirect(url_for('auth'))
     if request.method == 'POST':
         patient_id = request.form.get('patient_id', '')
@@ -167,12 +66,12 @@ def one_click():
         action = request.form.get('action', '')
 
         pdf_path, status, display_summary = generate_discharge_paper_one_click(
-            CALLBACK_MANAGER, patient_id, first_name, last_name
+            meldrx_api, patient_id, first_name, last_name
         )
 
-        if action == "display" and display_summary:
+        if action == "Display Summary" and display_summary:
             return render_template('oneclick.html', status=status, summary=display_summary)
-        elif action == "generate_pdf" and pdf_path:
+        elif action == "Generate PDF" and pdf_path:
             return send_file(pdf_path, as_attachment=True, download_name="discharge_summary.pdf")
         return render_template('oneclick.html', status=status, summary=display_summary)
 

utils/oneclick.py

@@ -1,67 +1,46 @@
 import os
-import json
 import logging
-from typing import Optional, Dict, Any
+from typing import Optional, Dict, Any, Tuple
 from huggingface_hub import InferenceClient
 from utils.meldrx import MeldRxAPI
 from utils.pdfutils import PDFGenerator
+from utils.responseparser import PatientDataExtractor
 from datetime import datetime
 
-# Set up logging
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
-# Initialize Hugging Face Inference Client
 HF_TOKEN = os.getenv("HF_TOKEN")
 if not HF_TOKEN:
-    raise ValueError("HF_TOKEN environment variable not set. Please set your Hugging Face API token.")
+    raise ValueError("HF_TOKEN environment variable not set.")
 client = InferenceClient(api_key=HF_TOKEN)
-MODEL_NAME = "meta-llama/Llama-3.3-70B-Instruct"  # Model to use for discharge summary generation
+MODEL_NAME = "meta-llama/Llama-3.3-70B-Instruct"
 
-def generate_ai_discharge_summary(patient_data: Dict[str, Any]) -> Optional[str]:
-    """
-    Generate a discharge summary using the Hugging Face Inference Client based on patient data.
-
-    Args:
-        patient_data (Dict[str, Any]): Patient data in FHIR JSON format.
-
-    Returns:
-        Optional[str]: Generated discharge summary text or None if generation fails.
-    """
+def generate_ai_discharge_summary(patient_dict: Dict[str, str]) -> Optional[str]:
     try:
-        # Extract relevant patient information
-        name = patient_data.get("name", [{}])[0]
-        full_name = f"{name.get('given', ['Unknown'])[0]} {name.get('family', 'Unknown')}"
-        gender = patient_data.get("gender", "Unknown").capitalize()
-        birth_date = patient_data.get("birthDate", "Unknown")
-        age = calculate_age(birth_date) if birth_date != "Unknown" else "Unknown"
-
-        # Placeholder for additional clinical data (e.g., diagnosis, treatment)
-        # In a real scenario, this would come from related FHIR resources like Encounter, Condition, etc.
         patient_info = (
-            f"Patient Name: {full_name}\n"
-            f"Gender: {gender}\n"
-            f"Age: {age}\n\n"
-            f"Presentation and Diagnosis:\n[Diagnosis data not provided in this snippet; assumed from related FHIR resources]\n\n"
-            f"Hospital Course:\n[Treatment data not provided in this snippet; assumed from related FHIR resources]\n\n"
-            f"Outcome:\n[Outcome data not provided in this snippet; assumed from related FHIR resources]"
+            f"Patient Name: {patient_dict['first_name']} {patient_dict['last_name']}\n"
+            f"Gender: {patient_dict['sex']}\n"
+            f"Age: {patient_dict['age']}\n"
+            f"Date of Birth: {patient_dict['dob']}\n"
+            f"Admission Date: {patient_dict['admission_date']}\n"
+            f"Discharge Date: {patient_dict['discharge_date']}\n\n"
+            f"Diagnosis:\n{patient_dict['diagnosis']}\n\n"
+            f"Medications:\n{patient_dict['medications']}\n\n"
+            f"Discharge Instructions:\n[Generated based on available data]"
         )
 
-        # Define the prompt for the AI model
         messages = [
-            {"role": "user", "content": ""},
             {
                 "role": "assistant",
                 "content": (
-                    "You are a senior expert medical health practitioner known for producing discharge papers. "
-                    "You will receive patient information and treatment details. Produce a complete discharge summary "
-                    "based on the information provided."
+                    "You are a senior medical practitioner tasked with creating discharge summaries. "
+                    "Generate a complete discharge summary based on the provided patient information."
                 )
             },
             {"role": "user", "content": patient_info}
         ]
 
-        # Generate discharge summary using streaming
         stream = client.chat.completions.create(
             model=MODEL_NAME,
             messages=messages,
@@ -83,92 +62,75 @@ def generate_ai_discharge_summary(patient_data: Dict[str, Any]) -> Optional[str]
         logger.error(f"Error generating AI discharge summary: {str(e)}")
         return None
 
-def calculate_age(birth_date: str) -> str:
-    """
-    Calculate age from birth date.
-
-    Args:
-        birth_date (str): Birth date in YYYY-MM-DD format.
-
-    Returns:
-        str: Calculated age or 'Unknown' if calculation fails.
-    """
-    try:
-        birth = datetime.strptime(birth_date, "%Y-%m-%d")
-        today = datetime.today()
-        age = today.year - birth.year - ((today.month, today.day) < (birth.month, birth.day))
-        return str(age)
-    except ValueError:
-        return "Unknown"
-
 def generate_discharge_paper_one_click(
     meldrx_api: MeldRxAPI,
     patient_id: str = None,
     first_name: str = None,
     last_name: str = None
-) -> tuple[Optional[str], str]:
-    """
-    Generate a discharge paper with AI content in one click.
-
-    Args:
-        meldrx_api (MeldRxAPI): Initialized and authenticated MeldRxAPI instance.
-        patient_id (str, optional): Patient ID to fetch specific patient data.
-        first_name (str, optional): First name for patient lookup if patient_id is not provided.
-        last_name (str, optional): Last name for patient lookup if patient_id is not provided.
-
-    Returns:
-        tuple[Optional[str], str]: (PDF file path, Status message)
-    """
+) -> Tuple[Optional[str], str, Optional[str]]:
     try:
-        # Check if already authenticated
         if not meldrx_api.access_token:
-            return None, "Error: Not authenticated. Please authenticate first in the 'Authenticate with MeldRx' tab."
+            if not meldrx_api.authenticate():
+                return None, "Error: Authentication failed. Please authenticate first.", None
+
+        patient_data = meldrx_api.get_patients()
+        if not patient_data or "entry" not in patient_data:
+            return None, "Error: Failed to fetch patient data.", None
 
-        # Fetch patient data
+        extractor = PatientDataExtractor(patient_data, format_type="json")
+        patients = extractor.get_all_patients()
+
+        if not patients:
+            return None, "Error: No patients found in the workspace.", None
+
+        patient_dict = None
         if patient_id:
-            patient_data = meldrx_api.get_patients()
-            if not patient_data or "entry" not in patient_data:
-                return None, "Error: Failed to fetch patient data by ID."
-            patients = [entry["resource"] for entry in patient_data.get("entry", [])]
-            patient = next((p for p in patients if p.get("id") == patient_id), None)
-            if not patient:
-                return None, f"Error: Patient with ID {patient_id} not found."
+            for p in patients:
+                extractor.set_patient_by_index(patients.index(p))
+                if extractor.get_id() == patient_id:
+                    patient_dict = p
+                    break
+            if not patient_dict:
+                return None, f"Error: Patient with ID {patient_id} not found.", None
+        elif first_name and last_name:
+            patient_dict = next(
+                (p for p in patients if
+                 p["first_name"].lower() == first_name.lower() and
+                 p["last_name"].lower() == last_name.lower()),
+                None
+            )
+            if not patient_dict:
+                return None, f"Error: Patient with name {first_name} {last_name} not found.", None
         else:
-            patient_data = meldrx_api.get_patients()
-            if not patient_data or "entry" not in patient_data:
-                return None, "Error: Failed to fetch patient data."
-            patients = [entry["resource"] for entry in patient_data.get("entry", [])]
-            if first_name and last_name:
-                patient = next(
-                    (p for p in patients if
-                     p.get("name", [{}])[0].get("given", [""])[0].lower() == first_name.lower() and
-                     p.get("name", [{}])[0].get("family", "").lower() == last_name.lower()),
-                    None
-                )
-                if not patient:
-                    return None, f"Error: Patient with name {first_name} {last_name} not found."
-            else:
-                patient = patients[0] if patients else None
-                if not patient:
-                    return None, "Error: No patients found in the workspace."
-
-        # Generate AI discharge summary
-        ai_content = generate_ai_discharge_summary(patient)
+            patient_dict = patients[0]
+
+        ai_content = generate_ai_discharge_summary(patient_dict)
         if not ai_content:
-            return None, "Error: Failed to generate AI discharge summary."
+            return None, "Error: Failed to generate AI discharge summary.", None
+
+        display_summary = (
+            f"<div style='color:#00FFFF; font-family: monospace;'>"
+            f"<strong>Discharge Summary Preview</strong><br>"
+            f"- Name: {patient_dict['first_name']} {patient_dict['last_name']}<br>"
+            f"- DOB: {patient_dict['dob']}, Age: {patient_dict['age']}, Sex: {patient_dict['sex']}<br>"
+            f"- Address: {patient_dict['address']}, {patient_dict['city']}, {patient_dict['state']} {patient_dict['zip_code']}<br>"
+            f"- Admission Date: {patient_dict['admission_date']}<br>"
+            f"- Discharge Date: {patient_dict['discharge_date']}<br>"
+            f"- Diagnosis: {patient_dict['diagnosis']}<br>"
+            f"- Medications: {patient_dict['medications']}<br>"
+            f"</div>"
+        )
 
-        # Generate PDF
         pdf_generator = PDFGenerator()
         pdf_path = pdf_generator.generate_pdf_from_text(
             ai_content,
-            f"discharge_summary_{patient.get('id', 'unknown')}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.pdf"
+            f"discharge_summary_{patient_id or 'unknown'}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.pdf"
         )
 
         if pdf_path:
-            return pdf_path, f"Success: Discharge paper generated for {patient.get('name', [{}])[0].get('given', ['Unknown'])[0]} {patient.get('name', [{}])[0].get('family', 'Unknown')}"
-        else:
-            return None, "Error: Failed to generate PDF."
+            return pdf_path, f"Success: Discharge paper generated for {patient_dict['first_name']} {patient_dict['last_name']}", display_summary
+        return None, "Error: Failed to generate PDF.", display_summary
 
     except Exception as e:
         logger.error(f"Error in one-click discharge generation: {str(e)}")
-        return None, f"Error: {str(e)}"
+        return None, f"Error: {str(e)}", None