Load custom struct from one node recursively

[ArthurBerettaVerkor]

Hello all, I made this function to load a custom struct from an OPC-UA server for only one node. I am using it to avoid the client.load_data_type_definitions(). I had a hard time with badly configured PLC that contain way too many custom structs to load, when I only needed a few to read.

To use it, just pass the node data type and the client to the function:

load_children_custom_struct_recursively(client, client.get_node(await node.read_data_type()))

Here is the script:

from typing import Any

from asyncua import Client, Node, ua
from asyncua.common.structures104 import _generate_object, clean_name, load_enums
from asyncua.ua.uaerrors import UaStatusCodeError
from loguru import logger


async def load_children_custom_struct_recursively(client: Client, data_type_node: Node):
    data_type_type_definition = await data_type_node.read_data_type_definition()
    _raise_on_not_structure_def(data_type_type_definition)
    for field in data_type_type_definition.Fields:
        field_data_type_node = client.get_node(field.DataType)
        try:
            if isinstance(
                await field_data_type_node.read_data_type_definition(),
                ua.StructureDefinition,
            ):
                await load_children_custom_struct_recursively(
                    client, field_data_type_node
                )
        except UaStatusCodeError:
            ...
    await _load_custom_struct(data_type_node)


def _raise_on_not_structure_def(data_type_definition: ua.DataTypeDefinition):
    if not isinstance(data_type_definition, ua.StructureDefinition):
        raise TypeError(
            "data_type_node must be an ua.StructureDefinition, got "
            + f"{type(data_type_definition)}"
        )


async def _load_custom_struct(node: Node) -> Any:
    data_type_definition = await node.read_data_type_definition()
    _raise_on_not_structure_def(data_type_definition)
    name = clean_name((await node.read_browse_name()).Name)

    env = await _generate_object(name, data_type_definition, data_type=node.nodeid)
    struct = env[name]
    ua.register_extension_object(
        name, data_type_definition.DefaultEncodingId, struct, node.nodeid
    )
    return env[name]

Is this function properly implemented, or did I miss something ? One thing I did not understand was the parent gathering in the asyncua.common.structures104.load_custom_struct function: why is it present ? And do I need it ?

Don't hesitate to correct me as I wish to use this function in production, so all critics are most welcome 😄

Kind regards,
AB

[schroeder-]

The code looks good, misses a lot of corner cases. Like renaming of basetypes (MyString = ua.String), optionfields, datatypes that reference them self and complicated datatype relationships. This stuff is not used that widely. All this makes our load_datatype_definiton complex.

Never used load_custom_struct, so no glue what it's purpose is.

[ArthurBerettaVerkor]

Hey, thank you for your answer, can you point me examples into the asyncua codebase to harden this function ?