Skip to content

Use Cases

ClinicalDecisionSupport

Bases: BaseUseCase

Implements EHR backend simulator for Clinical Decision Support (CDS)

PARAMETER DESCRIPTION
service_api

the function body to inject into the main service

TYPE: APIMethod DEFAULT: None

service_config

the config kwargs for the uvicorn server passed into service

TYPE: Dict DEFAULT: None

service

the service runner object

TYPE: Service DEFAULT: None

client

the client runner object

TYPE: BaseClient DEFAULT: None

See https://cds-hooks.org/ for specification

Source code in healthchain/use_cases/cds.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
class ClinicalDecisionSupport(BaseUseCase):
    """
    Implements EHR backend simulator for Clinical Decision Support (CDS)

    Parameters:
        service_api (APIMethod): the function body to inject into the main service
        service_config (Dict): the config kwargs for the uvicorn server passed into service
        service (Service): the service runner object
        client (BaseClient): the client runner object

    See https://cds-hooks.org/ for specification
    """

    def __init__(
        self,
        service_api: Optional[APIMethod] = None,
        service_config: Optional[Dict] = None,
        service: Optional[Service] = None,
        client: Optional[BaseClient] = None,
    ) -> None:
        super().__init__(
            service_api=service_api,
            service_config=service_config,
            service=service,
            client=client,
        )
        self._type = UseCaseType.cds
        self._strategy = ClinicalDecisionSupportStrategy()
        # do we need keys? just in case
        # TODO make configurable
        self._endpoints = {
            "info": Endpoint(
                path="/cds-services",
                method="GET",
                function=self.cds_discovery,
                api_protocol="REST",
            ),
            "service_mount": Endpoint(
                path="/cds-services/{id}",
                method="POST",
                function=self.cds_service,
                api_protocol="REST",
            ),
        }

    @property
    def description(self) -> str:
        return "Clinical decision support (HL7 CDS specification)"

    @property
    def type(self) -> UseCaseType:
        return self._type

    @property
    def strategy(self) -> BaseStrategy:
        return self._strategy

    @property
    def endpoints(self) -> Dict[str, Endpoint]:
        return self._endpoints

    def cds_discovery(self) -> CDSServiceInformation:
        """
        CDS discovery endpoint for FastAPI app, should be mounted to /cds-services
        """
        if self._client is None:
            log.warning("CDS 'client' not configured, check class init.")
            return CDSServiceInformation(services=[])

        service_info = CDSService(
            hook=self._client.workflow.value,
            description="A test CDS hook service.",
            id="1",
        )
        return CDSServiceInformation(services=[service_info])

    def cds_service(self, id: str, request: CDSRequest) -> CDSResponse:
        """
        CDS service endpoint for FastAPI app, mounted to /cds-services/{id}

        This method handles the execution of a specific CDS service. It validates the
        service configuration, checks the input parameters, executes the service
        function, and ensures the correct response type is returned.

        Args:
            id (str): The unique identifier of the CDS service to be executed.
            request (CDSRequest): The request object containing the input data for the CDS service.

        Returns:
            CDSResponse: The response object containing the cards generated by the CDS service.

        Raises:
            AssertionError: If the service function is not properly configured.
            TypeError: If the input or output types do not match the expected types.

        Note:
            This method performs several checks to ensure the integrity of the service:
            1. Verifies that the service API is configured.
            2. Validates the signature of the service function.
            3. Ensures the service function accepts a CDSRequest as its first argument.
            4. Verifies that the service function returns a CDSResponse.
        """
        # TODO: can register multiple services and fetch with id

        # Check service_api
        if self._service_api is None:
            log.warning("CDS 'service_api' not configured, check class init.")
            return CDSResponse(cards=[])

        # Check that the first argument of self._service_api.func is of type CDSRequest
        func_signature = inspect.signature(self._service_api.func)
        params = list(func_signature.parameters.values())
        if len(params) < 2:  # Only 'self' parameter
            raise AssertionError(
                "Service function must have at least one parameter besides 'self'"
            )
        first_param = params[1]  # Skip 'self'
        if first_param.annotation == inspect.Parameter.empty:
            log.warning(
                "Service function parameter has no type annotation. Expected CDSRequest."
            )
        elif first_param.annotation != CDSRequest:
            raise TypeError(
                f"Expected first argument of service function to be CDSRequest, but got {first_param.annotation}"
            )

        # Call the service function
        response = self._service_api.func(self, request)

        # Check that response is of type CDSResponse
        if not isinstance(response, CDSResponse):
            raise TypeError(f"Expected CDSResponse, but got {type(response).__name__}")

        return response

cds_discovery()

CDS discovery endpoint for FastAPI app, should be mounted to /cds-services

