Pathling Python

pip install pathling

from pathling import PathlingContext

pc = PathlingContext.create()

pc = PathlingContext.create(
    terminology_server_url='https://ontology.nhs.uk/production1/fhir',
    token_endpoint='https://ontology.nhs.uk/authorisation/auth/realms/nhs-digital-terminology/protocol/openid-connect/token',
    client_id='[client ID]',
    client_secret='[client secret]'
)

pc = PathlingContext.create(
    terminology_server_url="http://localhost:8081/fhir",
    terminology_verbose_request_logging=True,
    cache_override_expiry=2_628_000,
    cache_storage_type="disk",
    cache_storage_path=".local/tx-cache"
)

pc.spark  # Access the underlying Spark session
pc.spark.sparkContext.setLogLevel("DEBUG")  # Set log level

# Read each line from the NDJSON into a row within a Spark data set.
ndjson_dir = '/some/path/ndjson/'
json_resources = pc.spark.read.text(ndjson_dir)

# Convert the data set of strings into a structured FHIR data set.
patients = pc.encode(json_resources, 'Patient')

# Do some stuff.
patients.select('id', 'gender', 'birthDate').show()

data = pc.read.ndjson("/some/file/location")

# Read each Bundle into a row within a Spark data set.
bundles_dir = '/some/path/bundles/'
bundles = pc.spark.read.text(bundles_dir, wholetext=True)

# Convert the data set of strings into a structured FHIR data set.
patients = pc.encode_bundle(bundles, 'Patient')

# XML Bundles can be encoded using input type.
# patients = pc.encode_bundle(bundles, 'Patient', inputType=MimeType.FHIR_XML)

# Do some stuff.
patients.select('id', 'gender', 'birthDate').show()

# Read from previously persisted Delta tables.
data = pc.read.tables()

# Get available resource types.
data.resource_types()

# Read a specific resource.
patients = data.read('Patient')
patients.count()

result = data.view(
    resource="Patient",
    select=[
        {"column": [{"path": "getResourceKey()", "name": "patient_id"}]},
        {
            "forEach": "address",
            "column": [
                {"path": "line.join('\\n')", "name": "street"},
                {"path": "use", "name": "use"},
                {"path": "city", "name": "city"},
                {"path": "postalCode", "name": "zip"},
            ],
        },
    ],
)

display(result)

view_ds = datasource.view(
    resource='Patient',
    select=[
        {
            'column': [
                {'path': 'id', 'name': 'id'},
                {'path': 'gender', 'name': 'gender'},
                {'path': "telecom.where(system='phone').value ", 'name': 'phone_numbers', 'collection': True},
            ]
        }
    ],
    where=[
        {'path': "gender = 'male'"},
    ]
)

view_ds.show()

view_ds = datasource.view(
    resource='Patient',
    select=[
        {
            'forEach': 'name',
            'column': [
                {'path': 'use', 'name': 'name_use'},
                {'path': 'family', 'name': 'family_name'},
            ],
            'select': [
                {
                    'forEachOrNull': 'given',
                    'column': [
                        {'path': '$this', 'name': 'given_name'},
                    ],
                }
            ]
        },
    ]
)

view_ds.show()

from pathling import to_coding

# Convert a column containing codes into a Coding struct.
coding_column = to_coding(df.CODE, 'http://snomed.info/sct')

# With version.
coding_column = to_coding(df.CODE, 'http://snomed.info/sct', version='http://snomed.info/sct/32506021000036107/version/20250831')

from pathling import to_snomed_coding

# Shorthand for SNOMED CT codes.
coding_column = to_snomed_coding(df.CODE)
coding_column = to_snomed_coding(df.CODE, version='http://snomed.info/sct/32506021000036107/version/20250831')

from pathling.functions import to_loinc_coding

# Shorthand for LOINC codes.
coding_column = to_loinc_coding(df.CODE)

from pathling import Coding

# Create a Coding object.
coding = Coding('http://snomed.info/sct', '232208008')

# Create a SNOMED Coding using the factory method.
snomed_coding = Coding.of_snomed('232208008')

# Convert to a Spark literal column.
coding_literal = coding.to_literal()

from pathling import to_ecl_value_set

# Convert a SNOMED CT ECL expression into a FHIR ValueSet URI.
viral_infection_ecl = """
<< 64572001|Disease| : (
  << 370135005|Pathological process| = << 441862004|Infectious process|,
  << 246075003|Causative agent| = << 49872002|Virus|
)
"""
value_set_uri = to_ecl_value_set(viral_infection_ecl)

from pathling import member_of, to_snomed_coding, to_ecl_value_set

# Using an ECL expression.
result = csv.select(
    "CODE",
    "DESCRIPTION",
    member_of(
        to_snomed_coding(csv.CODE),
        to_ecl_value_set("<< 64572001")  # Viral disease
    ).alias("VIRAL_INFECTION")
)
result.show()

