asyncua.common.structures104

Complexity

Total Complexity

61

Size/Duplication

Total Lines	254
Duplicated Lines	0 %

Importance

Changes

0

9 Functions

Rating	Name	Duplication	Size	Complexity
C	get_default_value()	0	23	11
A	clean_name()	0	11	2
A	load_enums()	0	14	5
A	_recursive_parse()	0	8	3
A	_read_data_type_definition()	0	18	5
A	load_data_type_definitions()	0	13	4
A	make_enum_code()	0	20	2
B	_generate_object()	0	27	7
F	make_structure_code()	0	71	18

from enum import Enum
from datetime import datetime
import uuid
from enum import IntEnum
import logging
import re

from asyncua import ua

logger = logging.getLogger(__name__)


def clean_name(name):
    """
    Remove characters that might be present in  OPC UA structures
    but cannot be part of of Python class names
    """
    newname = re.sub(r'\W+', '_', name)
    newname = re.sub(r'^[0-9]+', r'_\g<0>', newname)

    if name != newname:
        logger.warning("renamed %s to %s due to Python syntax", name, newname)
    return newname


def get_default_value(uatype, enums=None):
    if enums is None:
        enums = {}
    if uatype == "String":
        return "None"
    elif uatype == "Guid":
        return "uuid.uuid4()"
    elif uatype in ("ByteString", "CharArray", "Char"):
        return b''
    elif uatype == "Boolean":
        return "True"
    elif uatype == "DateTime":
        return "datetime.utcnow()"
    elif uatype in ("Int16", "Int32", "Int64", "UInt16", "UInt32", "UInt64", "Double", "Float", "Byte", "SByte"):
        return 0
    elif uatype in enums:
        return f"ua.{uatype}({enums[uatype]})"
    elif hasattr(ua, uatype) and issubclass(getattr(ua, uatype), Enum):
        # We have an enum, try to initilize it correctly
        val = list(getattr(ua, uatype).__members__)[0]
        return f"ua.{uatype}.{val}"
    else:
        return f"ua.{uatype}()"


def make_structure_code(data_type, name, sdef):
    """
    given a StructureDefinition object, generate Python code
    """
    if sdef.StructureType not in (ua.StructureType.Structure, ua.StructureType.StructureWithOptionalFields):
        #if sdef.StructureType != ua.StructureType.Structure:
        raise NotImplementedError(f"Only StructureType implemented, not {ua.StructureType(sdef.StructureType).name} for node {name} with DataTypdeDefinition {sdef}")

    code = f"""

class {name}:

    '''
    {name} structure autogenerated from StructureDefinition object
    '''

    data_type = ua.NodeId({data_type.Identifier}, {data_type.NamespaceIndex})

"""
    counter = 0
    if sdef.StructureType == ua.StructureType.StructureWithOptionalFields:
        code += '    ua_switches = {\n'
        for field in sdef.Fields:

            if field.IsOptional:
                code += f"        '{field.Name}': ('Encoding', {counter}),\n"
                counter += 1
        code += "    }\n\n"

    code += '    ua_types = [\n'
    if sdef.StructureType == ua.StructureType.StructureWithOptionalFields:
        code += f"        ('Encoding', 'Byte'),\n"
    uatypes = []
    for field in sdef.Fields:
        prefix = 'ListOf' if field.ValueRank >= 1 else ''
        if field.DataType.NamespaceIndex == 0 and field.DataType.Identifier in ua.ObjectIdNames:
            uatype = ua.ObjectIdNames[field.DataType.Identifier]
        elif field.DataType in ua.extension_objects_by_datatype:
            uatype = ua.extension_objects_by_datatype[field.DataType].__name__
        elif field.DataType in ua.enums_by_datatype:
            uatype = ua.enums_by_datatype[field.DataType].__name__
        else:
            #FIXME: we are probably missing many custom tyes here based on builtin types
            #maybe we can use ua_utils.get_base_data_type()
            raise RuntimeError(f"Unknown datatype for field: {field} in structure:{name}, please report")
        if field.ValueRank >= 1 and uatype == 'Char':
            uatype = 'String'
        uatypes.append((field, uatype))
        code += f"        ('{field.Name}', '{prefix + uatype}'),\n"
    code += "    ]\n"
    code += f"""
    def __str__(self):
        vals = [f"{{name}}:{{val}}" for name, val in self.__dict__.items()]
        return f"{name}({{','.join(vals)}})"

    __repr__ = __str__

    def __init__(self):
"""
    if not sdef.Fields:
        code += "      pass"
    if sdef.StructureType == ua.StructureType.StructureWithOptionalFields:
        code += f"        self.Encoding = 0\n"
    for field, uatype in uatypes:
        if field.ValueRank >= 1:
            default_value = "[]"
        else:
            default_value = get_default_value(uatype)
        code += f"        self.{field.Name} = {default_value}\n"
    print("CODE", code)
    return code


async def _generate_object(name, sdef, data_type=None, env=None, enum=False):
    """
    generate Python code and execute in a new environment
    return a dict of structures {name: class}
    Rmw: Since the code is generated on the fly, in case of error the stack trace is
    not available and debugging is very hard...
    """
    if env is None:
        env = {}
    #  Add the required libraries to dict
    if "ua" not in env:
        env['ua'] = ua
    if "datetime" not in env:
        env['datetime'] = datetime
    if "uuid" not in env:
        env['uuid'] = uuid
    if "enum" not in env:
        env['IntEnum'] = IntEnum
    # generate classe add it to env dict
    if enum:
        code = make_enum_code(name, sdef)
    else:
        code = make_structure_code(data_type, name, sdef)
    logger.debug("Executing code: %s", code)
    print("CODE", code)
    exec(code, env)
    return env


class DataTypeSorter:
    def __init__(self, data_type, name, desc, sdef):
        self.data_type = data_type
        self.name = name
        self.desc = desc
        self.sdef = sdef
        self.encoding_id = self.sdef.DefaultEncodingId
        self.deps = [field.DataType for field in self.sdef.Fields]

    def __lt__(self, other):
        if self.desc.NodeId in other.deps:
            return True
        return False

    def __str__(self):
        return f"{self.__class__.__name__}({self.desc.NodeId, self.deps, self.encoding_id})"

    __repr__ = __str__


async def _recursive_parse(server, base_node, dtypes):
    for desc in await base_node.get_children_descriptions(refs=ua.ObjectIds.HasSubtype):
        sdef = await _read_data_type_definition(server, desc)
        if not sdef:
            continue
        name = clean_name(desc.BrowseName.Name)
        dtypes.append(DataTypeSorter(desc.NodeId, name, desc, sdef))
        await _recursive_parse(server, server.get_node(desc.NodeId), dtypes)


async def load_data_type_definitions(server, base_node=None):
    await load_enums(server)  # we need all enums to generate structure code
    if base_node is None:
        base_node = server.nodes.base_structure_type
    dtypes = []
    await _recursive_parse(server, base_node, dtypes)
    dtypes.sort()
    for dts in dtypes:
        try:
            env = await _generate_object(dts.name, dts.sdef, data_type=dts.data_type)
            ua.register_extension_object(dts.name, dts.encoding_id, env[dts.name], dts.desc.NodeId)
        except NotImplementedError:
            logger.exception("Structure type %s not implemented", dts.sdef)


async def _read_data_type_definition(server, desc):
    if desc.BrowseName.Name == "FilterOperand":
        #FIXME: find out why that one is not in ua namespace...
        return None
    # FIXME: this is fishy, we may have same name in different Namespaces
    if hasattr(ua, desc.BrowseName.Name):
        return None
    logger.warning("Registring data type %s %s", desc.NodeId, desc.BrowseName)
    node = server.get_node(desc.NodeId)
    try:
        sdef = await node.read_data_type_definition()
    except ua.uaerrors.BadAttributeIdInvalid:
        logger.warning("%s has no DataTypeDefinition atttribute", node)
        return None
    except Exception:
        logger.exception("Error getting datatype for node %s", node)
        return None
    return sdef


def make_enum_code(name, edef):
    """
    if node has a DataTypeDefinition arttribute, generate enum code
    """
    code = f"""

class {name}(IntEnum):

    '''
    {name} EnumInt autogenerated from EnumDefinition
    '''

"""

    for field in edef.Fields:
        name = field.Name
        value = field.Value
        code += f"    {name} = {value}\n"

    return code


async def load_enums(server, base_node=None):
    if base_node is None:
        base_node = server.nodes.enum_data_type
    for desc in await base_node.get_children_descriptions(refs=ua.ObjectIds.HasSubtype):
        if hasattr(ua, desc.BrowseName.Name):
            continue
        logger.warning("Registring Enum %s %s", desc.NodeId, desc.BrowseName)
        name = clean_name(desc.BrowseName.Name)
        sdef = await _read_data_type_definition(server, desc)
        if not sdef:
            continue
        node = server.get_node(desc.NodeId)
        env = await _generate_object(name, sdef, enum=True)
        ua.register_enum(name, desc.NodeId, env[name])