Building your own client

This guide provides a step by step tutorial on writing a client that can call out to an NLP API server using the NLP API Python SDK to process clinical documents.

Setup

Download the latest Python SDK zip file from the NLP API documentation Downloads page and unzip the contents.

unzip /path/to/emtellipro-python-sdk-7.2.0.zip

If you are using the command line, optionally create a virtual environment and then install the NLP API Python SDK.

$ python3 -m venv venv
$ source venv/bin/activate
(venv) $ pip install emtellipro-7.2.0-py3-none-any.whl

Set up the environment variables used in this guide for access to the NLP API server. You will need to set the following environment variables for the NLP API server and your access and secret keys. If you do not have access to the appropriate values for these variables please contact us.

export EMTELLIPRO_SERVER=''
export EMTELLIPRO_ACCESS_KEY=''
export EMTELLIPRO_SHARED_SECRET=''

If you are working within a Python notebook set up the environment variables in a .env file making sure you define the values for:

• EMTELLIPRO_SERVER
• EMTELLIPRO_ACCESS_KEY
• EMTELLIPRO_SHARED_SECRET

And use the dotenv library to load the environment variables:

%load_ext dotenv
%dotenv

Testing the connection

First we should ensure that your API keys are set up and working — to make sure you can use your API keys and can connect to the server, the quickest way is to use the built-in basic client to try and retrieve your user information:

emtellipro-db-client \
    get-user \
    --access-key $EMTELLIPRO_ACCESS_KEY \
    --shared-secret $EMTELLIPRO_SHARED_SECRET \
    --server $EMTELLIPRO_SERVER

If successful and your access is set up, this command will print out your username with a response like:

User with username: test_user

Processing plaintext documents

In this section, we provide a simple barebones client to call the NLP API server to process some documents and view the JSON result returned back to the client from the NLP API.

For the purposes of this example, we’ll assume your documents already exist as plaintext strings.

First we import the packages we need to write our simple Python client to call the NLP API to process documents.

from emtellipro import Emtellipro
from emtellipro.data import InputDocument

We first need to instantiate Emtellipro with the authentication details and the server we’re connecting to.

SERVER = os.environ['EMTELLIPRO_SERVER']
ACCESS_KEY = os.environ['EMTELLIPRO_ACCESS_KEY']
SHARED_SECRET = os.environ['EMTELLIPRO_SHARED_SECRET']
client = Emtellipro(
    server=SERVER,
    auth=(ACCESS_KEY, SHARED_SECRET),
)

To submit documents for processing, we need to add some metadata to them so that we can tell the NLP API some information about them and so we can match up which input document is associated with each annotated document in the results. We do this by creating InputDocument objects.

# these should be updated to your situation
texts = [
    "Sinus bradycardia. Otherwise no change from prior ECG.",
    "Long QTc interval.  Sinus bradycardia.  Low voltages are present in the limb leads. No prior available for comparison.",
    "Low QRS voltages in limb leads. Sinus rhythm. Nonspecific ST-T wave abnormality. First degree A-V block.",
]
category = 'cardiology'
subcategory = 'generic'
input_docs = []
for i, text in enumerate(texts):
    input_docs.append(
        InputDocument(
            id=i,
            category=category,
            subcategory=subcategory,
            data=text,
        ),
    )

input_doc = InputDocument(
    id=i,
    category=category,
    subcategory=subcategory,
    data=text,
    metadata={
        "hadm_id": "250322",
        "subject_dob": "1967-01-05 03:24:04",
        # the special 'extras' key allows storing arbitrary
        # key-value pairs
        "extras": [
            ("some arbitrary key", "the value"),
            ("another key", "another value"),
        ],
    },
)

Now that we have some simple input documents, we need to split them up into batches for processing. Processing documents in batches (rather than one at a time) ensures we process them more quickly. However, we cannot send them all as one big batch since that may go over the size limit of the NLP API instance we’re using (our example is small, so it’s not an issue here, but it’s a good habit to get into for larger projects).

To split the documents into batches, we can use batch_documents.

from emtellipro.utils import batch_documents
batches = batch_documents(input_docs)

Now we’re finally ready to process the documents using submit. This is where we can also set the list of processing features. Using None means “enable all”, but you can see FEATURES for the complete list of options. They must be passed as a list of strings.