# Using a predefined value set.
result = csv.select(
    "CODE",
    "DESCRIPTION",
    member_of(
        to_coding(csv.CODE, 'http://loinc.org'),
        'http://hl7.org/fhir/ValueSet/observation-vitalsignresult'
    ).alias("IS_VITAL_SIGN")
)

result = transformed_df.withColumn(
    "Viral Infection",
    pc.snomed.member_of(col("primary_diagnosis_concept"), "<< 64572001")
)

from pathling import translate, to_coding

# Translate SNOMED CT codes to Read CTV3.
result = pc.translate(
    csv,
    to_coding(csv.CODE, 'http://snomed.info/sct'),
    'http://snomed.info/sct/900000000000207008?fhir_cm=900000000000497000',
    output_column_name='READ_CODE'
)

# Extract just the code from the Coding struct.
result = result.withColumn('READ_CODE', result.READ_CODE.code)
result.select('CODE', 'DESCRIPTION', 'READ_CODE').show()

from pathling import subsumes, to_snomed_coding, Coding

# Test if codes are subsumed by a specific code.
# 232208008 |Ear, nose and throat disorder|
left_coding = Coding('http://snomed.info/sct', '232208008')
right_coding_column = to_snomed_coding(csv.CODE)

result = pc.subsumes(
    csv,
    'IS_ENT',
    left_coding=left_coding,
    right_coding_column=right_coding_column
)

result.select('CODE', 'DESCRIPTION', 'IS_ENT').show()

from pathling import subsumed_by

# Test if a code is subsumed by codes in a column.
result = pc.subsumed_by(
    csv,
    'IS_SUBTYPE',
    left_coding_column=to_snomed_coding(csv.CODE),
    right_coding=Coding('http://snomed.info/sct', '232208008')
)

from pathling import property_of, to_snomed_coding, PropertyType

# Get the parent codes for each code in the dataset.
parents = csv.withColumn(
    "PARENTS",
    property_of(to_snomed_coding(csv.CODE), "parent", PropertyType.CODE)
)

# Split each parent code into a separate row.
exploded_parents = parents.selectExpr(
    "CODE", "DESCRIPTION", "explode_outer(PARENTS) AS PARENT"
)

from pathling import display, to_snomed_coding

# Get the display term for parent codes.
with_displays = exploded_parents.withColumn(
    "PARENT_DISPLAY",
    display(to_snomed_coding(exploded_parents.PARENT))
)

with_displays.show()

transformed_df = source_df.withColumn(
    "Primary Diagnosis Term",
    pc.snomed.display(col("primary_diagnosis_concept"))
)

from pathling import designation, to_snomed_coding, Coding

# Get the synonyms for each code in the dataset.
# 900000000000013009 is the SNOMED CT "Synonym" designation use.
synonyms = csv.withColumn(
    "SYNONYMS",
    designation(
        to_snomed_coding(csv.CODE),
        Coding.of_snomed("900000000000013009")
    )
)

# Split each synonym into a separate row.
exploded_synonyms = synonyms.selectExpr(
    "CODE", "DESCRIPTION", "explode_outer(SYNONYMS) AS SYNONYM"
)

exploded_synonyms.show()

from pathling import PathlingContext, to_snomed_coding, to_ecl_value_set, member_of
from pyspark.sql.functions import col, when

pc = PathlingContext.create()

# Define value sets using ECL.
viral_infection_ecl = "<< 64572001"  # Viral disease
musculoskeletal_injury_ecl = "<< 263534002"  # Injury of musculoskeletal system
mental_health_ecl = "<< 40733004 |Mental state finding|"

# Add membership columns for each category.
categorised_df = df.withColumn(
    "Viral Infection",
    member_of(to_snomed_coding(col("diagnosis_code")), to_ecl_value_set(viral_infection_ecl))
).withColumn(
    "Musculoskeletal Injury",
    member_of(to_snomed_coding(col("diagnosis_code")), to_ecl_value_set(musculoskeletal_injury_ecl))
).withColumn(
    "Mental Health Problem",
    member_of(to_snomed_coding(col("diagnosis_code")), to_ecl_value_set(mental_health_ecl))
)

# Create mutually exclusive categories with hierarchy.
mutually_exclusive_df = categorised_df.withColumn(
    "Category",
    when(col("Viral Infection"), "Viral Infection")
    .when(~col("Viral Infection") & col("Musculoskeletal Injury"), "Musculoskeletal Injury")
    .when(~col("Viral Infection") & ~col("Musculoskeletal Injury") & col("Mental Health Problem"), "Mental Health Problem")
    .otherwise("Other")
)

from pathling import display, property_of, to_snomed_coding, PropertyType

# Add display terms to codes.
enriched_df = df.withColumn(
    "diagnosis_display",
    display(to_snomed_coding(df.diagnosis_code))
)

# Add parent codes.
with_parents = enriched_df.withColumn(
    "parent_codes",
    property_of(to_snomed_coding(df.diagnosis_code), "parent", PropertyType.CODE)
)

# Convert to Pandas.
pandas_df = result.toPandas()

# Use with plotting libraries.
import plotly.express as px

fig = px.bar(pandas_df, x="category", y="count")
fig.show()