|
| 1 | +""" |
| 2 | +parse simple structures from an xml tree |
| 3 | +We only support a subset of features but should be enough |
| 4 | +for custom structures |
| 5 | +""" |
| 6 | + |
| 7 | +import os |
| 8 | +import importlib |
| 9 | + |
| 10 | +from lxml import objectify |
| 11 | + |
| 12 | + |
| 13 | +from opcua.ua.ua_binary import Primitives |
| 14 | + |
| 15 | + |
| 16 | +def get_default_value(uatype): |
| 17 | + if uatype == "String": |
| 18 | + return "None" |
| 19 | + elif uatype == "Guid": |
| 20 | + return "uuid.uuid4()" |
| 21 | + elif uatype in ("ByteString", "CharArray", "Char"): |
| 22 | + return None |
| 23 | + elif uatype == "Boolean": |
| 24 | + return "True" |
| 25 | + elif uatype == "DateTime": |
| 26 | + return "datetime.utcnow()" |
| 27 | + elif uatype in ("Int8", "Int16", "Int32", "Int64", "UInt8", "UInt16", "UInt32", "UInt64", "Double", "Float", "Byte", "SByte"): |
| 28 | + return 0 |
| 29 | + else: |
| 30 | + return "ua." + uatype + "()" |
| 31 | + |
| 32 | + |
| 33 | +class Struct(object): |
| 34 | + def __init__(self, name): |
| 35 | + self.name = name |
| 36 | + self.fields = [] |
| 37 | + self.code = "" |
| 38 | + |
| 39 | + def get_code(self): |
| 40 | + if not self.fields: |
| 41 | + return """ |
| 42 | +
|
| 43 | +class {}(object): |
| 44 | + pass |
| 45 | +
|
| 46 | +""".format(self.name) |
| 47 | + self._make_constructor() |
| 48 | + self._make_from_binary() |
| 49 | + self._make_to_binary() |
| 50 | + return self.code |
| 51 | + |
| 52 | + def _make_constructor(self): |
| 53 | + self.code = """ |
| 54 | +
|
| 55 | +
|
| 56 | +class {0}(object): |
| 57 | + ''' |
| 58 | + {0} structure autogenerated from xml |
| 59 | + ''' |
| 60 | + def __init__(self, data=None): |
| 61 | + if data is not None: |
| 62 | + self._binary_init(data) |
| 63 | + return |
| 64 | +""".format(self.name) |
| 65 | + for field in self.fields: |
| 66 | + self.code += " self.{} = {}\n".format(field.name, field.value) |
| 67 | + |
| 68 | + def _make_from_binary(self): |
| 69 | + self.code += ''' |
| 70 | + @staticmethod |
| 71 | + def from_binary(data): |
| 72 | + return {}(data=data) |
| 73 | +
|
| 74 | + def _binary_init(self, data): |
| 75 | +'''.format(self.name) |
| 76 | + for field in self.fields: |
| 77 | + if hasattr(Primitives, field.uatype): |
| 78 | + if field.array: |
| 79 | + self.code += ' self.{} = ua.ua_binary.Primitives.{}.unpack_array(data)\n'.format(field.name, field.uatype) |
| 80 | + else: |
| 81 | + self.code += ' self.{} = ua.ua_binary.Primitives.{}.unpack(data)\n'.format(field.name, field.uatype) |
| 82 | + else: |
| 83 | + if field.array: |
| 84 | + self.code += ''' |
| 85 | + length = ua.ua_binary.Primitives.Int32.unpack(data) |
| 86 | + if length == -1: |
| 87 | + self.{0} = None |
| 88 | + else: |
| 89 | + self.{0} = [ua.{1}.from_binary(data) for _ in range(length)] |
| 90 | +'''.format(field.name, field.uatype) |
| 91 | + else: |
| 92 | + self.code += " self.{} = ua.{}.from_binary(data)\n".format(field.name, field.uatype) |
| 93 | + |
| 94 | + def _make_to_binary(self): |
| 95 | + self.code += ''' |
| 96 | + def to_binary(self): |
| 97 | + packet = [] |
| 98 | +''' |
| 99 | + for field in self.fields: |
| 100 | + if hasattr(Primitives, field.uatype): |
| 101 | + if field.array: |
| 102 | + self.code += ' packet.append(ua.ua_binary.Primitives.{}.pack_array(self.{}))\n'.format(field.uatype, field.name) |
| 103 | + else: |
| 104 | + self.code += ' packet.append(ua.ua_binary.Primitives.{}.pack(self.{}))\n'.format(field.uatype, field.name) |
| 105 | + else: |
| 106 | + if field.array: |
| 107 | + self.code += ''' |
| 108 | + if self.{0} is None: |
| 109 | + packet.append(ua.ua_binary.Primitives.Int32.pack(-1)) |
| 110 | + else: |
| 111 | + packet.append(ua.ua_binary.Primitives.Int32.pack(len(self.{0}))) |
| 112 | + for element in self.{0}: |
| 113 | + packet.append(element.to_binary()) |
| 114 | +'''.format(field.name) |
| 115 | + else: |
| 116 | + self.code += " packet.append(self.{}.to_binary())\n".format(field.name) |
| 117 | + self.code += ' return b"".join(packet)' |
| 118 | + |
| 119 | + |
| 120 | +class Field(object): |
| 121 | + def __init__(self, name): |
| 122 | + self.name = name |
| 123 | + self.uatype = None |
| 124 | + self.value = None |
| 125 | + self.array = False |
| 126 | + |
| 127 | + |
| 128 | +class StructGenerator(object): |
| 129 | + def __init__(self): |
| 130 | + self.model = [] |
| 131 | + |
| 132 | + def make_model_from_string(self, xml): |
| 133 | + obj = objectify.fromstring(xml) |
| 134 | + self._make_model(obj) |
| 135 | + |
| 136 | + def make_model_from_file(self, path): |
| 137 | + obj = objectify.parse(path) |
| 138 | + root = obj.getroot() |
| 139 | + self._make_model(root) |
| 140 | + |
| 141 | + def _make_model(self, root): |
| 142 | + for child in root.iter("{*}StructuredType"): |
| 143 | + struct = Struct(child.get("Name")) |
| 144 | + array = False |
| 145 | + for xmlfield in child.iter("{*}Field"): |
| 146 | + name = xmlfield.get("Name") |
| 147 | + if name.startswith("NoOf"): |
| 148 | + array = True |
| 149 | + continue |
| 150 | + field = Field(name) |
| 151 | + field.uatype = xmlfield.get("TypeName") |
| 152 | + if ":" in field.uatype: |
| 153 | + field.uatype = field.uatype.split(":")[1] |
| 154 | + field.value = get_default_value(field.uatype) |
| 155 | + if array: |
| 156 | + field.array = True |
| 157 | + field.value = [] |
| 158 | + array = False |
| 159 | + struct.fields.append(field) |
| 160 | + self.model.append(struct) |
| 161 | + |
| 162 | + def save_to_file(self, path): |
| 163 | + _file = open(path, "wt") |
| 164 | + self._make_header(_file) |
| 165 | + for struct in self.model: |
| 166 | + _file.write(struct.get_code()) |
| 167 | + _file.close() |
| 168 | + |
| 169 | + def save_and_import(self, path, append_to=None): |
| 170 | + """ |
| 171 | + save the new structures to a python file which be used later |
| 172 | + import the result and return resulting classes in a dict |
| 173 | + if append_to is a dict, the classes are added to the dict |
| 174 | + """ |
| 175 | + self.save_to_file(path) |
| 176 | + name = os.path.basename(path) |
| 177 | + name = os.path.splitext(name)[0] |
| 178 | + mymodule = importlib.import_module(name) |
| 179 | + if append_to is None: |
| 180 | + result = {} |
| 181 | + else: |
| 182 | + result = append_to |
| 183 | + for struct in self.model: |
| 184 | + result[struct.name] = getattr(mymodule, struct.name) |
| 185 | + return result |
| 186 | + |
| 187 | + def get_structures(self): |
| 188 | + ld = {} |
| 189 | + for struct in self.model: |
| 190 | + exec(struct.get_code(), ld) |
| 191 | + return ld |
| 192 | + |
| 193 | + def _make_header(self, _file): |
| 194 | + _file.write(""" |
| 195 | +''' |
| 196 | +THIS FILE IS AUTOGENERATED, DO NOT EDIT!!! |
| 197 | +''' |
| 198 | +
|
| 199 | +from datetime import datetime |
| 200 | +import uuid |
| 201 | +
|
| 202 | +from opcua import ua |
| 203 | +""") |
| 204 | + |
| 205 | + |
| 206 | + |
| 207 | + |
| 208 | +if __name__ == "__main__": |
| 209 | + import sys |
| 210 | + from IPython import embed |
| 211 | + sys.path.insert(0, ".") # necessary for import in current dir |
| 212 | + |
| 213 | + #xmlpath = "schemas/Opc.Ua.Types.bsd" |
| 214 | + xmlpath = "schemas/example.bsd" |
| 215 | + c = StructGenerator(xmlpath, "structures.py") |
| 216 | + c.run() |
| 217 | + import structures as s |
| 218 | + |
| 219 | + |
| 220 | + #sts = c.get_structures() |
| 221 | + embed() |
0 commit comments