Source code in healthchain/use_cases/cds.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
def cds_discovery(self) -> CDSServiceInformation:
    """
    CDS discovery endpoint for FastAPI app, should be mounted to /cds-services
    """
    if self._client is None:
        log.warning("CDS 'client' not configured, check class init.")
        return CDSServiceInformation(services=[])

    service_info = CDSService(
        hook=self._client.workflow.value,
        description="A test CDS hook service.",
        id="1",
    )
    return CDSServiceInformation(services=[service_info])

cds_service(id, request)

CDS service endpoint for FastAPI app, mounted to /cds-services/{id}

This method handles the execution of a specific CDS service. It validates the service configuration, checks the input parameters, executes the service function, and ensures the correct response type is returned.

PARAMETER DESCRIPTION
id

The unique identifier of the CDS service to be executed.

TYPE: str

request

The request object containing the input data for the CDS service.

TYPE: CDSRequest

RETURNS DESCRIPTION
CDSResponse

The response object containing the cards generated by the CDS service.

TYPE: CDSResponse

RAISES DESCRIPTION
AssertionError

If the service function is not properly configured.
TypeError

If the input or output types do not match the expected types.
Note

This method performs several checks to ensure the integrity of the service: 1. Verifies that the service API is configured. 2. Validates the signature of the service function. 3. Ensures the service function accepts a CDSRequest as its first argument. 4. Verifies that the service function returns a CDSResponse.

Source code in healthchain/use_cases/cds.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
def cds_service(self, id: str, request: CDSRequest) -> CDSResponse:
    """
    CDS service endpoint for FastAPI app, mounted to /cds-services/{id}

    This method handles the execution of a specific CDS service. It validates the
    service configuration, checks the input parameters, executes the service
    function, and ensures the correct response type is returned.

    Args:
        id (str): The unique identifier of the CDS service to be executed.
        request (CDSRequest): The request object containing the input data for the CDS service.

    Returns:
        CDSResponse: The response object containing the cards generated by the CDS service.

    Raises:
        AssertionError: If the service function is not properly configured.
        TypeError: If the input or output types do not match the expected types.

    Note:
        This method performs several checks to ensure the integrity of the service:
        1. Verifies that the service API is configured.
        2. Validates the signature of the service function.
        3. Ensures the service function accepts a CDSRequest as its first argument.
        4. Verifies that the service function returns a CDSResponse.
    """
    # TODO: can register multiple services and fetch with id

    # Check service_api
    if self._service_api is None:
        log.warning("CDS 'service_api' not configured, check class init.")
        return CDSResponse(cards=[])

    # Check that the first argument of self._service_api.func is of type CDSRequest
    func_signature = inspect.signature(self._service_api.func)
    params = list(func_signature.parameters.values())
    if len(params) < 2:  # Only 'self' parameter
        raise AssertionError(
            "Service function must have at least one parameter besides 'self'"
        )
    first_param = params[1]  # Skip 'self'
    if first_param.annotation == inspect.Parameter.empty:
        log.warning(
            "Service function parameter has no type annotation. Expected CDSRequest."
        )
    elif first_param.annotation != CDSRequest:
        raise TypeError(
            f"Expected first argument of service function to be CDSRequest, but got {first_param.annotation}"
        )

    # Call the service function
    response = self._service_api.func(self, request)

    # Check that response is of type CDSResponse
    if not isinstance(response, CDSResponse):
        raise TypeError(f"Expected CDSResponse, but got {type(response).__name__}")

    return response

ClinicalDecisionSupportStrategy

Bases: BaseStrategy

Handles the request construction and validation

Source code in healthchain/use_cases/cds.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
class ClinicalDecisionSupportStrategy(BaseStrategy):
    """
    Handles the request construction and validation
    """

    def __init__(self) -> None:
        self.api_protocol = ApiProtocol.rest
        self.context_mapping = {
            Workflow.order_select: OrderSelectContext,
            Workflow.order_sign: OrderSignContext,
            Workflow.patient_view: PatientViewContext,
            Workflow.encounter_discharge: EncounterDischargeContext,
        }

    @validate_workflow(UseCaseMapping.ClinicalDecisionSupport)
    def construct_request(self, data: CdsFhirData, workflow: Workflow) -> CDSRequest:
        """
        Constructs a HL7-compliant CDS request based on workflow.

        Parameters:
            data: FHIR data to be injected in request.
            workflow (Workflow): The CDS hook name, e.g. patient-view.

        Returns:
            CDSRequest: A Pydantic model that wraps a CDS request for REST

        Raises:
            ValueError: If the workflow is invalid or the data does not validate properly.
        """
        log.debug(f"Constructing CDS request for {workflow.value} from {data}")

        context_model = self.context_mapping.get(workflow, None)
        if context_model is None:
            raise ValueError(
                f"Invalid workflow {workflow.value} or workflow model not implemented."
            )
        if not isinstance(data, CdsFhirData):
            raise TypeError(
                f"CDS clients must return data of type CdsFhirData, not {type(data)}"
            )

        # i feel like theres a better way to do this
        request = CDSRequest(
            hook=workflow.value,
            context=context_model(**data.context),
            prefetch=data.model_dump_prefetch(),
        )

        return request

