asyncua.common.xmlimporter.XmlImporter._check_required_models() - Code Metrics - Inspection of "Check xmlimport requirements" - FreeOpcUa/opcua-asyncio - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#358)

unknown

created 2020-12-15 07:42 UTC

XmlImporter._check_required_models() C

↳ Parent: asyncua.common.xmlimporter

Complexity

Conditions

Size

Total Lines	19
Code Lines	18

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	11
eloc	18
nop	3
dl	0
loc	19
rs	5.4
c	0
b	0
f	0

How to fix Complexity

"""
add nodes defined in XML to address space
format is the one from opc-ua specification
"""
import logging
import uuid
from typing import Coroutine, Union, Dict
from copy import copy

from asyncua import ua
from .xmlparser import XMLParser, ua_type_to_python
from ..ua.uaerrors import UaError


_logger = logging.getLogger(__name__)


class XmlImporter:

    def __init__(self, server):
        self.parser = None
        self.server = server
        self.namespaces: Dict[int, int] = {}  #Dict[IndexInXml, IndexInServer]
        self.aliases: Dict[str, ua.NodeId] = {}
        self.refs = None

    async def _map_namespaces(self, namespaces_uris):
        """
        creates a mapping between the namespaces in the xml file and in the server.
        if not present the namespace is registered.
        """
        namespaces = {}
        for ns_index, ns_uri in enumerate(namespaces_uris):
            ns_server_index = await self.server.register_namespace(ns_uri)
            namespaces[ns_index + 1] = ns_server_index
        return namespaces

    def _map_aliases(self, aliases: dict):
        """
        maps the import aliases to the correct namespaces
        """
        aliases_mapped = {}
        for alias, node_id in aliases.items():
            aliases_mapped[alias] = self._to_migrated_nodeid(node_id)
        return aliases_mapped


    async def _get_existing_model_in_namespace(self):
        server_model_list = []
        server_namespaces_node = await self.server.nodes.namespaces.get_children()
        for model_node in server_namespaces_node:
            server_model_list.append({"ModelUri": await(await model_node.get_child("NamespaceUri")).read_value(),
                                      "Version": await(await model_node.get_child("NamespaceVersion")).read_value(),
                                      "PublicationDate": str(await(
                                          await model_node.get_child("NamespacePublicationDate")).read_value())})
        return server_model_list

    async def _check_required_models(self, xmlpath=None, xmlstring=None):
        req_models = self.parser.list_required_models(xmlpath, xmlstring)
        if not req_models:
            return None
        server_model_list = await self._get_existing_model_in_namespace()
        for model in server_model_list:
            for req_model in req_models:
                if model["ModelUri"] == req_model["ModelUri"] and \
                        model["PublicationDate"] >= req_model["PublicationDate"]:
                            if "Version" in model and "Version" in req_model:
                                if model["Version"] >= req_model["Version"]:
                                    req_models.remove(req_model)
                            else:
                                req_models.remove(req_model)
        if len(req_models):
            for missing_model in server_model_list:
                _logger.warning("Model is missing: ", missing_model)
            raise ValueError("Server doesn't satisfy required XML-Models. Import them first!")
        return None

    async def import_xml(self, xmlpath=None, xmlstring=None):
        """
        import xml and return added nodes
        """
        if not (bool(xmlpath) != bool(xmlstring)):
            raise ValueError("Expected either xmlpath or xmlstring, not both or neither.")
        _logger.info("Importing XML file %s", xmlpath)
        self.parser = XMLParser()
        await self._check_required_models(xmlpath, xmlstring)
        await self.parser.parse(xmlpath, xmlstring)
        self.namespaces = await self._map_namespaces(self.parser.get_used_namespaces())
        _logger.info("namespace map: %s", self.namespaces)
        self.aliases = self._map_aliases(self.parser.get_aliases())
        self.refs = []
        dnodes = self.parser.get_node_datas()
        dnodes = self.make_objects(dnodes)
        self._add_missing_parents(dnodes)
        nodes_parsed = self._sort_nodes_by_parentid(dnodes)
        nodes = []
        for nodedata in nodes_parsed:  # self.parser:
            try:
                node = await self._add_node_data(nodedata)
            except Exception:
                _logger.warning("failure adding node %s", nodedata)
                raise
            nodes.append(node)
        self.refs, remaining_refs = [], self.refs
        await self._add_references(remaining_refs)
        if len(self.refs):
            _logger.warning("The following references could not be imported and are probably broken: %s", self.refs)
        return nodes

    def _add_missing_parents(self, dnodes):
        missing = []
        childs = {}
        for nd in dnodes:
            if not nd.parent or nd.parent == nd.nodeid:
                missing.append(nd)
            for ref in nd.refs:
                if ref.forward:
                    if ref.reftype in [self.server.nodes.HasComponent.nodeid,
                                       self.server.nodes.HasProperty.nodeid,
                                       self.server.nodes.Organizes.nodeid]:
                        # if a node has several links, the last one will win
                        if ref.target in childs:
                            _logger.warning("overwriting parent target, shouldbe fixed",
                                            ref.target, nd.nodeid, ref.reftype, childs[ref.target])
                        childs[ref.target] = (nd.nodeid, ref.reftype)
        for nd in missing:
            if nd.nodeid in childs:
                target, reftype = childs[nd.nodeid]
                nd.parent = target
                nd.parentlink = reftype
                from IPython import embed
                embed()

    async def _add_node_data(self, nodedata) -> "Node":
        if nodedata.nodetype == "UAObject":
            node = await self.add_object(nodedata)
        elif nodedata.nodetype == "UAObjectType":
            node = await self.add_object_type(nodedata)
        elif nodedata.nodetype == "UAVariable":
            node = await self.add_variable(nodedata)
        elif nodedata.nodetype == "UAVariableType":
            node = await self.add_variable_type(nodedata)
        elif nodedata.nodetype == "UAReferenceType":
            node = await self.add_reference_type(nodedata)
        elif nodedata.nodetype == "UADataType":
            node = await self.add_datatype(nodedata)
        elif nodedata.nodetype == "UAMethod":
            node = await self.add_method(nodedata)
        else:
            raise ValueError(f"Not implemented node type: {nodedata.nodetype} ")
        return node

    def _get_server(self):
        if hasattr(self.server, "iserver"):
            return self.server.iserver.isession
        else:
            return self.server.uaclient

    async def _add_references(self, refs):
        res = await self._get_server().add_references(refs)

        for sc, ref in zip(res, refs):
            if not sc.is_good():
                self.refs.append(ref)

    def make_objects(self, node_data):
        new_nodes = []
        for node_datum in node_data:
            node_datum.nodeid = self._to_migrated_nodeid(node_datum.nodeid)
            node_datum.browsename = ua.QualifiedName.from_string(node_datum.browsename)
            if node_datum.parent:
                node_datum.parent = self._to_migrated_nodeid(node_datum.parent)
            if node_datum.parentlink:
                node_datum.parentlink = self._to_migrated_nodeid(node_datum.parentlink)
            if node_datum.typedef:
                node_datum.typedef = self._to_migrated_nodeid(node_datum.typedef)
            if node_datum.datatype:
                node_datum.datatype = self._to_migrated_nodeid(node_datum.datatype)
            new_nodes.append(node_datum)
            for ref in node_datum.refs:
                ref.reftype = self._to_migrated_nodeid(ref.reftype)
                ref.target = self._to_migrated_nodeid(ref.target)
            for field in node_datum.definitions:
                if field.datatype:
                    field.datatype = self._to_migrated_nodeid(field.datatype)
        return new_nodes

    def _migrate_ns(self, nodeid: ua.NodeId) -> ua.NodeId:
        """
        Check if the index of nodeid or browsename  given in the xml model file
        must be converted to a already existing namespace id based on the files
        namespace uri

        :returns: NodeId (str)
        """
        if nodeid.NamespaceIndex in self.namespaces:
            nodeid = copy(nodeid)
            nodeid.NamespaceIndex = self.namespaces[nodeid.NamespaceIndex]
        return nodeid

    def _get_add_node_item(self, obj):
        node = ua.AddNodesItem()
        node.RequestedNewNodeId = self._migrate_ns(obj.nodeid)
        node.BrowseName = self._migrate_ns(obj.browsename)
        _logger.info("Importing xml node (%s, %s) as (%s %s)", obj.browsename,
                     obj.nodeid, node.BrowseName, node.RequestedNewNodeId)
        node.NodeClass = getattr(ua.NodeClass, obj.nodetype[2:])
        if obj.parent and obj.parentlink:
            node.ParentNodeId = self._migrate_ns(obj.parent)
            node.ReferenceTypeId = self._migrate_ns(obj.parentlink)
        if obj.typedef:
            node.TypeDefinition = self._migrate_ns(obj.typedef)
        return node

    def _to_migrated_nodeid(self, nodeid: Union[ua.NodeId, None, str]) -> ua.NodeId:
        nodeid = self._to_nodeid(nodeid)
        return self._migrate_ns(nodeid)

    def _to_nodeid(self, nodeid: Union[ua.NodeId, None, str]) -> ua.NodeId:
        if isinstance(nodeid, ua.NodeId):
            return nodeid
        elif not nodeid:
            return ua.NodeId(ua.ObjectIds.String)
        elif "=" in nodeid:
            return ua.NodeId.from_string(nodeid)
        elif hasattr(ua.ObjectIds, nodeid):
            return ua.NodeId(getattr(ua.ObjectIds, nodeid))
        else:
            if nodeid in self.aliases:
                return self.aliases[nodeid]
            else:
                return ua.NodeId(getattr(ua.ObjectIds, nodeid))

    async def add_object(self, obj):
        node = self._get_add_node_item(obj)
        attrs = ua.ObjectAttributes()
        if obj.desc:
            attrs.Description = ua.LocalizedText(obj.desc)
        attrs.DisplayName = ua.LocalizedText(obj.displayname)
        attrs.EventNotifier = obj.eventnotifier
        node.NodeAttributes = attrs
        res = await self._get_server().add_nodes([node])
        await self._add_refs(obj)
        res[0].StatusCode.check()
        return res[0].AddedNodeId

    async def add_object_type(self, obj):
        node = self._get_add_node_item(obj)
        attrs = ua.ObjectTypeAttributes()
        if obj.desc:
            attrs.Description = ua.LocalizedText(obj.desc)
        attrs.DisplayName = ua.LocalizedText(obj.displayname)
        attrs.IsAbstract = obj.abstract
        node.NodeAttributes = attrs
        res = await self._get_server().add_nodes([node])
        await self._add_refs(obj)
        res[0].StatusCode.check()
        return res[0].AddedNodeId

    async def add_variable(self, obj):
        node = self._get_add_node_item(obj)
        attrs = ua.VariableAttributes()
        if obj.desc:
            attrs.Description = ua.LocalizedText(obj.desc)
        attrs.DisplayName = ua.LocalizedText(obj.displayname)
        attrs.DataType = obj.datatype
        if obj.value is not None:
            attrs.Value = self._add_variable_value(obj,)
        if obj.rank:
            attrs.ValueRank = obj.rank
        if obj.accesslevel:
            attrs.AccessLevel = obj.accesslevel
        if obj.useraccesslevel:
            attrs.UserAccessLevel = obj.useraccesslevel
        if obj.minsample:
            attrs.MinimumSamplingInterval = obj.minsample
        if obj.dimensions:
            attrs.ArrayDimensions = obj.dimensions
        node.NodeAttributes = attrs
        res = await self._get_server().add_nodes([node])
        await self._add_refs(obj)
        res[0].StatusCode.check()
        return res[0].AddedNodeId

    def _get_ext_class(self, name: str):
        if hasattr(ua, name):
            return getattr(ua, name)
        elif name in self.aliases.keys():
            nodeid = self.aliases[name]
            class_type = ua.uatypes.get_extensionobject_class_type(nodeid)
            if class_type:
                return class_type
            else:
                raise Exception("Error no extension class registered ", name, nodeid)
        else:
            raise Exception("Error no alias found for extension class", name)

    def _make_ext_obj(self, obj):
        ext = self._get_ext_class(obj.objname)()
        for name, val in obj.body:
            if not isinstance(val, list):
                raise Exception("Error val should be a list, this is a python-asyncua bug", name, type(val), val)
            else:
                for attname, v in val:
                    self._set_attr(ext, attname, v)
        return ext

    def _get_val_type(self, obj, attname: str):
        for name, uatype in obj.ua_types:
            if name == attname:
                return uatype
        raise UaError(f"Attribute '{attname}' defined in xml is not found in object '{obj}'")

    def _set_attr(self, obj, attname: str, val):
        # tow possible values:
        # either we get value directly
        # or a dict if it s an object or a list
        if isinstance(val, str):
            pval = ua_type_to_python(val, self._get_val_type(obj, attname))
            setattr(obj, attname, pval)
        else:
            # so we have either an object or a list...
            obj2 = getattr(obj, attname)
            if isinstance(obj2, ua.NodeId):  # NodeId representation does not follow common rules!!
                for attname2, v2 in val:
                    if attname2 == "Identifier":
                        if hasattr(ua.ObjectIds, v2):
                            obj2 = ua.NodeId(getattr(ua.ObjectIds, v2))
                        else:
                            obj2 = ua.NodeId.from_string(v2)
                        setattr(obj, attname, self._migrate_ns(obj2))
                        break
            elif not hasattr(obj2, "ua_types"):
                # we probably have a list
                my_list = []
                for vtype, v2 in val:
                    my_list.append(ua_type_to_python(v2, vtype))
                setattr(obj, attname, my_list)
            else:
                for attname2, v2 in val:
                    self._set_attr(obj2, attname2, v2)
                setattr(obj, attname, obj2)

    def _add_variable_value(self, obj):
        """
        Returns the value for a Variable based on the objects value type.
        """
        _logger.debug("Setting value with type %s and value %s", obj.valuetype, obj.value)
        if obj.valuetype == 'ListOfExtensionObject':
            values = []
            for ext in obj.value:
                extobj = self._make_ext_obj(ext)
                values.append(extobj)
            return ua.Variant(values, ua.VariantType.ExtensionObject)
        elif obj.valuetype == 'ListOfGuid':
            return ua.Variant([
                uuid.UUID(guid) for guid in obj.value
            ], getattr(ua.VariantType, obj.valuetype[6:]))
        elif obj.valuetype.startswith("ListOf"):
            vtype = obj.valuetype[6:]
            if hasattr(ua.ua_binary.Primitives, vtype):
                return ua.Variant(obj.value, getattr(ua.VariantType, vtype))
            elif vtype == "LocalizedText":
                return ua.Variant([getattr(ua, vtype)(text=item["Text"], locale=item["Locale"]) for item in obj.value])
            else:
                return ua.Variant([getattr(ua, vtype)(v) for v in obj.value])
        elif obj.valuetype == 'ExtensionObject':
            extobj = self._make_ext_obj(obj.value)
            return ua.Variant(extobj, getattr(ua.VariantType, obj.valuetype))
        elif obj.valuetype == 'Guid':
            return ua.Variant(uuid.UUID(obj.value), getattr(ua.VariantType, obj.valuetype))
        elif obj.valuetype == 'LocalizedText':
            ltext = ua.LocalizedText()
            for name, val in obj.value:
                if name == "Text":
                    ltext.Text = val
                elif name == "Locale":
                    ltext.Locale = val
                else:
                    _logger.warning("While parsing localizedText value, unkown element: %s with val: %s", name, val)
            return ua.Variant(ltext, ua.VariantType.LocalizedText)
        elif obj.valuetype == 'NodeId':
            return ua.Variant(ua.NodeId.from_string(obj.value))
        else:
            return ua.Variant(obj.value, getattr(ua.VariantType, obj.valuetype))

    async def add_variable_type(self, obj):
        node = self._get_add_node_item(obj)
        attrs = ua.VariableTypeAttributes()
        if obj.desc:
            attrs.Description = ua.LocalizedText(obj.desc)
        attrs.DisplayName = ua.LocalizedText(obj.displayname)
        attrs.DataType = obj.datatype
        if obj.value and len(obj.value) == 1:
            attrs.Value = obj.value[0]
        if obj.rank:
            attrs.ValueRank = obj.rank
        if obj.abstract:
            attrs.IsAbstract = obj.abstract
        if obj.dimensions:
            attrs.ArrayDimensions = obj.dimensions
        node.NodeAttributes = attrs
        res = await self._get_server().add_nodes([node])
        await self._add_refs(obj)
        res[0].StatusCode.check()
        return res[0].AddedNodeId

    async def add_method(self, obj):
        node = self._get_add_node_item(obj)
        attrs = ua.MethodAttributes()
        if obj.desc:
            attrs.Description = ua.LocalizedText(obj.desc)
        attrs.DisplayName = ua.LocalizedText(obj.displayname)
        if obj.accesslevel:
            attrs.AccessLevel = obj.accesslevel
        if obj.useraccesslevel:
            attrs.UserAccessLevel = obj.useraccesslevel
        if obj.minsample:
            attrs.MinimumSamplingInterval = obj.minsample
        if obj.dimensions:
            attrs.ArrayDimensions = obj.dimensions
        node.NodeAttributes = attrs
        res = await self._get_server().add_nodes([node])
        await self._add_refs(obj)
        res[0].StatusCode.check()
        return res[0].AddedNodeId

    async def add_reference_type(self, obj):
        node = self._get_add_node_item(obj)
        attrs = ua.ReferenceTypeAttributes()
        if obj.desc:
            attrs.Description = ua.LocalizedText(obj.desc)
        attrs.DisplayName = ua.LocalizedText(obj.displayname)
        if obj. inversename:
            attrs.InverseName = ua.LocalizedText(obj.inversename)
        if obj.abstract:
            attrs.IsAbstract = obj.abstract
        if obj.symmetric:
            attrs.Symmetric = obj.symmetric
        node.NodeAttributes = attrs
        res = await self._get_server().add_nodes([node])
        await self._add_refs(obj)
        res[0].StatusCode.check()
        return res[0].AddedNodeId

    async def add_datatype(self, obj):
        node = self._get_add_node_item(obj)
        attrs = ua.DataTypeAttributes()
        if obj.desc:
            attrs.Description = ua.LocalizedText(obj.desc)
        attrs.DisplayName = ua.LocalizedText(obj.displayname)
        if obj.abstract:
            attrs.IsAbstract = obj.abstract
        if not obj.definitions:
            pass
        else:
            if obj.parent == self.server.nodes.enum_data_type.nodeid:
                attrs.DataTypeDefinition = self._get_edef(node, obj)
            elif obj.parent == self.server.nodes.base_structure_type.nodeid:
                attrs.DataTypeDefinition = self._get_sdef(node, obj)
            else:
                parent_node = self.server.get_node(obj.parent)
                path = await parent_node.get_path()
                if self.server.nodes.base_structure_type in path:
                    attrs.DataTypeDefinition = self._get_sdef(node, obj)
                else:
                    _logger.warning("%s has datatypedefinition and path %s"
                                    " but we could not find out if this is a struct", obj, path)
        node.NodeAttributes = attrs
        res = await self._get_server().add_nodes([node])
        res[0].StatusCode.check()
        await self._add_refs(obj)
        return res[0].AddedNodeId

    async def _add_refs(self, obj):
        if not obj.refs:
            return
        refs = []
        for data in obj.refs:
            ref = ua.AddReferencesItem()
            ref.IsForward = data.forward
            ref.ReferenceTypeId = data.reftype
            ref.SourceNodeId = obj.nodeid
            ref.TargetNodeId = data.target
            refs.append(ref)
        await self._add_references(refs)

    def _get_edef(self, node, obj):
        if not obj.definitions:
            return None
        edef = ua.EnumDefinition()
        if obj.parent:
            edef.BaseDataType = obj.parent
        for field in obj.definitions:
            f = ua.EnumField()
            f.Name = field.name
            if field.dname:
                f.DisplayName = ua.LocalizedText(text=field.dname)
            else:
                f.DisplayName = ua.LocalizedText(text=field.name)
            f.Value = field.value
            f.Description = ua.LocalizedText(text=field.desc)
            edef.Fields.append(f)
        return edef

    def _get_sdef(self, node, obj):
        if not obj.definitions:
            return None
        sdef = ua.StructureDefinition()
        if obj.parent:
            sdef.BaseDataType = obj.parent
        for data in obj.refs:
            if data.reftype == self.server.nodes.HasEncoding.nodeid:
                # looks likebinary encodingisthe firt one...can someone confirm?
                sdef.DefaultEncodingId = data.target
                break
        optional = False
        for field in obj.definitions:
            f = ua.StructureField()
            f.Name = field.name
            f.DataType = field.datatype
            f.ValueRank = field.valuerank
            f.IsOptional = field.optional
            if f.IsOptional:
                optional = True
            f.ArrayDimensions = field.arraydim
            f.Description = ua.LocalizedText(text=field.desc)
            sdef.Fields.append(f)
        if optional:
            sdef.StructureType = ua.StructureType.StructureWithOptionalFields
        else:
            sdef.StructureType = ua.StructureType.Structure
        return sdef

    def _sort_nodes_by_parentid(self, ndatas):
        """
        Sort the list of nodes according their parent node in order to respect
        the dependency between nodes.

        :param nodes: list of NodeDataObjects
        :returns: list of sorted nodes
        """

        sorted_ndatas = []
        sorted_nodes_ids = []
        all_node_ids = [data.nodeid for data in ndatas]
        while ndatas:
            for ndata in ndatas[:]:
                if ndata.nodeid.NamespaceIndex not in self.namespaces.values() or \
                        ndata.parent is None or \
                        ndata.parent not in all_node_ids:
                    sorted_ndatas.append(ndata)
                    sorted_nodes_ids.append(ndata.nodeid)
                    ndatas.remove(ndata)
                else:
                    # Check if the nodes parent is already in the list of
                    # inserted nodes
                    if ndata.parent in sorted_nodes_ids:
                        sorted_ndatas.append(ndata)
                        sorted_nodes_ids.append(ndata.nodeid)
                        ndatas.remove(ndata)
        return sorted_ndatas


1			"""
2			add nodes defined in XML to address space
3			format is the one from opc-ua specification
4			"""
5			import logging
6			import uuid
7			from typing import Coroutine, Union, Dict
8			from copy import copy
9
10			from asyncua import ua
11			from .xmlparser import XMLParser, ua_type_to_python
12			from ..ua.uaerrors import UaError
13
14
15			_logger = logging.getLogger(__name__)
16
17
18			class XmlImporter:
19
20			def __init__(self, server):
21			self.parser = None
22			self.server = server
23			self.namespaces: Dict[int, int] = {} #Dict[IndexInXml, IndexInServer]
24			self.aliases: Dict[str, ua.NodeId] = {}
25			self.refs = None
26
27			async def _map_namespaces(self, namespaces_uris):
28			"""
29			creates a mapping between the namespaces in the xml file and in the server.
30			if not present the namespace is registered.
31			"""
32			namespaces = {}
33			for ns_index, ns_uri in enumerate(namespaces_uris):
34			ns_server_index = await self.server.register_namespace(ns_uri)
35			namespaces[ns_index + 1] = ns_server_index
36			return namespaces
37
38			def _map_aliases(self, aliases: dict):
39			"""
40			maps the import aliases to the correct namespaces
41			"""
42			aliases_mapped = {}
43			for alias, node_id in aliases.items():
44			aliases_mapped[alias] = self._to_migrated_nodeid(node_id)
45			return aliases_mapped
46
47
48			async def _get_existing_model_in_namespace(self):
49			server_model_list = []
50			server_namespaces_node = await self.server.nodes.namespaces.get_children()
51			for model_node in server_namespaces_node:
52			server_model_list.append({"ModelUri": await(await model_node.get_child("NamespaceUri")).read_value(),
53			"Version": await(await model_node.get_child("NamespaceVersion")).read_value(),
54			"PublicationDate": str(await(
55			await model_node.get_child("NamespacePublicationDate")).read_value())})
56			return server_model_list
57
58			async def _check_required_models(self, xmlpath=None, xmlstring=None):
59			req_models = self.parser.list_required_models(xmlpath, xmlstring)
60			if not req_models:
61			return None
62			server_model_list = await self._get_existing_model_in_namespace()
63			for model in server_model_list:
64			for req_model in req_models:
65			if model["ModelUri"] == req_model["ModelUri"] and \
66			model["PublicationDate"] >= req_model["PublicationDate"]:
67			if "Version" in model and "Version" in req_model:
68			if model["Version"] >= req_model["Version"]:
69			req_models.remove(req_model)
70			else:
71			req_models.remove(req_model)
72			if len(req_models):
73			for missing_model in server_model_list:
74			_logger.warning("Model is missing: ", missing_model)
75			raise ValueError("Server doesn't satisfy required XML-Models. Import them first!")
76			return None
77
78			async def import_xml(self, xmlpath=None, xmlstring=None):
79			"""
80			import xml and return added nodes
81			"""
82			if not (bool(xmlpath) != bool(xmlstring)):
83			raise ValueError("Expected either xmlpath or xmlstring, not both or neither.")
84			_logger.info("Importing XML file %s", xmlpath)
85			self.parser = XMLParser()
86			await self._check_required_models(xmlpath, xmlstring)
87			await self.parser.parse(xmlpath, xmlstring)
88			self.namespaces = await self._map_namespaces(self.parser.get_used_namespaces())
89			_logger.info("namespace map: %s", self.namespaces)
90			self.aliases = self._map_aliases(self.parser.get_aliases())
91			self.refs = []
92			dnodes = self.parser.get_node_datas()
93			dnodes = self.make_objects(dnodes)
94			self._add_missing_parents(dnodes)
95			nodes_parsed = self._sort_nodes_by_parentid(dnodes)
96			nodes = []
97			for nodedata in nodes_parsed: # self.parser:
98			try:
99			node = await self._add_node_data(nodedata)
100			except Exception:
101			_logger.warning("failure adding node %s", nodedata)
102			raise
103			nodes.append(node)
104			self.refs, remaining_refs = [], self.refs
105			await self._add_references(remaining_refs)
106			if len(self.refs):
107			_logger.warning("The following references could not be imported and are probably broken: %s", self.refs)
108			return nodes
109
110			def _add_missing_parents(self, dnodes):
111			missing = []
112			childs = {}
113			for nd in dnodes:
114			if not nd.parent or nd.parent == nd.nodeid:
115			missing.append(nd)
116			for ref in nd.refs:
117			if ref.forward:
118			if ref.reftype in [self.server.nodes.HasComponent.nodeid,
119			self.server.nodes.HasProperty.nodeid,
120			self.server.nodes.Organizes.nodeid]:
121			# if a node has several links, the last one will win
122			if ref.target in childs:
123			_logger.warning("overwriting parent target, shouldbe fixed",
124			ref.target, nd.nodeid, ref.reftype, childs[ref.target])
125			childs[ref.target] = (nd.nodeid, ref.reftype)
126			for nd in missing:
127			if nd.nodeid in childs:
128			target, reftype = childs[nd.nodeid]
129			nd.parent = target
130			nd.parentlink = reftype
131			from IPython import embed
132			embed()
133
134			async def _add_node_data(self, nodedata) -> "Node":
135			if nodedata.nodetype == "UAObject":
136			node = await self.add_object(nodedata)
137			elif nodedata.nodetype == "UAObjectType":
138			node = await self.add_object_type(nodedata)
139			elif nodedata.nodetype == "UAVariable":
140			node = await self.add_variable(nodedata)
141			elif nodedata.nodetype == "UAVariableType":
142			node = await self.add_variable_type(nodedata)
143			elif nodedata.nodetype == "UAReferenceType":
144			node = await self.add_reference_type(nodedata)
145			elif nodedata.nodetype == "UADataType":
146			node = await self.add_datatype(nodedata)
147			elif nodedata.nodetype == "UAMethod":
148			node = await self.add_method(nodedata)
149			else:
150			raise ValueError(f"Not implemented node type: {nodedata.nodetype} ")
151			return node
152
153			def _get_server(self):
154			if hasattr(self.server, "iserver"):
155			return self.server.iserver.isession
156			else:
157			return self.server.uaclient
158
159			async def _add_references(self, refs):
160			res = await self._get_server().add_references(refs)
161
162			for sc, ref in zip(res, refs):
163			if not sc.is_good():
164			self.refs.append(ref)
165
166			def make_objects(self, node_data):
167			new_nodes = []
168			for node_datum in node_data:
169			node_datum.nodeid = self._to_migrated_nodeid(node_datum.nodeid)
170			node_datum.browsename = ua.QualifiedName.from_string(node_datum.browsename)
171			if node_datum.parent:
172			node_datum.parent = self._to_migrated_nodeid(node_datum.parent)
173			if node_datum.parentlink:
174			node_datum.parentlink = self._to_migrated_nodeid(node_datum.parentlink)
175			if node_datum.typedef:
176			node_datum.typedef = self._to_migrated_nodeid(node_datum.typedef)
177			if node_datum.datatype:
178			node_datum.datatype = self._to_migrated_nodeid(node_datum.datatype)
179			new_nodes.append(node_datum)
180			for ref in node_datum.refs:
181			ref.reftype = self._to_migrated_nodeid(ref.reftype)
182			ref.target = self._to_migrated_nodeid(ref.target)
183			for field in node_datum.definitions:
184			if field.datatype:
185			field.datatype = self._to_migrated_nodeid(field.datatype)
186			return new_nodes
187
188			def _migrate_ns(self, nodeid: ua.NodeId) -> ua.NodeId:
189			"""
190			Check if the index of nodeid or browsename given in the xml model file
191			must be converted to a already existing namespace id based on the files
192			namespace uri
193
194			:returns: NodeId (str)
195			"""
196			if nodeid.NamespaceIndex in self.namespaces:
197			nodeid = copy(nodeid)
198			nodeid.NamespaceIndex = self.namespaces[nodeid.NamespaceIndex]
199			return nodeid
200
201			def _get_add_node_item(self, obj):
202			node = ua.AddNodesItem()
203			node.RequestedNewNodeId = self._migrate_ns(obj.nodeid)
204			node.BrowseName = self._migrate_ns(obj.browsename)
205			_logger.info("Importing xml node (%s, %s) as (%s %s)", obj.browsename,
206			obj.nodeid, node.BrowseName, node.RequestedNewNodeId)
207			node.NodeClass = getattr(ua.NodeClass, obj.nodetype[2:])
208			if obj.parent and obj.parentlink:
209			node.ParentNodeId = self._migrate_ns(obj.parent)
210			node.ReferenceTypeId = self._migrate_ns(obj.parentlink)
211			if obj.typedef:
212			node.TypeDefinition = self._migrate_ns(obj.typedef)
213			return node
214
215			def _to_migrated_nodeid(self, nodeid: Union[ua.NodeId, None, str]) -> ua.NodeId:
216			nodeid = self._to_nodeid(nodeid)
217			return self._migrate_ns(nodeid)
218
219			def _to_nodeid(self, nodeid: Union[ua.NodeId, None, str]) -> ua.NodeId:
220			if isinstance(nodeid, ua.NodeId):
221			return nodeid
222			elif not nodeid:
223			return ua.NodeId(ua.ObjectIds.String)
224			elif "=" in nodeid:
225			return ua.NodeId.from_string(nodeid)
226			elif hasattr(ua.ObjectIds, nodeid):
227			return ua.NodeId(getattr(ua.ObjectIds, nodeid))
228			else:
229			if nodeid in self.aliases:
230			return self.aliases[nodeid]
231			else:
232			return ua.NodeId(getattr(ua.ObjectIds, nodeid))
233
234			async def add_object(self, obj):
235			node = self._get_add_node_item(obj)
236			attrs = ua.ObjectAttributes()
237			if obj.desc:
238			attrs.Description = ua.LocalizedText(obj.desc)
239			attrs.DisplayName = ua.LocalizedText(obj.displayname)
240			attrs.EventNotifier = obj.eventnotifier
241			node.NodeAttributes = attrs
242			res = await self._get_server().add_nodes([node])
243			await self._add_refs(obj)
244			res[0].StatusCode.check()
245			return res[0].AddedNodeId
246
247			async def add_object_type(self, obj):
248			node = self._get_add_node_item(obj)
249			attrs = ua.ObjectTypeAttributes()
250			if obj.desc:
251			attrs.Description = ua.LocalizedText(obj.desc)
252			attrs.DisplayName = ua.LocalizedText(obj.displayname)
253			attrs.IsAbstract = obj.abstract
254			node.NodeAttributes = attrs
255			res = await self._get_server().add_nodes([node])
256			await self._add_refs(obj)
257			res[0].StatusCode.check()
258			return res[0].AddedNodeId
259
260			async def add_variable(self, obj):
261			node = self._get_add_node_item(obj)
262			attrs = ua.VariableAttributes()
263			if obj.desc:
264			attrs.Description = ua.LocalizedText(obj.desc)
265			attrs.DisplayName = ua.LocalizedText(obj.displayname)
266			attrs.DataType = obj.datatype
267			if obj.value is not None:
268			attrs.Value = self._add_variable_value(obj,)
269			if obj.rank:
270			attrs.ValueRank = obj.rank
271			if obj.accesslevel:
272			attrs.AccessLevel = obj.accesslevel
273			if obj.useraccesslevel:
274			attrs.UserAccessLevel = obj.useraccesslevel
275			if obj.minsample:
276			attrs.MinimumSamplingInterval = obj.minsample
277			if obj.dimensions:
278			attrs.ArrayDimensions = obj.dimensions
279			node.NodeAttributes = attrs
280			res = await self._get_server().add_nodes([node])
281			await self._add_refs(obj)
282			res[0].StatusCode.check()
283			return res[0].AddedNodeId
284
285			def _get_ext_class(self, name: str):
286			if hasattr(ua, name):
287			return getattr(ua, name)
288			elif name in self.aliases.keys():
289			nodeid = self.aliases[name]
290			class_type = ua.uatypes.get_extensionobject_class_type(nodeid)
291			if class_type:
292			return class_type
293			else:
294			raise Exception("Error no extension class registered ", name, nodeid)
295			else:
296			raise Exception("Error no alias found for extension class", name)
297
298			def _make_ext_obj(self, obj):
299			ext = self._get_ext_class(obj.objname)()
300			for name, val in obj.body:
301			if not isinstance(val, list):
302			raise Exception("Error val should be a list, this is a python-asyncua bug", name, type(val), val)
303			else:
304			for attname, v in val:
305			self._set_attr(ext, attname, v)
306			return ext
307
308			def _get_val_type(self, obj, attname: str):
309			for name, uatype in obj.ua_types:
310			if name == attname:
311			return uatype
312			raise UaError(f"Attribute '{attname}' defined in xml is not found in object '{obj}'")
313
314			def _set_attr(self, obj, attname: str, val):
315			# tow possible values:
316			# either we get value directly
317			# or a dict if it s an object or a list
318			if isinstance(val, str):
319			pval = ua_type_to_python(val, self._get_val_type(obj, attname))
320			setattr(obj, attname, pval)
321			else:
322			# so we have either an object or a list...
323			obj2 = getattr(obj, attname)
324			if isinstance(obj2, ua.NodeId): # NodeId representation does not follow common rules!!
325			for attname2, v2 in val:
326			if attname2 == "Identifier":
327			if hasattr(ua.ObjectIds, v2):
328			obj2 = ua.NodeId(getattr(ua.ObjectIds, v2))
329			else:
330			obj2 = ua.NodeId.from_string(v2)
331			setattr(obj, attname, self._migrate_ns(obj2))
332			break
333			elif not hasattr(obj2, "ua_types"):
334			# we probably have a list
335			my_list = []
336			for vtype, v2 in val:
337			my_list.append(ua_type_to_python(v2, vtype))
338			setattr(obj, attname, my_list)
339			else:
340			for attname2, v2 in val:
341			self._set_attr(obj2, attname2, v2)
342			setattr(obj, attname, obj2)
343
344			def _add_variable_value(self, obj):
345			"""
346			Returns the value for a Variable based on the objects value type.
347			"""
348			_logger.debug("Setting value with type %s and value %s", obj.valuetype, obj.value)
349			if obj.valuetype == 'ListOfExtensionObject':
350			values = []
351			for ext in obj.value:
352			extobj = self._make_ext_obj(ext)
353			values.append(extobj)
354			return ua.Variant(values, ua.VariantType.ExtensionObject)
355			elif obj.valuetype == 'ListOfGuid':
356			return ua.Variant([
357			uuid.UUID(guid) for guid in obj.value
358			], getattr(ua.VariantType, obj.valuetype[6:]))
359			elif obj.valuetype.startswith("ListOf"):
360			vtype = obj.valuetype[6:]
361			if hasattr(ua.ua_binary.Primitives, vtype):
362			return ua.Variant(obj.value, getattr(ua.VariantType, vtype))
363			elif vtype == "LocalizedText":
364			return ua.Variant([getattr(ua, vtype)(text=item["Text"], locale=item["Locale"]) for item in obj.value])
365			else:
366			return ua.Variant([getattr(ua, vtype)(v) for v in obj.value])
367			elif obj.valuetype == 'ExtensionObject':
368			extobj = self._make_ext_obj(obj.value)
369			return ua.Variant(extobj, getattr(ua.VariantType, obj.valuetype))
370			elif obj.valuetype == 'Guid':
371			return ua.Variant(uuid.UUID(obj.value), getattr(ua.VariantType, obj.valuetype))
372			elif obj.valuetype == 'LocalizedText':
373			ltext = ua.LocalizedText()
374			for name, val in obj.value:
375			if name == "Text":
376			ltext.Text = val
377			elif name == "Locale":
378			ltext.Locale = val
379			else:
380			_logger.warning("While parsing localizedText value, unkown element: %s with val: %s", name, val)
381			return ua.Variant(ltext, ua.VariantType.LocalizedText)
382			elif obj.valuetype == 'NodeId':
383			return ua.Variant(ua.NodeId.from_string(obj.value))
384			else:
385			return ua.Variant(obj.value, getattr(ua.VariantType, obj.valuetype))
386
387			async def add_variable_type(self, obj):
388			node = self._get_add_node_item(obj)
389			attrs = ua.VariableTypeAttributes()
390			if obj.desc:
391			attrs.Description = ua.LocalizedText(obj.desc)
392			attrs.DisplayName = ua.LocalizedText(obj.displayname)
393			attrs.DataType = obj.datatype
394			if obj.value and len(obj.value) == 1:
395			attrs.Value = obj.value[0]
396			if obj.rank:
397			attrs.ValueRank = obj.rank
398			if obj.abstract:
399			attrs.IsAbstract = obj.abstract
400			if obj.dimensions:
401			attrs.ArrayDimensions = obj.dimensions
402			node.NodeAttributes = attrs
403			res = await self._get_server().add_nodes([node])
404			await self._add_refs(obj)
405			res[0].StatusCode.check()
406			return res[0].AddedNodeId
407
408			async def add_method(self, obj):
409			node = self._get_add_node_item(obj)
410			attrs = ua.MethodAttributes()
411			if obj.desc:
412			attrs.Description = ua.LocalizedText(obj.desc)
413			attrs.DisplayName = ua.LocalizedText(obj.displayname)
414			if obj.accesslevel:
415			attrs.AccessLevel = obj.accesslevel
416			if obj.useraccesslevel:
417			attrs.UserAccessLevel = obj.useraccesslevel
418			if obj.minsample:
419			attrs.MinimumSamplingInterval = obj.minsample
420			if obj.dimensions:
421			attrs.ArrayDimensions = obj.dimensions
422			node.NodeAttributes = attrs
423			res = await self._get_server().add_nodes([node])
424			await self._add_refs(obj)
425			res[0].StatusCode.check()
426			return res[0].AddedNodeId
427
428			async def add_reference_type(self, obj):
429			node = self._get_add_node_item(obj)
430			attrs = ua.ReferenceTypeAttributes()
431			if obj.desc:
432			attrs.Description = ua.LocalizedText(obj.desc)
433			attrs.DisplayName = ua.LocalizedText(obj.displayname)
434			if obj. inversename:
435			attrs.InverseName = ua.LocalizedText(obj.inversename)
436			if obj.abstract:
437			attrs.IsAbstract = obj.abstract
438			if obj.symmetric:
439			attrs.Symmetric = obj.symmetric
440			node.NodeAttributes = attrs
441			res = await self._get_server().add_nodes([node])
442			await self._add_refs(obj)
443			res[0].StatusCode.check()
444			return res[0].AddedNodeId
445
446			async def add_datatype(self, obj):
447			node = self._get_add_node_item(obj)
448			attrs = ua.DataTypeAttributes()
449			if obj.desc:
450			attrs.Description = ua.LocalizedText(obj.desc)
451			attrs.DisplayName = ua.LocalizedText(obj.displayname)
452			if obj.abstract:
453			attrs.IsAbstract = obj.abstract
454			if not obj.definitions:
455			pass
456			else:
457			if obj.parent == self.server.nodes.enum_data_type.nodeid:
458			attrs.DataTypeDefinition = self._get_edef(node, obj)
459			elif obj.parent == self.server.nodes.base_structure_type.nodeid:
460			attrs.DataTypeDefinition = self._get_sdef(node, obj)
461			else:
462			parent_node = self.server.get_node(obj.parent)
463			path = await parent_node.get_path()
464			if self.server.nodes.base_structure_type in path:
465			attrs.DataTypeDefinition = self._get_sdef(node, obj)
466			else:
467			_logger.warning("%s has datatypedefinition and path %s"
468			" but we could not find out if this is a struct", obj, path)
469			node.NodeAttributes = attrs
470			res = await self._get_server().add_nodes([node])
471			res[0].StatusCode.check()
472			await self._add_refs(obj)
473			return res[0].AddedNodeId
474
475			async def _add_refs(self, obj):
476			if not obj.refs:
477			return
478			refs = []
479			for data in obj.refs:
480			ref = ua.AddReferencesItem()
481			ref.IsForward = data.forward
482			ref.ReferenceTypeId = data.reftype
483			ref.SourceNodeId = obj.nodeid
484			ref.TargetNodeId = data.target
485			refs.append(ref)
486			await self._add_references(refs)
487
488			def _get_edef(self, node, obj):
489			if not obj.definitions:
490			return None
491			edef = ua.EnumDefinition()
492			if obj.parent:
493			edef.BaseDataType = obj.parent
494			for field in obj.definitions:
495			f = ua.EnumField()
496			f.Name = field.name
497			if field.dname:
498			f.DisplayName = ua.LocalizedText(text=field.dname)
499			else:
500			f.DisplayName = ua.LocalizedText(text=field.name)
501			f.Value = field.value
502			f.Description = ua.LocalizedText(text=field.desc)
503			edef.Fields.append(f)
504			return edef
505
506			def _get_sdef(self, node, obj):
507			if not obj.definitions:
508			return None
509			sdef = ua.StructureDefinition()
510			if obj.parent:
511			sdef.BaseDataType = obj.parent
512			for data in obj.refs:
513			if data.reftype == self.server.nodes.HasEncoding.nodeid:
514			# looks likebinary encodingisthe firt one...can someone confirm?
515			sdef.DefaultEncodingId = data.target
516			break
517			optional = False
518			for field in obj.definitions:
519			f = ua.StructureField()
520			f.Name = field.name
521			f.DataType = field.datatype
522			f.ValueRank = field.valuerank
523			f.IsOptional = field.optional
524			if f.IsOptional:
525			optional = True
526			f.ArrayDimensions = field.arraydim
527			f.Description = ua.LocalizedText(text=field.desc)
528			sdef.Fields.append(f)
529			if optional:
530			sdef.StructureType = ua.StructureType.StructureWithOptionalFields
531			else:
532			sdef.StructureType = ua.StructureType.Structure
533			return sdef
534
535			def _sort_nodes_by_parentid(self, ndatas):
536			"""
537			Sort the list of nodes according their parent node in order to respect
538			the dependency between nodes.
539
540			:param nodes: list of NodeDataObjects
541			:returns: list of sorted nodes
542			"""
543
544			sorted_ndatas = []
545			sorted_nodes_ids = []
546			all_node_ids = [data.nodeid for data in ndatas]
547			while ndatas:
548			for ndata in ndatas[:]:
549			if ndata.nodeid.NamespaceIndex not in self.namespaces.values() or \
550			ndata.parent is None or \
551			ndata.parent not in all_node_ids:
552			sorted_ndatas.append(ndata)
553			sorted_nodes_ids.append(ndata.nodeid)
554			ndatas.remove(ndata)
555			else:
556			# Check if the nodes parent is already in the list of
557			# inserted nodes
558			if ndata.parent in sorted_nodes_ids:
559			sorted_ndatas.append(ndata)
560			sorted_nodes_ids.append(ndata.nodeid)
561			ndatas.remove(ndata)
562			return sorted_ndatas
563

FreeOpcUa / opcua-asyncio

Pull Request — master (#358)

XmlImporter._check_required_models() C

Complexity

Size

Duplication

Importance

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like