ccda_to_omop.value_transformations module

Transformation functions used by DERIVED fields in metadata configs.

Each function receives a dict of resolved argument values (looked up from the current output record by the parsing engine) and returns a single output value. Covers: concept ID / domain / source-concept lookups via the codemap, date and datetime casting, string concatenation, hash-based ID generation, and person / data-partner ID extraction.

class ccda_to_omop.value_transformations.VocabContext

Bases: object

Holds the vocabulary lookup maps used by transformation functions.

A single module-level instance (_context) is used at runtime. Tests can call reset() between cases to guarantee isolation without relying on import order or global keyword side-effects.

Full dependency injection is not yet possible because transformation functions are referenced by name in metadata configs and are called with only args_dict. This class is the first step: all mutable state lives here rather than as bare module globals, which eliminates the ‘global’ keyword and makes the state easy to inspect and reset.

ccda_to_omop.value_transformations.cast_as_concept_id(args_dict: dict[str, Any]) → int32 | None

Return args_dict[‘input’] as an int32 concept ID if type is ‘CD’ or ‘CE’, else None.

ccda_to_omop.value_transformations.cast_as_number(args_dict: dict[str, Any]) → int | None

Return args_dict[‘input’] as an int if type is ‘PQ’ (HL7 physical quantity), else None.

ccda_to_omop.value_transformations.cast_as_string(args_dict: dict[str, Any]) → str | None

Return args_dict[‘input’] as a string if type is ‘ST’ (HL7 string), else None.

ccda_to_omop.value_transformations.codemap_xwalk_concept_id(args_dict: dict[str, Any]) → int32 | None

expects: vocabulary_oid, concept_code returns: concept_id AS INTEGER (because that’s what’s in the table), not necessarily standard

If NMC is disallowed, it will return None instead of 0. Control this via set_allow_no_macthing_concept() in package_constant_access.

throws/raises when codemap_xwalk is None

ccda_to_omop.value_transformations.codemap_xwalk_domain_id(args_dict: dict[str, Any]) → str | None

expects: vocabulary_oid, concept_code returns: always returns domain_id throws/raises when codemap_xwalk is None

ccda_to_omop.value_transformations.codemap_xwalk_source_concept_id(args_dict: dict[str, Any]) → int32 | None

expects: vocabulary_oid, concept_code returns: unmapped concept_id AS INTEGER (because that’s what’s in the table), not necessarily standard throws/raises when codemap_xwalk is None

ccda_to_omop.value_transformations.concat_field_list_names(args_dict: dict[str, Any], data_dict: dict[str, Any]) → str

A DERIVED2 style function. Looks for a argument with the name of ‘args’ under the ‘argument_list’ brought in from the parse configuration. That list is a list of keys whose data we’re interested in fetching from the data_dict.

args_dict: the field’s paragraph in the parse configuration data_dict: the dictionary of values being built up for an OMOP row

by the parse configuration where all this comes from.

Returns: a joined list of the keys. The data_dict is unused.

ccda_to_omop.value_transformations.concat_field_list_values(args_dict: dict[str, Any], data_dict: dict[str, Any]) → str

A DERIVED2 style function. Looks for a argument with the name of ‘args’ under the ‘argument_list’ brought in from the parse configuration. That list is a list of keys whose data we’re interested in fetching from the data_dict.

args_dict: the field’s paragraph in the parse configuration data_dict: the dictionary of values being built up for an OMOP row

by the parse configuration where all this comes from.

Returns: a joined list of the data values associated with those keys.

ccda_to_omop.value_transformations.concat_fields(args_dict: dict[str, Any]) → str

A DERIVED style function. input key “delimiter” is a character to use to separate the fields following items in dict are the names of keys in the values to concat

returns one string, the concatenation of values corresponding to args 2-n, using arg 1 as a delimieter

ccda_to_omop.value_transformations.extract_day_of_birth(args_dict: dict[str, Any]) → int32

Extract day of birth from args_dict[‘date_object’] (assumes a date/datetime).

ccda_to_omop.value_transformations.extract_month_of_birth(args_dict: dict[str, Any]) → int32

Extract month of birth from args_dict[‘date_object’] (assumes a date/datetime).

ccda_to_omop.value_transformations.extract_year_of_birth(args_dict: dict[str, Any]) → int32

Extract year of birth from args_dict[‘date_object’] (assumes a date/datetime).

ccda_to_omop.value_transformations.get_codemap_dict() → dict[tuple[str, str], list[dict[str, int | str]]] | None

Return the currently loaded codemap crosswalk dictionary.

ccda_to_omop.value_transformations.get_data_partner_id(args_dict: dict[str, Any]) → int32

Returns Data Partner ID. Defaults to 0 if filename is not in map. Strictly returns an integer per the component contract.

ccda_to_omop.value_transformations.get_mspi_map() → dict[str, int] | None

ccda_to_omop.value_transformations.get_partner_map() → dict[str, int] | None

ccda_to_omop.value_transformations.map_filename_to_mspi(args_dict: dict[str, Any]) → int

Returns MSPI (person_id). Defaults to 0 if filename is not in map. Raises if the MSPI map has not been initialized.

ccda_to_omop.value_transformations.reset_context()

Reset all vocab maps to None. Call between test cases for isolation.

ccda_to_omop.value_transformations.set_codemap_dict(map: dict[tuple[str, str], list[dict[str, int | str]]] | None) → None

ccda_to_omop.value_transformations.set_mspi_map(m: dict[str, int] | None) → None

Initializes the MSPI (person_id) map on the executor.

ccda_to_omop.value_transformations.set_partner_map(m: dict[str, int] | None) → None

Initializes the partner map on the executor.

ccda_to_omop.value_transformations.transform_datetime_high(args) → datetime | None

Transforms a date-only string into a full ISO 8601 datetime defaulting to 23:59:59.

This function assumes the input is either in ISO 8601 (YYYY-MM-DD) or HL7 (YYYYMMDD) format. We can make this assumption because this transformation is typically called after ‘parseutils.parser’ has already standardized the raw input.

ccda_to_omop.value_transformations.transform_datetime_low(args) → datetime | None

Transforms a date-only string into a full ISO 8601 datetime defaulting to 00:00:00.

This function assumes the input is either in ISO 8601 (YYYY-MM-DD) or HL7 (YYYYMMDD) format. We can make this assumption because this transformation is typically called after ‘parseutils.parser’ has already standardized the raw input.