Skip to content

Interoperability Engine

InteropEngine

Generic interoperability engine for converting between healthcare formats

The InteropEngine provides capabilities for converting between different healthcare data format standards, such as HL7 FHIR, CDA, and HL7v2.

The engine uses a template-based approach for transformations, with templates stored in the configured template directory. Transformations are handled by format-specific parsers and generators that are lazily loaded as needed.

Configuration is handled through the config property, which provides direct access to the underlying ConfigManager instance. This allows for setting validation levels, changing environments, and accessing configuration values.

The engine supports registering custom parsers, generators, and validators to extend or override the default functionality.

Example

engine = InteropEngine()

Convert CDA to FHIR

fhir_resources = engine.to_fhir(cda_xml, src_format="cda")

Convert FHIR to CDA

cda_xml = engine.from_fhir(fhir_resources, dest_format="cda")

Access config directly:

engine.config.set_environment("production") engine.config.set_validation_level("warn") value = engine.config.get_config_value("cda.sections.problems.resource")

Access the template registry:

template = engine.template_registry.get_template("cda_fhir/condition") engine.template_registry.add_filter()

Register custom components:

engine.register_parser(FormatType.CDA, custom_parser) engine.register_generator(FormatType.FHIR, custom_generator)

Register custom configuration validators:

engine.register_cda_section_config_validator("Procedure", ProcedureSectionConfig) engine.register_cda_document_config_validator("CCD", CCDDocumentConfig)

Source code in healthchain/interop/engine.py 
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
class InteropEngine:
    """Generic interoperability engine for converting between healthcare formats

    The InteropEngine provides capabilities for converting between different
    healthcare data format standards, such as HL7 FHIR, CDA, and HL7v2.

    The engine uses a template-based approach for transformations, with templates
    stored in the configured template directory. Transformations are handled by
    format-specific parsers and generators that are lazily loaded as needed.

    Configuration is handled through the `config` property, which provides
    direct access to the underlying ConfigManager instance. This allows
    for setting validation levels, changing environments, and accessing
    configuration values.

    The engine supports registering custom parsers, generators, and validators
    to extend or override the default functionality.

    Example:
        engine = InteropEngine()

        # Convert CDA to FHIR
        fhir_resources = engine.to_fhir(cda_xml, src_format="cda")

        # Convert FHIR to CDA
        cda_xml = engine.from_fhir(fhir_resources, dest_format="cda")

        # Access config directly:
        engine.config.set_environment("production")
        engine.config.set_validation_level("warn")
        value = engine.config.get_config_value("cda.sections.problems.resource")

        # Access the template registry:
        template = engine.template_registry.get_template("cda_fhir/condition")
        engine.template_registry.add_filter()

        # Register custom components:
        engine.register_parser(FormatType.CDA, custom_parser)
        engine.register_generator(FormatType.FHIR, custom_generator)

        # Register custom configuration validators:
        engine.register_cda_section_config_validator("Procedure", ProcedureSectionConfig)
        engine.register_cda_document_config_validator("CCD", CCDDocumentConfig)
    """

    def __init__(
        self,
        config_dir: Optional[Path] = None,
        validation_level: str = ValidationLevel.STRICT,
        environment: Optional[str] = None,
    ):
        """Initialize the InteropEngine

        Args:
            config_dir: Base directory containing configuration files. If None, will search standard locations.
            validation_level: Level of configuration validation (strict, warn, ignore)
            environment: Optional environment to use (development, testing, production)
        """
        # Initialize configuration manager
        self.config = InteropConfigManager(config_dir, validation_level, environment)

        # Initialize template registry
        template_dir = config_dir / "templates"
        self.template_registry = TemplateRegistry(template_dir)

        # Create and register default filters
        # Get required configuration for filters
        mappings_dir = self.config.get_config_value("defaults.mappings_dir")
        if not mappings_dir:
            log.warning("No mappings directory configured, using default mappings")
            mappings_dir = "cda_default"
        mappings = self.config.get_mappings(mappings_dir)
        id_prefix = self.config.get_config_value("defaults.common.id_prefix")

        # Get default filters from the filters module
        default_filters = create_default_filters(mappings, id_prefix)
        self.template_registry.initialize(default_filters)

        # Component registries for lazy loading
        self._parsers = {}
        self._generators = {}

    # Lazy-loaded parsers
    @cached_property
    def cda_parser(self):
        """Lazily load the CDA parser"""
        return self._get_parser(FormatType.CDA)

    @cached_property
    def hl7v2_parser(self):
        """Lazily load the HL7v2 parser"""
        return self._get_parser(FormatType.HL7V2)

    # Lazy-loaded generators
    @cached_property
    def cda_generator(self):
        """Lazily load the CDA generator"""
        return self._get_generator(FormatType.CDA)

    @cached_property
    def fhir_generator(self):
        """Lazily load the FHIR generator"""
        return self._get_generator(FormatType.FHIR)

    @cached_property
    def hl7v2_generator(self):
        """Lazily load the HL7v2 generator"""
        return self._get_generator(FormatType.HL7V2)

    def _get_parser(self, format_type: FormatType):
        """Get or create a parser for the specified format

        Args:
            format_type: The format type to get a parser for (CDA or HL7v2)

        Returns:
            The parser instance for the specified format

        Raises:
            ValueError: If an unsupported format type is provided
        """
        if format_type not in self._parsers:
            if format_type == FormatType.CDA:
                parser = CDAParser(self.config)
                self._parsers[format_type] = parser
            elif format_type == FormatType.HL7V2:
                raise NotImplementedError("HL7v2 parser not implemented")
            else:
                raise ValueError(f"Unsupported parser format: {format_type}")

        return self._parsers[format_type]

    def _get_generator(self, format_type: FormatType):
        """Get or create a generator for the specified format

        Args:
            format_type: The format type to get a generator for (CDA, HL7v2, or FHIR)

        Returns:
            The generator instance for the specified format

        Raises:
            ValueError: If an unsupported format type is provided
        """
        if format_type not in self._generators:
            if format_type == FormatType.CDA:
                generator = CDAGenerator(self.config, self.template_registry)
                self._generators[format_type] = generator
            elif format_type == FormatType.HL7V2:
                raise NotImplementedError("HL7v2 generator not implemented")
            elif format_type == FormatType.FHIR:
                generator = FHIRGenerator(self.config, self.template_registry)
                self._generators[format_type] = generator
            else:
                raise ValueError(f"Unsupported generator format: {format_type}")

        return self._generators[format_type]

    def register_parser(self, format_type: FormatType, parser_instance):
        """Register a custom parser for a format type. This will replace the default parser for the format type.

        Args:
            format_type: The format type (CDA, HL7v2) to register the parser for
            parser_instance: The parser instance that implements the parsing logic

        Returns:
            InteropEngine: Returns self for method chaining

        Example:
            engine.register_parser(FormatType.CDA, CustomCDAParser())
        """
        self._parsers[format_type] = parser_instance
        return self

    def register_generator(self, format_type: FormatType, generator_instance):
        """Register a custom generator for a format type. This will replace the default generator for the format type.

        Args:
            format_type: The format type (CDA, HL7v2, FHIR) to register the generator for
            generator_instance: The generator instance that implements the generation logic

        Returns:
            InteropEngine: Returns self for method chaining

        Example:
            engine.register_generator(FormatType.CDA, CustomCDAGenerator())
        """
        self._generators[format_type] = generator_instance
        return self

    # TODO: make the config validator functions more generic
    def register_cda_section_config_validator(
        self, resource_type: str, template_model: BaseModel
    ) -> "InteropEngine":
        """Register a custom section config validator model for a resource type

        Args:
            resource_type: FHIR resource type (e.g., "Condition", "MedicationStatement") which converts to the CDA section
            template_model: Pydantic model for CDA section config validation

        Returns:
            Self for method chaining

        Example:
            # Register a config validator for the Problem section, which is converted from the Condition resource
            engine.register_cda_section_config_validator(
                "Condition", ProblemSectionConfig
            )
        """
        self.config.register_cda_section_config(resource_type, template_model)
        return self

    def register_cda_document_config_validator(
        self, document_type: str, document_model: BaseModel
    ) -> "InteropEngine":
        """Register a custom document validator model for a document type

        Args:
            document_type: Document type (e.g., "ccd", "discharge")
            document_model: Pydantic model for document validation

        Returns:
            Self for method chaining

        Example:
            # Register a config validator for the CCD document type
            engine.register_cda_document_validator(
                "ccd", CCDDocumentConfig
            )
        """
        self.config.register_cda_document_config(document_type, document_model)
        return self

    def to_fhir(
        self, src_data: str, src_format: Union[str, FormatType]
    ) -> List[Resource]:
        """Convert source data to FHIR resources

        Args:
            src_data: Input data as string (CDA XML or HL7v2 message)
            src_format: Source format type, either as string ("cda", "hl7v2")
                         or FormatType enum

        Returns:
            List[Resource]: List of FHIR resources generated from the source data

        Raises:
            ValueError: If src_format is not supported

        Example:
            # Convert CDA XML to FHIR resources
            fhir_resources = engine.to_fhir(cda_xml, src_format="cda")
        """
        src_format = validate_format(src_format)

        if src_format == FormatType.CDA:
            return self._cda_to_fhir(src_data)
        elif src_format == FormatType.HL7V2:
            return self._hl7v2_to_fhir(src_data)
        else:
            raise ValueError(f"Unsupported format: {src_format}")

    def from_fhir(
        self,
        resources: Union[List[Resource], Bundle],
        dest_format: Union[str, FormatType],
        **kwargs,
    ) -> str:
        """Convert FHIR resources to a target format

        Args:
            resources: List of FHIR resources to convert or a FHIR Bundle
            dest_format: Destination format type, either as string ("cda", "hl7v2")
                        or FormatType enum
            **kwargs: Additional arguments to pass to generator.
                     For CDA: document_type (str) - Type of CDA document (e.g. "ccd", "discharge")

        Returns:
            str: Converted data as string (CDA XML or HL7v2 message)

        Raises:
            ValueError: If dest_format is not supported

        Example:
            # Convert FHIR resources to CDA XML
            cda_xml = engine.from_fhir(fhir_resources, dest_format="cda")
        """
        dest_format = validate_format(dest_format)
        resources = normalize_resource_list(resources)

        if dest_format == FormatType.HL7V2:
            return self._fhir_to_hl7v2(resources, **kwargs)
        elif dest_format == FormatType.CDA:
            return self._fhir_to_cda(resources, **kwargs)
        else:
            raise ValueError(f"Unsupported format: {dest_format}")

    def _cda_to_fhir(self, xml: str, **kwargs) -> List[Resource]:
        """Convert CDA XML to FHIR resources

        Args:
            xml: CDA document as XML string
            **kwargs: Additional arguments to pass to parser and generator.

        Returns:
            List[Resource]: List of FHIR resources

        Raises:
            ValueError: If required mappings are missing or if sections are unsupported
        """
        # Get parser and generator (lazy loaded)
        parser = self.cda_parser
        generator = self.fhir_generator

        # Parse sections from CDA XML using the parser
        section_entries = parser.from_string(xml)

        # Process each section and convert entries to FHIR resources
        resources = []
        for section_key, entries in section_entries.items():
            section_resources = generator.transform(
                entries, src_format=FormatType.CDA, section_key=section_key
            )
            resources.extend(section_resources)

        return resources

    def _fhir_to_cda(self, resources: List[Resource], **kwargs) -> str:
        """Convert FHIR resources to CDA XML

        Args:
            resources: A list of FHIR resources
            **kwargs: Additional arguments to pass to generator.
                     Supported arguments:
                     - document_type: Type of CDA document (e.g. "CCD", "Discharge Summary")

        Returns:
            str: CDA document as XML string

        Raises:
            ValueError: If required mappings are missing or if resource types are unsupported
        """
        # Get generators (lazy loaded)
        cda_generator = self.cda_generator

        # Check for document type
        document_type = kwargs.get("document_type", "ccd")
        if document_type:
            log.info(f"Processing CDA document of type: {document_type}")

        # Get document configuration for this specific document type
        doc_config = self.config.get_cda_document_config(document_type)
        if not doc_config:
            raise ValueError(
                f"Invalid or missing document configuration for type: {document_type}"
            )

        return cda_generator.transform(resources, document_type=document_type)

    def _hl7v2_to_fhir(self, source_data: str) -> List[Resource]:
        """Convert HL7v2 to FHIR resources"""
        parser = self.hl7v2_parser
        generator = self.fhir_generator

        # Parse HL7v2 message using the parser
        message_entries = parser.from_string(source_data)

        # Process each message entry and convert to FHIR resources
        resources = []
        for message_key, entries in message_entries.items():
            resource_entries = generator.transform(
                entries, src_format=FormatType.HL7V2, message_key=message_key
            )
            resources.extend(resource_entries)

        return resources

    def _fhir_to_hl7v2(self, resources: List[Resource]) -> str:
        """Convert FHIR resources to HL7v2"""
        generator = self.hl7v2_generator

        # Process each resource and convert to HL7v2 message
        messages = []
        for resource in resources:
            message = generator.transform(resource)
            messages.append(message)

        return messages

