Helper functions

Resolver functions for IODD parsing.

This module contains helper functions that resolve references in IODD data structures, merging information from standard definitions and device-specific IODD files.

`resolve_errors(std_error_collection, device_error_collection, texts)`

Resolve errors from standard definitions and device error collection.

Combines referenced standard errors (code=128) with device-specific errors (code=129) into a unified dictionary.

Parameters:

Name	Type	Description	Default
`std_error_collection`	`IoddstandardErrorTypeCollectionT`	Standard error types from definitions XML.	required
`device_error_collection`	`ErrorTypeCollectionT \| None`	Device-specific error collection, may be None.	required
`texts`	`dict[str, str]`	Dictionary of resolved text strings.	required

Returns:

Type	Description
`dict[tuple[int, int], ResolvedError]`	Dictionary of resolved errors keyed by (code, additional_code) tuple.

Source code in src/iodd_parser/resolvers.py

def resolve_errors(
    std_error_collection: IoddstandardErrorTypeCollectionT,
    device_error_collection: ErrorTypeCollectionT | None,
    texts: dict[str, str],
) -> dict[tuple[int, int], ResolvedError]:
    """
    Resolve errors from standard definitions and device error collection.

    Combines referenced standard errors (code=128) with device-specific
    errors (code=129) into a unified dictionary.

    :param std_error_collection: Standard error types from definitions XML.
    :param device_error_collection: Device-specific error collection, may be None.
    :param texts: Dictionary of resolved text strings.
    :returns: Dictionary of resolved errors keyed by (code, additional_code) tuple.
    """
    errors: dict[tuple[int, int], ResolvedError] = {}

    if device_error_collection is None:
        return errors

    # Build lookup for standard errors by additional_code
    std_errors_lookup: dict[int, ErrorTypeT] = {err.additional_code: err for err in std_error_collection.error_type}

    # Add referenced standard errors (code=128)
    for ref in device_error_collection.std_error_type_ref:
        std_err = std_errors_lookup.get(ref.additional_code)
        if std_err is not None:
            key = (ref.code, ref.additional_code)
            errors[key] = ResolvedError(
                code=ref.code,
                additional_code=ref.additional_code,
                name=texts.get(std_err.name.text_id, std_err.name.text_id),
                description=(texts.get(std_err.description.text_id) if std_err.description else None),
            )

    # Add device-specific errors (code=129)
    for err in device_error_collection.error_type:
        key = (err.code, err.additional_code)
        errors[key] = ResolvedError(
            code=err.code,
            additional_code=err.additional_code,
            name=texts.get(err.name.text_id, err.name.text_id),
            description=(texts.get(err.description.text_id) if err.description else None),
        )

    return errors

`resolve_process_data(process_data_collection, texts, datatypes)`

Resolve process data from the ProcessDataCollection.

Process data defines the structure of cyclic data exchanged between master and device. Multiple ProcessData elements can exist with Condition elements for switching between configurations.

Parameters:

Name	Type	Description	Default
`process_data_collection`	`ProcessDataCollectionT \| None`	The process data collection, may be None.	required
`texts`	`dict[str, str]`	Dictionary of resolved text strings.	required
`datatypes`	`dict[str, DatatypeT]`	Dictionary of available datatypes.	required

Returns:

Type	Description
`dict[str, ResolvedProcessData]`	Dictionary of resolved process data keyed by ProcessData id.

Source code in src/iodd_parser/resolvers.py

