Skip to content

Legacy Adapter

Intent

Transform legacy data formats (HL7v2, CDA, DICOM SR) to FHIR while preserving clinical semantics and enabling incremental modernization of healthcare systems.

Forces

  • Legacy System Preservation: Healthcare organizations must modernize gradually while preserving existing clinical semantics and workflows.

Structure

The Legacy Adapter pattern provides semantic-preserving transformation of legacy healthcare data formats to FHIR using configurable mapping rules and terminology services.

Legacy Adapter Architecture

Key Components

Legacy Adapter

Main orchestration component that coordinates transformation workflow

Mapping Engine

Executes transformation rules using FHIR StructureMap and FHIRPath

Terminology Mapper

Maps codes between legacy and modern code systems

Validation Service

Validates transformed FHIR resources against profiles and business rules

Behavior

HL7v2 to FHIR Transformation

The following diagram shows the transformation workflow:

Legacy Adapter Sequence

Implementation Considerations

StructureMap-Based Transformation

FHIR StructureMap resource defining transformation rules from HL7v2 segments to FHIR resources with field mappings and terminology translations.

StructureMap-Based Transformation
{
  "resourceType": "StructureMap",
  "id": "hl7v2-adt-to-fhir",
  "name": "HL7v2 ADT to FHIR Bundle",
  "status": "active",
  "structure": [
    {
      "url": "http://hl7.org/fhir/v2/StructureDefinition/Message",
      "mode": "source"
    },
    {
      "url": "http://hl7.org/fhir/StructureDefinition/Bundle", 
      "mode": "target"
    }
  ],
  "group": [
    {
      "name": "main",
      "typeMode": "none",
      "rule": [
        {
          "name": "bundle",
          "source": [
            {
              "context": "src",
              "element": "MSH"
            }
          ],
          "target": [
            {
              "context": "bundle",
              "contextType": "variable",
              "element": "type",
              "transform": "copy",
              "parameter": [
                {
                  "valueString": "transaction"
                }
              ]
            }
          ],
          "rule": [
            {
              "name": "patient",
              "source": [
                {
                  "context": "src",
                  "element": "PID"
                }
              ],
              "target": [
                {
                  "context": "bundle",
                  "element": "entry",
                  "variable": "entry"
                }
              ],
              "dependent": [
                {
                  "name": "PatientTransform",
                  "variable": ["src", "entry"]
                }
              ]
            }
          ]
        }
      ]
    }
  ]
}

Python Implementation

Python implementation of the legacy adapter with HL7v2 parsing, segment mapping, terminology lookup, and FHIR resource construction.

Python Implementation
from dataclasses import dataclass
from typing import Dict, List, Optional, Any
from datetime import datetime
import hl7
from fhir.resources.bundle import Bundle
from fhir.resources.patient import Patient
from fhir.resources.encounter import Encounter

@dataclass
class TransformationContext:
    source_format: str
    target_format: str = "FHIR"
    terminology_mappings: Dict[str, str] = None
    validation_profiles: List[str] = None
    preserve_extensions: bool = True