cda_generator cached property

Lazily load the CDA generator

cda_parser cached property

Lazily load the CDA parser

fhir_generator cached property

Lazily load the FHIR generator

hl7v2_generator cached property

Lazily load the HL7v2 generator

hl7v2_parser cached property

Lazily load the HL7v2 parser

__init__(config_dir=None, validation_level=ValidationLevel.STRICT, environment=None)

Initialize the InteropEngine

PARAMETER DESCRIPTION
config_dir

Base directory containing configuration files. If None, will search standard locations.

TYPE: Optional[Path] DEFAULT: None

validation_level

Level of configuration validation (strict, warn, ignore)

TYPE: str DEFAULT: STRICT

environment

Optional environment to use (development, testing, production)

TYPE: Optional[str] DEFAULT: None

Source code in healthchain/interop/engine.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def __init__(
    self,
    config_dir: Optional[Path] = None,
    validation_level: str = ValidationLevel.STRICT,
    environment: Optional[str] = None,
):
    """Initialize the InteropEngine

    Args:
        config_dir: Base directory containing configuration files. If None, will search standard locations.
        validation_level: Level of configuration validation (strict, warn, ignore)
        environment: Optional environment to use (development, testing, production)
    """
    # Initialize configuration manager
    self.config = InteropConfigManager(config_dir, validation_level, environment)

    # Initialize template registry
    template_dir = config_dir / "templates"
    self.template_registry = TemplateRegistry(template_dir)

    # Create and register default filters
    # Get required configuration for filters
    mappings_dir = self.config.get_config_value("defaults.mappings_dir")
    if not mappings_dir:
        log.warning("No mappings directory configured, using default mappings")
        mappings_dir = "cda_default"
    mappings = self.config.get_mappings(mappings_dir)
    id_prefix = self.config.get_config_value("defaults.common.id_prefix")

    # Get default filters from the filters module
    default_filters = create_default_filters(mappings, id_prefix)
    self.template_registry.initialize(default_filters)

    # Component registries for lazy loading
    self._parsers = {}
    self._generators = {}

from_fhir(resources, dest_format, **kwargs)

Convert FHIR resources to a target format

PARAMETER DESCRIPTION
resources

List of FHIR resources to convert or a FHIR Bundle

TYPE: Union[List[Resource], Bundle]

dest_format

Destination format type, either as string ("cda", "hl7v2") or FormatType enum

TYPE: Union[str, FormatType]

**kwargs

Additional arguments to pass to generator. For CDA: document_type (str) - Type of CDA document (e.g. "ccd", "discharge")

DEFAULT: {}

RETURNS DESCRIPTION
str

Converted data as string (CDA XML or HL7v2 message)

TYPE: str

RAISES DESCRIPTION
ValueError

If dest_format is not supported
Example

Convert FHIR resources to CDA XML

cda_xml = engine.from_fhir(fhir_resources, dest_format="cda")

Source code in healthchain/interop/engine.py 
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
def from_fhir(
    self,
    resources: Union[List[Resource], Bundle],
    dest_format: Union[str, FormatType],
    **kwargs,
) -> str:
    """Convert FHIR resources to a target format

    Args:
        resources: List of FHIR resources to convert or a FHIR Bundle
        dest_format: Destination format type, either as string ("cda", "hl7v2")
                    or FormatType enum
        **kwargs: Additional arguments to pass to generator.
                 For CDA: document_type (str) - Type of CDA document (e.g. "ccd", "discharge")

    Returns:
        str: Converted data as string (CDA XML or HL7v2 message)

    Raises:
        ValueError: If dest_format is not supported

    Example:
        # Convert FHIR resources to CDA XML
        cda_xml = engine.from_fhir(fhir_resources, dest_format="cda")
    """
    dest_format = validate_format(dest_format)
    resources = normalize_resource_list(resources)

    if dest_format == FormatType.HL7V2:
        return self._fhir_to_hl7v2(resources, **kwargs)
    elif dest_format == FormatType.CDA:
        return self._fhir_to_cda(resources, **kwargs)
    else:
        raise ValueError(f"Unsupported format: {dest_format}")

register_cda_document_config_validator(document_type, document_model)

Register a custom document validator model for a document type

PARAMETER DESCRIPTION
document_type

Document type (e.g., "ccd", "discharge")

TYPE: str

document_model

Pydantic model for document validation

TYPE: BaseModel

RETURNS DESCRIPTION
InteropEngine

Self for method chaining
Example

Register a config validator for the CCD document type

engine.register_cda_document_validator( "ccd", CCDDocumentConfig )

Source code in healthchain/interop/engine.py 
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
def register_cda_document_config_validator(
    self, document_type: str, document_model: BaseModel
) -> "InteropEngine":
    """Register a custom document validator model for a document type

    Args:
        document_type: Document type (e.g., "ccd", "discharge")
        document_model: Pydantic model for document validation

    Returns:
        Self for method chaining

    Example:
        # Register a config validator for the CCD document type
        engine.register_cda_document_validator(
            "ccd", CCDDocumentConfig
        )
    """
    self.config.register_cda_document_config(document_type, document_model)
    return self

register_cda_section_config_validator(resource_type, template_model)

Register a custom section config validator model for a resource type

PARAMETER DESCRIPTION
resource_type

FHIR resource type (e.g., "Condition", "MedicationStatement") which converts to the CDA section

TYPE: str

template_model

Pydantic model for CDA section config validation

TYPE: BaseModel

RETURNS DESCRIPTION
InteropEngine

Self for method chaining
Example

Register a config validator for the Problem section, which is converted from the Condition resource

engine.register_cda_section_config_validator( "Condition", ProblemSectionConfig )

Source code in healthchain/interop/engine.py 
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
def register_cda_section_config_validator(
    self, resource_type: str, template_model: BaseModel
) -> "InteropEngine":
    """Register a custom section config validator model for a resource type

    Args:
        resource_type: FHIR resource type (e.g., "Condition", "MedicationStatement") which converts to the CDA section
        template_model: Pydantic model for CDA section config validation

    Returns:
        Self for method chaining

    Example:
        # Register a config validator for the Problem section, which is converted from the Condition resource
        engine.register_cda_section_config_validator(
            "Condition", ProblemSectionConfig
        )
    """
    self.config.register_cda_section_config(resource_type, template_model)
    return self

register_generator(format_type, generator_instance)

Register a custom generator for a format type. This will replace the default generator for the format type.

PARAMETER DESCRIPTION
format_type

The format type (CDA, HL7v2, FHIR) to register the generator for

TYPE: FormatType

generator_instance

The generator instance that implements the generation logic
RETURNS DESCRIPTION
InteropEngine

Returns self for method chaining
Example

engine.register_generator(FormatType.CDA, CustomCDAGenerator())

Source code in healthchain/interop/engine.py 
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
def register_generator(self, format_type: FormatType, generator_instance):
    """Register a custom generator for a format type. This will replace the default generator for the format type.

    Args:
        format_type: The format type (CDA, HL7v2, FHIR) to register the generator for
        generator_instance: The generator instance that implements the generation logic

    Returns:
        InteropEngine: Returns self for method chaining

    Example:
        engine.register_generator(FormatType.CDA, CustomCDAGenerator())
    """
    self._generators[format_type] = generator_instance
    return self

register_parser(format_type, parser_instance)

Register a custom parser for a format type. This will replace the default parser for the format type.

PARAMETER DESCRIPTION
format_type

The format type (CDA, HL7v2) to register the parser for

TYPE: FormatType

parser_instance

The parser instance that implements the parsing logic
RETURNS DESCRIPTION
InteropEngine

Returns self for method chaining
Example

engine.register_parser(FormatType.CDA, CustomCDAParser())

Source code in healthchain/interop/engine.py 
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
def register_parser(self, format_type: FormatType, parser_instance):
    """Register a custom parser for a format type. This will replace the default parser for the format type.

    Args:
        format_type: The format type (CDA, HL7v2) to register the parser for
        parser_instance: The parser instance that implements the parsing logic

    Returns:
        InteropEngine: Returns self for method chaining

    Example:
        engine.register_parser(FormatType.CDA, CustomCDAParser())
    """
    self._parsers[format_type] = parser_instance
    return self

to_fhir(src_data, src_format)

Convert source data to FHIR resources

PARAMETER DESCRIPTION
src_data

Input data as string (CDA XML or HL7v2 message)

TYPE: str

src_format

Source format type, either as string ("cda", "hl7v2") or FormatType enum

TYPE: Union[str, FormatType]

RETURNS DESCRIPTION
List[Resource]

List[Resource]: List of FHIR resources generated from the source data
RAISES DESCRIPTION
ValueError

If src_format is not supported
Example

Convert CDA XML to FHIR resources

fhir_resources = engine.to_fhir(cda_xml, src_format="cda")

Source code in healthchain/interop/engine.py 
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
def to_fhir(
    self, src_data: str, src_format: Union[str, FormatType]
) -> List[Resource]:
    """Convert source data to FHIR resources

    Args:
        src_data: Input data as string (CDA XML or HL7v2 message)
        src_format: Source format type, either as string ("cda", "hl7v2")
                     or FormatType enum

    Returns:
        List[Resource]: List of FHIR resources generated from the source data

    Raises:
        ValueError: If src_format is not supported

    Example:
        # Convert CDA XML to FHIR resources
        fhir_resources = engine.to_fhir(cda_xml, src_format="cda")
    """
    src_format = validate_format(src_format)

    if src_format == FormatType.CDA:
        return self._cda_to_fhir(src_data)
    elif src_format == FormatType.HL7V2:
        return self._hl7v2_to_fhir(src_data)
    else:
        raise ValueError(f"Unsupported format: {src_format}")

normalize_resource_list(resources)

Convert input resources to a normalized list format

Source code in healthchain/interop/engine.py 
24
25
26
27
28
29
30
31
32
33
def normalize_resource_list(
    resources: Union[Resource, List[Resource], Bundle],
) -> List[Resource]:
    """Convert input resources to a normalized list format"""
    if isinstance(resources, Bundle):
        return [entry.resource for entry in resources.entry if entry.resource]
    elif isinstance(resources, list):
        return resources
    else:
        return [resources]

InteropConfigManager for HealthChain Interoperability Engine

This module provides specialized configuration management for interoperability.

InteropConfigManager

Bases: ConfigManager

Specialized configuration manager for the interoperability module

Extends ConfigManager to handle CDA document and section template configurations. Provides functionality for:

  • Loading and validating interop configurations
  • Managing document and section templates
  • Registering custom validation models

Configuration structure: - Document templates (under "document") - Section templates (under "sections") - Default values and settings

Validation levels: - STRICT: Full validation (default) - WARN: Warning-only - IGNORE: No validation