construct_request(data, workflow)

Constructs a HL7-compliant CDS request based on workflow.

PARAMETER DESCRIPTION
data

FHIR data to be injected in request.

TYPE: CdsFhirData

workflow

The CDS hook name, e.g. patient-view.

TYPE: Workflow

RETURNS DESCRIPTION
CDSRequest

A Pydantic model that wraps a CDS request for REST

TYPE: CDSRequest

RAISES DESCRIPTION
ValueError

If the workflow is invalid or the data does not validate properly.
Source code in healthchain/use_cases/cds.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
@validate_workflow(UseCaseMapping.ClinicalDecisionSupport)
def construct_request(self, data: CdsFhirData, workflow: Workflow) -> CDSRequest:
    """
    Constructs a HL7-compliant CDS request based on workflow.

    Parameters:
        data: FHIR data to be injected in request.
        workflow (Workflow): The CDS hook name, e.g. patient-view.

    Returns:
        CDSRequest: A Pydantic model that wraps a CDS request for REST

    Raises:
        ValueError: If the workflow is invalid or the data does not validate properly.
    """
    log.debug(f"Constructing CDS request for {workflow.value} from {data}")

    context_model = self.context_mapping.get(workflow, None)
    if context_model is None:
        raise ValueError(
            f"Invalid workflow {workflow.value} or workflow model not implemented."
        )
    if not isinstance(data, CdsFhirData):
        raise TypeError(
            f"CDS clients must return data of type CdsFhirData, not {type(data)}"
        )

    # i feel like theres a better way to do this
    request = CDSRequest(
        hook=workflow.value,
        context=context_model(**data.context),
        prefetch=data.model_dump_prefetch(),
    )

    return request

CDSRequest

Bases: BaseModel

A model representing the data structure for a CDS service call, triggered by specific hooks within a healthcare application.

ATTRIBUTE DESCRIPTION
hook

The hook that triggered this CDS Service call. For example, 'patient-view'.

TYPE: str

hookInstance

A universally unique identifier for this particular hook call.

TYPE: UUID

fhirServer

The base URL of the CDS Client's FHIR server. This field is required if fhirAuthorization is provided.

TYPE: HttpUrl

fhirAuthorization

Optional authorization details providing a bearer access token for FHIR resources.

TYPE: Optional[FhirAuthorization]

context

Hook-specific contextual data required by the CDS service.

TYPE: Dict[str, Any]

prefetch

Optional FHIR data that was prefetched by the CDS Client.

TYPE: Optional[Dict[str, Any]]

Documentation: https://cds-hooks.org/specification/current/#http-request_1

Source code in healthchain/models/requests/cdsrequest.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
class CDSRequest(BaseModel):
    """
    A model representing the data structure for a CDS service call, triggered by specific hooks
    within a healthcare application.

    Attributes:
        hook (str): The hook that triggered this CDS Service call. For example, 'patient-view'.
        hookInstance (UUID): A universally unique identifier for this particular hook call.
        fhirServer (HttpUrl): The base URL of the CDS Client's FHIR server. This field is required if `fhirAuthorization` is provided.
        fhirAuthorization (Optional[FhirAuthorization]): Optional authorization details providing a bearer access token for FHIR resources.
        context (Dict[str, Any]): Hook-specific contextual data required by the CDS service.
        prefetch (Optional[Dict[str, Any]]): Optional FHIR data that was prefetched by the CDS Client.

    Documentation: https://cds-hooks.org/specification/current/#http-request_1
    """

    hook: str
    hookInstance: str = Field(default_factory=id_generator.generate_random_uuid)
    context: BaseHookContext
    fhirServer: Optional[HttpUrl] = None
    fhirAuthorization: Optional[FHIRAuthorization] = (
        None  # TODO: note this is required if fhirserver is given
    )
    prefetch: Optional[Dict[str, Any]] = (
        None  # fhir resource is passed either thru prefetched template of fhir server
    )
    extension: Optional[List[Dict[str, Any]]] = None

Action

Bases: BaseModel

Within a suggestion, all actions are logically AND'd together, such that a user selecting a suggestion selects all of the actions within it. When a suggestion contains multiple actions, the actions SHOULD be processed as per FHIR's rules for processing transactions with the CDS Client's fhirServer as the base url for the inferred full URL of the transaction bundle entries.

https://cds-hooks.org/specification/current/#action