# If you set this to None, all available features will be enabled; however,
# you usually want to limit it to the features you're actually interested
# in to limit processing time and storage use.
features = [
    'text',
    'snomed-ontology',
    'entity-measurement-unit',
    'entity-polarity',
    'measurement-relations',
    'qualifier-relations',
]
# Note that we're using a generator comprehension here because we don't
# want to submit all the batches at once; to avoid accidentally overloading
# the server, we'll process one batch at a time.
result_futures = (client.submit(b, features=features) for b in batches)

result_futures = [client.submit(b) for b in batches]

Each call to submit returns a ResultFuture object. This result future allows us to check the processing progress of the submission and retrieve the results when it’s done. It also provides methods for cancelling tasks.

Now we’ll iterate through all the result futures, waiting for them to complete, and retrieving the results (these will be Result objects).

results = []
for result_future in result_futures:
    print(f"Processing {result_future.task_id}")
    while not (done := result_future.done(timeout=1)):
        print(f"Progress: {done.progress * 100:.1f}%")
    # this contains Python objects representing the results returned
    results.append(result_future.result())

Now we can look at the JSON results we got from the NLP API.

import json
for res in results:
    # ResultFuture.json contains the parsed JSON
    print(json.dumps(res.json, indent=4))

The NLP API output JSON schema is available at this URL: /nlp/api

Explore Results using the NLP API Python SDK

Before you run the following code please run pip install pandas in your virtual environment.

The SDK provides functionality for parsing the JSON we get from the NLP API into Python objects; these allow easy exploration of the results and later storage in a database.

Recall that each result we got above was a Result object, which has some more attributes beyond just storing the unprocessed JSON: it also provides access to AnnotatedDocument instances representing each annotated document in the result.

# these will be the annotated docs for all the batches we processed
annotated_docs = []
for res in results:
    annotated_docs.extend(res.annotated_docs)

Once we have parsed the JSON results for each document in input_docs and stored it in annotated_docs in the example below we iterate through each document’s annotations and view information about the entities found by the NLP API a.k.a. the found entities (those entities that have a matched span in the document). The following code prints out the sentence in which the entity occurs, the matched span in the input document, the entity type, the SNOMED concept identifier, the SNOMED description and the polarity status (was it asserted or negated).

import pandas as pd
from emtellipro.data import SentenceLocation
from collections import defaultdict
def concat_spans(doc: InputDocument, spans):
    concat = []
    for span in spans:
        concat.append(doc.text[span.start:span.end])
    return "::".join(concat)
# fe is short for foundentity
fe_info = defaultdict(list)
docs_by_id = {d.id: d for d in input_docs}
for anno_doc in annotated_docs:
    input_doc = docs_by_id[anno_doc.id]
    for fe in anno_doc.found_entities:
        # only pick SNOMED information
        for concept in fe.concepts:
            if concept.ontology == 'snomed':
                for location in fe.locations['sentence']:
                    fe_info['sentence'].append(
                        concat_spans(input_doc, location.spans),
                    )
                    fe_info['foundentity'].append(
                        concat_spans(input_doc, fe.spans),
                    )
                    fe_info['entitytype'].append(
                        fe.type_name.get('snomed', '')
                    )
                    fe_info['section'].append(
                        fe.section_name,
                    )
                    fe_info['conceptid'].append(
                        concept.id,
                    )
                    # fsn = fully specified name from the ontology
                    fe_info['fsn'].append(
                        concept.description
                    )
                    fe_info['polarity'].append(
                        fe.polarity,
                    )
pd.DataFrame(fe_info)

For the sample input_docs above this returns the following pandas data frame:

For more details on the Python classes that represent the NLP API output please see the AnnotatedDocument.

Complete example

The following is a complete example for processing a set of input documents, printing the JSON and returning the results.

