add new generator for structures

Author: oroulet

opcua/common/structures_generator.py

@@ -0,0 +1,190 @@
+"""
+parse simple structures from an xml tree
+We only support a subset of features but should be enough
+for custom structures
+"""
+
+import sys
+from lxml import etree
+from lxml import objectify
+
+from IPython import embed
+
+
+from opcua.ua.ua_binary import Primitives1, Primitives
+#from opcua.common.ua_utils import get_default_value
+
+
+def get_default_value(uatype):
+    if uatype in ("String"):
+        return None 
+    elif uatype in ("ByteString", "CharArray", "Char"):
+        return None 
+    elif uatype in ("Boolean"):
+        return "True"
+    elif uatype in ("DateTime"):
+        return "datetime.utcnow()"
+    elif uatype in ("Int8", "Int16", "Int32", "Int64", "UInt8", "UInt16", "UInt32", "UInt64", "Double", "Float", "Byte", "SByte"):
+        return 0
+    elif uatype in ("ExtensionObject"):
+        return "None"
+    else:
+        return "ua." + uatype + "()"
+
+
+class Struct(object):
+    def __init__(self, name):
+        self.name = name
+        self.fields = []
+        self.code = ""
+
+    def get_code(self):
+        self._make_constructor()
+        self._make_from_binary()
+        self._make_to_binary()
+        return self.code
+
+    def _make_constructor(self):
+        self.code = """
+
+class {0}(object):
+    '''
+    {0} autogenerated from xml
+    '''
+    def __init__(self):
+""".format(self.name)
+        for field in self.fields:
+            self.code += "        self.{} = {}\n".format(field.name, field.value)
+
+    def _make_from_binary(self):
+        self.code += '\n    def from_binary(self, data):\n'
+        for field in self.fields:
+            if hasattr(Primitives, field.uatype):
+                if field.array:
+                    self.code += '        self.{} = ua.ua_binary.Primitives.{}.unpack_array(data)\n'.format(field.name, field.uatype)
+                else:
+                    self.code += '        self.{} = ua.ua_binary.Primitives.{}.unpack(data)\n'.format(field.name, field.uatype)
+            else:
+                if field.array:
+                    self.code += '''
+        length = ua.ua_binary.Primitives.Int32.unpack(data)
+        if length != -1:
+            self.{0} = [ua.{1}.from_binary(data) for _ in range(length)]
+'''.format(field.name, field.uatype)
+                else:
+                    self.code += "        self.{} = ua.{}.from_binary(data)\n".format(field.name, field.uatype)
+
+    def _make_to_binary(self):
+        self.code += '''
+    def to_binary(self):
+        packet = []
+'''
+        for field in self.fields:
+            if hasattr(Primitives, field.uatype):
+                if field.array:
+                    self.code += '        packet.append(ua.ua_binary.Primitives.{}.pack_array(self.{}))\n'.format(field.uatype, field.name)
+                else:
+                    self.code += '        packet.append(ua.ua_binary.Primitives.{}.pack(self.{}))\n'.format(field.uatype, field.name)
+            else:
+                if field.array:
+                    self.code += '''
+        if self.{0} is None:
+            packet.append(ua.ua_binary.Primitives.Int32.pack(-1))
+        else:
+            packet.append(ua.ua_binary.Primitives.Int32.pack(len(self.{0})))
+            for element in self.{0}:
+                packet.append(element.to_binary())
+'''.format(field.name, field.uatype)
+                else:
+                    self.code += "        packet.append(ua.{}.to_binary())\n".format(field.name)
+        self.code += '        return b"".join(packet)'
+
+
+
+
+
+class Field(object):
+    def __init__(self, name):
+        self.name = name
+        self.uatype = None
+        self.value = None
+        self.array = False
+
+
+class CodeGenerator(object):
+    def __init__(self, path, output):
+        self.path = path
+        self.output = output
+        self.model = []
+
+    def generate(self):
+        parser = etree.XMLParser(remove_blank_text=True, ns_clean=True)
+        root = etree.parse(self.path, parser)
+        embed()
+        for child in root.iter():
+            embed()
+
+    def _make_model(self):
+        obj = objectify.parse(self.path)
+        root = obj.getroot()
+        for child in root.iter("{*}StructuredType"):
+            struct = Struct(child.get("Name"))
+            array = False
+            for xmlfield in child.iter("{*}Field"):
+                name = xmlfield.get("Name")
+                if name.startswith("NoOf"):
+                    array = True
+                    continue
+                field = Field(name)
+                uatype = xmlfield.get("TypeName")
+                if uatype.startswith("opc"):
+                    field.uatype = uatype[4:]
+                else:
+                    field.uatype = uatype[3:]
+                field.value = get_default_value(field.uatype)
+                if array:
+                    field.array = True
+                    field.value = []
+                    array = False
+                struct.fields.append(field)
+            self.model.append(struct)
+
+    def run(self):
+        self._make_model()
+        self._file = open(self.output, "wt")
+        self._make_header()
+        for struct in self.model:
+            self._file.write(struct.get_code())
+        self._file.close()
+
+    def _make_header(self):
+        self._file.write("""
+'''
+THIS FILE IS AUTOGENERATED, DO NOT EDIT!!!
+'''
+
+from datetime import datetime
+
+from opcua import ua
+
+
+""")
+
+
+
+
+if __name__ == "__main__":
+
+    xmlpath = "/home/olivier/python-opcua/schemas/example.bsd"
+    #p = gm.Parser(xmlpath)
+    #model = p.parse()
+    #gm.add_basetype_members(model)
+    #gm.add_encoding_field(model)
+    #gm.remove_duplicates(model)
+    #gm.remove_vector_length(model)
+    #gm.split_requests(model)
+    #gm.fix_names(model)
+    c = CodeGenerator(xmlpath, "structures.py")
+    #c.generate()
+    c.run()
+    #embed()

opcua/common/ua_utils.py

@@ -258,3 +258,14 @@ def get_nodes_of_namespace(server, namespaces=None):
     nodes = [server.get_node(nodeid) for nodeid in server.iserver.aspace.keys()
              if nodeid.NamespaceIndex != 0 and nodeid.NamespaceIndex in namespace_indexes]
     return nodes
+
+
+def get_default_value(uatype):
+    if isinstance(uatype, ua.VariantType):
+        return ua.get_default_values(uatype)
+    elif hasattr(ua.VariantType, uatype):
+        return ua.get_default_value(getattr(ua.VariantType, uatype))
+    else:
+        return getattr(ua, uatype)()
+
+