Source code in healthchain/models/responses/cdsresponse.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
class Action(BaseModel):
    """
    Within a suggestion, all actions are logically AND'd together, such that a user selecting a
    suggestion selects all of the actions within it. When a suggestion contains multiple actions,
    the actions SHOULD be processed as per FHIR's rules for processing transactions with the CDS
    Client's fhirServer as the base url for the inferred full URL of the transaction bundle entries.

    https://cds-hooks.org/specification/current/#action
    """

    type: ActionTypeEnum
    description: str
    resource: Optional[Dict] = None
    resourceId: Optional[str] = None

    @model_validator(mode="after")
    def validate_action_type(self) -> Self:
        if self.type in [ActionTypeEnum.create, ActionTypeEnum.update]:
            assert (
                self.resource
            ), f"'resource' must be provided when type is '{self.type.value}'"
        else:
            assert (
                self.resourceId
            ), f"'resourceId' must be provided when type is '{self.type.value}'"

        return self

ActionTypeEnum

Bases: str, Enum

The type of action being performed

Source code in healthchain/models/responses/cdsresponse.py 
33
34
35
36
37
38
39
40
class ActionTypeEnum(str, Enum):
    """
    The type of action being performed
    """

    create = "create"
    update = "update"
    delete = "delete"

CDSResponse

Bases: BaseModel

Represents the response from a CDS service.

This class models the structure of a CDS Hooks response, which includes cards for displaying information or suggestions to the user, and optional system actions that can be executed automatically.

ATTRIBUTE DESCRIPTION
cards

A list of Card objects to be displayed to the end user. Default is an empty list.

TYPE: List[Card]

systemActions

A list of Action objects representing actions that the CDS Client should execute as part of performing the decision support requested. This field is optional.

TYPE: Optional[List[Action]]

For more information, see: https://cds-hooks.org/specification/current/#cds-service-response

Source code in healthchain/models/responses/cdsresponse.py 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
class CDSResponse(BaseModel):
    """
    Represents the response from a CDS service.

    This class models the structure of a CDS Hooks response, which includes
    cards for displaying information or suggestions to the user, and optional
    system actions that can be executed automatically.

    Attributes:
        cards (List[Card]): A list of Card objects to be displayed to the end user.
            Default is an empty list.
        systemActions (Optional[List[Action]]): A list of Action objects representing
            actions that the CDS Client should execute as part of performing
            the decision support requested. This field is optional.

    For more information, see:
    https://cds-hooks.org/specification/current/#cds-service-response
    """

    cards: List[Card] = []
    systemActions: Optional[List[Action]] = None

Card

Bases: BaseModel

Cards can provide a combination of information (for reading), suggested actions (to be applied if a user selects them), and links (to launch an app if the user selects them). The CDS Client decides how to display cards, but this specification recommends displaying suggestions using buttons, and links using underlined text.

https://cds-hooks.org/specification/current/#card-attributes

Source code in healthchain/models/responses/cdsresponse.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
class Card(BaseModel):
    """
    Cards can provide a combination of information (for reading), suggested actions
    (to be applied if a user selects them), and links (to launch an app if the user selects them).
    The CDS Client decides how to display cards, but this specification recommends displaying suggestions
    using buttons, and links using underlined text.

    https://cds-hooks.org/specification/current/#card-attributes
    """

    summary: str = Field(..., max_length=140)
    indicator: IndicatorEnum
    source: Source
    uuid: Optional[str] = None
    detail: Optional[str] = None
    suggestions: Optional[List[Suggestion]] = None
    selectionBehavior: Optional[SelectionBehaviorEnum] = None
    overrideReasons: Optional[List[SimpleCoding]] = None
    links: Optional[List[Link]] = None

    @model_validator(mode="after")
    def validate_suggestions(self) -> Self:
        if self.suggestions is not None:
            assert self.selectionBehavior, f"'selectionBehavior' must be given if 'suggestions' is present! Choose from {[v for v in SelectionBehaviorEnum.value]}"
        return self

IndicatorEnum

Bases: str, Enum

Urgency/importance of what Card conveys. Allowed values, in order of increasing urgency, are: info, warning, critical. The CDS Client MAY use this field to help make UI display decisions such as sort order or coloring.

Source code in healthchain/models/responses/cdsresponse.py 
 8
 9
10
11
12
13
14
15
16
17
class IndicatorEnum(str, Enum):
    """
    Urgency/importance of what Card conveys.
    Allowed values, in order of increasing urgency, are: info, warning, critical.
    The CDS Client MAY use this field to help make UI display decisions such as sort order or coloring.
    """

    info = "info"
    warning = "warning"
    critical = "critical"

Bases: BaseModel

  • CDS Client support for appContext requires additional coordination with the authorization server that is not described or specified in CDS Hooks nor SMART.

  • Autolaunchable is experimental

https://cds-hooks.org/specification/current/#link

