Refactor sandbox and add SandboxClient api (#154)

* Add new SandboxClient api

* Add dataset registry discovery

* Load preformed prefetch from path and replace DocumentReference with xml validation in clindoc and xml loading

* Add load_dotenv to cookbook examples

* Move data_generation to sandbox

* Refactor sandbox module

* Update docs

* Update tests

* Rename save_responses -> save_results

* Update api docs

* Move data loader work to new branch

Author: jenniferjiangkells

cookbook/cds_discharge_summarizer_hf_chat.py

@@ -1,17 +1,18 @@
-import healthchain as hc
+import os
+import getpass
+
 from healthchain.gateway import HealthChainAPI, CDSHooksService
 from healthchain.pipeline import SummarizationPipeline
-from healthchain.sandbox.use_cases import ClinicalDecisionSupport
-from healthchain.models import Prefetch, CDSRequest, CDSResponse
-from healthchain.data_generators import CdsDataGenerator
+from healthchain.models import CDSRequest, CDSResponse
 
 from langchain_huggingface.llms import HuggingFaceEndpoint
 from langchain_huggingface import ChatHuggingFace
 from langchain_core.prompts import PromptTemplate
 from langchain_core.output_parsers import StrOutputParser
 
-import getpass
-import os
+from dotenv import load_dotenv
+
+load_dotenv()
 
 
 if not os.getenv("HUGGINGFACEHUB_API_TOKEN"):
@@ -65,31 +66,37 @@ def discharge_summarizer(request: CDSRequest) -> CDSResponse:
 app.register_service(cds, path="/cds")
 
 
-@hc.sandbox(api="http://localhost:8000")
-class DischargeNoteSummarizer(ClinicalDecisionSupport):
-    def __init__(self):
-        super().__init__(path="/cds/cds-services/discharge-summarizer")
-        self.data_generator = CdsDataGenerator()
-
-    @hc.ehr(workflow="encounter-discharge")
-    def load_data_in_client(self) -> Prefetch:
-        data = self.data_generator.generate_prefetch(
-            free_text_path="data/discharge_notes.csv", column_name="text"
-        )
-        return data
-
-
 if __name__ == "__main__":
     import uvicorn
     import threading
 
+    from healthchain.sandbox import SandboxClient
+
     # Start the API server in a separate thread
     def start_api():
         uvicorn.run(app, port=8000)
 
     api_thread = threading.Thread(target=start_api, daemon=True)
     api_thread.start()
 
-    # Start the sandbox
-    summarizer = DischargeNoteSummarizer()
-    summarizer.start_sandbox()
+    # Create sandbox client and load test data
+    client = SandboxClient(
+        api_url="http://localhost:8000",
+        endpoint="/cds/cds-services/discharge-summarizer",
+    )
+    # Load discharge notes from CSV
+    client.load_free_text(
+        workflow="encounter-discharge",
+        csv_path="data/discharge_notes.csv",
+        column_name="text",
+    )
+    # Send requests and get responses
+    responses = client.send_requests()
+
+    # Save results
+    client.save_results("./output/")
+
+    try:
+        api_thread.join()
+    except KeyboardInterrupt:
+        pass

cookbook/cds_discharge_summarizer_hf_trf.py

@@ -1,12 +1,13 @@
-import healthchain as hc
+import os
+import getpass
+
 from healthchain.gateway import HealthChainAPI, CDSHooksService
 from healthchain.pipeline import SummarizationPipeline
-from healthchain.sandbox.use_cases import ClinicalDecisionSupport
-from healthchain.models import Prefetch, CDSRequest, CDSResponse
-from healthchain.data_generators import CdsDataGenerator
+from healthchain.models import CDSRequest, CDSResponse
 
-import getpass
-import os
+from dotenv import load_dotenv
+
+load_dotenv()
 
 
 if not os.getenv("HUGGINGFACEHUB_API_TOKEN"):
@@ -38,31 +39,37 @@ def discharge_summarizer(request: CDSRequest) -> CDSResponse:
 app.register_service(cds, path="/cds")
 
 
-@hc.sandbox(api="http://localhost:8000")
-class DischargeNoteSummarizer(ClinicalDecisionSupport):
-    def __init__(self):
-        super().__init__(path="/cds/cds-services/discharge-summarizer")
-        self.data_generator = CdsDataGenerator()
-
-    @hc.ehr(workflow="encounter-discharge")
-    def load_data_in_client(self) -> Prefetch:
-        data = self.data_generator.generate_prefetch(
-            free_text_path="data/discharge_notes.csv", column_name="text"
-        )
-        return data
-
-
 if __name__ == "__main__":
     import uvicorn
     import threading
 
+    from healthchain.sandbox import SandboxClient
+
     # Start the API server in a separate thread
     def start_api():
         uvicorn.run(app, port=8000)
 
     api_thread = threading.Thread(target=start_api, daemon=True)
     api_thread.start()
 
-    # Start the sandbox
-    summarizer = DischargeNoteSummarizer()
-    summarizer.start_sandbox()
+    # Create sandbox client and load test data
+    client = SandboxClient(
+        api_url="http://localhost:8000",
+        endpoint="/cds/cds-services/discharge-summarizer",
+    )
+    # Load discharge notes from CSV
+    client.load_free_text(
+        workflow="encounter-discharge",
+        csv_path="data/discharge_notes.csv",
+        column_name="text",
+    )
+    # Send requests and get responses
+    responses = client.send_requests()
+
+    # Save results
+    client.save_results("./output/")
+
+    try:
+        api_thread.join()
+    except KeyboardInterrupt:
+        pass

cookbook/notereader_clinical_coding_fhir.py

@@ -14,27 +14,23 @@
 """
 
 import logging
-import uvicorn
-import healthchain as hc
 
-from fhir.resources.documentreference import DocumentReference
 from spacy.tokens import Span
 from dotenv import load_dotenv
 
-from healthchain.fhir import create_document_reference, add_provenance_metadata
+from healthchain.fhir import add_provenance_metadata
 from healthchain.gateway.api import HealthChainAPI
 from healthchain.gateway.fhir import FHIRGateway
 from healthchain.gateway.clients.fhir.base import FHIRAuthConfig
 from healthchain.gateway.soap import NoteReaderService
 from healthchain.io import CdaAdapter, Document
 from healthchain.models import CdaRequest
 from healthchain.pipeline.medicalcodingpipeline import MedicalCodingPipeline
-from healthchain.sandbox.use_cases import ClinicalDocumentation
+
 
 # Suppress Spyne warnings
 logging.getLogger("spyne.model.complex").setLevel(logging.ERROR)
 
-
 load_dotenv()
 
 # Load configuration from environment variables
@@ -115,37 +111,16 @@ def ai_coding_workflow(request: CdaRequest):
     return app
 
 
-def create_sandbox():
-    @hc.sandbox(api="http://localhost:8000/")
-    class NotereaderSandbox(ClinicalDocumentation):
-        """Sandbox for testing clinical documentation workflows"""
-
-        def __init__(self):
-            super().__init__()
-            self.data_path = "./data/notereader_cda.xml"
-
-        @hc.ehr(workflow="sign-note-inpatient")
-        def load_clinical_document(self) -> DocumentReference:
-            """Load a sample CDA document for processing"""
-            with open(self.data_path, "r") as file:
-                xml_content = file.read()
-
-            return create_document_reference(
-                data=xml_content,
-                content_type="text/xml",
-                description="Sample CDA document from sandbox",
-            )
-
-    return NotereaderSandbox()
-
-
 # Create the app
 app = create_app()
 
 
 if __name__ == "__main__":
     import threading
+    import uvicorn
+
     from time import sleep
+    from healthchain.sandbox import SandboxClient
 
     # Start server
     def run_server():
@@ -155,9 +130,18 @@ def run_server():
     server_thread.start()
     sleep(2)  # Wait for startup
 
-    # Test sandbox
-    sandbox = create_sandbox()
-    sandbox.start_sandbox()
+    # Create sandbox client for testing
+    client = SandboxClient(
+        api_url="http://localhost:8000", endpoint="/notereader/fhir/", protocol="soap"
+    )
+    # Load clinical document from file
+    client.load_from_path("./data/notereader_cda.xml")
+
+    # Send request and save response
+    responses = client.send_requests()
+
+    # Save results
+    client.save_results("./output/")
 
     try:
         server_thread.join()

docs/api/clients.md

@@ -1,9 +1,11 @@
-# Use Cases
+# Sandbox Client
+
+::: healthchain.sandbox.sandboxclient.SandboxClient
+
+::: healthchain.sandbox.generators.cdsdatagenerator
 
-::: healthchain.sandbox.use_cases.cds
 ::: healthchain.models.requests.cdsrequest
 ::: healthchain.models.responses.cdsresponse
 
-::: healthchain.sandbox.use_cases.clindoc
 ::: healthchain.models.requests.cdarequest
 ::: healthchain.models.responses.cdaresponse

docs/api/data_generators.md

@@ -217,37 +217,21 @@ app.register_service(note_service, path="/notereader")
 
 ## Test with Sample Documents
 
-HealthChain provides a [sandbox utility](../reference/utilities/sandbox.md) which simulates the NoteReader workflow end-to-end. It loads your sample CDA document, sends it to your service via the configured endpoint, and saves the request/response exchange in an `output/` directory. This lets you test the complete integration locally before connecting to Epic.
+HealthChain provides a [sandbox client utility](../reference/utilities/sandbox.md) which simulates the NoteReader workflow end-to-end. It loads your sample CDA document, sends it to your service via the configured endpoint, and saves the request/response exchange in an `output/` directory. This lets you test the complete integration locally before connecting to Epic.
 
 ```python
-import healthchain as hc
-
-from healthchain.sandbox.use_cases import ClinicalDocumentation
-from healthchain.fhir import create_document_reference
-
-from fhir.resources.documentreference import DocumentReference
-
-def create_sandbox():
-    @hc.sandbox(api="http://localhost:8000/")
-    class NotereaderSandbox(ClinicalDocumentation):
-        """Sandbox for testing clinical documentation workflows"""
-        def __init__(self):
-            super().__init__()
-            self.data_path = "./data/notereader_cda.xml"
-
-        @hc.ehr(workflow="sign-note-inpatient")
-        def load_clinical_document(self) -> DocumentReference:
-            """Load a sample CDA document for processing"""
-            with open(self.data_path, "r") as file:
-                xml_content = file.read()
-
-            return create_document_reference(
-                data=xml_content,
-                content_type="text/xml",
-                description="Sample CDA document from sandbox",
-            )
-
-    return NotereaderSandbox()
+from healthchain.sandbox import SandboxClient
+
+# Create sandbox client for SOAP/CDA testing
+client = SandboxClient(
+    api_url="http://localhost:8000",
+    endpoint="/notereader/ProcessDocument",
+    workflow="sign-note-inpatient",
+    protocol="soap"
+)
+
+# Load sample CDA document
+client.load_from_path("./data/notereader_cda.xml")
 ```
 
 ## Run the Complete Example
@@ -256,13 +240,18 @@ Now for the moment of truth! Start your service and run the sandbox to see the c
 
 ```python
 import uvicorn
+import threading
 
-# Start the service on port 8000
-uvicorn.run(app)
+# Start the API server in a separate thread
+def start_api():
+    uvicorn.run(app, port=8000)
 
-# Create and start the sandbox
-sandbox = create_sandbox()
-sandbox.start_sandbox()
+api_thread = threading.Thread(target=start_api, daemon=True)
+api_thread.start()
+
+# Send requests and save responses with sandbox client
+client.send_requests()
+client.save_results("./output/")
 ```
 
 !!! abstract "What happens when you run this"