import os
import json
import typing
from emtellipro import Emtellipro
from emtellipro.data import InputDocument
from emtellipro.utils import batch_documents
def process(
    docs: typing.List[InputDocument],
    features=None,
):
    """
    Process the input documents and return the results for them.
    """
    server = os.environ['EMTELLIPRO_SERVER']
    access_key = os.environ['EMTELLIPRO_ACCESS_KEY']
    shared_secret = os.environ['EMTELLIPRO_SHARED_SECRET']
    client = Emtellipro(
        server=server,
        auth=(access_key, shared_secret),
    )
    for batch in batch_documents(docs):
        result_future = client.submit(batch, features=features)
        while not result_future.done(timeout=1):
            pass
        yield result_future.result()
def visualize(result):
    """
    Print JSON from the result
    """
    print(json.dumps(result.json, indent=4))

Process CCD documents

If you want to process CCDA documents with the NLP API, you need to extract the text from the XML first using ccd, and convert it to an InputDocument as before; the metadata from the CCD will be included in the input document object as metadata.

The ccd module contains functionality for loading CCDs from file objects or strings. See that module’s documentation for more details.

import pathlib
import emtellipro.ccd
# this is the same `input_docs` variable at the beginning of this tutorial
input_docs = []
ccd_path = pathlib.Path('../example-data/wright-ccd.xml')
with open(ccd_path) as fp:
    ccd_doc = emtellipro.ccd.load(fp)
    # alternatively:
    # ccd_doc = emtellipro.ccd.loads(fp.read())
    input_docs.append(
        ccd_doc.as_inputdoc(
            category='clinical',
            subcategory='ccd',
        ),
    )

The ccd module is helpful for working with CCD files directly (not just turning them into input documents), or if you just have the CCD file contents as strings (in which case you can parse them using loads).

However, if you’re just loading CCDs from a file path to convert them into input documents, you can use readfile which returns LoadedFile containing the input documents.

import emtellipro.load
input_docs = []
loaded_file = emtellipro.load.readfile(
    ccd_path,
    filetype='ccd',
    category='clinical',
    subcategory='ccd',
)
input_docs.extend(loaded_file.docs)

Now you can use that input_docs list to replace the list of plaintext documents at the top of this tutorial: Processing plaintext documents.

Or if you’re using the complete example:

annotated_docs = []
results = process(input_docs)
for res in results:
    visualize(res)
    annotated_docs.extend(res.annotated_docs)

Process PDF documents

This example shows you how to process PDF files using the Python SDK.

The readfile is used to load the PDF files into LoadedFile objects which contain the input documents. In the case of PDFs, there will only ever be one input document per loaded file.

import pathlib
import emtellipro.load
input_docs = []
pdf_path = pathlib.Path('../example-data/sample_ct_imagebased_report.pdf')
category = 'radiology'
subcategory = 'ct'
loaded_file = emtellipro.load.readfile(
    pdf_path,
    filetype='pdf',
    category=category,
    subcategory=subcategory,
)
input_docs.extend(loaded_file.docs)

Now you can use that input_docs list to replace the list of plaintext documents at the top of this tutorial: Processing plaintext documents.

Or if you’re using the complete example:

annotated_docs = []
results = process(input_docs)
for res in results:
    visualize(res)
    annotated_docs.extend(res.annotated_docs)

Process HL7 messages

We do have a HL7 listener that can accept HL7 messages and process the unstructured text data in those messages. However, we have not released this code along with the Python SDK yet. Please get in touch with us if you want to process HL7 messages.

Read documents from a database

The following sample code reads documents from a database and processes them using the NLP API. We are making some assumptions for this sample code which can be adapted to your own use case:

• The documents are in a MySQL or MariaDB database.
• The database has a table called gold_reports which contains the data we want to process.
• The category, subcategory columns are for the document category and subcategory (e.g. category is Radiology and subcategory is MR).
• The report column contains the text of the report we wish to process with the NLP API.
• The description column contains some information about the document.

The first step is to install a client that can connect to a database server. We are using pymysql in this tutorial because it is released under an MIT license.

pip install PyMySQL

First we import any additional Python packages we need:

import pymysql

This get_docs function executes a SQL query against the database and returns a list of documents that we want to process.

def get_docs(host, user, password, database, query):
    connection = pymysql.connect(
        host=host,
        user=user,
        password=password,
        database=database,
        cursorclass=pymysql.cursors.DictCursor,
    )
    rows = []
    with connection:
        with connection.cursor() as cursor:
            cursor.execute(query)
            rows = cursor.fetchall()
    return [r['text'] for r in rows]