Source code in healthchain/models/responses/cdsresponse.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
class Link(BaseModel):
    """
    * CDS Client support for appContext requires additional coordination with the authorization
    server that is not described or specified in CDS Hooks nor SMART.

    * Autolaunchable is experimental

    https://cds-hooks.org/specification/current/#link
    """

    label: str
    url: HttpUrl
    type: LinkTypeEnum
    appContext: Optional[str] = None
    autoLaunchable: Optional[bool]

    @model_validator(mode="after")
    def validate_link(self) -> Self:
        if self.appContext:
            assert (
                self.type == LinkTypeEnum.smart
            ), "'type' must be 'smart' for appContext to be valued."

        return self

LinkTypeEnum

Bases: str, Enum

The type of the given URL. There are two possible values for this field. A type of absolute indicates that the URL is absolute and should be treated as-is. A type of smart indicates that the URL is a SMART app launch URL and the CDS Client should ensure the SMART app launch URL is populated with the appropriate SMART launch parameters.

Source code in healthchain/models/responses/cdsresponse.py 
43
44
45
46
47
48
49
50
51
52
53
class LinkTypeEnum(str, Enum):
    """
    The type of the given URL. There are two possible values for this field.
    A type of absolute indicates that the URL is absolute and should be treated as-is.
    A type of smart indicates that the URL is a SMART app launch URL and the CDS Client
    should ensure the SMART app launch URL is populated with the appropriate SMART
    launch parameters.
    """

    absolute = "absolute"
    smart = "smart"

SelectionBehaviorEnum

Bases: str, Enum

Describes the intended selection behavior of the suggestions in the card. Allowed values are: at-most-one, indicating that the user may choose none or at most one of the suggestions; any, indicating that the end user may choose any number of suggestions including none of them and all of them. CDS Clients that do not understand the value MUST treat the card as an error.

Source code in healthchain/models/responses/cdsresponse.py 
20
21
22
23
24
25
26
27
28
29
30
class SelectionBehaviorEnum(str, Enum):
    """
    Describes the intended selection behavior of the suggestions in the card.
    Allowed values are: at-most-one, indicating that the user may choose none or
    at most one of the suggestions; any, indicating that the end user may choose
    any number of suggestions including none of them and all of them.
    CDS Clients that do not understand the value MUST treat the card as an error.
    """

    at_most_one = "at-most-one"
    any = "any"

SimpleCoding

Bases: BaseModel

The Coding data type captures the concept of a code. This coding type is a standalone data type in CDS Hooks modeled after a trimmed down version of the FHIR Coding data type.

Source code in healthchain/models/responses/cdsresponse.py 
82
83
84
85
86
87
88
89
90
class SimpleCoding(BaseModel):
    """
    The Coding data type captures the concept of a code. This coding type is a standalone data type
    in CDS Hooks modeled after a trimmed down version of the FHIR Coding data type.
    """

    code: str
    system: str
    display: Optional[str] = None

Source

Bases: BaseModel

Grouping structure for the Source of the information displayed on this card. The source should be the primary source of guidance for the decision support Card represents.

https://cds-hooks.org/specification/current/#source

Source code in healthchain/models/responses/cdsresponse.py 
137
138
139
140
141
142
143
144
145
146
147
148
class Source(BaseModel):
    """
    Grouping structure for the Source of the information displayed on this card.
    The source should be the primary source of guidance for the decision support Card represents.

    https://cds-hooks.org/specification/current/#source
    """

    label: str
    url: Optional[HttpUrl] = None
    icon: Optional[HttpUrl] = None
    topic: Optional[SimpleCoding] = None

Suggestion

Bases: BaseModel

Allows a service to suggest a set of changes in the context of the current activity (e.g. changing the dose of a medication currently being prescribed, for the order-sign activity). If suggestions are present, selectionBehavior MUST also be provided.

https://cds-hooks.org/specification/current/#suggestion

Source code in healthchain/models/responses/cdsresponse.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
class Suggestion(BaseModel):
    """
    Allows a service to suggest a set of changes in the context of the current activity
    (e.g. changing the dose of a medication currently being prescribed, for the order-sign activity).
    If suggestions are present, selectionBehavior MUST also be provided.

    https://cds-hooks.org/specification/current/#suggestion
    """

    label: str
    uuid: Optional[str] = None
    isRecommended: Optional[bool]
    actions: Optional[List[Action]] = []

ClinicalDocumentation

Bases: BaseUseCase

Implements EHR backend strategy for clinical documentation (NoteReader)

This class represents the backend strategy for clinical documentation using the NoteReader system. It inherits from the BaseUseCase class and provides methods for processing NoteReader documents.

ATTRIBUTE DESCRIPTION
service_api

The service API method to be used for processing the documents.

TYPE: Optional[APIMethod]

service_config

The configuration for the service.

TYPE: Optional[Dict]

service

The service to be used for processing the documents.

TYPE: Optional[Service]

client

The client to be used for communication with the service.

TYPE: Optional[BaseClient]

overwrite

Whether to overwrite existing data in the CDA document.

TYPE: bool