Source code in healthchain/interop/config_manager.py 
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
class InteropConfigManager(ConfigManager):
    """Specialized configuration manager for the interoperability module

    Extends ConfigManager to handle CDA document and section template configurations.
    Provides functionality for:

    - Loading and validating interop configurations
    - Managing document and section templates
    - Registering custom validation models

    Configuration structure:
    - Document templates (under "document")
    - Section templates (under "sections")
    - Default values and settings

    Validation levels:
    - STRICT: Full validation (default)
    - WARN: Warning-only
    - IGNORE: No validation
    """

    def __init__(
        self,
        config_dir,
        validation_level: str = ValidationLevel.STRICT,
        environment: Optional[str] = None,
    ):
        """Initialize the InteropConfigManager.

        Initializes the configuration manager with the interop module and validates
        the configuration. The interop module configuration must exist in the
        specified config directory.

        Args:
            config_dir: Base directory containing configuration files
            validation_level: Level of validation to perform. Default is STRICT.
                Can be STRICT, WARN, or IGNORE.
            environment: Optional environment name to load environment-specific configs.
                If provided, will load and merge environment-specific configuration.

        Raises:
            ValueError: If the interop module configuration is not found in config_dir.
        """
        # Initialize with "interop" as the fixed module
        super().__init__(config_dir, validation_level, module="interop")
        self.load(environment, skip_validation=True)

        if "interop" not in self._module_configs:
            raise ValueError(
                f"Interop module not found in configuration directory {config_dir}"
            )

        self.validate()

    def _find_cda_document_types(self) -> List[str]:
        """Find available CDA document types in the configs

        Returns:
            List of CDA document type strings
        """
        # Get document types from cda/document path
        doc_section = self._find_config_section(
            module_name="interop", section_path="cda/document"
        )

        # If no document section exists, return empty list
        if not doc_section:
            return []

        # Return the keys from the document section
        return list(doc_section.keys())

    def get_cda_section_configs(self, section_key: Optional[str] = None) -> Dict:
        """Get CDA section configuration(s).

        Retrieves section configurations from the loaded configs. When section_key is provided,
        retrieves configuration for a specific section; otherwise, returns all section configurations.
        Section configurations define how different CDA sections should be processed and mapped to
        FHIR resources.

        Args:
            section_key: Optional section identifier (e.g., "problems", "medications").
                         If provided, returns only that specific section's configuration.

        Returns:
            Dict: Dictionary mapping section keys to their configurations if section_key is None.
                  Single section configuration dict if section_key is provided.

        Raises:
            ValueError: If section_key is provided but not found in configurations
                       or if no sections are configured
        """
        # Get all sections
        sections = self._find_config_section(
            module_name="interop", section_path="cda/sections"
        )

        if not sections:
            raise ValueError("No CDA section configurations found")

        # If section_key is provided, return just that section
        if section_key is not None:
            if section_key not in sections:
                raise ValueError(f"Section configuration not found: {section_key}")

            # Basic validation that required fields exist
            section_config = sections[section_key]
            if "resource" not in section_config:
                raise ValueError(
                    f"Invalid section configuration for {section_key}: missing 'resource' field"
                )

            return section_config

        return sections

    def get_cda_document_config(self, document_type: str) -> Dict:
        """Get CDA document configuration for a specific document type.

        Retrieves the configuration for a CDA document type from the loaded configs.
        The configuration contains template settings and other document-specific parameters.

        Args:
            document_type: Type of document (e.g., "ccd", "discharge") to get config for

        Returns:
            Dict containing the document configuration

        Raises:
            ValueError: If document_type is not found or the configuration is invalid
        """
        document_config = self._find_config_section(
            module_name="interop", section_path=f"cda/document/{document_type}"
        )

        if not document_config:
            raise ValueError(
                f"Document configuration not found for type: {document_type}"
            )

        # Basic validation that required sections exist
        if "templates" not in document_config:
            raise ValueError(
                f"Invalid document configuration for {document_type}: missing 'templates' section"
            )

        # Return the validated config
        return document_config

    def validate(self) -> bool:
        """Validate that all required configurations are present for the interop module.

        Validates both section and document configurations according to their registered
        validation models. Section configs are required and will cause validation to fail
        if missing or invalid. Document configs are optional but will be validated if present.

        The validation behavior depends on the validation_level setting:
        - IGNORE: Always returns True without validating
        - WARN: Logs warnings for validation failures but returns True
        - ERROR: Returns False if any validation fails

        Returns:
            bool: True if validation passes or is ignored, False if validation fails
                 when validation_level is ERROR
        """
        if self._validation_level == ValidationLevel.IGNORE:
            return True

        is_valid = super().validate()

        # Validate section configs
        try:
            section_configs = self._find_config_section(
                module_name="interop", section_path="cda/sections"
            )
            if not section_configs:
                is_valid = self._handle_validation_error("No section configs found")
            else:
                # Validate each section config
                for section_key, section_config in section_configs.items():
                    result = validate_cda_section_config_model(
                        section_key, section_config
                    )
                    if not result:
                        is_valid = self._handle_validation_error(
                            f"Section config validation failed for key: {section_key}"
                        )
        except Exception as e:
            is_valid = self._handle_validation_error(
                f"Error validating section configs: {str(e)}"
            )

        # Validate document configs - but don't fail if no documents are configured
        # since some use cases might not require documents
        document_types = self._find_cda_document_types()
        for doc_type in document_types:
            try:
                doc_config = self._find_config_section(
                    module_name="interop", section_path=f"cda/document/{doc_type}"
                )
                if doc_config:
                    result = validate_cda_document_config_model(doc_type, doc_config)
                    if not result:
                        is_valid = self._handle_validation_error(
                            f"Document config validation failed for type: {doc_type}"
                        )
            except Exception as e:
                is_valid = self._handle_validation_error(
                    f"Error validating document config for {doc_type}: {str(e)}"
                )

        return is_valid

    def register_cda_section_config(
        self, resource_type: str, config_model: Type[BaseModel]
    ) -> None:
        """Register a validation model for a CDA section configuration.

        Registers a Pydantic model that will be used to validate configuration for a CDA section
        that maps to a specific FHIR resource type. The model defines the required and optional
        fields that should be present in the section configuration.

        Args:
            resource_type: FHIR resource type that the section maps to (e.g. "Condition")
            config_model: Pydantic model class that defines the validation schema for the section config
        """
        register_cda_section_template_config_model(resource_type, config_model)

    def register_cda_document_config(
        self, document_type: str, config_model: Type[BaseModel]
    ) -> None:
        """Register a validation model for a CDA document configuration.

        Registers a Pydantic model that will be used to validate configuration for a CDA document
        type. The model defines the required and optional fields that should be present in the
        document configuration.

        Args:
            document_type: Document type identifier (e.g., "ccd", "discharge")
            config_model: Pydantic model class that defines the validation schema for the document config
        """
        register_cda_document_template_config_model(document_type, config_model)

__init__(config_dir, validation_level=ValidationLevel.STRICT, environment=None)

Initialize the InteropConfigManager.

Initializes the configuration manager with the interop module and validates the configuration. The interop module configuration must exist in the specified config directory.

PARAMETER DESCRIPTION
config_dir

Base directory containing configuration files
validation_level

Level of validation to perform. Default is STRICT. Can be STRICT, WARN, or IGNORE.

TYPE: str DEFAULT: STRICT

environment

Optional environment name to load environment-specific configs. If provided, will load and merge environment-specific configuration.

TYPE: Optional[str] DEFAULT: None

RAISES DESCRIPTION
ValueError

If the interop module configuration is not found in config_dir.
Source code in healthchain/interop/config_manager.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def __init__(
    self,
    config_dir,
    validation_level: str = ValidationLevel.STRICT,
    environment: Optional[str] = None,
):
    """Initialize the InteropConfigManager.

    Initializes the configuration manager with the interop module and validates
    the configuration. The interop module configuration must exist in the
    specified config directory.

    Args:
        config_dir: Base directory containing configuration files
        validation_level: Level of validation to perform. Default is STRICT.
            Can be STRICT, WARN, or IGNORE.
        environment: Optional environment name to load environment-specific configs.
            If provided, will load and merge environment-specific configuration.

    Raises:
        ValueError: If the interop module configuration is not found in config_dir.
    """
    # Initialize with "interop" as the fixed module
    super().__init__(config_dir, validation_level, module="interop")
    self.load(environment, skip_validation=True)

    if "interop" not in self._module_configs:
        raise ValueError(
            f"Interop module not found in configuration directory {config_dir}"
        )

    self.validate()

get_cda_document_config(document_type)

Get CDA document configuration for a specific document type.

Retrieves the configuration for a CDA document type from the loaded configs. The configuration contains template settings and other document-specific parameters.

PARAMETER DESCRIPTION
document_type

Type of document (e.g., "ccd", "discharge") to get config for

TYPE: str

RETURNS DESCRIPTION
Dict

Dict containing the document configuration
RAISES DESCRIPTION
ValueError

If document_type is not found or the configuration is invalid
Source code in healthchain/interop/config_manager.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def get_cda_document_config(self, document_type: str) -> Dict:
    """Get CDA document configuration for a specific document type.

    Retrieves the configuration for a CDA document type from the loaded configs.
    The configuration contains template settings and other document-specific parameters.

    Args:
        document_type: Type of document (e.g., "ccd", "discharge") to get config for

    Returns:
        Dict containing the document configuration

    Raises:
        ValueError: If document_type is not found or the configuration is invalid
    """
    document_config = self._find_config_section(
        module_name="interop", section_path=f"cda/document/{document_type}"
    )

    if not document_config:
        raise ValueError(
            f"Document configuration not found for type: {document_type}"
        )

    # Basic validation that required sections exist
    if "templates" not in document_config:
        raise ValueError(
            f"Invalid document configuration for {document_type}: missing 'templates' section"
        )

    # Return the validated config
    return document_config

get_cda_section_configs(section_key=None)

Get CDA section configuration(s).

Retrieves section configurations from the loaded configs. When section_key is provided, retrieves configuration for a specific section; otherwise, returns all section configurations. Section configurations define how different CDA sections should be processed and mapped to FHIR resources.

PARAMETER DESCRIPTION
section_key

Optional section identifier (e.g., "problems", "medications"). If provided, returns only that specific section's configuration.

TYPE: Optional[str] DEFAULT: None

RETURNS DESCRIPTION
Dict

Dictionary mapping section keys to their configurations if section_key is None. Single section configuration dict if section_key is provided.

TYPE: Dict

RAISES DESCRIPTION
ValueError

If section_key is provided but not found in configurations or if no sections are configured
Source code in healthchain/interop/config_manager.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def get_cda_section_configs(self, section_key: Optional[str] = None) -> Dict:
    """Get CDA section configuration(s).

    Retrieves section configurations from the loaded configs. When section_key is provided,
    retrieves configuration for a specific section; otherwise, returns all section configurations.
    Section configurations define how different CDA sections should be processed and mapped to
    FHIR resources.

    Args:
        section_key: Optional section identifier (e.g., "problems", "medications").
                     If provided, returns only that specific section's configuration.

    Returns:
        Dict: Dictionary mapping section keys to their configurations if section_key is None.
              Single section configuration dict if section_key is provided.

    Raises:
        ValueError: If section_key is provided but not found in configurations
                   or if no sections are configured
    """
    # Get all sections
    sections = self._find_config_section(
        module_name="interop", section_path="cda/sections"
    )

    if not sections:
        raise ValueError("No CDA section configurations found")

    # If section_key is provided, return just that section
    if section_key is not None:
        if section_key not in sections:
            raise ValueError(f"Section configuration not found: {section_key}")

        # Basic validation that required fields exist
        section_config = sections[section_key]
        if "resource" not in section_config:
            raise ValueError(
                f"Invalid section configuration for {section_key}: missing 'resource' field"
            )

        return section_config

    return sections

register_cda_document_config(document_type, config_model)

Register a validation model for a CDA document configuration.

Registers a Pydantic model that will be used to validate configuration for a CDA document type. The model defines the required and optional fields that should be present in the document configuration.