We will retrieve the documents from the database using the following environment variables:

• MYSQL_HOST
• MYSQL_USER
• MYSQL_PASSWORD
• MYSQL_DATABASE

You can either set values for the above environment variables on the command line if you are running this code on the command line or you can save the values in a .env file if you are using a Python notebook and use the dotenv library to load the environment variables.

%load_ext dotenv
%dotenv

Next we query the database using the metadata stored in the data tables and process the documents retrieved using the NLP API.

query = """
    SELECT category, subcategory, report AS text
    FROM gold_reports
    WHERE category='Radiology'
        AND subcategory='MR'
        AND description LIKE '%KNEE%'
    LIMIT 2
    """
texts = get_docs(
    host=os.environ.get('MYSQL_HOST', '0.0.0.0'),
    user=os.environ.get('MYSQL_USER', ''),
    password=os.environ.get('MYSQL_PASSWORD', ''),
    database=os.environ.get('MYSQL_DATABASE', ''),
    query=query,
)
input_docs = [
    InputDocument(i, category='Radiology', subcategory='mr', data=text)
    for i, text in enumerate(texts)
]

Now you can use that input_docs list to replace the list of plaintext documents at the top of this tutorial: Processing plaintext documents.

Storing results in a database

The SDK provides a built-in client called the “database client” (see here). This client allows you process documents and store their results in SQL databases, CSV files, or JSON/JSONL files. The SDK also provides functionality for doing the same using Python.

The main entrypoint for doing this is the Database class. This class can create the necessary tables, migrate them when the schema changes, and store annotated documents in it.

Suppose we have a Postgres database running at locally. We’ll first connect to it and create all the tables we need.

import urllib.parse
import emtellipro.db
host = 'localhost'
port = '15432'
database_name = 'sdk-example'
user = 'postgres'
password = 'secret'
# if your password contains special characters, expecially @, it must be
# encoded so the URL can be valid
encoded_pass = urllib.parse.quote_plus(password)
url = f'postgresql://{user}:{encoded_pass}@{host}:{port}/{database_name}'
db = emtellipro.db.Database(url)
db.init()       # create tables
# OR
db.migrate()    # if tables already exist and only need be migrated

Now we have a database that’s ready to received annotated documents.

If you’ve been following this tutorial from the top, you should already have the input documents in input_docs and the annotated documents in annotated_docs. With those two lists, it’s very straightforward to save the documents in the database.

# we're starting a database transaction here
with db.begin() as conn:
    conn.save(input_docs, annotated_docs)

There are more parameters you can use to configure what extra information gets stored in the database. See save for details.

We can quickly confirm that the documents were stored in the database.

import sqlalchemy as sa
# we need to use text() here because the query is a string
query = sa.text("SELECT id FROM document")
with db.begin() as conn:
    result = conn.execute(query)
    for row in result:
        print("Saved document id:", row.id)

Handling document text

Often the text of the document on which NLP API’s entity spans are valid is the same as input document text (in cases where the input is plaintext or JSON containing plaintext).

However, if the input is a PDF file or if the infer-document-structure feature is enabled, then the NLP API will return the document text in its response; it is on the returned text that the spans are valid.

Because of this, the database saving code will always store the returned document text (when available), ensuring the spans are usable. When storing annotations yourself elsewhere, it’s important to remember to store the text from the annotated document as well.

Full-featured example

For a full featured example client using the techniques above, see the examples/end-to-end-notebook/ directory distributed with the SDK. It contains a client (client.py) that

1. reads documents from a Snowflake database,
2. submits them in batches to the NLP API,
3. retrieves results from the NLP API,
4. saves them to JSONL files, and
5. uploads the JSONL files to a target Snowflake database,

with steps 2-5 happening in parallel using a process pool.

See examples/end-to-end-notebook/README.md file for how it works, and how to use it from a notebook in a Databricks environment.

Next steps

Instead of building your own client, you might want to use the Python clients that we distribute as part of the Python SDK distribution especially the Python database client called emtellipro-db-client which has a lot of useful functionality that you might end up re-creating if you write a full-featured client of your own from scratch. See the database client documentation for more information.