Source code in healthchain/use_cases/clindoc.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
class ClinicalDocumentation(BaseUseCase):
    """
    Implements EHR backend strategy for clinical documentation (NoteReader)

    This class represents the backend strategy for clinical documentation using the NoteReader system.
    It inherits from the `BaseUseCase` class and provides methods for processing NoteReader documents.

    Attributes:
        service_api (Optional[APIMethod]): The service API method to be used for processing the documents.
        service_config (Optional[Dict]): The configuration for the service.
        service (Optional[Service]): The service to be used for processing the documents.
        client (Optional[BaseClient]): The client to be used for communication with the service.
        overwrite (bool): Whether to overwrite existing data in the CDA document.

    """

    def __init__(
        self,
        service_api: Optional[APIMethod] = None,
        service_config: Optional[Dict] = None,
        service: Optional[Service] = None,
        client: Optional[BaseClient] = None,
    ) -> None:
        super().__init__(
            service_api=service_api,
            service_config=service_config,
            service=service,
            client=client,
        )
        self._type = UseCaseType.clindoc
        self._strategy = ClinicalDocumentationStrategy()
        self._endpoints = {
            "service_mount": Endpoint(
                path="/notereader/",
                method="POST",
                function=self.process_notereader_document,
                api_protocol="SOAP",
            )
        }
        self.overwrite: bool = False

    @property
    def description(self) -> str:
        return "Clinical documentation (NoteReader)"

    @property
    def type(self) -> UseCaseType:
        return self._type

    @property
    def strategy(self) -> BaseStrategy:
        return self._strategy

    @property
    def endpoints(self) -> Dict[str, Endpoint]:
        return self._endpoints

    def process_notereader_document(self, request: CdaRequest) -> CdaResponse:
        """
        Process the NoteReader document using the configured service API.

        This method handles the execution of the NoteReader service. It validates the
        service configuration, checks the input parameters, executes the service
        function, and ensures the correct response type is returned.

        Args:
            request (CdaRequest): The request object containing the CDA document to be processed.

        Returns:
            CdaResponse: The response object containing the processed CDA document.

        Raises:
            AssertionError: If the service function is not properly configured.
            TypeError: If the output type does not match the expected CdaResponse type.

        Note:
            This method performs several checks to ensure the integrity of the service:
            1. Verifies that the service API is configured.
            2. Validates the signature of the service function.
            3. Ensures the service function accepts a CdaRequest as its argument.
            4. Verifies that the service function returns a CdaResponse.
        """
        # Check service_api
        if self._service_api is None:
            log.warning("'service_api' not configured, check class init.")
            return CdaResponse(document="")

        # Check service function signature
        signature = inspect.signature(self._service_api.func)
        params = list(signature.parameters.values())
        if len(params) < 2:  # Only 'self' parameter
            raise AssertionError(
                "Service function must have at least one parameter besides 'self'"
            )
        first_param = params[1]  # Skip 'self'
        if first_param.annotation == inspect.Parameter.empty:
            log.warning(
                "Service function parameter has no type annotation. Expected CdaRequest."
            )
        elif first_param.annotation != CdaRequest:
            raise TypeError(
                f"Expected first argument of service function to be CdaRequest, but got {first_param.annotation}"
            )

        # Call the service function
        response = self._service_api.func(self, request)

        # Check return type
        if not isinstance(response, CdaResponse):
            raise TypeError(
                f"Expected return type CdaResponse, got {type(response)} instead."
            )

        return response

process_notereader_document(request)

Process the NoteReader document using the configured service API.

This method handles the execution of the NoteReader service. It validates the service configuration, checks the input parameters, executes the service function, and ensures the correct response type is returned.

PARAMETER DESCRIPTION
request

The request object containing the CDA document to be processed.

TYPE: CdaRequest

RETURNS DESCRIPTION
CdaResponse

The response object containing the processed CDA document.

TYPE: CdaResponse

RAISES DESCRIPTION
AssertionError

If the service function is not properly configured.
TypeError

If the output type does not match the expected CdaResponse type.
Note

This method performs several checks to ensure the integrity of the service: 1. Verifies that the service API is configured. 2. Validates the signature of the service function. 3. Ensures the service function accepts a CdaRequest as its argument. 4. Verifies that the service function returns a CdaResponse.