PARAMETER DESCRIPTION
document_type

Document type identifier (e.g., "ccd", "discharge")

TYPE: str

config_model

Pydantic model class that defines the validation schema for the document config

TYPE: Type[BaseModel]

Source code in healthchain/interop/config_manager.py 
251
252
253
254
255
256
257
258
259
260
261
262
263
264
def register_cda_document_config(
    self, document_type: str, config_model: Type[BaseModel]
) -> None:
    """Register a validation model for a CDA document configuration.

    Registers a Pydantic model that will be used to validate configuration for a CDA document
    type. The model defines the required and optional fields that should be present in the
    document configuration.

    Args:
        document_type: Document type identifier (e.g., "ccd", "discharge")
        config_model: Pydantic model class that defines the validation schema for the document config
    """
    register_cda_document_template_config_model(document_type, config_model)

register_cda_section_config(resource_type, config_model)

Register a validation model for a CDA section configuration.

Registers a Pydantic model that will be used to validate configuration for a CDA section that maps to a specific FHIR resource type. The model defines the required and optional fields that should be present in the section configuration.

PARAMETER DESCRIPTION
resource_type

FHIR resource type that the section maps to (e.g. "Condition")

TYPE: str

config_model

Pydantic model class that defines the validation schema for the section config

TYPE: Type[BaseModel]

Source code in healthchain/interop/config_manager.py 
236
237
238
239
240
241
242
243
244
245
246
247
248
249
def register_cda_section_config(
    self, resource_type: str, config_model: Type[BaseModel]
) -> None:
    """Register a validation model for a CDA section configuration.

    Registers a Pydantic model that will be used to validate configuration for a CDA section
    that maps to a specific FHIR resource type. The model defines the required and optional
    fields that should be present in the section configuration.

    Args:
        resource_type: FHIR resource type that the section maps to (e.g. "Condition")
        config_model: Pydantic model class that defines the validation schema for the section config
    """
    register_cda_section_template_config_model(resource_type, config_model)

validate()

Validate that all required configurations are present for the interop module.

Validates both section and document configurations according to their registered validation models. Section configs are required and will cause validation to fail if missing or invalid. Document configs are optional but will be validated if present.

The validation behavior depends on the validation_level setting: - IGNORE: Always returns True without validating - WARN: Logs warnings for validation failures but returns True - ERROR: Returns False if any validation fails

RETURNS DESCRIPTION
bool

True if validation passes or is ignored, False if validation fails when validation_level is ERROR

TYPE: bool

Source code in healthchain/interop/config_manager.py 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
def validate(self) -> bool:
    """Validate that all required configurations are present for the interop module.

    Validates both section and document configurations according to their registered
    validation models. Section configs are required and will cause validation to fail
    if missing or invalid. Document configs are optional but will be validated if present.

    The validation behavior depends on the validation_level setting:
    - IGNORE: Always returns True without validating
    - WARN: Logs warnings for validation failures but returns True
    - ERROR: Returns False if any validation fails

    Returns:
        bool: True if validation passes or is ignored, False if validation fails
             when validation_level is ERROR
    """
    if self._validation_level == ValidationLevel.IGNORE:
        return True

    is_valid = super().validate()

    # Validate section configs
    try:
        section_configs = self._find_config_section(
            module_name="interop", section_path="cda/sections"
        )
        if not section_configs:
            is_valid = self._handle_validation_error("No section configs found")
        else:
            # Validate each section config
            for section_key, section_config in section_configs.items():
                result = validate_cda_section_config_model(
                    section_key, section_config
                )
                if not result:
                    is_valid = self._handle_validation_error(
                        f"Section config validation failed for key: {section_key}"
                    )
    except Exception as e:
        is_valid = self._handle_validation_error(
            f"Error validating section configs: {str(e)}"
        )

    # Validate document configs - but don't fail if no documents are configured
    # since some use cases might not require documents
    document_types = self._find_cda_document_types()
    for doc_type in document_types:
        try:
            doc_config = self._find_config_section(
                module_name="interop", section_path=f"cda/document/{doc_type}"
            )
            if doc_config:
                result = validate_cda_document_config_model(doc_type, doc_config)
                if not result:
                    is_valid = self._handle_validation_error(
                        f"Document config validation failed for type: {doc_type}"
                    )
        except Exception as e:
            is_valid = self._handle_validation_error(
                f"Error validating document config for {doc_type}: {str(e)}"
            )

    return is_valid

TemplateRegistry

Manages loading and accessing Liquid templates for the InteropEngine.

The TemplateRegistry handles loading Liquid template files from a directory and making them available for rendering. It supports custom filter functions that can be used within templates.

Key features: - Loads .liquid template files recursively from a directory - Supports adding custom filter functions - Provides template lookup by name - Validates template existence

Example

registry = TemplateRegistry(Path("templates")) registry.initialize({ "uppercase": str.upper, "lowercase": str.lower }) template = registry.get_template("cda_fhir/condition")

Source code in healthchain/interop/template_registry.py 
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
class TemplateRegistry:
    """Manages loading and accessing Liquid templates for the InteropEngine.

    The TemplateRegistry handles loading Liquid template files from a directory and making them
    available for rendering. It supports custom filter functions that can be used within templates.

    Key features:
    - Loads .liquid template files recursively from a directory
    - Supports adding custom filter functions
    - Provides template lookup by name
    - Validates template existence

    Example:
        registry = TemplateRegistry(Path("templates"))
        registry.initialize({
            "uppercase": str.upper,
            "lowercase": str.lower
        })
        template = registry.get_template("cda_fhir/condition")
    """

    def __init__(self, template_dir: Path):
        """Initialize the TemplateRegistry

        Args:
            template_dir: Directory containing template files
        """
        self.template_dir = template_dir
        self._templates = {}
        self._env = None
        self._filters = {}

        if not template_dir.exists():
            raise ValueError(f"Template directory not found: {template_dir}")

    def initialize(self, filters: Dict[str, Callable] = None) -> "TemplateRegistry":
        """Initialize the Liquid environment and load templates.

        This method sets up the Liquid template environment by:
        1. Storing any provided filter functions
        2. Creating the Liquid environment with the template directory
        3. Loading all template files from the directory

        The environment must be initialized before templates can be loaded or rendered.

        Args:
            filters: Optional dictionary mapping filter names to filter functions that can be used
                    in templates. For example: {"uppercase": str.upper}

        Returns:
            TemplateRegistry: Returns self for method chaining

        Raises:
            ValueError: If template directory does not exist or environment initialization fails
        """
        # Store initial filters
        if filters:
            self._filters.update(filters)

        self._create_environment()
        self._load_templates()
        return self

    def _create_environment(self) -> None:
        """Create and configure the Liquid environment with registered filters"""
        self._env = Environment(loader=FileSystemLoader(str(self.template_dir)))

        # Register all filters
        for name, func in self._filters.items():
            self._env.filters[name] = func

    def add_filter(self, name: str, filter_func: Callable) -> "TemplateRegistry":
        """Add a custom filter function

        Args:
            name: Name of the filter to use in templates
            filter_func: Filter function to register

        Returns:
            Self for method chaining
        """
        # Add to internal filter registry
        self._filters[name] = filter_func

        # If environment is already initialized, register the filter
        if self._env:
            self._env.filters[name] = filter_func

        return self

    def add_filters(self, filters: Dict[str, Callable]) -> "TemplateRegistry":
        """Add multiple custom filter functions

        Args:
            filters: Dictionary of filter names to filter functions

        Returns:
            Self for method chaining
        """
        for name, func in filters.items():
            self.add_filter(name, func)

        return self

    def _load_templates(self) -> None:
        """Load all Liquid template files from the template directory.

        This method recursively walks through the template directory and its subdirectories
        to find all .liquid template files. Each template is loaded into the environment
        and stored in the internal template registry using its full relative path (without extension)
        as the key (e.g., "cda_fhir/document"). This is not required but recommended for clarity.
        """
        if not self._env:
            raise ValueError("Environment not initialized. Call initialize() first.")

        # Walk through all subdirectories to find template files
        for template_file in self.template_dir.rglob("*.liquid"):
            rel_path = template_file.relative_to(self.template_dir)
            # Use full path without extension as the key (e.g., "cda_fhir/document")
            template_key = str(rel_path.with_suffix(""))

            try:
                template = self._env.get_template(str(rel_path))
                self._templates[template_key] = template
                log.debug(f"Loaded template: {template_key}")
            except Exception as e:
                log.error(f"Failed to load template {template_file}: {str(e)}")
                continue

        if not self._templates:
            raise ValueError(f"No templates found in {self.template_dir}")

        log.info(f"Loaded {len(self._templates)} templates")

    def get_template(self, template_key: str):
        """Get a template by key

        Args:
            template_key: Template identifier. Can be a full path (e.g., 'cda_fhir/document')
                or just a filename (e.g., 'document').

        Returns:
            The template object

        Raises:
            KeyError: If template not found
        """
        if template_key not in self._templates:
            raise KeyError(f"Template not found: {template_key}")

        return self._templates[template_key]

    def has_template(self, template_key: str) -> bool:
        """Check if a template exists

        Args:
            template_key: Template identifier. Can be a full path (e.g., 'cda_fhir/document')
                or just a filename (e.g., 'document').

        Returns:
            True if template exists, False otherwise
        """
        return template_key in self._templates

__init__(template_dir)

Initialize the TemplateRegistry

PARAMETER DESCRIPTION
template_dir

Directory containing template files

TYPE: Path

Source code in healthchain/interop/template_registry.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
def __init__(self, template_dir: Path):
    """Initialize the TemplateRegistry

    Args:
        template_dir: Directory containing template files
    """
    self.template_dir = template_dir
    self._templates = {}
    self._env = None
    self._filters = {}

    if not template_dir.exists():
        raise ValueError(f"Template directory not found: {template_dir}")

add_filter(name, filter_func)

Add a custom filter function

PARAMETER DESCRIPTION
name

Name of the filter to use in templates

TYPE: str

filter_func

Filter function to register

TYPE: Callable

RETURNS DESCRIPTION
TemplateRegistry

Self for method chaining
Source code in healthchain/interop/template_registry.py 
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def add_filter(self, name: str, filter_func: Callable) -> "TemplateRegistry":
    """Add a custom filter function

    Args:
        name: Name of the filter to use in templates
        filter_func: Filter function to register

    Returns:
        Self for method chaining
    """
    # Add to internal filter registry
    self._filters[name] = filter_func

    # If environment is already initialized, register the filter
    if self._env:
        self._env.filters[name] = filter_func

    return self

add_filters(filters)

Add multiple custom filter functions

PARAMETER DESCRIPTION
filters

Dictionary of filter names to filter functions

TYPE: Dict[str, Callable]

RETURNS DESCRIPTION
TemplateRegistry

Self for method chaining
Source code in healthchain/interop/template_registry.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
def add_filters(self, filters: Dict[str, Callable]) -> "TemplateRegistry":
    """Add multiple custom filter functions

    Args:
        filters: Dictionary of filter names to filter functions

    Returns:
        Self for method chaining
    """
    for name, func in filters.items():
        self.add_filter(name, func)

    return self

get_template(template_key)

Get a template by key

PARAMETER DESCRIPTION
template_key

Template identifier. Can be a full path (e.g., 'cda_fhir/document') or just a filename (e.g., 'document').

TYPE: str

RETURNS DESCRIPTION

The template object
RAISES DESCRIPTION
KeyError

If template not found
Source code in healthchain/interop/template_registry.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
def get_template(self, template_key: str):
    """Get a template by key

    Args:
        template_key: Template identifier. Can be a full path (e.g., 'cda_fhir/document')
            or just a filename (e.g., 'document').

    Returns:
        The template object

    Raises:
        KeyError: If template not found
    """
    if template_key not in self._templates:
        raise KeyError(f"Template not found: {template_key}")

    return self._templates[template_key]

has_template(template_key)

Check if a template exists

PARAMETER DESCRIPTION
template_key