def resolve_process_data(
    process_data_collection: ProcessDataCollectionT | None,
    texts: dict[str, str],
    datatypes: dict[str, DatatypeT],
) -> dict[str, ResolvedProcessData]:
    """
    Resolve process data from the ProcessDataCollection.

    Process data defines the structure of cyclic data exchanged between
    master and device. Multiple ProcessData elements can exist with
    Condition elements for switching between configurations.

    :param process_data_collection: The process data collection, may be None.
    :param texts: Dictionary of resolved text strings.
    :param datatypes: Dictionary of available datatypes.
    :returns: Dictionary of resolved process data keyed by ProcessData id.
    """
    if process_data_collection is None:
        return {}

    result: dict[str, ResolvedProcessData] = {}

    for pd in process_data_collection.process_data:
        # Extract condition information if present
        condition_var_id = None
        condition_subindex = None
        condition_value = None

        if pd.condition is not None:
            condition_var_id = pd.condition.variable_id
            condition_subindex = pd.condition.subindex
            condition_value = pd.condition.value

        # Resolve ProcessDataIn
        pd_in = None
        if pd.process_data_in is not None:
            pdi = pd.process_data_in
            raw_dt = _get_datatype_from_data_item(pdi, datatypes)
            pd_in = ResolvedProcessDataItem(
                id=pdi.id,
                bit_length=pdi.bit_length,
                name=texts.get(pdi.name.text_id, pdi.name.text_id),
                datatype=_resolve_datatype(raw_dt, texts, datatypes),
            )

        # Resolve ProcessDataOut
        pd_out = None
        if pd.process_data_out is not None:
            pdo = pd.process_data_out
            raw_dt = _get_datatype_from_data_item(pdo, datatypes)
            pd_out = ResolvedProcessDataItem(
                id=pdo.id,
                bit_length=pdo.bit_length,
                name=texts.get(pdo.name.text_id, pdo.name.text_id),
                datatype=_resolve_datatype(raw_dt, texts, datatypes),
            )

        result[pd.id] = ResolvedProcessData(
            id=pd.id,
            process_data_in=pd_in,
            process_data_out=pd_out,
            condition_variable_id=condition_var_id,
            condition_subindex=condition_subindex,
            condition_value=condition_value,
        )

    return result

`resolve_texts(loaded_definitions, loaded_units, device, lang, standard_lang_texts, device_lang_texts)`

Resolve text collections with language support.

Texts are resolved in the following priority order (later sources override earlier ones):

Primary language (English) from standard definitions
Primary language from device IODD
Primary language from standard unit definitions (English only)
Pre-loaded language-specific standard definitions file (if available)
Language sections within main standard definitions file
Language sections within main device IODD file
Device-specific language file (e.g., *-IODD1.1-de.xml) - highest priority

Parameters:

Name	Type	Description	Default
`loaded_definitions`	`IoddstandardDefinitions`	The loaded standard definitions.	required
`loaded_units`	`IoddstandardUnitDefinitions`	The loaded standard unit definitions.	required
`device`	`Iodevice`	The parsed device IODD.	required
`lang`	`str \| None`	Optional language code (e.g., "de", "fr").	required
`standard_lang_texts`	`dict[str, dict[str, str]]`	Pre-loaded language-specific standard definitions texts, mapping language code to text dictionaries.	required
`device_lang_texts`	`dict[str, dict[str, str]]`	Texts from device-specific language files, mapping language code to text dictionaries.	required

Returns:

Type	Description
`dict[str, str]`	Dictionary of resolved text strings keyed by text id.

Source code in src/iodd_parser/resolvers.py

def resolve_texts(
    loaded_definitions: IoddstandardDefinitions,
    loaded_units: IoddstandardUnitDefinitions,
    device: Iodevice,
    lang: str | None,
    standard_lang_texts: dict[str, dict[str, str]],
    device_lang_texts: dict[str, dict[str, str]],
) -> dict[str, str]:
    """
    Resolve text collections with language support.

    Texts are resolved in the following priority order (later sources override earlier ones):

    1. Primary language (English) from standard definitions
    2. Primary language from device IODD
    3. Primary language from standard unit definitions (English only)
    4. Pre-loaded language-specific standard definitions file (if available)
    5. Language sections within main standard definitions file
    6. Language sections within main device IODD file
    7. Device-specific language file (e.g., ``*-IODD1.1-de.xml``) - highest priority

    :param loaded_definitions: The loaded standard definitions.
    :param loaded_units: The loaded standard unit definitions.
    :param device: The parsed device IODD.
    :param lang: Optional language code (e.g., "de", "fr").
    :param standard_lang_texts: Pre-loaded language-specific standard definitions texts,
        mapping language code to text dictionaries.
    :param device_lang_texts: Texts from device-specific language files,
        mapping language code to text dictionaries.
    :returns: Dictionary of resolved text strings keyed by text id.
    """
    # Always start with the primary language (English) as base
    texts: dict[str, str] = {t.id: t.value for t in loaded_definitions.external_text_collection.primary_language.text}
    texts.update({t.id: t.value for t in device.external_text_collection.primary_language.text})
    texts.update({t.id: t.value for t in loaded_units.external_text_collection.primary_language.text})

    # If a specific language is requested, overlay those texts on top
    if lang:
        # First, apply pre-loaded language-specific standard definitions texts
        if lang in standard_lang_texts:
            texts.update(standard_lang_texts[lang])

        # Also check for language sections within the main definitions file (fallback)
        lang_dfs = next(
            (x for x in loaded_definitions.external_text_collection.language if _lang_id(x) == lang),
            None,
        )
        if lang_dfs is not None:
            texts.update({t.id: t.value for t in lang_dfs.text})

        # Check for language sections within the main device IODD file
        lang_dev = next(
            (x for x in device.external_text_collection.language if _lang_id(x) == lang),
            None,
        )
        if lang_dev is not None:
            texts.update({t.id: t.value for t in lang_dev.text})

        # Finally, overlay device-specific language file texts (highest priority)
        if lang in device_lang_texts:
            texts.update(device_lang_texts[lang])

    return texts

