asyncua.common.structures.Struct.__init__() - Code Metrics - Inspection of "Replace lxml with xml.etree.ElementTree (#346)" - FreeOpcUa/opcua-asyncio - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 9e5bc7...0261d3 )

by Olivier

created 2020-12-07 10:03 UTC

asyncua.common.structures.Struct.init() A

↳ Parent: asyncua.common.structures

Complexity

Conditions

Size

Total Lines	4
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	4
nop	2
dl	0
loc	4
rs	10
c	0
b	0
f	0

"""
Support for custom structures in client and server
We only support a subset of features but should be enough
for custom structures
"""

import uuid
import logging
# The next two imports are for generated code
from datetime import datetime
from enum import Enum, IntEnum, EnumMeta
from xml.etree import ElementTree as ET
from asyncua import ua

from .structures104 import get_default_value, clean_name

_logger = logging.getLogger(__name__)


class EnumType(object):
    def __init__(self, name):
        self.name = name
        self.fields = []
        self.typeid = None

    def __str__(self):
        return f"EnumType({self.name, self.fields})"
    __repr__ = __str__

    def get_code(self):
        code = """

class {0}(IntEnum):

    '''
    {0} EnumInt autogenerated from xml
    '''

""".format(self.name)

        for EnumeratedValue in self.fields:
            name = EnumeratedValue.Name
            value = EnumeratedValue.Value
            code += f"    {name} = {value}\n"

        return code


class EnumeratedValue:
    def __init__(self, name, value):
        if name == "None":
            name = "None_"
        name = name.replace(" ", "")
        self.Name = name
        self.Value = value


class Struct:
    def __init__(self, name):
        self.name = clean_name(name)
        self.fields = []
        self.typeid = None

    def __str__(self):
        return f"Struct(name={self.name}, fields={self.fields}"

    __repr__ = __str__

    def get_code(self):
        code = f"""

class {self.name}:

    '''
    {self.name} structure autogenerated from xml
    '''

"""
        code += '    ua_types = [\n'
        for field in self.fields:
            prefix = 'ListOf' if field.array else ''
            uatype = prefix + field.uatype
            if uatype == 'ListOfChar':
                uatype = 'String'
            code += f"        ('{field.name}', '{uatype}'),\n"
        code += "    ]"
        code += """
    def __str__(self):
        vals = [name + ": " + str(val) for name, val in self.__dict__.items()]
        return self.__class__.__name__ + "(" + ", ".join(vals) + ")"

    __repr__ = __str__

    def __init__(self):
"""
        if not self.fields:
            code += "      pass"
        for field in self.fields:
            code += f"        self.{field.name} = {field.value}\n"
        return code


class Field(object):
    def __init__(self, name):
        self.name = name
        self.uatype = None
        self.value = None
        self.array = False

    def __str__(self):
        return f"Field(name={self.name}, uatype={self.uatype})"

    __repr__ = __str__