Template identifier. Can be a full path (e.g., 'cda_fhir/document') or just a filename (e.g., 'document').

TYPE: str

RETURNS DESCRIPTION
bool

True if template exists, False otherwise
Source code in healthchain/interop/template_registry.py 
162
163
164
165
166
167
168
169
170
171
172
def has_template(self, template_key: str) -> bool:
    """Check if a template exists

    Args:
        template_key: Template identifier. Can be a full path (e.g., 'cda_fhir/document')
            or just a filename (e.g., 'document').

    Returns:
        True if template exists, False otherwise
    """
    return template_key in self._templates

initialize(filters=None)

Initialize the Liquid environment and load templates.

This method sets up the Liquid template environment by: 1. Storing any provided filter functions 2. Creating the Liquid environment with the template directory 3. Loading all template files from the directory

The environment must be initialized before templates can be loaded or rendered.

PARAMETER DESCRIPTION
filters

Optional dictionary mapping filter names to filter functions that can be used in templates. For example: {"uppercase": str.upper}

TYPE: Dict[str, Callable] DEFAULT: None

RETURNS DESCRIPTION
TemplateRegistry

Returns self for method chaining

TYPE: TemplateRegistry

RAISES DESCRIPTION
ValueError

If template directory does not exist or environment initialization fails
Source code in healthchain/interop/template_registry.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def initialize(self, filters: Dict[str, Callable] = None) -> "TemplateRegistry":
    """Initialize the Liquid environment and load templates.

    This method sets up the Liquid template environment by:
    1. Storing any provided filter functions
    2. Creating the Liquid environment with the template directory
    3. Loading all template files from the directory

    The environment must be initialized before templates can be loaded or rendered.

    Args:
        filters: Optional dictionary mapping filter names to filter functions that can be used
                in templates. For example: {"uppercase": str.upper}

    Returns:
        TemplateRegistry: Returns self for method chaining

    Raises:
        ValueError: If template directory does not exist or environment initialization fails
    """
    # Store initial filters
    if filters:
        self._filters.update(filters)

    self._create_environment()
    self._load_templates()
    return self

CDA Parser for HealthChain Interoperability Engine

This module provides functionality for parsing CDA XML documents.

CDAParser

Bases: BaseParser

Parser for CDA XML documents.

The CDAParser class provides functionality to parse Clinical Document Architecture (CDA) XML documents and extract structured data from their sections. It works in conjunction with the InteropConfigManager to identify and process sections based on configuration.

Key capabilities: - Parse complete CDA XML documents - Extract entries from configured sections based on template IDs or codes - Convert and validate section entries into structured dictionaries (xmltodict)

The parser uses configuration from InteropConfigManager to: - Identify sections by template ID or code - Map section contents to the appropriate data structures - Apply any configured transformations

ATTRIBUTE DESCRIPTION
config

Configuration manager instance

TYPE: InteropConfigManager

clinical_document

Currently loaded CDA document

TYPE: ClinicalDocument

Source code in healthchain/interop/parsers/cda.py 
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
class CDAParser(BaseParser):
    """Parser for CDA XML documents.

    The CDAParser class provides functionality to parse Clinical Document Architecture (CDA)
    XML documents and extract structured data from their sections. It works in conjunction with
    the InteropConfigManager to identify and process sections based on configuration.

    Key capabilities:
    - Parse complete CDA XML documents
    - Extract entries from configured sections based on template IDs or codes
    - Convert and validate section entries into structured dictionaries (xmltodict)

    The parser uses configuration from InteropConfigManager to:
    - Identify sections by template ID or code
    - Map section contents to the appropriate data structures
    - Apply any configured transformations

    Attributes:
        config (InteropConfigManager): Configuration manager instance
        clinical_document (ClinicalDocument): Currently loaded CDA document
    """

    def __init__(self, config: InteropConfigManager):
        """Initialize the CDA parser.

        Args:
            config: InteropConfigManager instance containing section configurations,
                   templates, and mapping rules for CDA document parsing
        """
        super().__init__(config)
        self.clinical_document = None

    def from_string(self, data: str) -> dict:
        """
        Parse input data and convert it to a structured format.

        Args:
            data: The CDA XML document string to parse

        Returns:
            A dictionary containing the parsed data structure with sections
        """
        return self.parse_document(data)

    def parse_document(self, xml: str) -> Dict[str, List[Dict]]:
        """Parse a complete CDA document and extract entries from all configured sections.

        This method parses a CDA XML document and extracts entries from each section that is
        defined in the configuration. It uses xmltodict to parse the XML into a dictionary
        and then processes each configured section to extract its entries.

        Args:
            xml: The CDA XML document string to parse

        Returns:
            Dict[str, List[Dict]]: Dictionary mapping section keys (e.g. "problems",
                "medications") to lists of entry dictionaries containing the parsed data
                from that section (xmltodict format).

        Raises:
            ValueError: If the XML string is empty or invalid
            Exception: If there is an error parsing the document or any section

        Example:
            >>> parser = CDAParser(config)
            >>> sections = parser.from_string(cda_xml)
            >>> problems = sections.get("problems", [])
        """
        section_entries = {}

        # Parse the document once
        try:
            doc_dict = xmltodict.parse(xml)
            self.clinical_document = ClinicalDocument(**doc_dict["ClinicalDocument"])
        except Exception as e:
            log.error(f"Error parsing CDA document: {str(e)}")
            return section_entries

        # Get section configurations
        sections = self.config.get_cda_section_configs()
        if not sections:
            log.warning("No sections found in configuration")
            return section_entries

        # Process each section from the configuration
        for section_key in sections.keys():
            try:
                entries = self._parse_section_entries_from_document(section_key)
                if entries:
                    section_entries[section_key] = entries
            except Exception as e:
                log.error(f"Failed to parse section {section_key}: {str(e)}")
                continue

        return section_entries

    def _parse_section_entries_from_document(self, section_key: str) -> List[Dict]:
        """Extract entries from a CDA section using an already parsed document.

        Args:
            section_key: Key identifying the section in the configuration (e.g. "problems",
                "medications"). Must match a section defined in the configuration.

        Returns:
            List[Dict]: List of entry dictionaries from the matched section. Each dictionary
                contains the parsed data from a single entry in the section. Returns an empty
                list if no entries are found or if an error occurs.

        Raises:
            ValueError: If no template_id or code is configured for the section_key, or if
                no matching section is found in the document.
            Exception: If there is an error parsing the section or its entries.
        """
        entries_dicts = []
        if not self.clinical_document:
            log.error("No document loaded. Call parse_document first.")
            return entries_dicts

        try:
            # Get all components
            components = self.clinical_document.component.structuredBody.component
            if not isinstance(components, list):
                components = [components]

            # Find matching section
            section = None
            for component in components:
                curr_section = component.section

                # Get template_id and code from config_manager
                template_id = self.config.get_config_value(
                    f"cda.sections.{section_key}.identifiers.template_id"
                )
                code = self.config.get_config_value(
                    f"cda.sections.{section_key}.identifiers.code"
                )

                if not template_id and not code:
                    raise ValueError(
                        f"No template_id or code found for section {section_key}: \
                            configure one of the following: \
                            cda.sections.{section_key}.identifiers.template_id \
                            or cda.sections.{section_key}.identifiers.code"
                    )

                if template_id and self._find_section_by_template_id(
                    curr_section, template_id
                ):
                    section = curr_section
                    break

                if code and self._find_section_by_code(curr_section, code):
                    section = curr_section
                    break

            if not section:
                log.warning(
                    f"Section with template_id: {template_id} or code: {code} not found in CDA document for key: {section_key}"
                )
                return entries_dicts

            # Check if this is a notes section (which doesn't have entries but has text) - temporary workaround
            if section_key == "notes":
                # For notes section, create a synthetic entry with the section's text content
                section_dict = section.model_dump(exclude_none=True, by_alias=True)
                log.debug(
                    f"Created synthetic entry for notes section with text: {type(section.text)}"
                )
                # Return the entire section as the entry for DocumentReference
                return [section_dict]

            # Get entries from section (normal case for other sections)
            if section.entry:
                entries_dicts = (
                    section.entry
                    if isinstance(section.entry, list)
                    else [section.entry]
                )
            else:
                log.warning(f"No entries found for section {section_key}")
                return entries_dicts

            # Convert entries to dictionaries
            entry_dicts = [
                entry.model_dump(exclude_none=True, by_alias=True)
                for entry in entries_dicts
                if entry
            ]

            log.debug(f"Found {len(entry_dicts)} entries in section {section_key}")

            return entry_dicts

        except Exception as e:
            log.error(f"Error parsing section {section_key}: {str(e)}")
            return entries_dicts

    def _find_section_by_template_id(self, section: Section, template_id: str) -> bool:
        """Returns True if section has matching template ID"""
        if not section.templateId:
            return False

        template_ids = (
            section.templateId
            if isinstance(section.templateId, list)
            else [section.templateId]
        )
        return any(tid.root == template_id for tid in template_ids)

    def _find_section_by_code(self, section: Section, code: str) -> bool:
        """Returns True if section has matching code"""
        return bool(section.code and section.code.code == code)

__init__(config)

Initialize the CDA parser.

PARAMETER DESCRIPTION
config

InteropConfigManager instance containing section configurations, templates, and mapping rules for CDA document parsing

TYPE: InteropConfigManager

Source code in healthchain/interop/parsers/cda.py 
41
42
43
44
45
46
47
48
49
def __init__(self, config: InteropConfigManager):
    """Initialize the CDA parser.

    Args:
        config: InteropConfigManager instance containing section configurations,
               templates, and mapping rules for CDA document parsing
    """
    super().__init__(config)
    self.clinical_document = None

from_string(data)

Parse input data and convert it to a structured format.

PARAMETER DESCRIPTION
data

The CDA XML document string to parse

TYPE: str

RETURNS DESCRIPTION
dict

A dictionary containing the parsed data structure with sections
Source code in healthchain/interop/parsers/cda.py 
51
52
53
54
55
56
57
58
59
60
61
def from_string(self, data: str) -> dict:
    """
    Parse input data and convert it to a structured format.

    Args:
        data: The CDA XML document string to parse

    Returns:
        A dictionary containing the parsed data structure with sections
    """
    return self.parse_document(data)

parse_document(xml)

Parse a complete CDA document and extract entries from all configured sections.

This method parses a CDA XML document and extracts entries from each section that is defined in the configuration. It uses xmltodict to parse the XML into a dictionary and then processes each configured section to extract its entries.

PARAMETER DESCRIPTION
xml

The CDA XML document string to parse

TYPE: str

RETURNS DESCRIPTION
Dict[str, List[Dict]]

Dict[str, List[Dict]]: Dictionary mapping section keys (e.g. "problems", "medications") to lists of entry dictionaries containing the parsed data from that section (xmltodict format).
RAISES DESCRIPTION
ValueError

If the XML string is empty or invalid
Exception

If there is an error parsing the document or any section
Example

parser = CDAParser(config) sections = parser.from_string(cda_xml) problems = sections.get("problems", [])

Source code in healthchain/interop/parsers/cda.py 
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def parse_document(self, xml: str) -> Dict[str, List[Dict]]:
    """Parse a complete CDA document and extract entries from all configured sections.

    This method parses a CDA XML document and extracts entries from each section that is
    defined in the configuration. It uses xmltodict to parse the XML into a dictionary
    and then processes each configured section to extract its entries.

    Args:
        xml: The CDA XML document string to parse

    Returns:
        Dict[str, List[Dict]]: Dictionary mapping section keys (e.g. "problems",
            "medications") to lists of entry dictionaries containing the parsed data
            from that section (xmltodict format).

    Raises:
        ValueError: If the XML string is empty or invalid
        Exception: If there is an error parsing the document or any section

    Example:
        >>> parser = CDAParser(config)
        >>> sections = parser.from_string(cda_xml)
        >>> problems = sections.get("problems", [])
    """
    section_entries = {}

    # Parse the document once
    try:
        doc_dict = xmltodict.parse(xml)
        self.clinical_document = ClinicalDocument(**doc_dict["ClinicalDocument"])
    except Exception as e:
        log.error(f"Error parsing CDA document: {str(e)}")
        return section_entries

    # Get section configurations
    sections = self.config.get_cda_section_configs()
    if not sections:
        log.warning("No sections found in configuration")
        return section_entries

    # Process each section from the configuration
    for section_key in sections.keys():
        try:
            entries = self._parse_section_entries_from_document(section_key)
            if entries:
                section_entries[section_key] = entries
        except Exception as e:
            log.error(f"Failed to parse section {section_key}: {str(e)}")
            continue

    return section_entries