class LegacyAdapter:
    def __init__(self, mapping_engine, terminology_mapper, validator, audit_service):
        self.mapping_engine = mapping_engine
        self.terminology_mapper = terminology_mapper
        self.validator = validator
        self.audit_service = audit_service

    async def transform_message(self, legacy_data: str, source_format: str, 
                              context: TransformationContext) -> Bundle:
        """Transform legacy message to FHIR Bundle"""

        try:
            # Parse legacy format
            parsed_data = await self._parse_legacy_data(legacy_data, source_format)

            # Apply transformation mappings
            fhir_bundle = await self.mapping_engine.transform(
                parsed_data, source_format, context
            )

            # Validate resulting FHIR resources
            validation_result = await self.validator.validate_bundle(
                fhir_bundle, context.validation_profiles
            )

            if not validation_result.is_valid:
                raise TransformationError(f"Validation failed: {validation_result.errors}")

            # Audit successful transformation
            await self._audit_transformation(legacy_data, fhir_bundle, context)

            return fhir_bundle

        except Exception as e:
            # Audit transformation failure
            await self._audit_transformation_error(legacy_data, str(e), context)
            raise

    async def _parse_legacy_data(self, data: str, format_type: str) -> Dict:
        """Parse legacy data format into structured representation"""

        if format_type == "HL7v2":
            return self._parse_hl7v2(data)
        elif format_type == "CDA":
            return self._parse_cda(data)
        elif format_type == "DICOM-SR":
            return self._parse_dicom_sr(data)
        else:
            raise ValueError(f"Unsupported format: {format_type}")

    def _parse_hl7v2(self, message: str) -> Dict:
        """Parse HL7v2 message"""

        parsed = hl7.parse(message)

        return {
            'message_type': str(parsed['MSH'][9][0]),
            'trigger_event': str(parsed['MSH'][9][1]),
            'patient': self._extract_patient_from_pid(parsed['PID']),
            'encounter': self._extract_encounter_from_pv1(parsed.get('PV1')),
            'observations': self._extract_observations_from_obx(parsed.segments('OBX')),
            'raw_segments': [str(seg) for seg in parsed]
        }

    def _extract_patient_from_pid(self, pid_segment) -> Dict:
        """Extract patient information from PID segment"""

        return {
            'identifiers': [
                {
                    'value': str(pid_segment[3][0]) if pid_segment[3] else None,
                    'system': str(pid_segment[3][0][3]) if pid_segment[3] and len(pid_segment[3][0]) > 3 else None
                }
            ],
            'name': {
                'family': str(pid_segment[5][0][0]) if pid_segment[5] else None,
                'given': [str(pid_segment[5][0][1])] if pid_segment[5] and len(pid_segment[5][0]) > 1 else []
            },
            'gender': str(pid_segment[8]) if pid_segment[8] else None,
            'birth_date': str(pid_segment[7]) if pid_segment[7] else None,
            'address': self._extract_address_from_xad(pid_segment[11]) if pid_segment[11] else None
        }

class MappingEngine:
    def __init__(self, terminology_mapper, structure_map_engine):
        self.terminology_mapper = terminology_mapper
        self.structure_map_engine = structure_map_engine

    async def transform(self, parsed_data: Dict, source_format: str, 
                       context: TransformationContext) -> Bundle:
        """Apply transformation rules to convert legacy data to FHIR"""

        if source_format == "HL7v2":
            return await self._transform_hl7v2_to_fhir(parsed_data, context)
        elif source_format == "CDA":
            return await self._transform_cda_to_fhir(parsed_data, context)
        else:
            raise ValueError(f"Transformation not implemented for {source_format}")

    async def _transform_hl7v2_to_fhir(self, data: Dict, context: TransformationContext) -> Bundle:
        """Transform HL7v2 message to FHIR Bundle"""

        bundle = Bundle(type="transaction", entry=[])

        # Transform patient
        if 'patient' in data:
            patient = await self._create_fhir_patient(data['patient'], context)
            bundle.entry.append({
                'resource': patient,
                'request': {
                    'method': 'POST',
                    'url': 'Patient'
                }
            })

        # Transform encounter
        if 'encounter' in data and data['encounter']:
            encounter = await self._create_fhir_encounter(data['encounter'], context)
            bundle.entry.append({
                'resource': encounter,
                'request': {
                    'method': 'POST',
                    'url': 'Encounter'
                }
            })

        # Transform observations
        for obs_data in data.get('observations', []):
            observation = await self._create_fhir_observation(obs_data, context)
            bundle.entry.append({
                'resource': observation,
                'request': {
                    'method': 'POST',
                    'url': 'Observation'
                }
            })

        return bundle

    async def _create_fhir_patient(self, patient_data: Dict, context: TransformationContext) -> Patient:
        """Create FHIR Patient from HL7v2 PID data"""

        patient = Patient()

        # Map identifiers
        if patient_data.get('identifiers'):
            patient.identifier = []
            for id_data in patient_data['identifiers']:
                if id_data['value']:
                    patient.identifier.append({
                        'value': id_data['value'],
                        'system': id_data.get('system')
                    })

        # Map name
        if patient_data.get('name'):
            name_data = patient_data['name']
            patient.name = [{
                'family': name_data.get('family'),
                'given': name_data.get('given', [])
            }]

        # Map gender with terminology mapping
        if patient_data.get('gender'):
            mapped_gender = await self.terminology_mapper.map_code(
                patient_data['gender'],
                'http://terminology.hl7.org/CodeSystem/v2-0001',  # HL7v2 Administrative Sex
                'http://hl7.org/fhir/administrative-gender'  # FHIR Administrative Gender
            )
            patient.gender = mapped_gender.code if mapped_gender else None

        # Map birth date
        if patient_data.get('birth_date'):
            birth_date = self._parse_hl7_date(patient_data['birth_date'])
            if birth_date:
                patient.birthDate = birth_date

        return patient