class StructGenerator(object):
    def __init__(self):
        self.model = []

    def make_model_from_string(self, xml):
        obj = ET.fromstring(xml)
        self._make_model(obj)

    def make_model_from_file(self, path):
        obj = ET.parse(path)
        root = obj.getroot()
        self._make_model(root)

    def _make_model(self, root):
        enums = {}
        for child in root:
            if child.tag.endswith("EnumeratedType"):
                intenum = EnumType(child.get("Name"))
                for xmlfield in child:
                    if xmlfield.tag.endswith("EnumeratedValue"):
                        name = xmlfield.get("Name")
                        value = xmlfield.get("Value")
                        enumvalue = EnumeratedValue(name, value)
                        intenum.fields.append(enumvalue)
                        enums[child.get("Name")] = value
                self.model.append(intenum)

        for child in root:
            if child.tag.endswith("StructuredType"):
                struct = Struct(child.get("Name"))
                array = False
                # these lines can be reduced in >= Python3.8 with root.iterfind("{*}Field") and similar
                for xmlfield in child:
                    if xmlfield.tag.endswith("Field"):
                        name = xmlfield.get("Name")
                        if name.startswith("NoOf"):
                            array = True
                            continue
                        field = Field(clean_name(name))
                        field.uatype = xmlfield.get("TypeName")
                        if ":" in field.uatype:
                            field.uatype = field.uatype.split(":")[1]
                        field.uatype = clean_name(field.uatype)
                        field.value = get_default_value(field.uatype, enums)
                        if array:
                            field.array = True
                            field.value = []
                            array = False
                        struct.fields.append(field)
                self.model.append(struct)

    def save_to_file(self, path, register=False):
        _file = open(path, "w+")
        self._make_header(_file)
        for struct in self.model:
            _file.write(struct.get_code())
        if register:
            _file.write(self._make_registration())
        _file.close()

    def _make_registration(self):
        code = "\n\n"
        for struct in self.model:
            code += f"ua.register_extension_object('{struct.name}'," \
                    f" ua.NodeId.from_string('{struct.typeid}'), {struct.name})\n"
        return code

    def get_python_classes(self, env=None):
        return _generate_python_class(self.model, env=env)

    def _make_header(self, _file):
        _file.write("""
'''
THIS FILE IS AUTOGENERATED, DO NOT EDIT!!!
'''

from datetime import datetime
import uuid

from asyncua import ua
""")

    def set_typeid(self, name, typeid):
        for struct in self.model:
            if struct.name == name:
                struct.typeid = typeid
                return


async def load_type_definitions(server, nodes=None):
    """
    Download xml from given variable node defining custom structures.
    If no node is given, attemps to import variables from all nodes under
    "0:OPC Binary"
    the code is generated and imported on the fly. If you know the structures
    are not going to be modified it might be interresting to copy the generated files
    and include them in you code
    """
    if nodes is None:
        nodes = []
        for desc in await server.nodes.opc_binary.get_children_descriptions():
            if desc.BrowseName != ua.QualifiedName("Opc.Ua"):
                nodes.append(server.get_node(desc.NodeId))

    structs_dict = {}
    generators = []
    for node in nodes:
        xml = await node.read_value()
        generator = StructGenerator()
        generators.append(generator)
        generator.make_model_from_string(xml)
        # generate and execute new code on the fly
        generator.get_python_classes(structs_dict)
        # same but using a file that is imported. This can be usefull for debugging library
        # name = node.read_browse_name().Name
        # Make sure structure names do not contain charaters that cannot be used in Python class file names
        # name = clean_name(name)
        # name = "structures_" + node.read_browse_name().Name
        # generator.save_and_import(name + ".py", append_to=structs_dict)

        # register classes
        # every children of our node should represent a class
        for ndesc in await node.get_children_descriptions():
            ndesc_node = server.get_node(ndesc.NodeId)
            ref_desc_list = await ndesc_node.get_references(refs=ua.ObjectIds.HasDescription, direction=ua.BrowseDirection.Inverse)
            if ref_desc_list:  # some server put extra things here
                name = clean_name(ndesc.BrowseName.Name)
                if name not in structs_dict:
                    _logger.warning("%s is found as child of binary definition node but is not found in xml", name)
                    continue
                nodeid = ref_desc_list[0].NodeId
                ua.register_extension_object(name, nodeid, structs_dict[name])
                # save the typeid if user want to create static file for type definitnion
                generator.set_typeid(name, nodeid.to_string())

        for key, val in structs_dict.items():
            if isinstance(val, EnumMeta) and key != "IntEnum":
                setattr(ua, key, val)

    return generators, structs_dict