`resolve_units(unit_definitions, texts)`

Resolve units from standard unit definitions.

Parameters:

Name	Type	Description	Default
`unit_definitions`	`IoddstandardUnitDefinitions`	The parsed standard unit definitions XML.	required
`texts`	`dict[str, str]`	Dictionary of resolved text strings.	required

Returns:

Type	Description
`dict[int, ResolvedUnit]`	Dictionary of resolved units keyed by unit code.

Source code in src/iodd_parser/resolvers.py

def resolve_units(
    unit_definitions: IoddstandardUnitDefinitions,
    texts: dict[str, str],
) -> dict[int, ResolvedUnit]:
    """
    Resolve units from standard unit definitions.

    :param unit_definitions: The parsed standard unit definitions XML.
    :param texts: Dictionary of resolved text strings.
    :returns: Dictionary of resolved units keyed by unit code.
    """
    units: dict[int, ResolvedUnit] = {}
    for unit in unit_definitions.unit_collection.unit:
        units[unit.code] = ResolvedUnit(
            code=unit.code,
            abbreviation=unit.abbr,
            name=texts.get(unit.text_id, unit.text_id),
        )
    return units

`resolve_user_interface(user_interface, texts)`

Resolve the UserInterface element to a ResolvedUserInterface.

The UserInterface contains: - ProcessDataRefCollection (optional): Display info for process data - MenuCollection: All menu definitions - Three role-based MenuSets: Observer, Maintenance, Specialist

Each role has fixed top-level menus (Identification, Parameter, Observation, Diagnosis) that reference menus from the MenuCollection.

Parameters:

Name	Type	Description	Default
`user_interface`	`UserInterfaceT`	The user interface element from DeviceFunction.	required
`texts`	`dict[str, str]`	Dictionary of resolved text strings.	required

Returns:

Type	Description
`ResolvedUserInterface`	Resolved user interface with all menus and role assignments.

Source code in src/iodd_parser/resolvers.py

def resolve_user_interface(
    user_interface: UserInterfaceT,
    texts: dict[str, str],
) -> ResolvedUserInterface:
    """
    Resolve the UserInterface element to a ResolvedUserInterface.

    The UserInterface contains:
    - ProcessDataRefCollection (optional): Display info for process data
    - MenuCollection: All menu definitions
    - Three role-based MenuSets: Observer, Maintenance, Specialist

    Each role has fixed top-level menus (Identification, Parameter,
    Observation, Diagnosis) that reference menus from the MenuCollection.

    :param user_interface: The user interface element from DeviceFunction.
    :param texts: Dictionary of resolved text strings.
    :returns: Resolved user interface with all menus and role assignments.
    """
    # Resolve all menus from MenuCollection
    menus: dict[str, ResolvedMenu] = {}
    for menu in user_interface.menu_collection.menu:
        menus[menu.id] = _resolve_menu(menu, texts)

    # Resolve the three role menu sets
    observer_menu_set = _resolve_menu_set(user_interface.observer_role_menu_set)
    maintenance_menu_set = _resolve_menu_set(user_interface.maintenance_role_menu_set)
    specialist_menu_set = _resolve_menu_set(user_interface.specialist_role_menu_set)

    # Resolve process data references
    process_data_refs = _resolve_process_data_refs(user_interface.process_data_ref_collection)

    return ResolvedUserInterface(
        menus=menus,
        observer_role_menu_set=observer_menu_set,
        maintenance_role_menu_set=maintenance_menu_set,
        specialist_role_menu_set=specialist_menu_set,
        process_data_refs=process_data_refs,
    )

`resolve_variables(std_variable_collection, device_variable_collection, texts, datatypes)`

Resolve variables from standard definitions and device variable collection.

Variables come from three sources:

StdVariableRef: references to standard variables from IODD-StandardDefinitions1.1.xml
DirectParameterOverlay: device-specific data within DirectParameter page
Variable: vendor-specific variables with a device-specific index

Parameters:

Name	Type	Description	Default
`std_variable_collection`	`IoddstandardVariableCollectionT`	Standard variables from definitions XML.	required
`device_variable_collection`	`VariableCollectionT`	Device variable collection from IODD.	required
`texts`	`dict[str, str]`	Dictionary of resolved text strings.	required
`datatypes`	`dict[str, DatatypeT]`	Dictionary of available datatypes.	required

Returns:

Type	Description
`dict[str, ResolvedVariable]`	Dictionary of resolved variables keyed by variable id.

Source code in src/iodd_parser/resolvers.py

def resolve_variables(
    std_variable_collection: IoddstandardVariableCollectionT,
    device_variable_collection: VariableCollectionT,
    texts: dict[str, str],
    datatypes: dict[str, DatatypeT],
) -> dict[str, ResolvedVariable]:
    """
    Resolve variables from standard definitions and device variable collection.

    Variables come from three sources:

    - **StdVariableRef**: references to standard variables from
      IODD-StandardDefinitions1.1.xml
    - **DirectParameterOverlay**: device-specific data within DirectParameter page
    - **Variable**: vendor-specific variables with a device-specific index

    :param std_variable_collection: Standard variables from definitions XML.
    :param device_variable_collection: Device variable collection from IODD.
    :param texts: Dictionary of resolved text strings.
    :param datatypes: Dictionary of available datatypes.
    :returns: Dictionary of resolved variables keyed by variable id.
    """
    variables: dict[str, ResolvedVariable] = {}

    # Build lookup for standard variables by id
    std_vars_lookup: dict[str, IoddstandardVariableT] = {var.id: var for var in std_variable_collection.variable}

    # 1. Add referenced standard variables (StdVariableRef)
    for ref in device_variable_collection.std_variable_ref:
        std_var = std_vars_lookup.get(ref.id)
        if std_var is None:
            continue

        raw_datatype = _get_datatype(std_var, datatypes)
        resolved_datatype = _resolve_datatype(raw_datatype, texts, datatypes)

        variables[ref.id] = ResolvedVariable(
            id=ref.id,
            index=std_var.index,
            datatype=resolved_datatype,
            name=texts.get(std_var.name.text_id, std_var.name.text_id),
            description=(texts.get(std_var.description.text_id) if std_var.description else None),
            access_rights=std_var.access_rights,
            dynamic=std_var.dynamic,
            modifies_other_variables=std_var.modifies_other_variables,
            # excludedFromDataStorage can be overridden by StdVariableRef
            excluded_from_data_storage=(ref.excluded_from_data_storage or std_var.excluded_from_data_storage),
            default_value=ref.default_value,
            fixed_length_restriction=ref.fixed_length_restriction,
            record_item_info=std_var.record_item_info,
        )

    # 2. Add DirectParameterOverlay if present (index=1 for V_DirectParameters_2)
    if device_variable_collection.direct_parameter_overlay is not None:
        overlay = device_variable_collection.direct_parameter_overlay
        raw_datatype = _get_datatype(overlay, datatypes)
        resolved_datatype = _resolve_datatype(raw_datatype, texts, datatypes)

        variables[overlay.id] = ResolvedVariable(
            id=overlay.id,
            index=1,  # DirectParameterOverlay maps to index 1 (V_DirectParameters_2)
            datatype=resolved_datatype,
            name=texts.get(overlay.name.text_id, overlay.name.text_id),
            description=(texts.get(overlay.description.text_id) if overlay.description else None),
            access_rights=overlay.access_rights,
            dynamic=overlay.dynamic,
            modifies_other_variables=overlay.modifies_other_variables,
            excluded_from_data_storage=overlay.excluded_from_data_storage,
            record_item_info=overlay.record_item_info,
        )

    # 3. Add vendor-specific variables (Variable)
    for var in device_variable_collection.variable:
        raw_datatype = _get_datatype(var, datatypes)
        resolved_datatype = _resolve_datatype(raw_datatype, texts, datatypes)

        variables[var.id] = ResolvedVariable(
            id=var.id,
            index=var.index,
            datatype=resolved_datatype,
            name=texts.get(var.name.text_id, var.name.text_id),
            description=(texts.get(var.description.text_id) if var.description else None),
            access_rights=var.access_rights,
            dynamic=var.dynamic,
            modifies_other_variables=var.modifies_other_variables,
            excluded_from_data_storage=var.excluded_from_data_storage,
            default_value=var.default_value,
            record_item_info=var.record_item_info,
        )

    return variables