|
|
|
from flask import Flask, render_template, request, send_file, redirect, url_for |
|
import os |
|
import logging |
|
from utils.meldrx import MeldRxAPI |
|
from utils.oneclick import generate_discharge_paper_one_click |
|
|
|
logging.basicConfig(level=logging.INFO) |
|
logger = logging.getLogger(__name__) |
|
|
|
app = Flask(__name__) |
|
|
|
|
|
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" |
|
|
|
|
|
meldrx_api = MeldRxAPI( |
|
client_id=CLIENT_ID, |
|
client_secret=CLIENT_SECRET, |
|
workspace_id=WORKSPACE_ID, |
|
redirect_uri=REDIRECT_URI |
|
) |
|
|
|
@app.route('/') |
|
def index(): |
|
return render_template('index.html') |
|
|
|
@app.route('/auth', methods=['GET', 'POST']) |
|
def auth(): |
|
if request.method == 'POST': |
|
auth_code = request.form.get('auth_code') |
|
if auth_code and 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(): |
|
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=meldrx_api.get_authorization_url(), auth_result="Callback failed") |
|
|
|
@app.route('/dashboard', methods=['GET']) |
|
def dashboard(): |
|
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('/oneclick', methods=['GET', 'POST']) |
|
def one_click(): |
|
if not meldrx_api.access_token: |
|
return redirect(url_for('auth')) |
|
|
|
if request.method == 'POST': |
|
patient_id = request.form.get('patient_id', '').strip() |
|
first_name = request.form.get('first_name', '').strip() |
|
last_name = request.form.get('last_name', '').strip() |
|
action = request.form.get('action', '') |
|
|
|
pdf_path, status, display_summary = generate_discharge_paper_one_click( |
|
meldrx_api, patient_id, first_name, last_name |
|
) |
|
|
|
if action == "Display Summary" and display_summary: |
|
return render_template('oneclick.html', |
|
status=status, |
|
summary=display_summary.replace('\n', '<br>')) |
|
elif action == "Generate PDF" and pdf_path: |
|
return send_file(pdf_path, |
|
as_attachment=True, |
|
download_name=f"discharge_summary_{patient_id or 'patient'}.pdf") |
|
return render_template('oneclick.html', |
|
status=status, |
|
summary=display_summary.replace('\n', '<br>') if display_summary else None) |
|
|
|
return render_template('oneclick.html') |
|
|
|
if __name__ == '__main__': |
|
port = int(os.getenv("PORT", 7860)) |
|
app.run(debug=False, host='0.0.0.0', port=port) |
|
|
|
|
|
|
|
from typing import Tuple, Optional |
|
from .meldrx import MeldRxAPI |
|
from .responseparser import PatientDataExtractor |
|
from .pdfutils import PDFGenerator |
|
import logging |
|
import json |
|
|
|
logger = logging.getLogger(__name__) |
|
|
|
def generate_discharge_paper_one_click( |
|
api: MeldRxAPI, |
|
patient_id: str = "", |
|
first_name: str = "", |
|
last_name: str = "" |
|
) -> Tuple[Optional[str], str, Optional[str]]: |
|
""" |
|
Generate a discharge summary PDF and upload it to MeldRx patient data. |
|
|
|
Args: |
|
api: Initialized MeldRxAPI instance |
|
patient_id: Optional patient ID filter |
|
first_name: Optional first name filter |
|
last_name: Optional last name filter |
|
|
|
Returns: |
|
Tuple of (pdf_path, status_message, display_summary) |
|
""" |
|
try: |
|
|
|
patients_data = api.get_patients() |
|
if not patients_data or "entry" not in patients_data: |
|
return None, "Failed to fetch patient data", None |
|
|
|
|
|
extractor = PatientDataExtractor(patients_data, "json") |
|
|
|
|
|
matching_patients = [] |
|
matching_indices = [] |
|
for i in range(len(extractor.patients)): |
|
extractor.set_patient_by_index(i) |
|
patient_data = extractor.get_patient_dict() |
|
|
|
if (not patient_id or patient_data["id"] == patient_id) and \ |
|
(not first_name or patient_data["first_name"].lower() == first_name.lower()) and \ |
|
(not last_name or patient_data["last_name"].lower() == last_name.lower()): |
|
matching_patients.append(patient_data) |
|
matching_indices.append(i) |
|
|
|
if not matching_patients: |
|
return None, "No matching patients found", None |
|
|
|
|
|
patient_data = matching_patients[0] |
|
patient_index = matching_indices[0] |
|
extractor.set_patient_by_index(patient_index) |
|
|
|
|
|
summary_text = format_discharge_summary(patient_data) |
|
|
|
|
|
pdf_gen = PDFGenerator() |
|
filename = f"discharge_{patient_data['id'] or 'summary'}_{patient_data['last_name']}.pdf" |
|
pdf_path = pdf_gen.generate_pdf_from_text(summary_text, filename) |
|
|
|
|
|
if pdf_path: |
|
|
|
discharge_summary_json = json.dumps(summary_text) |
|
|
|
|
|
patient_resource = extractor.patients[patient_index] |
|
|
|
|
|
if "extension" not in patient_resource: |
|
patient_resource["extension"] = [] |
|
|
|
patient_resource["extension"].append({ |
|
"url": "http://example.com/fhir/StructureDefinition/discharge-summary", |
|
"valueString": discharge_summary_json |
|
}) |
|
|
|
|
|
update_success = api.update_fhir_patient(patient_data["id"], patient_resource) |
|
|
|
status_message = "Discharge summary generated and uploaded successfully" if update_success \ |
|
else "Discharge summary generated but failed to upload to MeldRx" |
|
|
|
return pdf_path, status_message, summary_text |
|
|
|
return None, "Failed to generate PDF", summary_text |
|
|
|
except Exception as e: |
|
logger.error(f"Error in one-click discharge generation: {str(e)}") |
|
return None, f"Error: {str(e)}", None |
|
|
|
def format_discharge_summary(patient_data: dict) -> str: |
|
"""Format patient data into a discharge summary text.""" |
|
summary = [ |
|
"DISCHARGE SUMMARY", |
|
"", |
|
"PATIENT INFORMATION", |
|
f"Name: {patient_data['name_prefix']} {patient_data['first_name']} {patient_data['last_name']}", |
|
f"Date of Birth: {patient_data['dob']}", |
|
f"Age: {patient_data['age']}", |
|
f"Gender: {patient_data['sex']}", |
|
f"Patient ID: {patient_data['id']}", |
|
"", |
|
"CONTACT INFORMATION", |
|
f"Address: {patient_data['address']}", |
|
f"City: {patient_data['city']}, {patient_data['state']} {patient_data['zip_code']}", |
|
f"Phone: {patient_data['phone']}", |
|
"", |
|
"ADMISSION INFORMATION", |
|
f"Admission Date: {patient_data['admission_date']}", |
|
f"Discharge Date: {patient_data['discharge_date']}", |
|
f"Diagnosis: {patient_data['diagnosis']}", |
|
"", |
|
"MEDICATIONS", |
|
f"{patient_data['medications']}", |
|
"", |
|
"PHYSICIAN INFORMATION", |
|
f"Physician: Dr. {patient_data['doctor_first_name']} {patient_data['doctor_last_name']}", |
|
f"Hospital: {patient_data['hospital_name']}", |
|
f"Address: {patient_data['doctor_address']}", |
|
f"City: {patient_data['doctor_city']}, {patient_data['doctor_state']} {patient_data['doctor_zip']}", |
|
] |
|
|
|
return "\n".join(line for line in summary if line.strip() or line == "") |