Skip to content

Convert Between Healthcare Data Formats

Level: Intermediate

Convert between CDA, HL7v2, and FHIR formats using HealthChain's interoperability engine. This recipe covers the most common conversion scenarios for integrating legacy healthcare systems with modern FHIR-based applications.

The InteropEngine provides a unified interface for bidirectional format conversion, handling the complexity of mapping between different healthcare data standards.

Setup

Install HealthChain:

pip install healthchain

Create an interoperability engine:

from healthchain.interop import create_interop, FormatType
from pathlib import Path
import json

engine = create_interop()

Converting CDA to FHIR

Parse a CDA document and extract FHIR resources:

cda_xml = """
<ClinicalDocument xmlns="urn:hl7-org:v3">
  <templateId root="2.16.840.1.113883.10.20.22.1.2"/>
  <id root="2.16.840.1.113883.19.5.99999.1"/>
  <code code="34133-9" displayName="Summarization of Episode Note"
        codeSystem="2.16.840.1.113883.6.1" codeSystemName="LOINC"/>
  <title>Example CDA Document</title>
  <effectiveTime value="20150519"/>
  <confidentialityCode code="N" codeSystem="2.16.840.1.113883.5.25"/>
  <languageCode code="en-US"/>
  <recordTarget>
    <patientRole>
      <id extension="12345" root="2.16.840.1.113883.19.5.99999.2"/>
      <patient>
        <name>
          <given>John</given>
          <family>Smith</family>
        </name>
        <administrativeGenderCode code="M" codeSystem="2.16.840.1.113883.5.1"/>
        <birthTime value="19701101"/>
      </patient>
    </patientRole>
  </recordTarget>
  <component>
    <structuredBody>
      <component>
        <section>
          <templateId root="2.16.840.1.113883.10.20.1.11"/>
          <code code="11450-4" codeSystem="2.16.840.1.113883.6.1"
                displayName="Problem List"/>
          <title>Problems</title>
          <text>Hypertension</text>
          <entry>
            <observation classCode="OBS" moodCode="EVN">
              <templateId root="2.16.840.1.113883.10.20.1.28"/>
              <templateId root="2.16.840.1.113883.10.20.1.54"/>
              <id root="2.16.840.1.113883.19.5.99999.3"/>
              <code code="64572001" displayName="Hypertension"
                    codeSystem="2.16.840.1.113883.6.96" codeSystemName="SNOMED CT"/>
              <statusCode code="completed"/>
              <effectiveTime value="20180101"/>
              <value xsi:type="CD" code="64572001" displayName="Hypertension"
                     codeSystem="2.16.840.1.113883.6.96" codeSystemName="SNOMED CT"
                     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"/>
            </observation>
          </entry>
        </section>
      </component>
    </structuredBody>
  </component>
</ClinicalDocument>
"""

# Convert CDA to FHIR resources
fhir_resources = engine.to_fhir(cda_xml, src_format=FormatType.CDA)

# Inspect the results
for resource in fhir_resources:
    print(f"Resource Type: {resource.resource_type}")
    print(json.dumps(resource.dict(), indent=2))

Converting FHIR to CDA

Generate a CDA document from FHIR resources:

from healthchain.fhir.r4b import Condition, Patient

# Create FHIR resources
patient = Patient(
    resourceType="Patient",
    id="patient-1",
    name=[{"family": "Smith", "given": ["John"]}],
    gender="male",
    birthDate="1970-11-01"
)

condition = Condition(
    resourceType="Condition",
    id="condition-1",
    subject={"reference": "Patient/patient-1"},
    code={
        "coding": [
            {
                "system": "http://snomed.info/sct",
                "code": "38341003",
                "display": "Hypertension"
            }
        ],
        "text": "Hypertension"
    },
    clinicalStatus={
        "coding": [
            {
                "system": "http://terminology.hl7.org/CodeSystem/condition-clinical",
                "code": "active"
            }
        ]
    },
    verificationStatus={
        "coding": [
            {
                "system": "http://terminology.hl7.org/CodeSystem/condition-ver-status",
                "code": "confirmed"
            }
        ]
    },
    onsetDateTime="2018-01-01"
)

# Convert to CDA
resources = [patient, condition]
cda_document = engine.from_fhir(resources, dest_format=FormatType.CDA)

print(cda_document)

Converting HL7v2 to FHIR

Parse an HL7v2 message and extract FHIR resources:

hl7v2_message = """
MSH|^~\&|EPIC|EPICADT|SMS|SMSADT|199912271408|CHARRIS|ADT^A01|1817457|D|2.5|
PID|1||PATID1234^5^M11^ADT1^MR^GOOD HEALTH HOSPITAL~123456789^^^USSSA^SS||EVERYMAN^ADAM^A^III||19610615|M||C|2222 HOME STREET^^GREENSBORO^NC^27401-1020|GL|(555) 555-2004|(555)555-2004||S||PATID12345001^2^M10^ADT1^AN^A|444333333|987654^NC|
NK1|1|NUCLEAR^NELDA^W|SPO||(555)555-3333||EC|||||||||||||||||||||||||
PV1|1|I|2000^2012^01||||004777^ATTEND^AARON^A|||SUR||||ADM|A0|
"""

# Convert HL7v2 to FHIR resources
fhir_resources = engine.to_fhir(hl7v2_message, src_format=FormatType.HL7V2)

for resource in fhir_resources:
    print(f"Resource Type: {resource.resource_type}")
    print(json.dumps(resource.dict(), indent=2))

Converting FHIR to HL7v2

Generate an HL7v2 message from FHIR resources:

from healthchain.fhir.r4b import Patient, Encounter

patient = Patient(
    resourceType="Patient",
    id="patient-1",
    name=[{"family": "Everyman", "given": ["Adam", "A"], "suffix": ["III"]}],
    gender="male",
    birthDate="1961-06-15",
    identifier=[
        {
            "system": "urn:oid:1.2.36.146.595.217.0.1",
            "value": "PATID1234"
        }
    ],
    address=[
        {
            "line": ["2222 Home Street"],
            "city": "Greensboro",
            "state": "NC",
            "postalCode": "27401"
        }
    ],
    telecom=[
        {
            "system": "phone",
            "value": "(555) 555-2004",
            "use": "home"
        }
    ]
)

encounter = Encounter(
    resourceType="Encounter",
    id="encounter-1",
    status="finished",
    class_fhir={
        "system": "http://terminology.hl7.org/CodeSystem/v3-ActCode",
        "code": "IMP",
        "display": "inpatient encounter"
    },
    subject={
        "reference": "Patient/patient-1"
    }
)

# Convert to HL7v2
resources = [patient, encounter]
hl7v2_message = engine.from_fhir(resources, dest_format=FormatType.HL7V2)

print(hl7v2_message)

Saving Results

Save converted data to files:

output_dir = Path("./output")
output_dir.mkdir(exist_ok=True)

# Save FHIR resources as JSON
for resource in fhir_resources:
    filename = f"{resource.resource_type.lower()}_{resource.id}.json"
    with open(output_dir / filename, "w") as f:
        json.dump(resource.dict(), f, indent=2)

# Save CDA document as XML
with open(output_dir / "document.xml", "w") as f:
    f.write(cda_document)

# Save HL7v2 message
with open(output_dir / "message.hl7", "w") as f:
    f.write(hl7v2_message)

Next Steps