Merge pull request #4 from dotimplement/feature/mock-ehr

Feature/mock ehr

.github/workflows/ci.yml

@@ -24,6 +24,9 @@ jobs:
         with:
           python-version: ${{matrix.python-version}}
           poetry-version: ${{matrix.poetry-version}}
+      - name: Install package
+        run: |
+          poetry install --no-interaction
       - name: Lint with Ruff
         run: |
           poetry run ruff format

.pre-commit-config.yaml

@@ -0,0 +1,19 @@
+# See https://pre-commit.com for more information
+# See https://pre-commit.com/hooks.html for more hooks
+repos:
+-   repo: https://github.com/pre-commit/pre-commit-hooks
+    rev: v3.2.0
+    hooks:
+    -   id: trailing-whitespace
+    -   id: end-of-file-fixer
+    -   id: check-yaml
+    -   id: check-added-large-files
+- repo: https://github.com/astral-sh/ruff-pre-commit
+  # Ruff version.
+  rev: v0.4.2
+  hooks:
+    # Run the linter.
+    - id: ruff
+      args: [ --fix ]
+    # Run the formatter.
+    - id: ruff-format

README.md

@@ -1,8 +1,8 @@
 ![GitHub License](https://img.shields.io/github/license/:user/:repo)
 
-# DoppelData
+# HealthChain
 
-A package to help Healthtech developers generate synthetic data in the Fhir data standard.
+Simplify prototyping and testing LLM applications in healthcare context.
 
 ## Quickstart
 

docs/index.md

@@ -1 +1 @@
-# DoppelData
+# DoppelData

example_use.py

@@ -0,0 +1,56 @@
+from healthchain.use_cases.cds import ClinicalDecisionSupport
+from healthchain.decorators import ehr
+import dataclasses
+import uuid
+
+
+def Run():
+    # ehr = EHR()
+    # ehr = EHR.from_doppeldata(data)
+    # ehr = EHR.from_path(path)
+
+    # ehr.UseCase = ClinicalDecisionSupport()
+    # print(ehr.UseCase)
+
+    # ehr.add_database(data)
+
+    # ehr.send_request("http://0.0.0.0:8000", Workflow("patient-view"))
+    # ehr.send_request("http://0.0.0.0:8000", Workflow("notereader-sign-inpatient"))
+
+    @dataclasses.dataclass
+    class synth_data:
+        context: dict
+        uuid: str
+        prefetch: dict
+
+    # @sandbox(use_case=ClinicalDecisionSupport())
+    class myCDS:
+        def __init__(self) -> None:
+            self.data_generator = None
+            self.use_case = ClinicalDecisionSupport()
+
+        # decorator sets up an instance of ehr configured with use case CDS
+        @ehr(workflow="patient-view", num=5)
+        def load_data(self, data_spec):
+            # data = "hello, " + data_spec
+            data = synth_data(
+                context={"userId": "Practitioner/123", "patientId": data_spec},
+                uuid=str(uuid.uuid4()),
+                prefetch={},
+            )
+            return data
+
+        # @service(langserve=True)
+        # def llm(self):
+        #     chain = llm | output_parser
+        #     return chain
+
+    cds = myCDS()
+    ehr_client = cds.load_data("123")
+    request = ehr_client.request_data
+    for i in range(len(request)):
+        print(request[i].model_dump_json(exclude_none=True))
+
+
+if __name__ == "__main__":
+    Run()

healthchain/__init__.py

@@ -0,0 +1,6 @@
+import logging
+from .utils.logger import ColorLogger
+
+
+logging.setLoggerClass(ColorLogger)
+logger = logging.getLogger(__name__)

healthchain/base.py

@@ -0,0 +1,83 @@
+from abc import ABC, abstractmethod
+from enum import Enum
+from typing import Dict
+
+
+# a workflow is a specific event that may occur in an EHR that triggers a request to server
+class Workflow(Enum):
+    patient_view = "patient-view"
+    order_select = "order-select"
+    order_sign = "order-sign"
+    encounter_discharge = "encounter-discharge"
+    notereader_sign_inpatient = "notereader-sign-inpatient"
+    notereader_sign_outpatient = "notereader-sign-outpatient"
+
+
+class UseCaseType(Enum):
+    cds = "ClinicalDecisionSupport"
+    clindoc = "ClinicalDocumentation"
+
+
+class UseCaseMapping(Enum):
+    ClinicalDecisionSupport = (
+        "patient-view",
+        "order-select",
+        "order-sign",
+        "encounter-discharge",
+    )
+    ClinicalDocumentation = ("notereader-sign-inpatient", "notereader-sign-outpatient")
+
+    def __init__(self, *workflows):
+        self.allowed_workflows = workflows
+
+
+def is_valid_workflow(use_case: UseCaseMapping, workflow: Workflow) -> bool:
+    return workflow.value in use_case.allowed_workflows
+
+
+def validate_workflow(use_case: UseCaseMapping):
+    def decorator(func):
+        def wrapper(*args, **kwargs):
+            if len(kwargs) > 0:
+                workflow = kwargs.get("workflow")
+            else:
+                for arg in args:
+                    if type(arg) == Workflow:
+                        workflow = arg
+            if not is_valid_workflow(use_case, workflow):
+                raise ValueError(f"Invalid workflow {workflow} for UseCase {use_case}")
+            return func(*args, **kwargs)
+
+        return wrapper
+
+    return decorator
+
+
+class BaseClient(ABC):
+    """Base client class
+    A client can be an EHR or CPOE etc.
+    The basic operation is that it sends data in a specified standard.
+    """
+
+    @abstractmethod
+    def send_request(self) -> None:
+        """
+        Sends a request to AI service
+        """
+
+
+class BaseUseCase(ABC):
+    """
+    Abstract class for a specific use case of an EHR object
+    Use cases will differ by:
+    - the data it accepts (FHIR or CDA)
+    - the format of the request it constructs (CDS Hook or NoteReader workflows)
+    """
+
+    @abstractmethod
+    def _validate_data(self, data) -> bool:
+        pass
+
+    @abstractmethod
+    def construct_request(self, data, workflow: Workflow) -> Dict:
+        pass

healthchain/clients.py

@@ -0,0 +1,66 @@
+import logging
+import requests
+
+from typing import Any, Callable, List, Dict
+
+from .base import BaseUseCase, BaseClient, Workflow
+
+log = logging.getLogger(__name__)
+
+
+class EHRClient(BaseClient):
+    def __init__(
+        self, func: Callable[..., Any], workflow: Workflow, use_case: BaseUseCase
+    ):
+        """
+        Initializes the EHRClient with a data generator function and optional workflow and use case.
+
+        Parameters:
+            func (Callable[..., Any]): A function to generate data for requests.
+            workflow ([Workflow]): The workflow context to apply to the data generator.
+            use_case ([BaseUseCase]): The strategy object to construct requests based on the generated data.
+            Should be a subclass of BaseUseCase. Example - ClinicalDecisionSupport()
+        """
+        self.data_generator_func: Callable[..., Any] = func
+        self.workflow: Workflow = workflow
+        self.use_case: BaseUseCase = use_case
+        self.request_data: List[Dict] = []
+
+    def generate_request(self, *args: Any, **kwargs: Any) -> None:
+        """
+        Generates a request using the data produced by the data generator function,
+        and appends it to the internal request queue.
+
+            Parameters:
+                *args (Any): Positional arguments passed to the data generator function.
+                **kwargs (Any): Keyword arguments passed to the data generator function.
+
+            Raises:
+                ValueError: If the use case is not configured.
+        """
+        data = self.data_generator_func(*args, **kwargs)
+        self.request_data.append(self.use_case.construct_request(data, self.workflow))
+
+    def send_request(self, url: str) -> List[Dict]:
+        """
+        Sends all queued requests to the specified URL and collects the responses.
+
+            Parameters:
+                url (str): The URL to which the requests will be sent.
+            Returns:
+                List[dict]: A list of JSON responses from the server.
+            Notes:
+                This method logs errors rather than raising them, to avoid interrupting the batch processing of requests.
+        """
+        json_responses: List[Dict] = []
+        for request in self.request_data:
+            try:
+                response = requests.post(
+                    url=url, data=request.model_dump_json(exclude_none=True)
+                )
+                json_responses.append(response.json())
+            except Exception as e:
+                log.error(f"Error sending request: {e}")
+                json_responses.append({})
+
+        return json_responses

healthchain/decorators.py

@@ -0,0 +1,71 @@
+import logging
+
+from functools import wraps
+from typing import Any, TypeVar, Optional, Callable, Union
+
+from .base import Workflow, BaseUseCase, UseCaseType
+from .clients import EHRClient
+
+log = logging.getLogger(__name__)
+
+F = TypeVar("F", bound=Callable)
+
+
+# TODO: add validator and error handling
+def ehr(
+    func: Optional[F] = None, *, workflow: Workflow, num: int = 1
+) -> Union[Callable[..., Any], Callable[[F], F]]:
+    """
+    A decorator that wraps around a data generator function and returns an EHRClient
+
+    Parameters:
+        func (Optional[Callable]): The function to be decorated. If None, this allows the decorator to
+                                   be used with arguments.
+        workflow ([str]): The workflow identifier which should match an item in the Workflow enum.
+                                  This specifies the context in which the EHR function will operate.
+        num (int): The number of requests to generate in the queue; defaults to 1.
+
+    Returns:
+        Callable: A decorated callable that incorporates EHR functionality or the decorator itself
+                  if 'func' is None, allowing it to be used as a parameterized decorator.
+
+    Raises:
+        ValueError: If the workflow does not correspond to any defined enum or if use case is not configured.
+        NotImplementedError: If the use case class is not one of the supported types.
+
+    Example:
+        @ehr(workflow='patient-view', num=2)
+        def generate_data(self, config):
+            # Function implementation
+    """
+
+    def decorator(func: F) -> F:
+        @wraps(func)
+        def wrapper(self: BaseUseCase, *args: Any, **kwargs: Any) -> EHRClient:
+            use_case = getattr(self, "use_case", None)
+            if use_case is None:
+                raise ValueError(
+                    f"Use case not configured! Check {type(self)} is a valid strategy."
+                )
+
+            try:
+                workflow_enum = Workflow(workflow)
+            except ValueError as e:
+                raise ValueError(
+                    f"{e}: please select from {[x.value for x in Workflow]}"
+                )
+
+            if use_case.__class__.__name__ in [e.value for e in UseCaseType]:
+                method = EHRClient(func, workflow=workflow_enum, use_case=use_case)
+                for _ in range(num):
+                    method.generate_request(self, *args, **kwargs)
+            else:
+                raise NotImplementedError
+            return method
+
+        return wrapper
+
+    if func is None:
+        return decorator
+    else:
+        return decorator(func)

healthchain/models/hooks/basehookcontext.py

@@ -0,0 +1,7 @@
+from pydantic import BaseModel
+from abc import ABC
+
+
+class BaseHookContext(BaseModel, ABC):
+    userId: str
+    patientId: str

healthchain/models/hooks/encounterdischarge.py

@@ -0,0 +1,32 @@
+from pydantic import Field
+
+from .basehookcontext import BaseHookContext
+
+
+class EncounterDischargeContext(BaseHookContext):
+    """
+    Workflow:
+    This hook is triggered during the discharge process for typically inpatient encounters. It can be invoked
+    at any point from the start to the end of the discharge process. The purpose is to allow hook services to
+    intervene in various aspects of the discharge decision. This includes verifying discharge medications,
+    ensuring continuity of care planning, and verifying necessary documentation for discharge processing.
+
+    Attributes:
+        userId (str): REQUIRED. The ID of the current user, expected to be a Practitioner or PractitionerRole.
+                      For example, 'Practitioner/123'.
+        patientId (str): REQUIRED. The FHIR Patient.id of the patient being discharged.
+        encounterId (str): REQUIRED. The FHIR Encounter.id of the encounter being ended.
+
+    Documentation: https://cds-hooks.org/hooks/encounter-discharge/
+    """
+
+    userId: str = Field(
+        ...,
+        description="The ID of the current user, expected to be in the format 'Practitioner/123'.",
+    )
+    patientId: str = Field(
+        ..., description="The FHIR Patient.id of the patient being discharged."
+    )
+    encounterId: str = Field(
+        ..., description="The FHIR Encounter.id of the encounter being ended."
+    )