add callback manager , model app

callbackmanager.py

@@ -3,35 +3,22 @@ from meldrx import MeldRxAPI
 import json
 import os
 
-# CallbackManager class to handle OAuth callbacks
 class CallbackManager:
     def __init__(self, redirect_uri: str, client_secret: str = None):
-        """
-        Initialize the CallbackManager with MeldRx API credentials.
-        The client_id and workspace_id are retrieved from environment variables APPID and WORKSPACE_URL.
-
-        Args:
-            redirect_uri (str): The redirect URI for the Gradio app (e.g., Space URL + /callback).
-            client_secret (str, optional): The client secret (for confidential clients).
-        """
-        client_id = os.getenv("APPID")  # Retrieve client_id from APPID secret
+        client_id = os.getenv("APPID")
         if not client_id:
             raise ValueError("APPID environment variable not set. Please configure it in your Space secrets.")
-        
-        workspace_id = os.getenv("WORKSPACE_URL")  # Retrieve workspace_id from WORKSPACE_URL secret
+        workspace_id = os.getenv("WORKSPACE_URL")
         if not workspace_id:
             raise ValueError("WORKSPACE_URL environment variable not set. Please configure it in your Space secrets.")
-        
         self.api = MeldRxAPI(client_id, client_secret, workspace_id, redirect_uri)
         self.auth_code = None
         self.access_token = None
 
     def get_auth_url(self) -> str:
-        """Generate and return the SMART on FHIR authorization URL."""
         return self.api.get_authorization_url()
 
     def set_auth_code(self, code: str) -> str:
-        """Set the authorization code and attempt to get an access token."""
         self.auth_code = code
         if self.api.authenticate_with_code(code):
             self.access_token = self.api.access_token
@@ -39,7 +26,6 @@ class CallbackManager:
         return "Authentication failed. Please check the code."
 
     def get_patient_data(self) -> str:
-        """Retrieve patient data using the access token."""
         if not self.access_token:
             return "Not authenticated. Please provide a valid authorization code first."
         patients = self.api.get_patients()
@@ -47,7 +33,6 @@ class CallbackManager:
             return json.dumps(patients, indent=2) if patients else "No patient data returned."
         return "Failed to retrieve patient data."
 
-# Discharge form display function (unchanged)
 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,