class TerminologyMapper:
    def __init__(self, terminology_service):
        self.terminology_service = terminology_service
        self.mapping_cache = {}

    async def map_code(self, source_code: str, source_system: str, target_system: str):
        """Map code from source to target terminology system"""

        cache_key = f"{source_system}#{source_code}->{target_system}"

        # Check cache first
        if cache_key in self.mapping_cache:
            return self.mapping_cache[cache_key]

        # Use FHIR terminology service for mapping
        mapping_result = await self.terminology_service.translate_code(
            code=source_code,
            system=source_system,
            target_system=target_system
        )

        # Cache result
        self.mapping_cache[cache_key] = mapping_result

        return mapping_result

class SemanticPreservationValidator:
    """Validates that transformations preserve clinical semantics"""

    def __init__(self):
        self.semantic_rules = self._load_semantic_rules()

    async def validate_transformation(self, source_data: Dict, transformed_fhir: Bundle) -> Dict:
        """Validate that transformation preserves clinical meaning"""

        validation_results = {
            'semantic_preservation': True,
            'warnings': [],
            'errors': []
        }

        # Check that essential clinical data is preserved
        source_patient_id = self._extract_patient_id(source_data)
        fhir_patient_ids = self._extract_fhir_patient_ids(transformed_fhir)

        if source_patient_id not in fhir_patient_ids:
            validation_results['errors'].append(
                f"Patient ID {source_patient_id} not preserved in transformation"
            )
            validation_results['semantic_preservation'] = False

        # Check terminology mappings
        for rule in self.semantic_rules:
            result = await self._apply_semantic_rule(rule, source_data, transformed_fhir)
            if not result.valid:
                validation_results['warnings'].extend(result.warnings)
                validation_results['errors'].extend(result.errors)

        return validation_results

# Example configuration for HL7v2 ADT transformation
transformation_config = {
    "source_format": "HL7v2",
    "message_types": ["ADT^A01", "ADT^A04", "ADT^A08"],
    "terminology_mappings": {
        "gender": {
            "source_system": "http://terminology.hl7.org/CodeSystem/v2-0001",
            "target_system": "http://hl7.org/fhir/administrative-gender",
            "mappings": {
                "M": "male",
                "F": "female", 
                "O": "other",
                "U": "unknown"
            }
        },
        "encounter_class": {
            "source_system": "http://terminology.hl7.org/CodeSystem/v2-0004",
            "target_system": "http://terminology.hl7.org/CodeSystem/v3-ActCode",
            "mappings": {
                "I": "IMP",  # Inpatient
                "O": "AMB",  # Outpatient -> Ambulatory
                "E": "EMER", # Emergency
                "R": "IMP"   # Recurring -> Inpatient
            }
        }
    },
    "validation_profiles": [
        "http://hl7.org/fhir/us/core/StructureDefinition/us-core-patient",
        "http://hl7.org/fhir/us/core/StructureDefinition/us-core-encounter"
    ]
}

  • Broker: Broker routes requests to Legacy Adapter for systems not yet modernized
  • Audit & Provenance Chain: All legacy transformations are logged with provenance tracking
  • Security Strategy: Legacy credentials are mapped to FHIR security context
  • Capability Facade: Capability Facade exposes legacy system capabilities as FHIR CapabilityStatement

Benefits

  • Semantic Preservation: Maintains clinical meaning during transformation
  • Incremental Migration: Enables gradual modernization without disruption
  • Standards Compliance: Produces valid FHIR resources with proper profiles
  • Terminology Mapping: Handles code system translations automatically
  • Audit Trail: Complete tracking of transformation decisions and outcomes

Trade-offs

  • Complexity: Sophisticated mapping rules and semantic validation required
  • Performance: Transformation overhead for each legacy message
  • Data Fidelity: Some legacy data may not map cleanly to FHIR
  • Maintenance: Mapping rules need updates as standards evolve
  • Validation Overhead: Semantic preservation checking adds processing time

References

Incremental Approach

Start with the most critical message types and gradually expand coverage. Use FHIR extensions to preserve legacy data that doesn't map cleanly to standard FHIR elements.

Semantic Loss

Always validate that clinical meaning is preserved during transformation. Consider maintaining original data alongside transformed versions for audit and verification purposes.