CDA Generator for HealthChain Interoperability Engine

This module provides functionality for generating CDA documents.

CDAGenerator

Bases: BaseGenerator

Handles generation of CDA documents from FHIR resources.

This class provides functionality to convert FHIR resources into CDA (Clinical Document Architecture) documents using configurable templates. It handles the mapping of resources to appropriate CDA sections, rendering of entries and sections, and generation of the final XML document.

Example

generator = CDAGenerator(config_manager, template_registry)

Convert FHIR resources to CDA XML document

cda_xml = generator.transform( resources=fhir_resources, document_type="ccd" )

Source code in healthchain/interop/generators/cda.py 
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
class CDAGenerator(BaseGenerator):
    """Handles generation of CDA documents from FHIR resources.

    This class provides functionality to convert FHIR resources into CDA (Clinical Document Architecture)
    documents using configurable templates. It handles the mapping of resources to appropriate CDA sections,
    rendering of entries and sections, and generation of the final XML document.

    Example:
        generator = CDAGenerator(config_manager, template_registry)

        # Convert FHIR resources to CDA XML document
        cda_xml = generator.transform(
            resources=fhir_resources,
            document_type="ccd"
        )
    """

    def transform(self, resources, **kwargs) -> str:
        """Transform FHIR resources to CDA format.

        Args:
            resources: List of FHIR resources
            **kwargs:
                document_type: Type of CDA document

        Returns:
            str: CDA document as XML string
        """
        # TODO: add validation
        document_type = kwargs.get("document_type", "ccd")
        return self.generate_document_from_fhir_resources(resources, document_type)

    def generate_document_from_fhir_resources(
        self,
        resources: List[Resource],
        document_type: str,
        validate: bool = True,
    ) -> str:
        """Generate a complete CDA document from FHIR resources

        This method handles the entire process of generating a CDA document:
        1. Mapping FHIR resources to CDA sections (config)
        2. Rendering sections (template)
        3. Generating the final document (template)

        Args:
            resources: FHIR resources to include in the document
            document_type: Type of document to generate
            validate: Whether to validate the CDA document (default: True)

        Returns:
            CDA document as XML string
        """
        mapped_entries = self._get_mapped_entries(resources, document_type)
        sections = self._render_sections(mapped_entries, document_type)

        # Generate final CDA document
        return self._render_document(sections, document_type, validate=validate)

    def _render_entry(
        self,
        resource: Resource,
        config_key: str,
    ) -> Optional[Dict]:
        """Render a single entry for a resource

        Args:
            resource: FHIR resource
            config_key: Key identifying the section

        Returns:
            Dictionary representation of the rendered entry (xmltodict)
        """
        try:
            # Get validated section configuration
            section_config = self.config.get_cda_section_configs(config_key)

            timestamp_format = self.config.get_config_value(
                "defaults.common.timestamp", "%Y%m%d"
            )
            timestamp = datetime.now().strftime(format=timestamp_format)

            id_format = self.config.get_config_value(
                "defaults.common.reference_name", "#{uuid}name"
            )
            reference_name = id_format.replace("{uuid}", str(uuid.uuid4())[:8])

            # Create context
            context = {
                "timestamp": timestamp,
                "text_reference_name": reference_name,
                "resource": resource.model_dump(),
                "config": section_config,
            }

            # Get template and render
            template = self.get_template_from_section_config(config_key, "entry")
            if template is None:
                log.error(f"Required entry template for '{config_key}' not found")
                return None

            return self.render_template(template, context)

        except Exception as e:
            log.error(f"Failed to render {config_key} entry: {str(e)}")
            return None

    def _get_mapped_entries(
        self, resources: List[Resource], document_type: str = None
    ) -> Dict:
        """Map FHIR resources to CDA section entries by resource type.

        Args:
            resources: List of FHIR resources to map to CDA entries
            document_type: Optional document type to determine which sections to include

        Returns:
            Dictionary mapping section keys (e.g. 'problems', 'medications') to lists of
            their rendered CDA entries. For example:
            {
                'problems': [<rendered condition entry>, ...],
                'medications': [<rendered medication entry>, ...]
            }
        """
        # Get included sections from document config if document_type is provided
        include_sections = None
        if document_type:
            include_sections = self.config.get_config_value(
                f"cda.document.{document_type}.structure.body.include_sections"
            )
            if include_sections:
                log.info(
                    f"Generating sections: {include_sections} for document type {document_type}"
                )

        section_entries = {}
        for resource in resources:
            # Find matching section for resource type
            resource_type = resource.__class__.__name__
            all_configs = self.config.get_cda_section_configs()
            section_key = _find_section_key_for_resource_type(
                resource_type, all_configs
            )
            if not section_key:
                log.error(f"No section config found for resource type: {resource_type}")
                continue

            # Skip if section is not included in the document config
            if include_sections and section_key not in include_sections:
                log.info(
                    f"Skipping section {section_key} as it's not in include_sections"
                )
                continue

            entry = self._render_entry(resource, section_key)
            if entry:
                section_entries.setdefault(section_key, []).append(entry)

        return section_entries

    def _render_sections(self, mapped_entries: Dict, document_type: str) -> List[Dict]:
        """Render all sections with their entries

        Args:
            mapped_entries: Dictionary mapping section keys to their entries
            document_type: Type of document to generate

        Returns:
            List of formatted section dictionaries

        Raises:
            ValueError: If section configurations or templates are not found
        """
        sections = []

        try:
            # Get validated section configurations
            section_configs = self.config.get_cda_section_configs()
        except ValueError as e:
            log.error(f"Error getting section configs: {str(e)}")
            raise ValueError(f"Failed to load section configurations: {str(e)}")

        # Get section template name from config
        section_template_name = self.config.get_config_value(
            f"cda.document.{document_type}.templates.section"
        )
        if not section_template_name:
            raise ValueError(
                f"No section template found for document type: {document_type}"
            )

        # Get the section template
        section_template = self.get_template(section_template_name)
        if not section_template:
            raise ValueError(f"Required template '{section_template_name}' not found")

        # Render each section that has entries
        for section_key, section_config in section_configs.items():
            entries = mapped_entries.get(section_key, [])
            if entries:
                try:
                    # Special handling for notes section, bit of a hack for now
                    if section_key == "notes":
                        # For DocumentReference, the generated entries already contain the full
                        # section structure, so we need to extract the section directly
                        if (
                            len(entries) > 0
                            and "component" in entries[0]
                            and "section" in entries[0]["component"]
                        ):
                            # Just extract the first section (we don't support multiple notes sections yet)
                            section_data = entries[0]["component"]["section"]
                            sections.append({"section": section_data})
                            continue

                    # Regular handling for other resource types
                    context = {
                        "entries": entries,
                        "config": section_config,
                    }
                    rendered = self.render_template(section_template, context)
                    if rendered:
                        sections.append(rendered)
                except Exception as e:
                    log.error(f"Failed to render section {section_key}: {str(e)}")

        return sections

    def _render_document(
        self,
        sections: List[Dict],
        document_type: str,
        validate: bool = True,
    ) -> str:
        """Generate the final CDA document

        Args:
            sections: List of formatted section dictionaries
            document_type: Type of document to generate
            validate: Whether to validate the CDA document

        Returns:
            CDA document as XML string

        Raises:
            ValueError: If document configuration or template is not found
        """
        try:
            # Get validated document configuration
            config = self.config.get_cda_document_config(document_type)
        except ValueError as e:
            log.error(f"Error getting document config: {str(e)}")
            raise ValueError(f"Failed to load document configuration: {str(e)}")

        # Get document template name from config
        document_template_name = self.config.get_config_value(
            f"cda.document.{document_type}.templates.document"
        )
        if not document_template_name:
            raise ValueError(
                f"No document template found for document type: {document_type}"
            )

        # Get the document template
        document_template = self.get_template(document_template_name)
        if not document_template:
            raise ValueError(f"Required template '{document_template_name}' not found")

        # Create document context
        # TODO: modify this as bundle metadata is not extracted
        context = {
            "config": config,
            "sections": sections,
        }

        rendered = self.render_template(document_template, context)
        if validate:
            if "ClinicalDocument" not in rendered:
                log.error(
                    "Unable to validate document structure: missing ClinicalDocument"
                )
                out_dict = rendered
            else:
                validated = ClinicalDocument(**rendered["ClinicalDocument"])
                out_dict = {
                    "ClinicalDocument": validated.model_dump(
                        exclude_none=True, exclude_unset=True, by_alias=True
                    )
                }
        else:
            out_dict = rendered

        # Get XML formatting options
        pretty_print = self.config.get_config_value(
            f"cda.document.{document_type}.rendering.xml.pretty_print", True
        )
        encoding = self.config.get_config_value(
            f"cda.document.{document_type}.rendering.xml.encoding", "UTF-8"
        )

        # Generate XML without preprocessor
        xml_string = xmltodict.unparse(out_dict, pretty=pretty_print, encoding=encoding)

        # Replace text elements containing < or > with CDATA sections
        # This regex matches <text>...</text> tags where content has HTML entities
        def replace_with_cdata(match):
            content = match.group(1)
            # Only process if it contains HTML entities
            if "&lt;" in content or "&gt;" in content:
                # Convert HTML entities back to characters
                import html

                decoded = html.unescape(content)
                return f"<text><![CDATA[{decoded}]]></text>"
            return f"<text>{content}</text>"

        xml_string = re.sub(
            r"<text>(.*?)</text>", replace_with_cdata, xml_string, flags=re.DOTALL
        )

        # Fix self-closing tags
        return re.sub(r"(<(\w+)(\s+[^>]*?)?)></\2>", r"\1/>", xml_string)

generate_document_from_fhir_resources(resources, document_type, validate=True)

Generate a complete CDA document from FHIR resources

This method handles the entire process of generating a CDA document: 1. Mapping FHIR resources to CDA sections (config) 2. Rendering sections (template) 3. Generating the final document (template)

PARAMETER DESCRIPTION
resources

FHIR resources to include in the document

TYPE: List[Resource]

document_type

Type of document to generate

TYPE: str

validate

Whether to validate the CDA document (default: True)

TYPE: bool DEFAULT: True

RETURNS DESCRIPTION
str

CDA document as XML string
Source code in healthchain/interop/generators/cda.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def generate_document_from_fhir_resources(
    self,
    resources: List[Resource],
    document_type: str,
    validate: bool = True,
) -> str:
    """Generate a complete CDA document from FHIR resources

    This method handles the entire process of generating a CDA document:
    1. Mapping FHIR resources to CDA sections (config)
    2. Rendering sections (template)
    3. Generating the final document (template)

    Args:
        resources: FHIR resources to include in the document
        document_type: Type of document to generate
        validate: Whether to validate the CDA document (default: True)

    Returns:
        CDA document as XML string
    """
    mapped_entries = self._get_mapped_entries(resources, document_type)
    sections = self._render_sections(mapped_entries, document_type)

    # Generate final CDA document
    return self._render_document(sections, document_type, validate=validate)

transform(resources, **kwargs)

Transform FHIR resources to CDA format.

PARAMETER DESCRIPTION
resources

List of FHIR resources
**kwargs

document_type: Type of CDA document

DEFAULT: {}

RETURNS DESCRIPTION
str

CDA document as XML string

TYPE: str

Source code in healthchain/interop/generators/cda.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def transform(self, resources, **kwargs) -> str:
    """Transform FHIR resources to CDA format.

    Args:
        resources: List of FHIR resources
        **kwargs:
            document_type: Type of CDA document

    Returns:
        str: CDA document as XML string
    """
    # TODO: add validation
    document_type = kwargs.get("document_type", "ccd")
    return self.generate_document_from_fhir_resources(resources, document_type)

FHIR Generator for HealthChain Interoperability Engine

This module provides functionality for generating FHIR resources from templates.

FHIRGenerator

Bases: BaseGenerator

Handles generation of FHIR resources from templates.

This class provides functionality to convert CDA section entries into FHIR resources using configurable templates. It handles validation, required field population, and error handling during the conversion process.

Key features: - Template-based conversion of CDA entries (xmltodict format) to FHIR resources - Automatic population of required FHIR fields based on configuration for common resource types like Condition, MedicationStatement, AllergyIntolerance - Validation of generated FHIR resources

Example

generator = FHIRGenerator(config_manager, template_registry)

Convert CDA problem entries to FHIR Condition resources

problems = generator.generate_resources_from_cda_section_entries( entries=problem_entries, section_key="problems" # from configs )

Source code in healthchain/interop/generators/fhir.py 
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
class FHIRGenerator(BaseGenerator):
    """Handles generation of FHIR resources from templates.

    This class provides functionality to convert CDA section entries into FHIR resources
    using configurable templates. It handles validation, required field population, and
    error handling during the conversion process.

    Key features:
    - Template-based conversion of CDA entries (xmltodict format) to FHIR resources
    - Automatic population of required FHIR fields based on configuration for
        common resource types like Condition, MedicationStatement, AllergyIntolerance
    - Validation of generated FHIR resources

    Example:
        generator = FHIRGenerator(config_manager, template_registry)

        # Convert CDA problem entries to FHIR Condition resources
        problems = generator.generate_resources_from_cda_section_entries(
            entries=problem_entries,
            section_key="problems"  # from configs
        )
    """

    def transform(self, data, **kwargs) -> str:
        """Transform input data to FHIR resources.

        Args:
            data: List of entries from source format
            **kwargs:
                src_format: The source format type (FormatType.CDA or FormatType.HL7V2)
                section_key: For CDA, the section key
                message_key: For HL7v2, the message key

        Returns:
            List[Resource]: FHIR resources
        """
        src_format = kwargs.get("src_format")
        if src_format == FormatType.CDA:
            return self.generate_resources_from_cda_section_entries(
                data, kwargs.get("section_key")
            )
        elif src_format == FormatType.HL7V2:
            return self.generate_resources_from_hl7v2_entries(
                data, kwargs.get("message_key")
            )
        else:
            raise ValueError(f"Unsupported source format: {src_format}")

    def generate_resources_from_cda_section_entries(
        self, entries: List[Dict], section_key: str
    ) -> List[Dict]:
        """
        Convert CDA section entries into FHIR resources using configured templates.

        This method processes entries from a CDA section and generates corresponding FHIR
        resources based on templates and configuration. It handles validation and error
        checking during the conversion process.

        Args:
            entries: List of CDA section entries in xmltodict format to convert
            section_key: Configuration key identifying the section (e.g. "problems", "medications")
                Used to look up templates and resource type mappings

        Returns:
            List of validated FHIR resource dictionaries. Empty list if conversion fails.

        Example:
            # Convert problem list entries to FHIR Condition resources
            conditions = generator.generate_resources_from_cda_section_entries(
                problem_entries, "problems"
            )
        """
        if not section_key:
            log.error(
                "No section key provided for CDA section entries: data needs to be in the format \
                      '{<section_key>}: {<section_entries>}'"
            )
            return []

        resources = []
        template = self.get_template_from_section_config(section_key, "resource")

        if not template:
            log.error(f"No resource template found for section {section_key}")
            return resources

        resource_type = self.config.get_config_value(
            f"cda.sections.{section_key}.resource"
        )
        if not resource_type:
            log.error(f"No resource type specified for section {section_key}")
            return resources

        for entry in entries:
            try:
                # Convert entry to FHIR resource dictionary
                resource_dict = self._render_resource_from_entry(
                    entry, section_key, template
                )
                if not resource_dict:
                    continue

                log.debug(f"Rendered FHIR resource: {resource_dict}")

                resource = self._validate_fhir_resource(resource_dict, resource_type)

                if resource:
                    resources.append(resource)

            except Exception as e:
                log.error(f"Failed to convert entry in section {section_key}: {str(e)}")
                continue

        return resources

    def _render_resource_from_entry(
        self, entry: Dict, section_key: str, template: Type[Template]
    ) -> Optional[Dict]:
        """Renders a FHIR resource dictionary from a CDA entry using templates.

        Args:
            entry: CDA entry dictionary
            section_key: Section identifier (e.g. "problems")
            template: Template to use for rendering

        Returns:
            FHIR resource dictionary or None if rendering fails
        """
        try:
            # Get validated section configuration
            try:
                section_config = self.config.get_cda_section_configs(section_key)
            except ValueError as e:
                log.error(
                    f"Failed to get CDA section config for {section_key}: {str(e)}"
                )
                return None

            # Create context with entry data and config
            context = {"entry": entry, "config": section_config}

            # Render template with context
            return self.render_template(template, context)

        except Exception as e:
            log.error(f"Failed to render resource for section {section_key}: {str(e)}")
            return None

    def _validate_fhir_resource(
        self, resource_dict: Dict, resource_type: str
    ) -> Optional[Resource]:
        """Validates and creates a FHIR resource from a dictionary.
        Adds required fields.

        Args:
            resource_dict: FHIR resource dictionary
            resource_type: FHIR resource type

        Returns:
            FHIR resource or None if validation fails
        """

        try:
            resource_dict = self._add_required_fields(resource_dict, resource_type)
            resource = create_resource_from_dict(resource_dict, resource_type)
            if resource:
                return resource
        except Exception as e:
            log.error(f"Failed to validate FHIR resource: {str(e)}")
            return None

    def _add_required_fields(self, resource_dict: Dict, resource_type: str) -> Dict:
        """Add required fields to FHIR resource dictionary based on resource type.
        Currently only supports Condition, MedicationStatement, and AllergyIntolerance.

        Args:
            resource_dict: Dictionary representation of the resource
            resource_type: Type of FHIR resource

        Returns:
            Dict: Resource dictionary with required fields added
        """
        # Add common fields
        id_prefix = self.config.get_config_value("defaults.common.id_prefix", "hc-")
        if "id" not in resource_dict:
            resource_dict["id"] = f"{id_prefix}{str(uuid.uuid4())}"

        # Get default values from configuration if available
        default_subject = self.config.get_config_value("defaults.common.subject")

        # Add resource-specific required fields
        if resource_type == "Condition":
            if "subject" not in resource_dict:
                resource_dict["subject"] = default_subject

            if "clinicalStatus" not in resource_dict:
                default_status = self.config.get_config_value(
                    "defaults.resources.Condition.clinicalStatus"
                )
                resource_dict["clinicalStatus"] = default_status
        elif resource_type == "MedicationStatement":
            if "subject" not in resource_dict:
                resource_dict["subject"] = default_subject
            if "status" not in resource_dict:
                default_status = self.config.get_config_value(
                    "defaults.resources.MedicationStatement.status"
                )
                resource_dict["status"] = default_status
        elif resource_type == "AllergyIntolerance":
            if "patient" not in resource_dict:
                resource_dict["patient"] = default_subject
            if "clinicalStatus" not in resource_dict:
                default_status = self.config.get_config_value(
                    "defaults.resources.AllergyIntolerance.clinicalStatus"
                )
                resource_dict["clinicalStatus"] = default_status

        return resource_dict

    def generate_resources_from_hl7v2_entries(
        self, entries: List[Dict], message_key: str
    ) -> List[Dict]:
        """
        Convert HL7v2 message entries into FHIR resources.
        This is a placeholder implementation.

        Args:
            entries: List of HL7v2 message entries to convert
            message_key: Key identifying the message type

        Returns:
            List of FHIR resources
        """
        log.warning(
            "FHIR resource generation from HL7v2 is a placeholder implementation"
        )
        return []

generate_resources_from_cda_section_entries(entries, section_key)

Convert CDA section entries into FHIR resources using configured templates.

This method processes entries from a CDA section and generates corresponding FHIR resources based on templates and configuration. It handles validation and error checking during the conversion process.

PARAMETER DESCRIPTION
entries

List of CDA section entries in xmltodict format to convert

TYPE: List[Dict]

section_key

Configuration key identifying the section (e.g. "problems", "medications") Used to look up templates and resource type mappings

TYPE: str

RETURNS DESCRIPTION
List[Dict]

List of validated FHIR resource dictionaries. Empty list if conversion fails.
Example

Convert problem list entries to FHIR Condition resources

conditions = generator.generate_resources_from_cda_section_entries( problem_entries, "problems" )

Source code in healthchain/interop/generators/fhir.py 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def generate_resources_from_cda_section_entries(
    self, entries: List[Dict], section_key: str
) -> List[Dict]:
    """
    Convert CDA section entries into FHIR resources using configured templates.

    This method processes entries from a CDA section and generates corresponding FHIR
    resources based on templates and configuration. It handles validation and error
    checking during the conversion process.

    Args:
        entries: List of CDA section entries in xmltodict format to convert
        section_key: Configuration key identifying the section (e.g. "problems", "medications")
            Used to look up templates and resource type mappings

    Returns:
        List of validated FHIR resource dictionaries. Empty list if conversion fails.

    Example:
        # Convert problem list entries to FHIR Condition resources
        conditions = generator.generate_resources_from_cda_section_entries(
            problem_entries, "problems"
        )
    """
    if not section_key:
        log.error(
            "No section key provided for CDA section entries: data needs to be in the format \
                  '{<section_key>}: {<section_entries>}'"
        )
        return []

    resources = []
    template = self.get_template_from_section_config(section_key, "resource")

    if not template:
        log.error(f"No resource template found for section {section_key}")
        return resources

    resource_type = self.config.get_config_value(
        f"cda.sections.{section_key}.resource"
    )
    if not resource_type:
        log.error(f"No resource type specified for section {section_key}")
        return resources

    for entry in entries:
        try:
            # Convert entry to FHIR resource dictionary
            resource_dict = self._render_resource_from_entry(
                entry, section_key, template
            )
            if not resource_dict:
                continue

            log.debug(f"Rendered FHIR resource: {resource_dict}")

            resource = self._validate_fhir_resource(resource_dict, resource_type)

            if resource:
                resources.append(resource)

        except Exception as e:
            log.error(f"Failed to convert entry in section {section_key}: {str(e)}")
            continue

    return resources

generate_resources_from_hl7v2_entries(entries, message_key)

Convert HL7v2 message entries into FHIR resources. This is a placeholder implementation.

PARAMETER DESCRIPTION
entries

List of HL7v2 message entries to convert

TYPE: List[Dict]

message_key

Key identifying the message type

TYPE: str

RETURNS DESCRIPTION
List[Dict]

List of FHIR resources
Source code in healthchain/interop/generators/fhir.py 
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
def generate_resources_from_hl7v2_entries(
    self, entries: List[Dict], message_key: str
) -> List[Dict]:
    """
    Convert HL7v2 message entries into FHIR resources.
    This is a placeholder implementation.

    Args:
        entries: List of HL7v2 message entries to convert
        message_key: Key identifying the message type

    Returns:
        List of FHIR resources
    """
    log.warning(
        "FHIR resource generation from HL7v2 is a placeholder implementation"
    )
    return []

transform(data, **kwargs)

Transform input data to FHIR resources.

PARAMETER DESCRIPTION
data

List of entries from source format
**kwargs

src_format: The source format type (FormatType.CDA or FormatType.HL7V2) section_key: For CDA, the section key message_key: For HL7v2, the message key

DEFAULT: {}

RETURNS DESCRIPTION
str

List[Resource]: FHIR resources
Source code in healthchain/interop/generators/fhir.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def transform(self, data, **kwargs) -> str:
    """Transform input data to FHIR resources.

    Args:
        data: List of entries from source format
        **kwargs:
            src_format: The source format type (FormatType.CDA or FormatType.HL7V2)
            section_key: For CDA, the section key
            message_key: For HL7v2, the message key

    Returns:
        List[Resource]: FHIR resources
    """
    src_format = kwargs.get("src_format")
    if src_format == FormatType.CDA:
        return self.generate_resources_from_cda_section_entries(
            data, kwargs.get("section_key")
        )
    elif src_format == FormatType.HL7V2:
        return self.generate_resources_from_hl7v2_entries(
            data, kwargs.get("message_key")
        )
    else:
        raise ValueError(f"Unsupported source format: {src_format}")

Configuration validators for HealthChain

This module provides validation models and utilities for configuration files.

AllergySectionTemplateConfig

Bases: SectionTemplateConfigBase

Template configuration for Allergy Section

Source code in healthchain/config/validators.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
class AllergySectionTemplateConfig(SectionTemplateConfigBase):
    """Template configuration for Allergy Section"""

    act: ComponentTemplateConfig
    allergy_obs: ComponentTemplateConfig
    reaction_obs: Optional[ComponentTemplateConfig] = None
    severity_obs: Optional[ComponentTemplateConfig] = None
    clinical_status_obs: ComponentTemplateConfig

    @field_validator("allergy_obs")
    @classmethod
    def validate_allergy_obs(cls, v):
        required_fields = {"code", "code_system", "status_code"}
        missing = required_fields - set(v.model_dump(exclude_unset=True).keys())
        if missing:
            raise ValueError(f"allergy_obs missing required fields: {missing}")
        return v

CcdDocumentConfig

Bases: DocumentConfigBase

Configuration model specific to CCD documents

Source code in healthchain/config/validators.py 
193
194
195
196
class CcdDocumentConfig(DocumentConfigBase):
    """Configuration model specific to CCD documents"""

    allowed_sections: List[str] = ["problems", "medications", "allergies", "notes"]

ComponentTemplateConfig

Bases: BaseModel

Generic template for CDA/FHIR component configuration

Source code in healthchain/config/validators.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
class ComponentTemplateConfig(BaseModel):
    """Generic template for CDA/FHIR component configuration"""

    template_id: Union[List[str], str]
    code: Optional[str] = None
    code_system: Optional[str] = "2.16.840.1.113883.6.1"
    code_system_name: Optional[str] = "LOINC"
    display_name: Optional[str] = None
    status_code: Optional[str] = "active"
    class_code: Optional[str] = None
    mood_code: Optional[str] = None
    type_code: Optional[str] = None
    inversion_ind: Optional[bool] = None
    value: Optional[Dict[str, Any]] = None

    model_config = ConfigDict(extra="allow")

DocumentConfigBase

Bases: BaseModel

Generic document configuration model

Source code in healthchain/config/validators.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
class DocumentConfigBase(BaseModel):
    """Generic document configuration model"""

    type_id: Dict[str, Any]
    code: Dict[str, Any]
    confidentiality_code: Dict[str, Any]
    language_code: Optional[str] = "en-US"
    templates: Optional[Dict[str, Any]] = None
    structure: Optional[Dict[str, Any]] = None
    defaults: Optional[Dict[str, Any]] = None
    rendering: Optional[Dict[str, Any]] = None

    @field_validator("type_id")
    @classmethod
    def validate_type_id(cls, v):
        if not isinstance(v, dict) or "root" not in v:
            raise ValueError("type_id must contain 'root' field")
        return v

    @field_validator("code")
    @classmethod
    def validate_code(cls, v):
        if not isinstance(v, dict) or "code" not in v or "code_system" not in v:
            raise ValueError("code must contain 'code' and 'code_system' fields")
        return v

    @field_validator("confidentiality_code")
    @classmethod
    def validate_confidentiality_code(cls, v):
        if not isinstance(v, dict) or "code" not in v:
            raise ValueError("confidentiality_code must contain 'code' field")
        return v

    @field_validator("templates")
    @classmethod
    def validate_templates(cls, v):
        if not isinstance(v, dict) or "section" not in v or "document" not in v:
            raise ValueError("templates must contain 'section' and 'document' fields")
        return v

    model_config = ConfigDict(extra="allow")

MedicationSectionTemplateConfig

Bases: SectionTemplateConfigBase

Template configuration for SubstanceAdministration Section

Source code in healthchain/config/validators.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
class MedicationSectionTemplateConfig(SectionTemplateConfigBase):
    """Template configuration for SubstanceAdministration Section"""

    substance_admin: ComponentTemplateConfig
    manufactured_product: ComponentTemplateConfig
    clinical_status_obs: ComponentTemplateConfig

    @field_validator("substance_admin")
    @classmethod
    def validate_substance_admin(cls, v):
        if not v.status_code:
            raise ValueError("substance_admin requires status_code")
        return v

NoteSectionTemplateConfig

Bases: SectionTemplateConfigBase

Template configuration for Notes Section

Source code in healthchain/config/validators.py 
199
200
201
202
203
204
205
206
207
208
209
210
211
class NoteSectionTemplateConfig(SectionTemplateConfigBase):
    """Template configuration for Notes Section"""

    note_section: ComponentTemplateConfig

    @field_validator("note_section")
    @classmethod
    def validate_note_section(cls, v):
        required_fields = {"template_id", "code", "code_system", "status_code"}
        missing = required_fields - set(v.model_dump(exclude_unset=True).keys())
        if missing:
            raise ValueError(f"note_section missing required fields: {missing}")
        return v

ProblemSectionTemplateConfig

Bases: SectionTemplateConfigBase

Template configuration for Problem Section

Source code in healthchain/config/validators.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
class ProblemSectionTemplateConfig(SectionTemplateConfigBase):
    """Template configuration for Problem Section"""

    act: ComponentTemplateConfig
    problem_obs: ComponentTemplateConfig
    clinical_status_obs: ComponentTemplateConfig

    @field_validator("problem_obs")
    @classmethod
    def validate_problem_obs(cls, v):
        required_fields = {"code", "code_system", "status_code"}
        missing = required_fields - set(v.model_dump(exclude_unset=True).keys())
        if missing:
            raise ValueError(f"problem_obs missing required fields: {missing}")
        return v

    @field_validator("clinical_status_obs")
    @classmethod
    def validate_clinical_status(cls, v):
        required_fields = {"code", "code_system", "status_code"}
        missing = required_fields - set(v.model_dump(exclude_unset=True).keys())
        if missing:
            raise ValueError(f"clinical_status_obs missing required fields: {missing}")
        return v

RenderingConfig

Bases: BaseModel

Configuration for section rendering

Source code in healthchain/config/validators.py 
51
52
53
54
55
56
57
class RenderingConfig(BaseModel):
    """Configuration for section rendering"""

    narrative: Optional[Dict[str, Any]] = None
    entry: Optional[Dict[str, Any]] = None

    model_config = ConfigDict(extra="allow")

SectionBaseConfig

Bases: BaseModel

Base model for all section configurations

Source code in healthchain/config/validators.py 
60
61
62
63
64
65
66
67
68
69
class SectionBaseConfig(BaseModel):
    """Base model for all section configurations"""

    resource: str
    resource_template: str
    entry_template: str
    identifiers: SectionIdentifiersConfig
    rendering: Optional[RenderingConfig] = None

    model_config = ConfigDict(extra="allow")

SectionIdentifiersConfig

Bases: BaseModel

Section identifiers validation

Source code in healthchain/config/validators.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
class SectionIdentifiersConfig(BaseModel):
    """Section identifiers validation"""

    template_id: str
    code: str
    code_system: Optional[str] = "2.16.840.1.113883.6.1"
    code_system_name: Optional[str] = "LOINC"
    display: str
    clinical_status: Optional[Dict[str, str]] = None
    reaction: Optional[Dict[str, str]] = None
    severity: Optional[Dict[str, str]] = None

    model_config = ConfigDict(extra="allow")

SectionTemplateConfigBase

Bases: BaseModel

Base class for section template configurations

Source code in healthchain/config/validators.py 
77
78
79
80
81
82
83
84
85
86
87
class SectionTemplateConfigBase(BaseModel):
    """Base class for section template configurations"""

    def validate_component_fields(self, component, required_fields):
        """Helper method to validate required fields in a component"""
        missing = required_fields - set(component.model_dump(exclude_unset=True).keys())
        if missing:
            raise ValueError(
                f"{component.__class__.__name__} missing required fields: {missing}"
            )
        return component

validate_component_fields(component, required_fields)

Helper method to validate required fields in a component

Source code in healthchain/config/validators.py 
80
81
82
83
84
85
86
87
def validate_component_fields(self, component, required_fields):
    """Helper method to validate required fields in a component"""
    missing = required_fields - set(component.model_dump(exclude_unset=True).keys())
    if missing:
        raise ValueError(
            f"{component.__class__.__name__} missing required fields: {missing}"
        )
    return component

create_cda_section_validator(resource_type, template_model)

Create a section validator for a specific resource type

Source code in healthchain/config/validators.py 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
def create_cda_section_validator(
    resource_type: str, template_model: Type[BaseModel]
) -> Type[BaseModel]:
    """Create a section validator for a specific resource type"""

    class DynamicSectionConfig(SectionBaseConfig):
        template: Dict[str, Any]

        @field_validator("template")
        @classmethod
        def validate_template(cls, v):
            try:
                template_model(**v)
            except ValidationError as e:
                raise ValueError(f"Template validation failed: {str(e)}")
            return v

    DynamicSectionConfig.__name__ = f"{resource_type}SectionConfig"
    return DynamicSectionConfig

register_cda_document_template_config_model(document_type, document_model)

Register a custom document model

Source code in healthchain/config/validators.py 
317
318
319
320
321
322
def register_cda_document_template_config_model(
    document_type: str, document_model: Type[BaseModel]
) -> None:
    """Register a custom document model"""
    CDA_DOCUMENT_CONFIG_REGISTRY[document_type.lower()] = document_model
    logger.info(f"Registered custom document model for {document_type}")

register_cda_section_template_config_model(resource_type, template_model)

Register a custom template model for a section

Source code in healthchain/config/validators.py 
306
307
308
309
310
311
312
313
314
def register_cda_section_template_config_model(
    resource_type: str, template_model: Type[BaseModel]
) -> None:
    """Register a custom template model for a section"""
    CDA_SECTION_CONFIG_REGISTRY[resource_type] = template_model
    SECTION_VALIDATORS[resource_type] = create_cda_section_validator(
        resource_type, template_model
    )
    logger.info(f"Registered custom template model for {resource_type}")

validate_cda_document_config_model(document_type, document_config)

Validate a document configuration

Source code in healthchain/config/validators.py 
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
def validate_cda_document_config_model(
    document_type: str, document_config: Dict[str, Any]
) -> bool:
    """Validate a document configuration"""
    validator = CDA_DOCUMENT_CONFIG_REGISTRY.get(document_type.lower())
    if not validator:
        logger.warning(f"No specific validator for document type: {document_type}")
        return True

    try:
        validator(**document_config)
        return True
    except ValidationError as e:
        logger.error(f"Document validation failed for {document_type}: {str(e)}")
        return False

validate_cda_section_config_model(section_key, section_config)

Validate a section configuration

Source code in healthchain/config/validators.py 
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
def validate_cda_section_config_model(
    section_key: str, section_config: Dict[str, Any]
) -> bool:
    """Validate a section configuration"""
    resource_type = section_config.get("resource")
    if not resource_type:
        logger.error(f"Section '{section_key}' is missing 'resource' field")
        return False

    validator = SECTION_VALIDATORS.get(resource_type)
    if not validator:
        # TODO: Pass validation level to this
        logger.warning(f"No specific validator for resource type: {resource_type}")
        return True

    try:
        validator(**section_config)
        return True
    except ValidationError as e:
        logger.error(f"Section validation failed for {resource_type}: {str(e)}")
        return False