@@ -88,17 +73,16 @@ def display_form(
     """
     return form
 
-# Initialize CallbackManager, fetching client_id and workspace_id from secrets
+# Initialize CallbackManager
 CALLBACK_MANAGER = CallbackManager(
     redirect_uri="https://multitransformer-discharge-guard.hf.space/callback",
-    client_secret=None  # Replace with actual secret if using a confidential client
+    client_secret=None  # Replace with actual secret if needed
 )
 
-# Gradio interface (unchanged)
+
+
 with gr.Blocks() as demo:
     gr.Markdown("# Patient Discharge Form with MeldRx Integration")
-
-    # Authentication Section
     with gr.Tab("Authenticate with MeldRx"):
         gr.Markdown("## SMART on FHIR Authentication")
         auth_url_output = gr.Textbox(label="Authorization URL", value=CALLBACK_MANAGER.get_auth_url(), interactive=False)
@@ -108,77 +92,54 @@ with gr.Blocks() as demo:
         auth_result = gr.Textbox(label="Authentication Result")
         patient_data_button = gr.Button("Fetch Patient Data")
         patient_data_output = gr.Textbox(label="Patient Data")
-
-        auth_submit.click(
-            fn=CALLBACK_MANAGER.set_auth_code,
-            inputs=auth_code_input,
-            outputs=auth_result
-        )
-        patient_data_button.click(
-            fn=CALLBACK_MANAGER.get_patient_data,
-            inputs=None,
-            outputs=patient_data_output
-        )
-
-    # Discharge Form Section
+        auth_submit.click(fn=CALLBACK_MANAGER.set_auth_code, inputs=auth_code_input, outputs=auth_result)
+        patient_data_button.click(fn=CALLBACK_MANAGER.get_patient_data, inputs=None, outputs=patient_data_output)
     with gr.Tab("Discharge Form"):
         gr.Markdown("## Patient Details")
         with gr.Row():
             first_name = gr.Textbox(label="First Name")
             last_name = gr.Textbox(label="Last Name")
             middle_initial = gr.Textbox(label="Middle Initial")
-        
         with gr.Row():
             dob = gr.Textbox(label="Date of Birth")
             age = gr.Textbox(label="Age")
             sex = gr.Textbox(label="Sex")
-        
         address = gr.Textbox(label="Address")
         with gr.Row():
             city = gr.Textbox(label="City")
             state = gr.Textbox(label="State")
             zip_code = gr.Textbox(label="Zip Code")
-        
         gr.Markdown("## Primary Healthcare Professional Details")
         with gr.Row():
             doctor_first_name = gr.Textbox(label="Doctor's First Name")
             doctor_last_name = gr.Textbox(label="Doctor's Last Name")
             doctor_middle_initial = gr.Textbox(label="Middle Initial")
-        
         hospital_name = gr.Textbox(label="Hospital/Clinic Name")
         doctor_address = gr.Textbox(label="Address")
         with gr.Row():
             doctor_city = gr.Textbox(label="City")
             doctor_state = gr.Textbox(label="State")
             doctor_zip = gr.Textbox(label="Zip Code")
-        
         gr.Markdown("## Admission and Discharge Details")
         with gr.Row():
             admission_date = gr.Textbox(label="Date of Admission")
             referral_source = gr.Textbox(label="Source of Referral")
-        
         admission_method = gr.Textbox(label="Method of Admission")
         with gr.Row():
             discharge_date = gr.Textbox(label="Date of Discharge")
             discharge_reason = gr.Radio(["Treated", "Transferred", "Discharge Against Advice", "Patient Died"], label="Discharge Reason")
-        
         date_of_death = gr.Textbox(label="Date of Death (if applicable)")
-        
         gr.Markdown("## Diagnosis & Procedures")
         diagnosis = gr.Textbox(label="Diagnosis")
         procedures = gr.Textbox(label="Operation & Procedures")
-        
         gr.Markdown("## Medication Details")
         medications = gr.Textbox(label="Medication on Discharge")
-        
         gr.Markdown("## Prepared By")
         with gr.Row():
             preparer_name = gr.Textbox(label="Name")
             preparer_job_title = gr.Textbox(label="Job Title")
-        
         submit = gr.Button("Generate Form")
         output = gr.Markdown()
-        
         submit.click(
             display_form,
             inputs=[

meldrx.py

@@ -1,23 +1,12 @@
 import requests
 import json
+import base64
+import hashlib
+import secrets
 from typing import Optional, Dict, Any
 
 class MeldRxAPI:
-    """
-    A Python class to interact with the MeldRx API, including the MIPS API.
-    Provides methods to authenticate, retrieve/update patients, create virtual workspaces,
-    and interact with encounter data.
-    """
-
     def __init__(self, client_id: str, client_secret: str, workspace_id: str, redirect_uri: str):
-        """
-            Initialize the MeldRxAPI class with client credentials and workspace ID.
-
-            Args:
-                client_id (str): The client ID (App ID) obtained from MeldRx Developer Portal.
-                client_secret (str): The client secret for authentication.
-                workspace_id (str): The workspace slug/ID to interact with.
-        """
         self.base_url = "https://app.meldrx.com"
         self.api_base_url = f"{self.base_url}/api"
         self.fhir_base_url = f"{self.api_base_url}/fhir/{workspace_id}"
@@ -29,22 +18,93 @@ class MeldRxAPI:
         self.workspace_id = workspace_id
         self.redirect_uri = redirect_uri
         self.access_token = None
+        self.code_verifier = None  # Store the code_verifier for later use
         self.session = requests.Session()
 
-    def authenticate(self) -> bool:
-        """
-        Authenticate with the MeldRx API to obtain an access token.
+    def _generate_code_verifier(self) -> str:
+        """Generate a random code_verifier string (43-128 characters)."""
+        self.code_verifier = secrets.token_urlsafe(32)  # Generates a 43-character safe string
+        return self.code_verifier
 
-        Returns:
-            bool: True if authentication is successful, False otherwise.
-        """
+    def _generate_code_challenge(self, code_verifier: str) -> str:
+        """Generate a code_challenge from the code_verifier using SHA-256."""
+        sha256_hash = hashlib.sha256(code_verifier.encode('utf-8')).digest()
+        code_challenge = base64.urlsafe_b64encode(sha256_hash).decode('utf-8').rstrip('=')
+        return code_challenge
+
+    def authenticate(self) -> bool:
         payload = {
             "grant_type": "client_credentials",
             "client_id": self.client_id,
             "client_secret": self.client_secret
         }
         headers = {"Content-Type": "application/x-www-form-urlencoded"}
+        try:
+            response = self.session.post(self.token_url, data=payload, headers=headers)
+            response.raise_for_status()
+            token_data = response.json()
+            self.access_token = token_data.get("access_token")
+            if not self.access_token:
+                raise ValueError("No access token received from the server.")
+            return True
+        except requests.RequestException as e:
+            print(f"Authentication failed: {e}")
+            return False
+        except ValueError as e:
+            print(f"Authentication error: {e}")
+            return False
 
+    def _get_headers(self) -> Dict[str, str]:
+        headers = {"Content-Type": "application/json"}
+        if self.access_token:
+            headers["Authorization"] = f"Bearer {self.access_token}"
+        return headers
+
+    def get_patients(self) -> Optional[Dict[str, Any]]:
+        url = f"{self.fhir_base_url}/Patient"
+        if not self.access_token and not self.authenticate():
+            print("Cannot proceed without authentication.")
+            return None
+        try:
+            response = self.session.get(url, headers=self._get_headers())
+            response.raise_for_status()
+            return response.json() if response.text else {}
+        except requests.RequestException as e:
+            print(f"Failed to retrieve patients: {e}")
+            return None
+
+    def get_authorization_url(self, scope: str = "patient/*.read openid profile", state: str = "random_state") -> str:
+        """Generate the SMART on FHIR authorization URL with PKCE."""
+        code_verifier = self._generate_code_verifier()
+        code_challenge = self._generate_code_challenge(code_verifier)
+        params = {
+            "response_type": "code",
+            "client_id": self.client_id,
+            "redirect_uri": self.redirect_uri,
+            "scope": scope,
+            "state": state,
+            "aud": self.fhir_base_url,
+            "code_challenge": code_challenge,
+            "code_challenge_method": "S256"  # MeldRx likely expects SHA-256
+        }
+        query_string = "&".join(f"{k}={v}" for k, v in params.items())
+        return f"{self.authorize_url}?{query_string}"
+
+    def authenticate_with_code(self, auth_code: str) -> bool:
+        """Exchange an authorization code for an access token, including the code_verifier."""
+        if not self.code_verifier:
+            print("Code verifier not set. Generate an authorization URL first.")
+            return False
+        payload = {
+            "grant_type": "authorization_code",
+            "code": auth_code,
+            "redirect_uri": self.redirect_uri,
+            "client_id": self.client_id,
+            "code_verifier": self.code_verifier  # Include the code_verifier
+        }
+        if self.client_secret:
+            payload["client_secret"] = self.client_secret
+        headers = {"Content-Type": "application/x-www-form-urlencoded"}
         try:
             response = self.session.post(self.token_url, data=payload, headers=headers)
             response.raise_for_status()