ccda_to_omop.visit_reconciliation module

Visit_reconciliation.py

This module contains functions for linking domain rows to visits as well as functions to differentiate visit_occurrence from visit_detail.

Main entry points are: - assign_visit_occurrence_ids_to_events() - assign_visit_detail_ids_to_events() - reclassify_nested_visit_occurrences_as_detail()

This code processes visit data to create a hierarchical structure where inpatient parent visits (<= 1 year duration) are kept in visit_occurrence and their nested child visits are moved to visit_detail.

Process: 1. Identify inpatient parent visits with duration <= 1 year 2. Find visits temporally nested within each parent 3. Create visit_detail records for nested children 4. Return updated visit_occurrence (parents + standalone) and new visit_detail list

ccda_to_omop.visit_reconciliation.assign_visit_detail_ids_to_events(data_dict: dict[str, list[dict[str, None | str | float | int | int32 | int64 | datetime | date] | None] | None])

visit_detail FK reconciliation: Match clinical domain events to visit_detail records.

This assigns visit_detail_id to events that occur during nested visits. Events must already have visit_occurrence_id set (from assign_visit_occurrence_ids_to_events).

For events that fall within multiple nested visits, chooses the most specific (smallest duration) visit_detail.

Parameters:

data_dict – Dictionary with config_name → list of records (omop_dict)

ccda_to_omop.visit_reconciliation.assign_visit_occurrence_ids_to_events(data_dict: dict[str, list[dict[str, None | str | float | int | int32 | int64 | datetime | date] | None] | None])

Assign visit_occurrence_id FKs to all clinical event records in data_dict.

Iterates over all configs in data_dict whose domain is one of Measurement, Observation, Condition, Procedure, Drug, or Device, and calls reconcile_visit_FK_with_specific_domain() to match each event to a visit.

Expects visit records to be under the 'Visit' config key (after visit hierarchy processing has merged Visit and Visit_encompassingEncounter). Removes the temporary __visit_candidates key from all records after reconciliation is complete.

Parameters:

data_dict – Dict mapping config names to lists of record dicts, as returned by parse_doc().

ccda_to_omop.visit_reconciliation.create_visit_detail_record(visit_dict: dict[str, None | str | float | int | int32 | int64 | datetime | date], top_level_parent_id: int64, immediate_parent_id: int64 | None = None) → dict[str, None | str | float | int | int32 | int64 | datetime | date]

Convert a visit (visit_occurrence) record into visit_detail format.

Parameters:

• visit_dict – Visit record to convert

• top_level_parent_id – The top-level visit_occurrence_id

• immediate_parent_id – The immediate parent’s visit_detail_id (or None for Layer 2)

Returns:

Dictionary in visit_detail format

ccda_to_omop.visit_reconciliation.find_most_specific_parent(child_dict: dict[str, None | str | float | int | int32 | int64 | datetime | date], potential_parents: list[dict[str, None | str | float | int | int32 | int64 | datetime | date]]) → int64 | None

Find the most specific (shortest duration, most immediate) parent for a child visit.

When multiple parents overlap and contain a child, choose the parent with the shortest duration as it represents the most specific/immediate context.

However, if the child has multiple containing parents that are at the same hierarchy level (i.e., the parents don’t contain each other), returns None to avoid ambiguity. The child visit will remain at its current level.

Parameters:

• child_dict – Child visit record

• potential_parents – List of potential parent visit records

Returns:

• No parent found

• Multiple parents at the same hierarchy level exist (ambiguous)

Return type:

visit_occurrence_id of the most specific parent, or None if

ccda_to_omop.visit_reconciliation.get_visit_detail_duration(visit_detail_dict: dict[str, None | str | float | int | int32 | int64 | datetime | date]) → float

Calculate duration of a visit_detail in days.

Parameters:

visit_detail_dict – Visit_detail record

Returns:

Duration in days (float)

ccda_to_omop.visit_reconciliation.get_visit_duration_days(visit_dict: dict[str, None | str | float | int | int32 | int64 | datetime | date]) → float | None

Calculate visit duration in days.

Parameters:

