For the best learning experience, run this tutorial interactively, via one of the environment setup options. Use the above button depending on your chosen setup option.
This tutorial explores how to work with genomic data in FHIR for clinical decision support (CDS) applications. We’ll use the Synthea Coherent Data Set to retrieve and analyze genomic reports and observations, ultimately implementing a CDS service that provides actionable recommendations based on a patient’s genetic profile.
In this tutorial, we’re using a FHIR server located at http://localhost:8080/fhir, but any FHIR server loaded with appropriate data can be used. For instructions on setting up your own test server, see Standing up a FHIR Testing Server.
This tutorial offers three difficulty levels to accommodate different experience levels:
| Level | Focus Areas | Recommended For |
|---|---|---|
| Beginner | FHIR server connection, Basic genomics report retrieval | Those new to FHIR Genomics |
| Intermediate | Variant analysis, Data transformation | Those familiar with genetic terminology and FHIR |
| Advanced | CDS implementation, Clinical filtering | Experienced developers implementing genomic CDS systems |
Prerequisites: Basic knowledge of Python, FHIR resources, and genomics terminology is recommended.
In this section, you’ll learn how to:
First, let’s set up our connection to the FHIR server:
# Load dependency
import requests, os
fhir_server = os.environ.get('FHIR_SERVER')
print(f"Using FHIR server: {fhir_server}")
# Check if the server is running and connection is successful
response = requests.get(f"{fhir_server}/metadata")
print(f"Server status: {response.status_code}")Using FHIR server: http://localhost:8080/fhir
Server status: 200The metadata endpoint (/metadata) is a special FHIR endpoint that returns the server’s capability statement - a structured document that describes what the server can do. When we query this endpoint:
This is a lightweight way to validate connectivity before attempting more complex queries.
If the server is responsive (code 200), proceed with the next code block.
# Import required libraries
import pandas as pd
import json
import re
import os
import requests
from fhir_pyrate import Pirate
from IPython.display import display, HTML
# Initialize FHIR-PYrate
search = Pirate(
auth=None,
base_url=fhir_server,
print_request_url=True,
)Now let’s retrieve basic genetic diagnostic reports. The LOINC code 55232-3 refers to Genetic analysis summary panel, which is commonly used to identify genomic reports. To learn more about FHIRPath and mapping columns, refer to the intermediate level FHIR analysis exercise.
# Fetch genomics reports
print("Retrieving basic genetic diagnostic reports...")
reports_df = search.steal_bundles_to_dataframe(
resource_type="DiagnosticReport",
request_params={"code": "http://loinc.org|55232-3", "_count": 10},
num_pages=1,
fhir_paths=[
("id", "id"),
("patient", "subject.reference"),
("status", "status"),
("code_display", "code.coding[0].display"),
("issued", "issued"),
],
)
# Display the first few genomics reports
print("First few genomics reports:")
display(reports_df.head())Retrieving basic genetic diagnostic reports...--------------------------------------------------------------------------- OptionalImportError Traceback (most recent call last) Cell In[3], line 3 1 # Fetch genomics reports 2 print("Retrieving basic genetic diagnostic reports...") ----> 3 reports_df = search.steal_bundles_to_dataframe( 4 resource_type="DiagnosticReport", 5 request_params={"code": "http://loinc.org|55232-3", "_count": 10}, 6 num_pages=1, 7 fhir_paths=[ 8 ("id", "id"), 9 ("patient", "subject.reference"), 10 ("status", "status"), 11 ("code_display", "code.coding[0].display"), 12 ("issued", "issued"), 13 ], 14 ) 16 # Display the first few genomics reports 17 print("First few genomics reports:") File ~/work/fhir-for-research/fhir-for-research/.venv/lib/python3.14/site-packages/fhir_pyrate/pirate.py:282, in Pirate.steal_bundles_to_dataframe(self, resource_type, request_params, num_pages, process_function, fhir_paths, build_df_after_query) 251 def steal_bundles_to_dataframe( 252 self, 253 resource_type: str, (...) 258 build_df_after_query: bool = False, 259 ) -> Union[pd.DataFrame, Dict[str, pd.DataFrame]]: 260 """ 261 Execute a request, iterates through the result pages, and builds a DataFrame with their 262 information. The DataFrames are either built after each (...) 280 returned. 281 """ --> 282 return self._query_to_dataframe(self._get_bundles)( 283 resource_type=resource_type, 284 request_params=request_params, 285 num_pages=num_pages, 286 silence_tqdm=False, 287 process_function=process_function, 288 fhir_paths=fhir_paths, 289 build_df_after_query=build_df_after_query, 290 disable_multiprocessing_build=True, 291 always_return_dict=False, 292 ) File ~/work/fhir-for-research/fhir-for-research/.venv/lib/python3.14/site-packages/fhir_pyrate/pirate.py:1441, in Pirate._query_to_dataframe.<locals>.wrap(process_function, fhir_paths, build_df_after_query, disable_multiprocessing_build, always_return_dict, *args, **kwargs) 1436 if fhir_paths is not None: 1437 logger.info( 1438 f"The selected process_function {process_function.__name__} will be " 1439 f"overwritten." 1440 ) -> 1441 process_function = self._set_up_fhirpath_function(fhir_paths) 1442 return self._bundles_to_dataframe( 1443 bundles=bundles_function( 1444 *args, **kwargs, tqdm_df_build=not build_df_after_query (...) 1449 always_return_dict=always_return_dict, 1450 ) File ~/work/fhir-for-research/fhir-for-research/.venv/lib/python3.14/site-packages/fhir_pyrate/pirate.py:1349, in Pirate._set_up_fhirpath_function(self, fhir_paths) 1331 if ( 1332 re.search(pattern=rf"{token}[\.\[]|[\.\]]{token}$", string=path) 1333 is not None 1334 ): 1335 warnings.warn( 1336 f"You are using the term {token} in of your FHIR path {path}. " 1337 f"Please keep in mind that this token can be used a function according " (...) 1346 stacklevel=2, 1347 ) 1348 compiled_paths = [ -> 1349 (name, fhirpathpy.compile(path=path)) for name, path in fhir_paths_with_name 1350 ] 1351 return partial(parse_fhir_path, compiled_fhir_paths=compiled_paths) File ~/work/fhir-for-research/fhir-for-research/.venv/lib/python3.14/site-packages/fhir_pyrate/util/imports.py:106, in optional_import.<locals>._LazyRaise.__getattr__(self, name) 101 def __getattr__(self, name: str) -> str: 102 """ 103 Raise: 104 OptionalImportError: When you call this method. 105 """ --> 106 raise self._exception File ~/work/fhir-for-research/fhir-for-research/.venv/lib/python3.14/site-packages/fhir_pyrate/util/imports.py:59, in optional_import(module, name, descriptor, allow_namespace_pkg) 57 actual_cmd = f"import {module}" 58 try: ---> 59 the_module = import_module(module) 60 if not allow_namespace_pkg: 61 is_namespace = getattr(the_module, "__file__", None) is None and hasattr( 62 the_module, "__path__" 63 ) File /opt/hostedtoolcache/Python/3.14.2/x64/lib/python3.14/importlib/__init__.py:88, in import_module(name, package) 86 break 87 level += 1 ---> 88 return _bootstrap._gcd_import(name[level:], package, level) File <frozen importlib._bootstrap>:1398, in _gcd_import(name, package, level) 1396 if level > 0: 1397 name = _resolve_name(name, package, level) -> 1398 return _find_and_load(name, _gcd_import) File <frozen importlib._bootstrap>:1371, in _find_and_load(name, import_) 1369 module = sys.modules.get(name, _NEEDS_LOADING) 1370 if module is _NEEDS_LOADING: -> 1371 return _find_and_load_unlocked(name, import_) 1373 # Optimization: only call _bootstrap._lock_unlock_module() if 1374 # module.__spec__._initializing is True. 1375 # NOTE: because of this, initializing must be set *before* 1376 # putting the new module in sys.modules. 1377 _lock_unlock_module(name) File <frozen importlib._bootstrap>:1342, in _find_and_load_unlocked(name, import_) 1340 parent_spec._uninitialized_submodules.append(child) 1341 try: -> 1342 module = _load_unlocked(spec) 1343 finally: 1344 if parent_spec: File <frozen importlib._bootstrap>:938, in _load_unlocked(spec) 935 raise ImportError('missing loader', name=spec.name) 936 # A namespace package so do nothing. 937 else: --> 938 spec.loader.exec_module(module) 939 except: 940 try: File <frozen importlib._bootstrap_external>:759, in _LoaderBasics.exec_module(self, module) 756 if code is None: 757 raise ImportError(f'cannot load module {module.__name__!r} when ' 758 'get_code() returns None') --> 759 _bootstrap._call_with_frames_removed(exec, code, module.__dict__) File <frozen importlib._bootstrap>:491, in _call_with_frames_removed(f, *args, **kwds) 483 def _call_with_frames_removed(f, *args, **kwds): 484 """remove_importlib_frames in import.c will always remove sequences 485 of importlib frames that end with a call to this function 486 (...) 489 module code) 490 """ --> 491 return f(*args, **kwds) File ~/work/fhir-for-research/fhir-for-research/.venv/lib/python3.14/site-packages/fhirpathpy/__init__.py:2 1 from fhirpathpy.engine.invocations.constants import constants ----> 2 from fhirpathpy.parser import parse 3 from fhirpathpy.engine import do_eval 4 from fhirpathpy.engine.util import arraify, get_data, set_paths File ~/work/fhir-for-research/fhir-for-research/.venv/lib/python3.14/site-packages/fhirpathpy/parser/__init__.py:2 1 import sys ----> 2 from antlr4 import * 3 from antlr4.tree.Tree import ParseTreeWalker 4 from antlr4.error.ErrorListener import ErrorListener File ~/work/fhir-for-research/fhir-for-research/.venv/lib/python3.14/site-packages/antlr4/__init__.py:6 4 from antlr4.StdinStream import StdinStream 5 from antlr4.BufferedTokenStream import TokenStream ----> 6 from antlr4.CommonTokenStream import CommonTokenStream 7 from antlr4.Lexer import Lexer 8 from antlr4.Parser import Parser File ~/work/fhir-for-research/fhir-for-research/.venv/lib/python3.14/site-packages/antlr4/CommonTokenStream.py:33 1 # 2 # Copyright (c) 2012-2017 The ANTLR Project. All rights reserved. 3 # Use of this file is governed by the BSD 3-clause license that (...) 29 # channel.</p> 30 #/ 32 from antlr4.BufferedTokenStream import BufferedTokenStream ---> 33 from antlr4.Lexer import Lexer 34 from antlr4.Token import Token 37 class CommonTokenStream(BufferedTokenStream): File ~/work/fhir-for-research/fhir-for-research/.venv/lib/python3.14/site-packages/antlr4/Lexer.py:12 1 # Copyright (c) 2012-2017 The ANTLR Project. All rights reserved. 2 # Use of this file is governed by the BSD 3-clause license that 3 # can be found in the LICENSE.txt file in the project root. (...) 9 # of speed. 10 #/ 11 from io import StringIO ---> 12 from typing.io import TextIO 13 import sys 14 from antlr4.CommonTokenFactory import CommonTokenFactory OptionalImportError: import fhirpathpy (No module named 'typing.io'; 'typing' is not a package).
This tutorial introduces key FHIR genomics concepts from the HL7 FHIR Genomics Reporting IG:
| Category | Description | FHIR Resource |
|---|---|---|
| Genomic Report | Container for all genomic results | DiagnosticReport |
| Genomic Findings | Variants, genotypes, and haplotypes | Observation profiles |
| Genomic Implications | Clinical impact of findings | Observation (linked via derivedFrom) |
| Recommended Actions | Follow-up or treatment suggestions | Task |
Let’s examine what information we got back. The reports DataFrame contains basic information about genetic diagnostic reports, including:
In this section, we’ll:
Let’s retrieve diagnostic reports with their related observations in a single query:
# Retrieve reports with related observations
print("Retrieving genetic reports with observations...")
results = search.steal_bundles_to_dataframe(
resource_type="DiagnosticReport",
request_params={
"code": "http://loinc.org|55232-3", # LOINC code for "Genetic analysis summary panel"
"_include": "DiagnosticReport:result", # Include the Observation resources
"_count": 10
},
num_pages=1
)
# Extract the different resource types from the results
reports_df = results.get("DiagnosticReport", pd.DataFrame())
observations_df = results.get("Observation", pd.DataFrame())
# Display basic report information
print("\nRetrieved genetic analysis reports:")
display(reports_df[["id", "subject_reference", "status", "issued"]].head())
# Display observation information
print("\nRetrieved genetic observations:")
print(f"Found {len(observations_df)} observations")
display(observations_df.head(3))Retrieving genetic reports with observations...
http://localhost:8080/fhir/DiagnosticReport?_count=10&_include=DiagnosticReport:result&code=http://loinc.org|55232-3Query & Build DF (DiagnosticReport): 0%| | 0/1 [00:00<?, ?it/s]Query & Build DF (DiagnosticReport): 100%|██████████| 1/1 [00:00<00:00, 140.74it/s]
Retrieved genetic analysis reports:| id | subject_reference | status | issued | |
|---|---|---|---|---|
| 0 | a31d8d94-6acd-a5f0-618a-f1b38746f9e4 | Patient/df860bc2-1943-237f-7445-ed960a1ef069 | final | 1992-02-10T08:04:51.390-05:00 |
| 1 | 1e5e18ae-bad4-3a45-b4e3-d2c7b7da66d9 | Patient/7d9ba758-f0b8-3fc3-befa-f0e8e8fb6935 | final | 2007-11-23T16:23:59.928-05:00 |
| 2 | 53db09ba-0414-61e1-78d2-02b20a7092b8 | Patient/837e80f6-a7a5-77f8-36aa-c7b8ff002c4b | final | 2001-02-25T10:39:54.395-05:00 |
| 3 | c012aa9a-fe79-40bc-00ae-162c9906c6f8 | Patient/39533e4a-f6f2-a144-ab37-6500460250dc | final | 2017-02-21T03:33:40.769-05:00 |
| 4 | 27e51912-a805-26dc-b9c8-d88f131969a6 | Patient/68c3ae0b-e298-62b7-5d3a-7936fd998fe0 | final | 1989-10-03T04:33:40.769-04:00 |
Retrieved genetic observations:
Found 159 observations| resourceType | id | meta_versionId | meta_lastUpdated | meta_source | status | category_0_coding_0_system | code_coding_0_system | code_coding_0_code | code_coding_0_display | code_text | subject_reference | encounter_reference | effectiveDateTime | issued | |
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| 0 | Observation | 6387bd0f-10f7-0691-c2a9-11618eafd3cb | 1 | 2026-02-20T20:44:23.251+00:00 | #g90y2GnxQt64tyro | final | http://terminology.hl7.org/CodeSystem/observat... | http://loinc.org | 69548-6 | The SELE gene exhibits a variation of 'Uncerta... | The SELE gene exhibits a variation of 'Uncerta... | Patient/837e80f6-a7a5-77f8-36aa-c7b8ff002c4b | Encounter/8c2ce4f9-3f70-c405-09d1-48f5992f22e9 | 2001-02-25T10:39:54-05:00 | 2001-02-25T10:39:54.395-05:00 |
| 1 | Observation | a31c8fc2-0017-0b05-2d2e-49121d2ffb1d | 1 | 2026-02-20T20:44:23.251+00:00 | #g90y2GnxQt64tyro | final | http://terminology.hl7.org/CodeSystem/observat... | http://loinc.org | 69548-6 | The SELE gene exhibits a variation of 'Uncerta... | The SELE gene exhibits a variation of 'Uncerta... | Patient/837e80f6-a7a5-77f8-36aa-c7b8ff002c4b | Encounter/8c2ce4f9-3f70-c405-09d1-48f5992f22e9 | 2001-02-25T10:39:54-05:00 | 2001-02-25T10:39:54.395-05:00 |
| 2 | Observation | a5f5e9e0-948f-d6df-8ddc-57818c851e6d | 1 | 2026-02-20T20:44:23.251+00:00 | #g90y2GnxQt64tyro | final | http://terminology.hl7.org/CodeSystem/observat... | http://loinc.org | 69548-6 | The PSRC1 gene exhibits a variation of 'Uncert... | The PSRC1 gene exhibits a variation of 'Uncert... | Patient/837e80f6-a7a5-77f8-36aa-c7b8ff002c4b | Encounter/8c2ce4f9-3f70-c405-09d1-48f5992f22e9 | 2001-02-25T10:39:54-05:00 | 2001-02-25T10:39:54.395-05:00 |
When working with the Synthea Coherent Data Set:
_include=DiagnosticReport:result to retrieve related observationsNow, let’s extract genetic variant information from the observations:
# Extract genetic variant information from observations
print("\nExtracting genetic variant information...")
# Create a list to store variant information
variants = []
# Process each observation to extract key information
for _, row in observations_df.iterrows():
# Check if text content is available
if 'code_text' in row and not pd.isna(row['code_text']):
display_text = str(row['code_text'])
# Extract genetic information using regular expressions
gene_match = re.search(r'The (\S+) gene', display_text)
significance_match = re.search(r"variation of '([^']+)'", display_text)
variant_match = re.search(r'index (\S+) is', display_text)
risks_match = re.search(r'risk of: (.+)\.', display_text)
# Create a record with the extracted information
variant_info = {
'patient': row['subject_reference'],
'gene': gene_match.group(1) if gene_match else 'Unknown',
'clinical_significance': significance_match.group(1) if significance_match else 'Unknown',
'variant_id': variant_match.group(1) if variant_match else 'Unknown',
'associated_risks': risks_match.group(1) if risks_match else 'Unknown'
}
variants.append(variant_info)
# Convert to DataFrame for easier analysis
variants_df = pd.DataFrame(variants)
# Display the extracted variants
print("\nExtracted genetic variants:")
display(variants_df.head(5))
Extracting genetic variant information...
Extracted genetic variants:| patient | gene | clinical_significance | variant_id | associated_risks | |
|---|---|---|---|---|---|
| 0 | Patient/837e80f6-a7a5-77f8-36aa-c7b8ff002c4b | SELE | Uncertain | rs5361_FS1 | Ischemic Stroke, Arterial Occlusive Disease an... |
| 1 | Patient/837e80f6-a7a5-77f8-36aa-c7b8ff002c4b | SELE | Uncertain | rs5361_FS2 | Ischemic Stroke, Arterial Occlusive Disease an... |
| 2 | Patient/837e80f6-a7a5-77f8-36aa-c7b8ff002c4b | PSRC1 | Uncertain | rs599839_FS1 | Cardiovascular Issues, Dsyslipidemia, Heart Di... |
| 3 | Patient/837e80f6-a7a5-77f8-36aa-c7b8ff002c4b | AGT | Risk Factor | rs699_FS1 | Ischemic Stroke, Pulmonary Hypertension and Ca... |
| 4 | Patient/837e80f6-a7a5-77f8-36aa-c7b8ff002c4b | AGT | Risk Factor | rs699_FS2 | Ischemic Stroke, Pulmonary Hypertension and Ca... |
In this section, we’ll:
First, let’s filter for variants with significant clinical impact:
# Filter for variants with significant clinical impact
print("Filtering for clinically significant variants...")
pathogenic_variants = variants_df[
variants_df['clinical_significance'].str.contains('Pathogenic|Risk Factor', na=False)
]
print(f"Found {len(pathogenic_variants)} variants with clinical significance")
display(pathogenic_variants.head())Filtering for clinically significant variants...
Found 25 variants with clinical significance| patient | gene | clinical_significance | variant_id | associated_risks | |
|---|---|---|---|---|---|
| 3 | Patient/837e80f6-a7a5-77f8-36aa-c7b8ff002c4b | AGT | Risk Factor | rs699_FS1 | Ischemic Stroke, Pulmonary Hypertension and Ca... |
| 4 | Patient/837e80f6-a7a5-77f8-36aa-c7b8ff002c4b | AGT | Risk Factor | rs699_FS2 | Ischemic Stroke, Pulmonary Hypertension and Ca... |
| 12 | Patient/837e80f6-a7a5-77f8-36aa-c7b8ff002c4b | ADD1 | Risk Factor | rs4961_FS1 | Cardiovascular Issues, High LDL, Hemorrhagic S... |
| 13 | Patient/837e80f6-a7a5-77f8-36aa-c7b8ff002c4b | ADD1 | Risk Factor | rs4961_FS2 | Cardiovascular Issues, High LDL, Hemorrhagic S... |
| 18 | Patient/837e80f6-a7a5-77f8-36aa-c7b8ff002c4b | PON1 | Risk Factor | rs662_FS1 | Cardiovascular Issues, Mitral valve issues, Ds... |
Now that we’ve analyzed genetic variants and generated recommendations, let’s explore how to deploy these as a clinical decision support service using CDS Hooks.
CDS Hooks is a standard for integrating clinical decision support into electronic health record (EHR) systems at the point of care. It allows our genomic analysis recommendations to be delivered to clinicians at the right time and in the right context.
For a comprehensive introduction to CDS Hooks, please refer to CDS Hooks Introduction, which covers the fundamental concepts, workflow, and implementation details.
CDS Hooks provides a framework for:
Common Hook Points:
The CDS Hooks Workflow
When integrated into a real clinical environment, the CDS Hooks workflow would look like this:
patient-view hookNow, let’s create a CDS Hooks service to serve our genomic recommendations. This service will:
The service exposes two endpoints:
We will use Flask, a lightweight application framework for writing web applications and services using Python.
# Import necessary libraries
from flask import Flask, jsonify, request
import threading
import time
import re
import pandas as pd
# Choose a port for the Flask server
flask_port = 3299
flask_server_thread = None
# Create a Flask application
app = Flask(__name__)
# CDS Hooks discovery endpoint - this tells EHRs what services we provide
@app.route('/cds-services', methods=['GET'])
def discovery():
return jsonify({
'services': [
{
'id': 'genomics-advisor',
'hook': 'patient-view',
'title': 'Genomics Clinical Advisor',
'description': 'Provides recommendations based on genetic findings',
# Note: In a production environment, prefetch could be used for optimization
# but is omitted here for simplicity
}
]
})
# CDS Hooks service endpoint - this returns recommendations for a specific patient
@app.route('/cds-services/genomics-advisor', methods=['POST'])
def genomics_advisor():
# Get data from the request
request_data = request.json
context = request_data.get('context', {})
patient_id = context.get('patientId')
print(f"Processing request for patient: {patient_id}")
try:
# Query for this specific patient's genetic reports
diagnostic_reports = search.steal_bundles_to_dataframe(
resource_type="DiagnosticReport",
request_params={
"code": "http://loinc.org|55232-3", # Genetic analysis summary panel
"subject": patient_id,
"_count": 100
},
num_pages=1
)
# Check if we got results
if isinstance(diagnostic_reports, dict) and "DiagnosticReport" in diagnostic_reports:
reports_df = diagnostic_reports["DiagnosticReport"]
else:
# The data might be directly in the DataFrame
reports_df = diagnostic_reports
if reports_df.empty:
print("No genetic reports found for this patient")
pathogenic_variants = pd.DataFrame()
else:
print(f"Found {len(reports_df)} genetic reports for patient {patient_id}")
# Extract genetic information from flattened result columns
variants = []
import re
# Identify all result display columns
display_columns = [col for col in reports_df.columns if 'result' in col and 'display' in col]
print(f"Found {len(display_columns)} result display columns")
# Process each result display column
for col in display_columns:
for _, row in reports_df.iterrows():
if pd.notna(row[col]):
display_text = str(row[col])
# Extract information using regex
gene_match = re.search(r'The (\S+) gene', display_text)
significance_match = re.search(r"variation of '([^']+)'", display_text)
variant_match = re.search(r'index (\S+) is', display_text)
risks_match = re.search(r'risk of: (.+)\.', display_text)
if gene_match or significance_match:
variant_info = {
'gene': gene_match.group(1) if gene_match else 'Unknown',
'clinical_significance': significance_match.group(1) if significance_match else 'Unknown',
'variant_id': variant_match.group(1) if variant_match else 'Unknown',
'associated_risks': risks_match.group(1) if risks_match else 'Unknown'
}
variants.append(variant_info)
# Convert to DataFrame and filter for pathogenic variants
variants_df = pd.DataFrame(variants)
print(f"Extracted {len(variants_df)} variants")
if not variants_df.empty:
pathogenic_variants = variants_df[
variants_df['clinical_significance'].str.contains('Pathogenic|Risk Factor', case=False)
]
print(f"Found {len(pathogenic_variants)} pathogenic variants")
else:
print("No variants could be extracted")
pathogenic_variants = pd.DataFrame()
except Exception as e:
import traceback
print(f"Error querying FHIR server: {e}")
print("Traceback:")
traceback.print_exc()
pathogenic_variants = pd.DataFrame()
# Generate recommendations based on pathogenic variants
recommendations = []
if not pathogenic_variants.empty:
# Check for specific gene variants
genes = pathogenic_variants['gene'].unique()
print(f"Found genes for recommendations: {genes}")
# Known genes with clinical recommendations
gene_recommendations = {
'APOE': {
'title': 'APOE Pathogenic Variant Detected',
'detail': 'Consider lipid panel and cardiovascular risk assessment',
'source': 'Clinical Practice Guidelines',
'urgency': 'high'
},
'BRCA1': {
'title': 'BRCA1 Pathogenic Variant Detected',
'detail': 'Consider cancer risk assessment and screening',
'source': 'NCCN Guidelines',
'urgency': 'high'
},
'BRCA2': {
'title': 'BRCA2 Pathogenic Variant Detected',
'detail': 'Consider cancer risk assessment and screening',
'source': 'NCCN Guidelines',
'urgency': 'high'
},
'PON1': {
'title': 'PON1 Risk Variant Detected',
'detail': 'Consider cardiovascular risk assessment and lipid-lowering therapy',
'source': 'Cardiovascular Guidelines',
'urgency': 'medium'
},
'ADRB3': {
'title': 'ADRB3 Risk Variant Detected',
'detail': 'Associated with metabolic disorders and cardiovascular risk',
'source': 'Metabolic Risk Guidelines',
'urgency': 'medium'
},
'CCL2': {
'title': 'CCL2 Pathogenic Variant Detected',
'detail': 'Associated with inflammatory processes and stroke risk',
'source': 'Stroke Risk Guidelines',
'urgency': 'high'
},
'FTO': {
'title': 'FTO Risk Variant Detected',
'detail': 'Associated with obesity and insulin resistance',
'source': 'Metabolic Guidelines',
'urgency': 'medium'
}
}
# Add recommendations for each significant gene
for gene in genes:
if gene in gene_recommendations:
recommendations.append(gene_recommendations[gene])
print(f"Added recommendation for gene: {gene}")
# Check for risk categories in all variants
all_risks = ' '.join(pathogenic_variants['associated_risks'].dropna())
# Add general variant recommendations if no specific ones
if not recommendations and len(pathogenic_variants) > 0:
unique_genes = ', '.join(genes)
recommendations.append({
'title': 'Genetic Variants Detected',
'detail': f'Clinically significant variants found in genes: {unique_genes}',
'source': 'Genetic Analysis',
'urgency': 'medium'
})
print(f"Added general recommendation for genes: {unique_genes}")
else:
print("No pathogenic variants found, no recommendations generated")
# Convert recommendations to CDS cards
cards = []
for rec in recommendations:
cards.append({
'summary': f"{rec['title']}",
'indicator': 'warning' if rec['urgency'] == 'high' else 'info',
'detail': rec['detail'],
'source': {
'label': rec['source']
}
})
# If no recommendations were found, provide a default card
if not cards:
cards = [{
'summary': f'No Significant Genetic Findings - {test_patient_id}',
'indicator': 'info',
'detail': 'No pathogenic or likely pathogenic variants detected in this patient\'s genetic analysis.',
'source': {
'label': 'Genomics Service'
}
}]
# Return the CDS Hooks response
return jsonify({'cards': cards})
# Start Flask in a background thread
def run_flask_in_thread():
from werkzeug.serving import run_simple
run_simple('localhost', flask_port, app, use_reloader=False, use_debugger=False)
# Start the Flask server
flask_server_thread = threading.Thread(target=run_flask_in_thread)
flask_server_thread.daemon = True # Thread will exit when the notebook exits
flask_server_thread.start()
# Give the server a moment to start
time.sleep(1)
print(f"CDS Hooks service running at http://localhost:{flask_port}/cds-services")INFO:werkzeug:WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead.
* Running on http://localhost:3299
INFO:werkzeug:Press CTRL+C to quitCDS Hooks service running at http://localhost:3299/cds-servicesNow that our CDS Hooks service is running, let’s test it by simulating how an EHR would interact with it. When a clinician opens a patient’s record in an EHR system, the EHR sends a request to all registered CDS services, allowing them to provide relevant recommendations.
# Let's directly query for genetic reports and work with a patient that definitely has genetic data
print("Querying for genetic diagnostic reports...")
genetic_reports_dict = search.steal_bundles_to_dataframe(
resource_type="DiagnosticReport",
request_params={
"code": "http://loinc.org|55232-3", # Genetic analysis summary panel
"_count": 5,
"_include": "DiagnosticReport:result", # Include the Observation resources
},
num_pages=1,
fhir_paths=[
("id", "id"),
("subject", "subject.reference"),
("status", "status"),
("issued", "issued"),
]
)
# Display diagnostic reports
genetic_reports_df = genetic_reports_dict["DiagnosticReport"]
print(f"Found {len(genetic_reports_df)} genetic reports")
display(genetic_reports_df.head())
# Display observations count if available
if "Observation" in genetic_reports_dict:
observations_df = genetic_reports_dict["Observation"]
print(f"Found {len(observations_df)} genetic observations")
# Select a patient of interest, as if a clinician selected a patient from the EHR
test_patient_id = genetic_reports_df.loc[genetic_reports_df["subject"] == "Patient/df860bc2-1943-237f-7445-ed960a1ef069", "subject"].values[0]
print(f"Selected patient with genetic data: {test_patient_id}")
print(f"Testing with patient ID: {test_patient_id}")Querying for genetic diagnostic reports...--------------------------------------------------------------------------- OptionalImportError Traceback (most recent call last) Cell In[9], line 3 1 # Let's directly query for genetic reports and work with a patient that definitely has genetic data 2 print("Querying for genetic diagnostic reports...") ----> 3 genetic_reports_dict = search.steal_bundles_to_dataframe( 4 resource_type="DiagnosticReport", 5 request_params={ 6 "code": "http://loinc.org|55232-3", # Genetic analysis summary panel 7 "_count": 5, 8 "_include": "DiagnosticReport:result", # Include the Observation resources 9 }, 10 num_pages=1, 11 fhir_paths=[ 12 ("id", "id"), 13 ("subject", "subject.reference"), 14 ("status", "status"), 15 ("issued", "issued"), 16 ] 17 ) 19 # Display diagnostic reports 20 genetic_reports_df = genetic_reports_dict["DiagnosticReport"] File ~/work/fhir-for-research/fhir-for-research/.venv/lib/python3.14/site-packages/fhir_pyrate/pirate.py:282, in Pirate.steal_bundles_to_dataframe(self, resource_type, request_params, num_pages, process_function, fhir_paths, build_df_after_query) 251 def steal_bundles_to_dataframe( 252 self, 253 resource_type: str, (...) 258 build_df_after_query: bool = False, 259 ) -> Union[pd.DataFrame, Dict[str, pd.DataFrame]]: 260 """ 261 Execute a request, iterates through the result pages, and builds a DataFrame with their 262 information. The DataFrames are either built after each (...) 280 returned. 281 """ --> 282 return self._query_to_dataframe(self._get_bundles)( 283 resource_type=resource_type, 284 request_params=request_params, 285 num_pages=num_pages, 286 silence_tqdm=False, 287 process_function=process_function, 288 fhir_paths=fhir_paths, 289 build_df_after_query=build_df_after_query, 290 disable_multiprocessing_build=True, 291 always_return_dict=False, 292 ) File ~/work/fhir-for-research/fhir-for-research/.venv/lib/python3.14/site-packages/fhir_pyrate/pirate.py:1441, in Pirate._query_to_dataframe.<locals>.wrap(process_function, fhir_paths, build_df_after_query, disable_multiprocessing_build, always_return_dict, *args, **kwargs) 1436 if fhir_paths is not None: 1437 logger.info( 1438 f"The selected process_function {process_function.__name__} will be " 1439 f"overwritten." 1440 ) -> 1441 process_function = self._set_up_fhirpath_function(fhir_paths) 1442 return self._bundles_to_dataframe( 1443 bundles=bundles_function( 1444 *args, **kwargs, tqdm_df_build=not build_df_after_query (...) 1449 always_return_dict=always_return_dict, 1450 ) File ~/work/fhir-for-research/fhir-for-research/.venv/lib/python3.14/site-packages/fhir_pyrate/pirate.py:1349, in Pirate._set_up_fhirpath_function(self, fhir_paths) 1331 if ( 1332 re.search(pattern=rf"{token}[\.\[]|[\.\]]{token}$", string=path) 1333 is not None 1334 ): 1335 warnings.warn( 1336 f"You are using the term {token} in of your FHIR path {path}. " 1337 f"Please keep in mind that this token can be used a function according " (...) 1346 stacklevel=2, 1347 ) 1348 compiled_paths = [ -> 1349 (name, fhirpathpy.compile(path=path)) for name, path in fhir_paths_with_name 1350 ] 1351 return partial(parse_fhir_path, compiled_fhir_paths=compiled_paths) File ~/work/fhir-for-research/fhir-for-research/.venv/lib/python3.14/site-packages/fhir_pyrate/util/imports.py:106, in optional_import.<locals>._LazyRaise.__getattr__(self, name) 101 def __getattr__(self, name: str) -> str: 102 """ 103 Raise: 104 OptionalImportError: When you call this method. 105 """ --> 106 raise self._exception [... skipping hidden 1 frame] Cell In[3], line 3 1 # Fetch genomics reports 2 print("Retrieving basic genetic diagnostic reports...") ----> 3 reports_df = search.steal_bundles_to_dataframe( 4 resource_type="DiagnosticReport", 5 request_params={"code": "http://loinc.org|55232-3", "_count": 10}, 6 num_pages=1, 7 fhir_paths=[ 8 ("id", "id"), 9 ("patient", "subject.reference"), 10 ("status", "status"), 11 ("code_display", "code.coding[0].display"), 12 ("issued", "issued"), 13 ], 14 ) 16 # Display the first few genomics reports 17 print("First few genomics reports:") File ~/work/fhir-for-research/fhir-for-research/.venv/lib/python3.14/site-packages/fhir_pyrate/pirate.py:282, in Pirate.steal_bundles_to_dataframe(self, resource_type, request_params, num_pages, process_function, fhir_paths, build_df_after_query) 251 def steal_bundles_to_dataframe( 252 self, 253 resource_type: str, (...) 258 build_df_after_query: bool = False, 259 ) -> Union[pd.DataFrame, Dict[str, pd.DataFrame]]: 260 """ 261 Execute a request, iterates through the result pages, and builds a DataFrame with their 262 information. The DataFrames are either built after each (...) 280 returned. 281 """ --> 282 return self._query_to_dataframe(self._get_bundles)( 283 resource_type=resource_type, 284 request_params=request_params, 285 num_pages=num_pages, 286 silence_tqdm=False, 287 process_function=process_function, 288 fhir_paths=fhir_paths, 289 build_df_after_query=build_df_after_query, 290 disable_multiprocessing_build=True, 291 always_return_dict=False, 292 ) File ~/work/fhir-for-research/fhir-for-research/.venv/lib/python3.14/site-packages/fhir_pyrate/pirate.py:1441, in Pirate._query_to_dataframe.<locals>.wrap(process_function, fhir_paths, build_df_after_query, disable_multiprocessing_build, always_return_dict, *args, **kwargs) 1436 if fhir_paths is not None: 1437 logger.info( 1438 f"The selected process_function {process_function.__name__} will be " 1439 f"overwritten." 1440 ) -> 1441 process_function = self._set_up_fhirpath_function(fhir_paths) 1442 return self._bundles_to_dataframe( 1443 bundles=bundles_function( 1444 *args, **kwargs, tqdm_df_build=not build_df_after_query (...) 1449 always_return_dict=always_return_dict, 1450 ) File ~/work/fhir-for-research/fhir-for-research/.venv/lib/python3.14/site-packages/fhir_pyrate/pirate.py:1349, in Pirate._set_up_fhirpath_function(self, fhir_paths) 1331 if ( 1332 re.search(pattern=rf"{token}[\.\[]|[\.\]]{token}$", string=path) 1333 is not None 1334 ): 1335 warnings.warn( 1336 f"You are using the term {token} in of your FHIR path {path}. " 1337 f"Please keep in mind that this token can be used a function according " (...) 1346 stacklevel=2, 1347 ) 1348 compiled_paths = [ -> 1349 (name, fhirpathpy.compile(path=path)) for name, path in fhir_paths_with_name 1350 ] 1351 return partial(parse_fhir_path, compiled_fhir_paths=compiled_paths) File ~/work/fhir-for-research/fhir-for-research/.venv/lib/python3.14/site-packages/fhir_pyrate/util/imports.py:106, in optional_import.<locals>._LazyRaise.__getattr__(self, name) 101 def __getattr__(self, name: str) -> str: 102 """ 103 Raise: 104 OptionalImportError: When you call this method. 105 """ --> 106 raise self._exception File ~/work/fhir-for-research/fhir-for-research/.venv/lib/python3.14/site-packages/fhir_pyrate/util/imports.py:59, in optional_import(module, name, descriptor, allow_namespace_pkg) 57 actual_cmd = f"import {module}" 58 try: ---> 59 the_module = import_module(module) 60 if not allow_namespace_pkg: 61 is_namespace = getattr(the_module, "__file__", None) is None and hasattr( 62 the_module, "__path__" 63 ) File /opt/hostedtoolcache/Python/3.14.2/x64/lib/python3.14/importlib/__init__.py:88, in import_module(name, package) 86 break 87 level += 1 ---> 88 return _bootstrap._gcd_import(name[level:], package, level) File <frozen importlib._bootstrap>:1398, in _gcd_import(name, package, level) 1396 if level > 0: 1397 name = _resolve_name(name, package, level) -> 1398 return _find_and_load(name, _gcd_import) File <frozen importlib._bootstrap>:1371, in _find_and_load(name, import_) 1369 module = sys.modules.get(name, _NEEDS_LOADING) 1370 if module is _NEEDS_LOADING: -> 1371 return _find_and_load_unlocked(name, import_) 1373 # Optimization: only call _bootstrap._lock_unlock_module() if 1374 # module.__spec__._initializing is True. 1375 # NOTE: because of this, initializing must be set *before* 1376 # putting the new module in sys.modules. 1377 _lock_unlock_module(name) File <frozen importlib._bootstrap>:1342, in _find_and_load_unlocked(name, import_) 1340 parent_spec._uninitialized_submodules.append(child) 1341 try: -> 1342 module = _load_unlocked(spec) 1343 finally: 1344 if parent_spec: File <frozen importlib._bootstrap>:938, in _load_unlocked(spec) 935 raise ImportError('missing loader', name=spec.name) 936 # A namespace package so do nothing. 937 else: --> 938 spec.loader.exec_module(module) 939 except: 940 try: File <frozen importlib._bootstrap_external>:759, in _LoaderBasics.exec_module(self, module) 756 if code is None: 757 raise ImportError(f'cannot load module {module.__name__!r} when ' 758 'get_code() returns None') --> 759 _bootstrap._call_with_frames_removed(exec, code, module.__dict__) File <frozen importlib._bootstrap>:491, in _call_with_frames_removed(f, *args, **kwds) 483 def _call_with_frames_removed(f, *args, **kwds): 484 """remove_importlib_frames in import.c will always remove sequences 485 of importlib frames that end with a call to this function 486 (...) 489 module code) 490 """ --> 491 return f(*args, **kwds) File ~/work/fhir-for-research/fhir-for-research/.venv/lib/python3.14/site-packages/fhirpathpy/__init__.py:2 1 from fhirpathpy.engine.invocations.constants import constants ----> 2 from fhirpathpy.parser import parse 3 from fhirpathpy.engine import do_eval 4 from fhirpathpy.engine.util import arraify, get_data, set_paths File ~/work/fhir-for-research/fhir-for-research/.venv/lib/python3.14/site-packages/fhirpathpy/parser/__init__.py:2 1 import sys ----> 2 from antlr4 import * 3 from antlr4.tree.Tree import ParseTreeWalker 4 from antlr4.error.ErrorListener import ErrorListener File ~/work/fhir-for-research/fhir-for-research/.venv/lib/python3.14/site-packages/antlr4/__init__.py:6 4 from antlr4.StdinStream import StdinStream 5 from antlr4.BufferedTokenStream import TokenStream ----> 6 from antlr4.CommonTokenStream import CommonTokenStream 7 from antlr4.Lexer import Lexer 8 from antlr4.Parser import Parser File ~/work/fhir-for-research/fhir-for-research/.venv/lib/python3.14/site-packages/antlr4/CommonTokenStream.py:33 1 # 2 # Copyright (c) 2012-2017 The ANTLR Project. All rights reserved. 3 # Use of this file is governed by the BSD 3-clause license that (...) 29 # channel.</p> 30 #/ 32 from antlr4.BufferedTokenStream import BufferedTokenStream ---> 33 from antlr4.Lexer import Lexer 34 from antlr4.Token import Token 37 class CommonTokenStream(BufferedTokenStream): File ~/work/fhir-for-research/fhir-for-research/.venv/lib/python3.14/site-packages/antlr4/Lexer.py:12 1 # Copyright (c) 2012-2017 The ANTLR Project. All rights reserved. 2 # Use of this file is governed by the BSD 3-clause license that 3 # can be found in the LICENSE.txt file in the project root. (...) 9 # of speed. 10 #/ 11 from io import StringIO ---> 12 from typing.io import TextIO 13 import sys 14 from antlr4.CommonTokenFactory import CommonTokenFactory OptionalImportError: import fhirpathpy (No module named 'typing.io'; 'typing' is not a package).
In production, EHR systems would send requests directly to your CDS Hooks service. Since we’re developing locally, we need to manually mock these requests to test our service without connecting to a real EHR system.
# Create a request with the selected patient
mock_request = {
"hook": "patient-view",
"hookInstance": "d1577c69-dfbe-44ad-ba6d-3e05e953b2ea",
"context": {
"patientId": test_patient_id
}
}
# Send the request to our CDS service
print("Sending request to CDS service...")
service_response = requests.post(
f"http://localhost:{flask_port}/cds-services/genomics-advisor",
json=mock_request
)
# Display the JSON response from our service
print("\nCDS Service Response:")
print(json.dumps(service_response.json(), indent=2))--------------------------------------------------------------------------- NameError Traceback (most recent call last) Cell In[10], line 6 1 # Create a request with the selected patient 2 mock_request = { 3 "hook": "patient-view", 4 "hookInstance": "d1577c69-dfbe-44ad-ba6d-3e05e953b2ea", 5 "context": { ----> 6 "patientId": test_patient_id 7 } 8 } 10 # Send the request to our CDS service 11 print("Sending request to CDS service...") NameError: name 'test_patient_id' is not defined
The response contains CDS cards with our recommendations. These cards would typically be displayed directly in the EHR interface to help clinicians make informed decisions. Let’s create a visual representation of how these cards might appear in an EHR system:
# Extract the CDS cards from the response
cards = service_response.json().get('cards', [])
# Create a simple HTML visualization to show how these cards would appear in an EHR
html_output = f"""
<div style="font-family: Arial, sans-serif; border: 1px solid #ccc; border-radius: 5px; padding: 15px; max-width: 700px; margin: 0 auto;">
<div style="border-bottom: 1px solid #ddd; padding-bottom: 10px; margin-bottom: 15px;">
<h3 style="margin-top: 0;">EHR Patient View: {test_patient_id}</h3>
<div style="color: #666; font-size: 0.9em;">Genomic Decision Support</div>
</div>
<div style="background-color: #f8f9fa; border: 1px solid #e9ecef; padding: 10px; margin-bottom: 15px; border-radius: 3px;">
<p style="margin: 0; font-size: 0.9em; color: #6c757d;">
<strong>Demo Note:</strong> This is a simulation of how CDS Hooks would appear in an EHR.
The buttons shown below are not functional in this demo. In a real implementation,
clinicians would be able to click these buttons to take recommended actions.
</p>
</div>
"""
# Add each card to the visualization
if not cards:
html_output += "<p>No clinical decision support recommendations available for this patient.</p>"
else:
for card in cards:
# Set card color based on importance
border_color = "#007bff" # Default blue
if card.get('indicator') == "warning":
border_color = "#ff9800" # Orange for warnings
elif card.get('indicator') == "critical":
border_color = "#dc3545" # Red for critical alerts
# Create the card HTML
html_output += f"""
<div style="border-left: 4px solid {border_color}; padding: 10px 15px; margin-bottom: 10px; background-color: #f8f9fa;">
<h4 style="margin-top: 0;">{card.get('summary', '')}</h4>
<p>{card.get('detail', '')}</p>
"""
# Add suggestion buttons
if 'suggestions' in card:
html_output += '<div>'
for suggestion in card.get('suggestions', []):
html_output += f'<span style="display: inline-block; background: #e9ecef; border: 1px solid #ced4da; padding: 5px 10px; border-radius: 3px; margin-top: 8px; margin-right: 5px; cursor: pointer;">{suggestion.get("label", "")}</span>'
html_output += '</div>'
# Add source information
if 'source' in card:
html_output += f'<div style="color: #6c757d; font-size: 0.85em; margin-top: 5px;">Source: {card.get("source", {}).get("label", "")}</div>'
html_output += '</div>'
html_output += "</div>"
# Display the HTML
print("\nVisualization of CDS Cards in EHR Interface:")
display(HTML(html_output))--------------------------------------------------------------------------- NameError Traceback (most recent call last) Cell In[11], line 2 1 # Extract the CDS cards from the response ----> 2 cards = service_response.json().get('cards', []) 4 # Create a simple HTML visualization to show how these cards would appear in an EHR 5 html_output = f""" 6 <div style="font-family: Arial, sans-serif; border: 1px solid #ccc; border-radius: 5px; padding: 15px; max-width: 700px; margin: 0 auto;"> 7 <div style="border-bottom: 1px solid #ddd; padding-bottom: 10px; margin-bottom: 15px;"> (...) 18 </div> 19 """ NameError: name 'service_response' is not defined
Congratulations on completing this tutorial on genomic clinical decision support! We’ve explored how to build a CDS service that can identify clinically relevant genomic variants and deliver actionable recommendations to clinicians at the point of care.
In this tutorial, we’ve covered:
Moving a prototype CDS service to production involves significant security, compliance, and safety concerns. Improper implementation could pose risks to patient care and data security.
To extend this work in a production environment, you would need to:
Remember that clinical decision support systems directly impact patient care—making robust implementation both an ethical and legal obligation.
For more information, see: