#!/usr/bin/env python # -*- coding: utf-8 -*- # This Source Code Form is subject to the terms of the Mozilla Public # License, v. 2.0. If a copy of the MPL was not distributed with this # file, You can obtain one at http://mozilla.org/MPL/2.0/. ### ### Author: Chris Iatrou (ichrispa@core-vector.net) ### Version: rev 13 ### ### This program was created for educational purposes and has been ### contributed to the open62541 project by the author. All licensing ### terms for this source is inherited by the terms and conditions ### specified for by the open62541 project (see the projects readme ### file for more information on the MPLv2 terms and restrictions). ### ### This program is not meant to be used in a production environment. The ### author is not liable for any complications arising due to the use of ### this program. ### from __future__ import print_function import sys from struct import pack as structpack from collections import deque from time import struct_time, strftime, strptime, mktime import logging; logger = logging.getLogger(__name__) from ua_builtin_types import *; from ua_node_types import *; from ua_constants import *; def getNextElementNode(xmlvalue): if xmlvalue == None: return None xmlvalue = xmlvalue.nextSibling while not xmlvalue == None and not xmlvalue.nodeType == xmlvalue.ELEMENT_NODE: xmlvalue = xmlvalue.nextSibling return xmlvalue class NodeSet(): """ Class holding and managing a set of OPCUA nodes. This class handles parsing XML description of namespaces, instantiating nodes, linking references, graphing the namespace and compiling a binary representation. Note that nodes assigned to this class are not restricted to having a single namespace ID. This class represents the entire physical address space of the binary representation and all nodes that are to be included in that segment of memory. """ nodes = [] nodeids = {} aliases = {} __linkLater__ = [] __binaryIndirectPointers__ = [] name = "" knownNodeTypes = "" namespaceIdentifiers = {} # list of 'int':'string' giving different namespace an array-mapable name def __init__(self, name): self.nodes = [] self.knownNodeTypes = ['variable', 'object', 'method', 'referencetype', \ 'objecttype', 'variabletype', 'methodtype', \ 'datatype', 'referencetype', 'aliases'] self.name = name self.nodeids = {} self.aliases = {} self.namespaceIdentifiers = {} self.__binaryIndirectPointers__ = [] def addNamespace(self, numericId, stringURL): self.namespaceIdentifiers[numericId] = stringURL def linkLater(self, pointer): """ Called by nodes or references who have parsed an XML reference to a node represented by a string. No return value XML String representations of references have the form 'i=xy' or 'ns=1;s="This unique Node"'. Since during the parsing of this attribute only a subset of nodes are known/parsed, this reference string cannot be linked when encountered. References register themselves with the namespace to have their target attribute (string) parsed by linkOpenPointers() when all nodes are created, so that target can be dereferenced an point to an actual node. """ self.__linkLater__.append(pointer) def getUnlinkedPointers(self): """ Return the list of references registered for linking during the next call of linkOpenPointers() """ return self.__linkLater__ def unlinkedItemCount(self): """ Returns the number of unlinked references that will be processed during the next call of linkOpenPointers() """ return len(self.__linkLater__) def buildAliasList(self, xmlelement): """ Parses the XML Element present in must XML NodeSet definitions. No return value Contents the Alias element are stored in a dictionary for further dereferencing during pointer linkage (see linkOpenPointer()). """ if not xmlelement.tagName == "Aliases": logger.error("XMLElement passed is not an Aliaslist") return for al in xmlelement.childNodes: if al.nodeType == al.ELEMENT_NODE: if al.hasAttribute("Alias"): aliasst = al.getAttribute("Alias") if sys.version_info[0] < 3: aliasnd = unicode(al.firstChild.data) else: aliasnd = al.firstChild.data if not aliasst in self.aliases: self.aliases[aliasst] = aliasnd logger.debug("Added new alias \"" + str(aliasst) + "\" == \"" + str(aliasnd) + "\"") else: if self.aliases[aliasst] != aliasnd: logger.error("Alias definitions for " + aliasst + " differ. Have " + self.aliases[aliasst] + " but XML defines " + aliasnd + ". Keeping current definition.") def getNodeByBrowseName(self, idstring): """ Returns the first node in the nodelist whose browseName matches idstring. """ return next((n for n in self.nodes if idstring==str(n.browseName())), None) def getNodeByIDString(self, idstring): """ Returns the first node in the nodelist whose id string representation matches idstring. """ return next((n for n in self.nodes if idstring==str(n.id())), None) def createNode(self, ndtype, xmlelement): """ createNode is instantiates a node described by xmlelement, its type being defined by the string ndtype. No return value If the xmlelement is an , the contents will be parsed and stored for later dereferencing during pointer linking (see linkOpenPointers). Recognized types are: * UAVariable * UAObject * UAMethod * UAView * UAVariableType * UAObjectType * UAMethodType * UAReferenceType * UADataType For every recognized type, an appropriate node class is added to the node list of the namespace. The NodeId of the given node is created and parsing of the node attributes and elements is delegated to the parseXML() and parseXMLSubType() functions of the instantiated class. If the NodeID attribute is non-unique in the node list, the creation is deferred and an error is logged. """ if not isinstance(xmlelement, dom.Element): logger.error( "Error: Can not create node from invalid XMLElement") return # An ID is mandatory for everything but aliases! id = None for idname in ['NodeId', 'NodeID', 'nodeid']: if xmlelement.hasAttribute(idname): id = xmlelement.getAttribute(idname) if ndtype == 'aliases': self.buildAliasList(xmlelement) return elif id == None: logger.info( "Error: XMLElement has no id, node will not be created!") return else: id = opcua_node_id_t(id) if str(id) in self.nodeids: # Normal behavior: Do not allow duplicates, first one wins #logger.error( "XMLElement with duplicate ID " + str(id) + " found, node will not be created!") #return # Open62541 behavior for header generation: Replace the duplicate with the new node logger.info( "XMLElement with duplicate ID " + str(id) + " found, node will be replaced!") nd = self.getNodeByIDString(str(id)) self.nodes.remove(nd) self.nodeids.pop(str(nd.id())) node = None if (ndtype == 'variable'): node = opcua_node_variable_t(id, self) elif (ndtype == 'object'): node = opcua_node_object_t(id, self) elif (ndtype == 'method'): node = opcua_node_method_t(id, self) elif (ndtype == 'objecttype'): node = opcua_node_objectType_t(id, self) elif (ndtype == 'variabletype'): node = opcua_node_variableType_t(id, self) elif (ndtype == 'methodtype'): node = opcua_node_methodType_t(id, self) elif (ndtype == 'datatype'): node = opcua_node_dataType_t(id, self) elif (ndtype == 'referencetype'): node = opcua_node_referenceType_t(id, self) else: logger.error( "No node constructor for type " + ndtype) if node != None: node.parseXML(xmlelement) self.nodes.append(node) self.nodeids[str(node.id())] = node def removeNodeById(self, nodeId): nd = self.getNodeByIDString(nodeId) if nd == None: return False logger.debug("Removing nodeId " + str(nodeId)) self.nodes.remove(nd) if nd.getInverseReferences() != None: for ref in nd.getInverseReferences(): src = ref.target(); src.removeReferenceToNode(nd) return True def registerBinaryIndirectPointer(self, node): """ Appends a node to the list of nodes that should be contained in the first 765 bytes (255 pointer slots a 3 bytes) in the binary representation (indirect referencing space). This function is reserved for references and dataType pointers. """ if not node in self.__binaryIndirectPointers__: self.__binaryIndirectPointers__.append(node) return self.__binaryIndirectPointers__.index(node) def getBinaryIndirectPointerIndex(self, node): """ Returns the slot/index of a pointer in the indirect referencing space (first 765 Bytes) of the binary representation. """ if not node in self.__binaryIndirectPointers__: return -1 return self.__binaryIndirectPointers__.index(node) def parseXML(self, xmldoc): """ Reads an XML Namespace definition and instantiates node. No return value parseXML open the file xmldoc using xml.dom.minidom and searches for the first UANodeSet Element. For every Element encountered, createNode is called to instantiate a node of the appropriate type. """ typedict = {} UANodeSet = dom.parse(xmldoc).getElementsByTagName("UANodeSet") if len(UANodeSet) == 0: logger.error( "Error: No NodeSets found") return if len(UANodeSet) != 1: logger.error( "Error: Found more than 1 Nodeset in XML File") UANodeSet = UANodeSet[0] for nd in UANodeSet.childNodes: if nd.nodeType != nd.ELEMENT_NODE: continue ndType = nd.tagName.lower() if ndType[:2] == "ua": ndType = ndType[2:] elif not ndType in self.knownNodeTypes: logger.warn("XML Element or NodeType " + ndType + " is unknown and will be ignored") continue if not ndType in typedict: typedict[ndType] = 1 else: typedict[ndType] = typedict[ndType] + 1 self.createNode(ndType, nd) logger.debug("Currently " + str(len(self.nodes)) + " nodes in address space. Type distribution for this run was: " + str(typedict)) def linkOpenPointers(self): """ Substitutes symbolic NodeIds in references for actual node instances. No return value References that have registered themselves with linkLater() to have their symbolic NodeId targets ("ns=2;i=32") substituted for an actual node will be iterated by this function. For each reference encountered in the list of unlinked/open references, the target string will be evaluated and searched for in the node list of this namespace. If found, the target attribute of the reference will be substituted for the found node. If a reference fails to get linked, it will remain in the list of unlinked references. The individual items in this list can be retrieved using getUnlinkedPointers(). """ linked = [] logger.debug( str(self.unlinkedItemCount()) + " pointers need to get linked.") for l in self.__linkLater__: targetLinked = False if not l.target() == None and not isinstance(l.target(), opcua_node_t): if isinstance(l.target(),str) or isinstance(l.target(),unicode): # If is not a node ID, it should be an alias. Try replacing it # with a proper node ID if l.target() in self.aliases: l.target(self.aliases[l.target()]) # If the link is a node ID, try to find it hopening that no ass has # defined more than one kind of id for that sucker if l.target()[:2] == "i=" or l.target()[:2] == "g=" or \ l.target()[:2] == "b=" or l.target()[:2] == "s=" or \ l.target()[:3] == "ns=" : tgt = self.getNodeByIDString(str(l.target())) if tgt == None: logger.error("Failed to link pointer to target (node not found) " + l.target()) else: l.target(tgt) targetLinked = True else: logger.error("Failed to link pointer to target (target not Alias or Node) " + l.target()) else: logger.error("Failed to link pointer to target (don't know dummy type + " + str(type(l.target())) + " +) " + str(l.target())) else: logger.error("Pointer has null target: " + str(l)) referenceLinked = False if not l.referenceType() == None: if l.referenceType() in self.aliases: l.referenceType(self.aliases[l.referenceType()]) tgt = self.getNodeByIDString(str(l.referenceType())) if tgt == None: logger.error("Failed to link reference type to target (node not found) " + l.referenceType()) else: l.referenceType(tgt) referenceLinked = True else: referenceLinked = True if referenceLinked == True and targetLinked == True: linked.append(l) # References marked as "not forward" must be inverted (removed from source # node, assigned to target node and relinked) logger.warn("Inverting reference direction for all references with isForward==False attribute (is this correct!?)") for n in self.nodes: for r in n.getReferences(): if r.isForward() == False: tgt = r.target() if isinstance(tgt, opcua_node_t): nref = opcua_referencePointer_t(n, parentNode=tgt) nref.referenceType(r.referenceType()) tgt.addReference(nref) # Create inverse references for all nodes logger.debug("Updating all referencedBy fields in nodes for inverse lookups.") for n in self.nodes: n.updateInverseReferences() for l in linked: self.__linkLater__.remove(l) if len(self.__linkLater__) != 0: logger.warn(str(len(self.__linkLater__)) + " could not be linked.") def sanitize(self): remove = [] logger.debug("Sanitizing nodes and references...") for n in self.nodes: if n.sanitize() == False: remove.append(n) if not len(remove) == 0: logger.warn(str(len(remove)) + " nodes will be removed because they failed sanitation.") # FIXME: Some variable ns=0 nodes fail because they don't have DataType fields... # How should this be handles!? logger.warn("Not actually removing nodes... it's unclear if this is valid or not") def getRoot(self): """ Returns the first node instance with the browseName "Root". """ return self.getNodeByBrowseName("Root") def buildEncodingRules(self): """ Calls buildEncoding() for all DataType nodes (opcua_node_dataType_t). No return value """ stat = {True: 0, False: 0} for n in self.nodes: if isinstance(n, opcua_node_dataType_t): n.buildEncoding() stat[n.isEncodable()] = stat[n.isEncodable()] + 1 logger.debug("Type definitions built/passed: " + str(stat)) def allocateVariables(self): for n in self.nodes: if isinstance(n, opcua_node_variable_t): n.allocateValue() def getSubTypesOf(self, tdNodes = None, currentNode = None, hasSubtypeRefNode = None): # If this is a toplevel call, collect the following information as defaults if tdNodes == None: tdNodes = [] if currentNode == None: currentNode = self.getNodeByBrowseName("HasTypeDefinition") tdNodes.append(currentNode) if len(tdNodes) < 1: return [] if hasSubtypeRefNode == None: hasSubtypeRefNode = self.getNodeByBrowseName("HasSubtype") if hasSubtypeRefNode == None: return tdNodes # collect all subtypes of this node for ref in currentNode.getReferences(): if ref.isForward() and ref.referenceType().id() == hasSubtypeRefNode.id(): tdNodes.append(ref.target()) self.getTypeDefinitionNodes(tdNodes=tdNodes, currentNode = ref.target(), hasSubtypeRefNode=hasSubtypeRefNode) return tdNodes def printDotGraphWalk(self, depth=1, filename="out.dot", rootNode=None, followInverse = False, excludeNodeIds=[]): """ Outputs a graphiz/dot description the nodes centered around rootNode. References beginning from rootNode will be followed for depth steps. If "followInverse = True" is passed, then inverse (not Forward) references will also be followed. Nodes can be excluded from the graph by passing a list of NodeIds as string representation using excludeNodeIds (ex ["i=53", "ns=2;i=453"]). Output is written into filename to be parsed by dot/neato/srfp... """ iter = depth processed = [] if rootNode == None or \ not isinstance(rootNode, opcua_node_t) or \ not rootNode in self.nodes: root = self.getRoot() else: root = rootNode file=open(filename, 'w+') if root == None: return file.write("digraph ns {\n") file.write(root.printDot()) refs=[] if followInverse == True: refs = root.getReferences(); # + root.getInverseReferences() else: for ref in root.getReferences(): if ref.isForward(): refs.append(ref) while iter > 0: tmp = [] for ref in refs: if isinstance(ref.target(), opcua_node_t): tgt = ref.target() if not str(tgt.id()) in excludeNodeIds: if not tgt in processed: file.write(tgt.printDot()) processed.append(tgt) if ref.isForward() == False and followInverse == True: tmp = tmp + tgt.getReferences(); # + tgt.getInverseReferences() elif ref.isForward() == True : tmp = tmp + tgt.getReferences(); refs = tmp iter = iter - 1 file.write("}\n") file.close() def getSubTypesOf2(self, node): re = [node] for ref in node.getReferences(): if isinstance(ref.target(), opcua_node_t): if ref.referenceType().displayName() == "HasSubtype" and ref.isForward(): re = re + self.getSubTypesOf2(ref.target()) return re def reorderNodesMinDependencies(self, printedExternally): #Kahn's algorithm #https://algocoding.wordpress.com/2015/04/05/topological-sorting-python/ relevant_types = ["HierarchicalReferences", "HasComponent"] temp = [] for t in relevant_types: temp = temp + self.getSubTypesOf2(self.getNodeByBrowseName(t)) relevant_types = temp in_degree = { u : 0 for u in self.nodes } # determine in-degree for u in self.nodes: # of each node if u not in printedExternally: for ref in u.getReferences(): if isinstance(ref.target(), opcua_node_t): if(ref.referenceType() in relevant_types and ref.isForward()): in_degree[ref.target()] += 1 Q = deque() # collect nodes with zero in-degree for u in in_degree: if in_degree[u] == 0: Q.appendleft(u) L = [] # list for order of nodes while Q: u = Q.pop() # choose node of zero in-degree L.append(u) # and 'remove' it from graph for ref in u.getReferences(): if isinstance(ref.target(), opcua_node_t): if(ref.referenceType() in relevant_types and ref.isForward()): in_degree[ref.target()] -= 1 if in_degree[ref.target()] == 0: Q.appendleft(ref.target()) if len(L) == len(self.nodes): self.nodes = L else: # if there is a cycle, logger.error("Node graph is circular on the specified references") self.nodes = L + [x for x in self.nodes if x not in L] return