Source code in healthchain/use_cases/clindoc.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
def process_notereader_document(self, request: CdaRequest) -> CdaResponse:
    """
    Process the NoteReader document using the configured service API.

    This method handles the execution of the NoteReader service. It validates the
    service configuration, checks the input parameters, executes the service
    function, and ensures the correct response type is returned.

    Args:
        request (CdaRequest): The request object containing the CDA document to be processed.

    Returns:
        CdaResponse: The response object containing the processed CDA document.

    Raises:
        AssertionError: If the service function is not properly configured.
        TypeError: If the output type does not match the expected CdaResponse type.

    Note:
        This method performs several checks to ensure the integrity of the service:
        1. Verifies that the service API is configured.
        2. Validates the signature of the service function.
        3. Ensures the service function accepts a CdaRequest as its argument.
        4. Verifies that the service function returns a CdaResponse.
    """
    # Check service_api
    if self._service_api is None:
        log.warning("'service_api' not configured, check class init.")
        return CdaResponse(document="")

    # Check service function signature
    signature = inspect.signature(self._service_api.func)
    params = list(signature.parameters.values())
    if len(params) < 2:  # Only 'self' parameter
        raise AssertionError(
            "Service function must have at least one parameter besides 'self'"
        )
    first_param = params[1]  # Skip 'self'
    if first_param.annotation == inspect.Parameter.empty:
        log.warning(
            "Service function parameter has no type annotation. Expected CdaRequest."
        )
    elif first_param.annotation != CdaRequest:
        raise TypeError(
            f"Expected first argument of service function to be CdaRequest, but got {first_param.annotation}"
        )

    # Call the service function
    response = self._service_api.func(self, request)

    # Check return type
    if not isinstance(response, CdaResponse):
        raise TypeError(
            f"Expected return type CdaResponse, got {type(response)} instead."
        )

    return response

ClinicalDocumentationStrategy

Bases: BaseStrategy

Handles the request construction and validation of a NoteReader CDA file

Source code in healthchain/use_cases/clindoc.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
class ClinicalDocumentationStrategy(BaseStrategy):
    """
    Handles the request construction and validation of a NoteReader CDA file
    """

    def __init__(self) -> None:
        self.api_protocol: ApiProtocol = ApiProtocol.soap
        self.soap_envelope: Dict = self._load_soap_envelope()

    def _load_soap_envelope(self):
        data = pkgutil.get_data("healthchain", "templates/soap_envelope.xml")
        return xmltodict.parse(data.decode("utf-8"))

    def construct_cda_xml_document(self):
        """
        This function should wrap FHIR data from CcdFhirData into a template CDA file (dep. vendor
        TODO: implement this function
        """
        pass

    @validate_workflow(UseCaseMapping.ClinicalDocumentation)
    def construct_request(self, data: CcdData, workflow: Workflow) -> CdaRequest:
        """
        Constructs a CDA request for clinical documentation use cases (NoteReader)

        Parameters:
            data: CDA data to be injected in the request
            workflow (Workflow): The NoteReader workflow type, e.g. notereader-sign-inpatient

        Returns:
            CdaRequest: A Pydantic model that wraps CDA data for SOAP request

        Raises:
            ValueError: If the workflow is invalid or the data does not validate properly.
        """
        # TODO: handle converting fhir data from data generator to cda
        # TODO: handle different workflows
        if data.cda_xml is not None:
            # Encode the cda xml in base64
            encoded_xml = base64.b64encode(data.cda_xml.encode("utf-8")).decode("utf-8")

            # Make a copy of the SOAP envelope template
            soap_envelope = self.soap_envelope.copy()

            # Insert encoded cda in the Document section
            if not insert_at_key(soap_envelope, "urn:Document", encoded_xml):
                raise ValueError(
                    "Key 'urn:Document' missing from SOAP envelope template!"
                )
            request = CdaRequest.from_dict(soap_envelope)

            return request
        else:
            log.warning(
                "Data generation methods for CDA documents not implemented yet!"
            )

construct_cda_xml_document()