def _generate_python_class(model, env=None):
    """
    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 = ua.__dict__
    #  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 classes one by one and add them to dict
    for element in model:
        code = element.get_code()
        exec(code, env)
    return env


async def load_enums(server, env=None, force=False):
    """
    Read enumeration data types on server and generate python Enums in ua scope for them
    """
    model = []

    for desc in await server.nodes.enum_data_type.get_children_descriptions(refs=ua.ObjectIds.HasSubtype):
        enum_name = desc.BrowseName.Name
        enum_node = server.get_node(desc.NodeId)
        if not force and hasattr(ua, enum_name):
            _logger.debug("Enum type %s is already in ua namespace, ignoring", enum_name)
            continue
        c = None
        for child_desc in await enum_node.get_children_descriptions(refs=ua.ObjectIds.HasProperty):
            child_node = server.get_node(child_desc.NodeId)
            if child_desc.BrowseName.Name == "EnumStrings":
                c = await _get_enum_strings(enum_name, child_node)
            elif child_desc.BrowseName.Name == "EnumValues":
                c = await _get_enum_values(enum_name, server.get_node(child_desc.NodeId))
            else:
                _logger.warning("Unexpected children of node %s: %s", desc, child_desc)
        if c is not None:
            model.append(c)
    return _generate_python_class(model, env=env)


async def _get_enum_values(name, node):
    val = await node.read_value()
    c = EnumType(name)
    c.fields = [EnumeratedValue(enumval.DisplayName.Text, enumval.Value) for enumval in val]
    return c


async def _get_enum_strings(name, node):
    val = await node.read_value()
    c = EnumType(name)
    c.fields = [EnumeratedValue(st.Text, idx) for idx, st in enumerate(val)]
    return c


1			"""
2			Support for custom structures in client and server
3			We only support a subset of features but should be enough
4			for custom structures
5			"""
6
7			import uuid
8			import logging
9			# The next two imports are for generated code
10			from datetime import datetime
11			from enum import Enum, IntEnum, EnumMeta
12			from xml.etree import ElementTree as ET
13			from asyncua import ua
14
15			from .structures104 import get_default_value, clean_name
16
17			_logger = logging.getLogger(__name__)
18
19
20			class EnumType(object):
21			def __init__(self, name):
22			self.name = name
23			self.fields = []
24			self.typeid = None
25
26			def __str__(self):
27			return f"EnumType({self.name, self.fields})"
28			__repr__ = __str__
29
30			def get_code(self):
31			code = """
32
33			class {0}(IntEnum):
34
35			'''
36			{0} EnumInt autogenerated from xml
37			'''
38
39			""".format(self.name)
40
41			for EnumeratedValue in self.fields:
42			name = EnumeratedValue.Name
43			value = EnumeratedValue.Value
44			code += f" {name} = {value}\n"
45
46			return code
47
48
49			class EnumeratedValue:
50			def __init__(self, name, value):
51			if name == "None":
52			name = "None_"
53			name = name.replace(" ", "")
54			self.Name = name
55			self.Value = value
56
57
58			class Struct:
59			def __init__(self, name):
60			self.name = clean_name(name)
61			self.fields = []
62			self.typeid = None
63
64			def __str__(self):
65			return f"Struct(name={self.name}, fields={self.fields}"
66
67			__repr__ = __str__
68
69			def get_code(self):
70			code = f"""
71
72			class {self.name}:
73
74			'''
75			{self.name} structure autogenerated from xml
76			'''
77
78			"""
79			code += ' ua_types = [\n'
80			for field in self.fields:
81			prefix = 'ListOf' if field.array else ''
82			uatype = prefix + field.uatype
83			if uatype == 'ListOfChar':
84			uatype = 'String'
85			code += f" ('{field.name}', '{uatype}'),\n"
86			code += " ]"
87			code += """
88			def __str__(self):
89			vals = [name + ": " + str(val) for name, val in self.__dict__.items()]
90			return self.__class__.__name__ + "(" + ", ".join(vals) + ")"
91
92			__repr__ = __str__
93
94			def __init__(self):
95			"""
96			if not self.fields:
97			code += " pass"
98			for field in self.fields:
99			code += f" self.{field.name} = {field.value}\n"
100			return code
101
102
103			class Field(object):
104			def __init__(self, name):
105			self.name = name
106			self.uatype = None
107			self.value = None
108			self.array = False
109
110			def __str__(self):
111			return f"Field(name={self.name}, uatype={self.uatype})"
112
113			__repr__ = __str__
114
115
116			class StructGenerator(object):
117			def __init__(self):
118			self.model = []
119
120			def make_model_from_string(self, xml):
121			obj = ET.fromstring(xml)
122			self._make_model(obj)
123
124			def make_model_from_file(self, path):
125			obj = ET.parse(path)
126			root = obj.getroot()
127			self._make_model(root)
128
129			def _make_model(self, root):
130			enums = {}
131			for child in root:
132			if child.tag.endswith("EnumeratedType"):
133			intenum = EnumType(child.get("Name"))
134			for xmlfield in child:
135			if xmlfield.tag.endswith("EnumeratedValue"):
136			name = xmlfield.get("Name")
137			value = xmlfield.get("Value")
138			enumvalue = EnumeratedValue(name, value)
139			intenum.fields.append(enumvalue)
140			enums[child.get("Name")] = value
141			self.model.append(intenum)
142
143			for child in root:
144			if child.tag.endswith("StructuredType"):
145			struct = Struct(child.get("Name"))
146			array = False
147			# these lines can be reduced in >= Python3.8 with root.iterfind("{*}Field") and similar
148			for xmlfield in child:
149			if xmlfield.tag.endswith("Field"):
150			name = xmlfield.get("Name")
151			if name.startswith("NoOf"):
152			array = True
153			continue
154			field = Field(clean_name(name))
155			field.uatype = xmlfield.get("TypeName")
156			if ":" in field.uatype:
157			field.uatype = field.uatype.split(":")[1]
158			field.uatype = clean_name(field.uatype)
159			field.value = get_default_value(field.uatype, enums)
160			if array:
161			field.array = True
162			field.value = []
163			array = False
164			struct.fields.append(field)
165			self.model.append(struct)
166
167			def save_to_file(self, path, register=False):
168			_file = open(path, "w+")
169			self._make_header(_file)
170			for struct in self.model:
171			_file.write(struct.get_code())
172			if register:
173			_file.write(self._make_registration())
174			_file.close()
175
176			def _make_registration(self):
177			code = "\n\n"
178			for struct in self.model:
179			code += f"ua.register_extension_object('{struct.name}'," \
180			f" ua.NodeId.from_string('{struct.typeid}'), {struct.name})\n"
181			return code
182
183			def get_python_classes(self, env=None):
184			return _generate_python_class(self.model, env=env)
185
186			def _make_header(self, _file):
187			_file.write("""
188			'''
189			THIS FILE IS AUTOGENERATED, DO NOT EDIT!!!
190			'''
191
192			from datetime import datetime
193			import uuid
194
195			from asyncua import ua
196			""")
197
198			def set_typeid(self, name, typeid):
199			for struct in self.model:
200			if struct.name == name:
201			struct.typeid = typeid
202			return
203
204
205			async def load_type_definitions(server, nodes=None):
206			"""
207			Download xml from given variable node defining custom structures.
208			If no node is given, attemps to import variables from all nodes under
209			"0:OPC Binary"
210			the code is generated and imported on the fly. If you know the structures
211			are not going to be modified it might be interresting to copy the generated files
212			and include them in you code
213			"""
214			if nodes is None:
215			nodes = []
216			for desc in await server.nodes.opc_binary.get_children_descriptions():
217			if desc.BrowseName != ua.QualifiedName("Opc.Ua"):
218			nodes.append(server.get_node(desc.NodeId))
219
220			structs_dict = {}
221			generators = []
222			for node in nodes:
223			xml = await node.read_value()
224			generator = StructGenerator()
225			generators.append(generator)
226			generator.make_model_from_string(xml)
227			# generate and execute new code on the fly
228			generator.get_python_classes(structs_dict)
229			# same but using a file that is imported. This can be usefull for debugging library
230			# name = node.read_browse_name().Name
231			# Make sure structure names do not contain charaters that cannot be used in Python class file names
232			# name = clean_name(name)
233			# name = "structures_" + node.read_browse_name().Name
234			# generator.save_and_import(name + ".py", append_to=structs_dict)
235
236			# register classes
237			# every children of our node should represent a class
238			for ndesc in await node.get_children_descriptions():
239			ndesc_node = server.get_node(ndesc.NodeId)
240			ref_desc_list = await ndesc_node.get_references(refs=ua.ObjectIds.HasDescription, direction=ua.BrowseDirection.Inverse)
241			if ref_desc_list: # some server put extra things here
242			name = clean_name(ndesc.BrowseName.Name)
243			if name not in structs_dict:
244			_logger.warning("%s is found as child of binary definition node but is not found in xml", name)
245			continue
246			nodeid = ref_desc_list[0].NodeId
247			ua.register_extension_object(name, nodeid, structs_dict[name])
248			# save the typeid if user want to create static file for type definitnion
249			generator.set_typeid(name, nodeid.to_string())
250
251			for key, val in structs_dict.items():
252			if isinstance(val, EnumMeta) and key != "IntEnum":
253			setattr(ua, key, val)
254
255			return generators, structs_dict
256
257
258			def _generate_python_class(model, env=None):
259			"""
260			generate Python code and execute in a new environment
261			return a dict of structures {name: class}
262			Rmw: Since the code is generated on the fly, in case of error the stack trace is
263			not available and debugging is very hard...
264			"""
265			if env is None:
266			env = ua.__dict__
267			# Add the required libraries to dict
268			if "ua" not in env:
269			env['ua'] = ua
270			if "datetime" not in env:
271			env['datetime'] = datetime
272			if "uuid" not in env:
273			env['uuid'] = uuid
274			if "enum" not in env:
275			env['IntEnum'] = IntEnum
276			# generate classes one by one and add them to dict
277			for element in model:
278			code = element.get_code()
279			exec(code, env)
280			return env
281
282
283			async def load_enums(server, env=None, force=False):
284			"""
285			Read enumeration data types on server and generate python Enums in ua scope for them
286			"""
287			model = []
288
289			for desc in await server.nodes.enum_data_type.get_children_descriptions(refs=ua.ObjectIds.HasSubtype):
290			enum_name = desc.BrowseName.Name
291			enum_node = server.get_node(desc.NodeId)
292			if not force and hasattr(ua, enum_name):
293			_logger.debug("Enum type %s is already in ua namespace, ignoring", enum_name)
294			continue
295			c = None
296			for child_desc in await enum_node.get_children_descriptions(refs=ua.ObjectIds.HasProperty):
297			child_node = server.get_node(child_desc.NodeId)
298			if child_desc.BrowseName.Name == "EnumStrings":
299			c = await _get_enum_strings(enum_name, child_node)
300			elif child_desc.BrowseName.Name == "EnumValues":
301			c = await _get_enum_values(enum_name, server.get_node(child_desc.NodeId))
302			else:
303			_logger.warning("Unexpected children of node %s: %s", desc, child_desc)
304			if c is not None:
305			model.append(c)
306			return _generate_python_class(model, env=env)
307
308
309			async def _get_enum_values(name, node):
310			val = await node.read_value()
311			c = EnumType(name)
312			c.fields = [EnumeratedValue(enumval.DisplayName.Text, enumval.Value) for enumval in val]
313			return c
314
315
316			async def _get_enum_strings(name, node):
317			val = await node.read_value()
318			c = EnumType(name)
319			c.fields = [EnumeratedValue(st.Text, idx) for idx, st in enumerate(val)]
320			return c
321

FreeOpcUa / opcua-asyncio

Push — master ( 9e5bc7...0261d3 )

asyncua.common.structures.Struct.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

asyncua.common.structures.Struct.init() A