1
2
3
4
5
6
7
8
9
10
11
12
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
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
83
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
|
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.
import asyncio
import os
import datetime
from azure.core.credentials import AzureKeyCredential
from azure.healthinsights.clinicalmatching.aio import ClinicalMatchingClient
from azure.healthinsights.clinicalmatching import models
"""
FILE: sample_match_trials_fhir_async.py
DESCRIPTION:
Trial Eligibility Assessment for a Custom Trial.
Trial Matcher can be used to understand the gaps of eligibility criteria for a specific patient for a given clinical
trial. In this case, the trial is not taken from clinicaltrials.gov, however the trial is a custom trial that might
be not published clinicaltrials.gov yet. The custom trial eligibility criteria section is provided as an input to
the Trial Matcher.
In this use case, the patient clinical information is provided to the Trial Matcher as a FHIR bundle.
Note that the Trial Matcher configuration include reference to the FHIR Server where the patient FHIR bundle is
located.
USAGE:
python sample_match_trials_fhir_async.py
Set the environment variables with your own values before running the sample:
1) HEALTHINSIGHTS_KEY - your source from Health Insights API key.
2) HEALTHINSIGHTS_ENDPOINT - the endpoint to your source Health Insights resource.
"""
class HealthInsightsSamples:
async def match_trials_async(self) -> None:
KEY = os.environ["HEALTHINSIGHTS_KEY"]
ENDPOINT = os.environ["HEALTHINSIGHTS_ENDPOINT"]
# Create aTrial Matcher client
# <client>
trial_matcher_client = ClinicalMatchingClient(endpoint=ENDPOINT,
credential=AzureKeyCredential(KEY))
# </client>
# Construct Patient
# <PatientConstructor>
patient1 = self.get_patient_from_fhir_patient()
# </PatientConstructor>
# Create registry filter
registry_filters = models.ClinicalTrialRegistryFilter()
# Limit the trial to a specific patient condition ("Non-small cell lung cancer")
registry_filters.conditions = ["Non-small cell lung cancer"]
# Limit the clinical trial to a certain phase, phase 1
registry_filters.phases = [models.ClinicalTrialPhase.PHASE1]
# Specify the clinical trial registry source as ClinicalTrials.Gov
registry_filters.sources = [models.ClinicalTrialSource.CLINICALTRIALS_GOV]
# Limit the clinical trial to a certain location, in this case California, USA
registry_filters.facility_locations = [models.GeographicLocation(country_or_region="United States",
city="Gilbert",
state="Arizona")]
# Limit the trial to a specific study type, interventional
registry_filters.study_types = [models.ClinicalTrialStudyType.INTERVENTIONAL]
# Construct ClinicalTrial instance and attach the registry filter to it.
clinical_trials = models.ClinicalTrials(registry_filters=[registry_filters])
# Create TrialMatcherRequest
configuration = models.TrialMatcherModelConfiguration(clinical_trials=clinical_trials)
trial_matcher_data = models.TrialMatcherData(patients=[patient1], configuration=configuration)
# Health Insights Trial match trials
try:
poller = await trial_matcher_client.begin_match_trials(trial_matcher_data)
trial_matcher_result = await poller.result()
self.print_results(trial_matcher_result)
except Exception as ex:
print(str(ex))
return
# print match trials (eligible/ineligible)
@staticmethod
def print_results(trial_matcher_result):
if trial_matcher_result.status == models.JobStatus.SUCCEEDED:
tm_results = trial_matcher_result.results
for patient_result in tm_results.patients:
print(f"Inferences of Patient {patient_result.id}")
for tm_inferences in patient_result.inferences:
print(f"Trial Id {tm_inferences.id}")
print(f"Type: {str(tm_inferences.type)} Value: {tm_inferences.value}")
print(f"Description {tm_inferences.description}")
else:
tm_errors = trial_matcher_result.errors
if tm_errors is not None:
for error in tm_errors:
print(f"{error.code} : {error.message}")
def get_patient_from_fhir_patient(self) -> models.PatientRecord:
patient_info = models.PatientInfo(sex=models.PatientInfoSex.MALE, birth_date=datetime.date(1965, 12, 26))
patient_data = models.PatientDocument(type=models.DocumentType.FHIR_BUNDLE,
id="Consultation-14-Demo",
content=models.DocumentContent(
source_type=models.DocumentContentSourceType.INLINE,
value=self.get_patient_doc_content()),
clinical_type=models.ClinicalDocumentType.CONSULTATION)
return models.PatientRecord(id="patient_id", info=patient_info, data=[patient_data])
@staticmethod
def get_patient_doc_content() -> str:
samples_dir = os.path.dirname(os.path.realpath(__file__))
data_location = os.path.join(samples_dir, "../sample_data/match_trial_fhir_data.json")
with open(data_location, 'r', encoding='utf-8-sig') as f:
content = f.read()
return content
async def main():
sample = HealthInsightsSamples()
await sample.match_trials_async()
if __name__ == "__main__":
asyncio.run(main())
|