This function should wrap FHIR data from CcdFhirData into a template CDA file (dep. vendor TODO: implement this function

Source code in healthchain/use_cases/clindoc.py 
39
40
41
42
43
44
def construct_cda_xml_document(self):
    """
    This function should wrap FHIR data from CcdFhirData into a template CDA file (dep. vendor
    TODO: implement this function
    """
    pass

construct_request(data, workflow)

Constructs a CDA request for clinical documentation use cases (NoteReader)

PARAMETER DESCRIPTION
data

CDA data to be injected in the request

TYPE: CcdData

workflow

The NoteReader workflow type, e.g. notereader-sign-inpatient

TYPE: Workflow

RETURNS DESCRIPTION
CdaRequest

A Pydantic model that wraps CDA data for SOAP request

TYPE: CdaRequest

RAISES DESCRIPTION
ValueError

If the workflow is invalid or the data does not validate properly.
Source code in healthchain/use_cases/clindoc.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
@validate_workflow(UseCaseMapping.ClinicalDocumentation)
def construct_request(self, data: CcdData, workflow: Workflow) -> CdaRequest:
    """
    Constructs a CDA request for clinical documentation use cases (NoteReader)

    Parameters:
        data: CDA data to be injected in the request
        workflow (Workflow): The NoteReader workflow type, e.g. notereader-sign-inpatient

    Returns:
        CdaRequest: A Pydantic model that wraps CDA data for SOAP request

    Raises:
        ValueError: If the workflow is invalid or the data does not validate properly.
    """
    # TODO: handle converting fhir data from data generator to cda
    # TODO: handle different workflows
    if data.cda_xml is not None:
        # Encode the cda xml in base64
        encoded_xml = base64.b64encode(data.cda_xml.encode("utf-8")).decode("utf-8")

        # Make a copy of the SOAP envelope template
        soap_envelope = self.soap_envelope.copy()

        # Insert encoded cda in the Document section
        if not insert_at_key(soap_envelope, "urn:Document", encoded_xml):
            raise ValueError(
                "Key 'urn:Document' missing from SOAP envelope template!"
            )
        request = CdaRequest.from_dict(soap_envelope)

        return request
    else:
        log.warning(
            "Data generation methods for CDA documents not implemented yet!"
        )

CdaRequest

Bases: BaseModel

Source code in healthchain/models/requests/cdarequest.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
class CdaRequest(BaseModel):
    document: str

    @classmethod
    def from_dict(cls, data: Dict):
        """
        Loads data from dict (xmltodict format)
        """
        return cls(document=xmltodict.unparse(data))

    def model_dump(self, *args, **kwargs) -> Dict:
        """
        Dumps document as dict with xmltodict
        """
        return xmltodict.parse(self.document)

    def model_dump_xml(self, *args, **kwargs) -> str:
        """
        Decodes and dumps document as an xml string
        """
        xml_dict = xmltodict.parse(self.document)
        document = search_key(xml_dict, "urn:Document")
        if document is None:
            log.warning("Coudln't find document under namespace 'urn:Document")
            return ""
        cda = base64.b64decode(document).decode("UTF-8")

        return cda

from_dict(data) classmethod

Loads data from dict (xmltodict format)

Source code in healthchain/models/requests/cdarequest.py 
16
17
18
19
20
21
@classmethod
def from_dict(cls, data: Dict):
    """
    Loads data from dict (xmltodict format)
    """
    return cls(document=xmltodict.unparse(data))

model_dump(*args, **kwargs)

Dumps document as dict with xmltodict

Source code in healthchain/models/requests/cdarequest.py 
23
24
25
26
27
def model_dump(self, *args, **kwargs) -> Dict:
    """
    Dumps document as dict with xmltodict
    """
    return xmltodict.parse(self.document)

model_dump_xml(*args, **kwargs)

Decodes and dumps document as an xml string

Source code in healthchain/models/requests/cdarequest.py 
29
30
31
32
33
34
35
36
37
38
39
40
def model_dump_xml(self, *args, **kwargs) -> str:
    """
    Decodes and dumps document as an xml string
    """
    xml_dict = xmltodict.parse(self.document)
    document = search_key(xml_dict, "urn:Document")
    if document is None:
        log.warning("Coudln't find document under namespace 'urn:Document")
        return ""
    cda = base64.b64decode(document).decode("UTF-8")

    return cda

CdaResponse

Bases: BaseModel

Source code in healthchain/models/responses/cdaresponse.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
class CdaResponse(BaseModel):
    document: str
    error: Optional[str] = None

    @classmethod
    def from_dict(cls, data: Dict):
        """
        Loads data from dict (xmltodict format)
        """
        return cls(document=xmltodict.unparse(data))

    def model_dump(self, *args, **kwargs) -> Dict:
        """
        Dumps document as dict with xmltodict
        """
        return xmltodict.parse(self.document)

    def model_dump_xml(self, *args, **kwargs) -> str:
        """
        Decodes and dumps document as an xml string
        """
        xml_dict = xmltodict.parse(self.document)
        document = search_key(xml_dict, "tns:Document")
        if document is None:
            log.warning("Couldn't find document under namespace 'tns:Document")
            return ""

        cda = base64.b64decode(document).decode("UTF-8")

        return cda

from_dict(data) classmethod

Loads data from dict (xmltodict format)

Source code in healthchain/models/responses/cdaresponse.py 
18
19
20
21
22
23
@classmethod
def from_dict(cls, data: Dict):
    """
    Loads data from dict (xmltodict format)
    """
    return cls(document=xmltodict.unparse(data))

model_dump(*args, **kwargs)

Dumps document as dict with xmltodict

Source code in healthchain/models/responses/cdaresponse.py 
25
26
27
28
29
def model_dump(self, *args, **kwargs) -> Dict:
    """
    Dumps document as dict with xmltodict
    """
    return xmltodict.parse(self.document)

model_dump_xml(*args, **kwargs)

Decodes and dumps document as an xml string

Source code in healthchain/models/responses/cdaresponse.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
def model_dump_xml(self, *args, **kwargs) -> str:
    """
    Decodes and dumps document as an xml string
    """
    xml_dict = xmltodict.parse(self.document)
    document = search_key(xml_dict, "tns:Document")
    if document is None:
        log.warning("Couldn't find document under namespace 'tns:Document")
        return ""

    cda = base64.b64decode(document).decode("UTF-8")

    return cda