visit_dict – Dictionary containing visit record

Returns:

Duration in days, or None if dates are missing

ccda_to_omop.visit_reconciliation.identify_inpatient_parents(visit_list: list[dict[str, None | str | float | int | int32 | int64 | datetime | date]]) → list[dict[str, None | str | float | int | int32 | int64 | datetime | date]]

Identify inpatient parent visits that are meaningful and time-bounded.

Criteria: - visit_concept_id is in INPATIENT_CONCEPT_IDS - Duration between start and end is <= 1 year - Has valid start and end datetimes

Parameters:

visit_list – List of visit record dictionaries

Returns:

List containing only eligible inpatient parent visits

ccda_to_omop.visit_reconciliation.is_temporally_contained(child_dict: dict[str, None | str | float | int | int32 | int64 | datetime | date], parent_dict: dict[str, None | str | float | int | int32 | int64 | datetime | date]) → bool

Check if child visit is temporally contained within parent visit.

Parameters:

• child_dict – Child visit record

• parent_dict – Parent visit record

Returns:

True if child is fully contained within parent timeframe

ccda_to_omop.visit_reconciliation.reclassify_nested_visit_occurrences_as_detail(omop_dict: dict[str, list[dict[str, None | str | float | int | int32 | int64 | datetime | date]] | None]) → dict[str, list[dict[str, None | str | float | int | int32 | int64 | datetime | date]] | None]

Main entry point for visit hierarchy processing. Merges visits from both ‘Visit’ and ‘Visit_encompassingEncounter’ configs, deduplicates them, processes hierarchy, and creates visit_detail records.

Strategy: 1. Collect visits from both configs 2. Deduplicate (keep ‘Visit’ when duplicate exists with ‘Visit_encompassingEncounter’) 3. Process merged list as unified hierarchy 4. Put results in ‘Visit’ config and create ‘VISITDETAIL_visit_occurrence’ config

Parameters:

omop_dict – Dictionary of config_name → list of records

Returns:

• ‘Visit’: top-level visit_occurrence records (merged from both configs)

• ’VISITDETAIL_visit_occurrence’: visit_detail records

• ’Visit_encompassingEncounter’: removed (merged into ‘Visit’)

Return type:

Updated omop_dict with

ccda_to_omop.visit_reconciliation.reconcile_visit_FK_with_specific_domain(domain: str, domain_dict: list[dict[str, None | str | float | int | int32 | int64 | datetime | date]] | None, visit_dict: list[dict[str, None | str | float | int | int32 | int64 | datetime | date]] | None)

Assign visit_occurrence_id to each event record in a clinical domain.

Matches each event in domain_dict to a visit in visit_dict by checking whether the event’s date falls within the visit’s start/end window. If exactly one visit matches, its visit_occurrence_id is written into the event record. If zero or multiple visits match, a warning is logged and the field is left unset (or a __visit_candidates key is added for ambiguous cases).

Parameters:

• domain – OMOP domain name (e.g. 'Measurement', 'Condition'). Must be a key in the module-level domain_dates config dict.

• domain_dict – List of clinical event record dicts to be matched.

• visit_dict – List of visit_occurrence record dicts to match against.

ccda_to_omop.visit_reconciliation.reconcile_visit_detail_FK_with_specific_domain(domain: str, domain_dict: list[dict[str, None | str | float | int | int32 | int64 | datetime | date]] | None, visit_detail_dict: list[dict[str, None | str | float | int | int32 | int64 | datetime | date]] | None)

Match events to visit_detail records by temporal containment. Choose the most specific (smallest duration) matching visit_detail.

Parameters:

• domain – Domain name (e.g., ‘Measurement’, ‘Condition’)

• domain_dict – List of event records to reconcile

• visit_detail_dict – List of visit_detail records

ccda_to_omop.visit_reconciliation.strip_tz(dt)

Remove timezone info from a datetime, leaving date objects unchanged.

Parameters:

dt – A datetime.datetime, datetime.date, or None value.

Returns:

The same value with tzinfo removed if it was a timezone-aware datetime; otherwise the original value unchanged.