asyncua.common.structures104.make_structure_code() - Code Metrics - Inspection of "Fixes" - FreeOpcUa/opcua-asyncio - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#238)

by Olivier

created 2020-07-22 09:15 UTC

asyncua.common.structures104.make_structure_code() F

↳ Parent: asyncua.common.structures104

Complexity

Conditions

Size

Total Lines	63
Code Lines	38

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	16
eloc	38
nop	2
dl	0
loc	63
rs	2.4
c	0
b	0
f	0

How to fix Long Method Complexity

from enum import Enum
from datetime import datetime
import uuid
from enum import IntEnum
import logging
import re

from asyncua import ua

logger = logging.getLogger(__name__)


def clean_name(name):
    """
    Remove characters that might be present in  OPC UA structures
    but cannot be part of of Python class names
    """
    newname = re.sub(r'\W+', '_', name)
    newname = re.sub(r'^[0-9]+', r'_\g<0>', newname)

    if name != newname:
        logger.warning("renamed %s to %s due to Python syntax", name, newname)
    return newname


def get_default_value(uatype, enums=None):
    if enums is None:
        enums = {}
    if uatype == "String":
        return "None"
    elif uatype == "Guid":
        return "uuid.uuid4()"
    elif uatype in ("ByteString", "CharArray", "Char"):
        return b''
    elif uatype == "Boolean":
        return "True"
    elif uatype == "DateTime":
        return "datetime.utcnow()"
    elif uatype in ("Int16", "Int32", "Int64", "UInt16", "UInt32", "UInt64", "Double", "Float", "Byte", "SByte"):
        return 0
    elif uatype in enums:
        return f"ua.{uatype}({enums[uatype]})"
    elif hasattr(ua, uatype) and issubclass(getattr(ua, uatype), Enum):
        # We have an enum, try to initilize it correctly
        val = list(getattr(ua, uatype).__members__)[0]
        return f"ua.{uatype}.{val}"
    else:
        return f"ua.{uatype}()"


def make_structure_code(name, sdef):
    """
    given a StructureDefinition object, generate Python code
    """
    if sdef.StructureType not in (ua.StructureType.Structure, ua.StructureType.StructureWithOptionalFields):
        #if sdef.StructureType != ua.StructureType.Structure:
        raise NotImplementedError(f"Only StructureType implemented, not {ua.StructureType(sdef.StructureType).name} for node {name} with DataTypdeDefinition {sdef}")

    code = f"""

class {name}:

    '''
    {name} structure autogenerated from StructureDefinition object
    '''

"""
    counter = 0
    if sdef.StructureType == ua.StructureType.StructureWithOptionalFields:
        code += '    ua_switches = {\n'
        for field in sdef.Fields:

            if field.IsOptional:
                code += f"        '{field.Name}': ('Encoding', {counter}),\n"
                counter += 1
        code += "    }\n\n"

    code += '    ua_types = [\n'
    uatypes = []
    for field in sdef.Fields:
        prefix = 'ListOf' if field.ValueRank >= 1 else ''
        if field.DataType.NamespaceIndex == 0 and field.DataType.Identifier in ua.ObjectIdNames:
            uatype = ua.ObjectIdNames[field.DataType.Identifier]
        elif field.DataType in ua.extension_objects_by_datatype:
            uatype = ua.extension_objects_by_datatype[field.DataType].__name__
        elif field.DataType in ua.enums_by_datatype:
            uatype = ua.enums_by_datatype[field.DataType].__name__
        else:
            raise RuntimeError(f"Unknown datatype for field: {field} in structure:{name}")
        if field.ValueRank >= 1 and uatype == 'Char':
            uatype = 'String'
        uatypes.append((field, uatype))
        code += f"        ('{field.Name}', '{prefix + uatype}'),\n"
    code += "    ]\n"
    code += f"""
    def __str__(self):
        vals = [name + ": " + str(val) for name, val in self.__dict__.items()]
        return "{name}(" + ", ".join(vals) + ")"

    __repr__ = __str__

    def __init__(self):
"""
    if not sdef.Fields:
        code += "      pass"
    for field, uatype in uatypes:
        if field.ValueRank >= 1:
            default_value = "[]"
        else:
            default_value = get_default_value(uatype)
        code += f"        self.{field.Name} = {default_value}\n"
    print("CODE", code)
    return code


async def _generate_object(name, sdef, env=None, enum=False):
    """
    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 = {}
    #  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 classe add it to env dict
    if enum:
        code = make_enum_code(name, sdef)
    else:
        code = make_structure_code(name, sdef)
    logger.debug("Executing code: %s", code)
    print("CODE", code)
    exec(code, env)
    return env


class DataTypeSorter:
    def __init__(self, name, desc, sdef):
        self.name = name
        self.desc = desc
        self.sdef = sdef
        self.encoding_id = self.sdef.DefaultEncodingId
        self.deps = [field.DataType for field in self.sdef.Fields]

    def __lt__(self, other):
        if self.desc.NodeId in other.deps:
            return True
        return False

    def __str__(self):
        return f"{self.__class__.__name__}({self.desc.NodeId, self.deps, self.encoding_id})"

    __repr__ = __str__


async def _recursive_parse(server, base_node, dtypes):
    for desc in await base_node.get_children_descriptions(refs=ua.ObjectIds.HasSubtype):
        sdef = await _read_data_type_definition(server, desc)
        if not sdef:
            continue
        name = clean_name(desc.BrowseName.Name)
        dtypes.append(DataTypeSorter(name, desc, sdef))
        await _recursive_parse(server, server.get_node(desc.NodeId), dtypes)


async def load_data_type_definitions(server, base_node=None):
    await load_enums(server)  # we need all enums to generate structure code
    if base_node is None:
        base_node = server.nodes.base_structure_type
    dtypes = []
    await _recursive_parse(server, base_node, dtypes)
    for dts in dtypes:
        env = await _generate_object(dts.name, dts.sdef)
        ua.register_extension_object(dts.name, dts.encoding_id, env[dts.name], dts.desc.NodeId)


async def _read_data_type_definition(server, desc):
    if desc.BrowseName.Name == "FilterOperand":
        #FIXME: find out why that one is not in ua namespace...
        return None
    if hasattr(ua, desc.BrowseName.Name):
        return None
    logger.warning("Registring data type %s %s", desc.NodeId, desc.BrowseName)
    node = server.get_node(desc.NodeId)
    try:
        sdef = await node.read_data_type_definition()
    except ua.uaerrors.BadAttributeIdInvalid:
        logger.warning("%s has no DataTypeDefinition atttribute", node)
        return None
    except Exception:
        logger.exception("Error getting datatype for node %s", node)
        return None
    return sdef


def make_enum_code(name, edef):
    """
    if node has a DataTypeDefinition arttribute, generate enum code
    """
    code = f"""

class {name}(IntEnum):

    '''
    {name} EnumInt autogenerated from EnumDefinition
    '''

"""

    for field in edef.Fields:
        name = field.Name
        value = field.Value
        code += f"    {name} = {value}\n"

    return code


async def load_enums(server, base_node=None):
    if base_node is None:
        base_node = server.nodes.enum_data_type
    for desc in await base_node.get_children_descriptions(refs=ua.ObjectIds.HasSubtype):
        if hasattr(ua, desc.BrowseName.Name):
            continue
        logger.warning("Registring Enum %s %s", desc.NodeId, desc.BrowseName)
        name = clean_name(desc.BrowseName.Name)
        print("LOOKING AT ENUM", name)
        sdef = await _read_data_type_definition(server, desc)
        if not sdef:
            continue
        print("SDEF", sdef)
        node = server.get_node(desc.NodeId)
        env = await _generate_object(name, sdef, enum=True)
        ua.register_enum(name, desc.NodeId, env[name])


1			from enum import Enum
2			from datetime import datetime
3			import uuid
4			from enum import IntEnum
5			import logging
6			import re
7
8			from asyncua import ua
9
10			logger = logging.getLogger(__name__)
11
12
13			def clean_name(name):
14			"""
15			Remove characters that might be present in OPC UA structures
16			but cannot be part of of Python class names
17			"""
18			newname = re.sub(r'\W+', '_', name)
19			newname = re.sub(r'^[0-9]+', r'_\g<0>', newname)
20
21			if name != newname:
22			logger.warning("renamed %s to %s due to Python syntax", name, newname)
23			return newname
24
25
26			def get_default_value(uatype, enums=None):
27			if enums is None:
28			enums = {}
29			if uatype == "String":
30			return "None"
31			elif uatype == "Guid":
32			return "uuid.uuid4()"
33			elif uatype in ("ByteString", "CharArray", "Char"):
34			return b''
35			elif uatype == "Boolean":
36			return "True"
37			elif uatype == "DateTime":
38			return "datetime.utcnow()"
39			elif uatype in ("Int16", "Int32", "Int64", "UInt16", "UInt32", "UInt64", "Double", "Float", "Byte", "SByte"):
40			return 0
41			elif uatype in enums:
42			return f"ua.{uatype}({enums[uatype]})"
43			elif hasattr(ua, uatype) and issubclass(getattr(ua, uatype), Enum):
44			# We have an enum, try to initilize it correctly
45			val = list(getattr(ua, uatype).__members__)[0]
46			return f"ua.{uatype}.{val}"
47			else:
48			return f"ua.{uatype}()"
49
50
51			def make_structure_code(name, sdef):
52			"""
53			given a StructureDefinition object, generate Python code
54			"""
55			if sdef.StructureType not in (ua.StructureType.Structure, ua.StructureType.StructureWithOptionalFields):
56			#if sdef.StructureType != ua.StructureType.Structure:
57			raise NotImplementedError(f"Only StructureType implemented, not {ua.StructureType(sdef.StructureType).name} for node {name} with DataTypdeDefinition {sdef}")
58
59			code = f"""
60
61			class {name}:
62
63			'''
64			{name} structure autogenerated from StructureDefinition object
65			'''
66
67			"""
68			counter = 0
69			if sdef.StructureType == ua.StructureType.StructureWithOptionalFields:
70			code += ' ua_switches = {\n'
71			for field in sdef.Fields:
72
73			if field.IsOptional:
74			code += f" '{field.Name}': ('Encoding', {counter}),\n"
75			counter += 1
76			code += " }\n\n"
77
78			code += ' ua_types = [\n'
79			uatypes = []
80			for field in sdef.Fields:
81			prefix = 'ListOf' if field.ValueRank >= 1 else ''
82			if field.DataType.NamespaceIndex == 0 and field.DataType.Identifier in ua.ObjectIdNames:
83			uatype = ua.ObjectIdNames[field.DataType.Identifier]
84			elif field.DataType in ua.extension_objects_by_datatype:
85			uatype = ua.extension_objects_by_datatype[field.DataType].__name__
86			elif field.DataType in ua.enums_by_datatype:
87			uatype = ua.enums_by_datatype[field.DataType].__name__
88			else:
89			raise RuntimeError(f"Unknown datatype for field: {field} in structure:{name}")
90			if field.ValueRank >= 1 and uatype == 'Char':
91			uatype = 'String'
92			uatypes.append((field, uatype))
93			code += f" ('{field.Name}', '{prefix + uatype}'),\n"
94			code += " ]\n"
95			code += f"""
96			def __str__(self):
97			vals = [name + ": " + str(val) for name, val in self.__dict__.items()]
98			return "{name}(" + ", ".join(vals) + ")"
99
100			__repr__ = __str__
101
102			def __init__(self):
103			"""
104			if not sdef.Fields:
105			code += " pass"
106			for field, uatype in uatypes:
107			if field.ValueRank >= 1:
108			default_value = "[]"
109			else:
110			default_value = get_default_value(uatype)
111			code += f" self.{field.Name} = {default_value}\n"
112			print("CODE", code)
113			return code
114
115
116			async def _generate_object(name, sdef, env=None, enum=False):
117			"""
118			generate Python code and execute in a new environment
119			return a dict of structures {name: class}
120			Rmw: Since the code is generated on the fly, in case of error the stack trace is
121			not available and debugging is very hard...
122			"""
123			if env is None:
124			env = {}
125			# Add the required libraries to dict
126			if "ua" not in env:
127			env['ua'] = ua
128			if "datetime" not in env:
129			env['datetime'] = datetime
130			if "uuid" not in env:
131			env['uuid'] = uuid
132			if "enum" not in env:
133			env['IntEnum'] = IntEnum
134			# generate classe add it to env dict
135			if enum:
136			code = make_enum_code(name, sdef)
137			else:
138			code = make_structure_code(name, sdef)
139			logger.debug("Executing code: %s", code)
140			print("CODE", code)
141			exec(code, env)
142			return env
143
144
145			class DataTypeSorter:
146			def __init__(self, name, desc, sdef):
147			self.name = name
148			self.desc = desc
149			self.sdef = sdef
150			self.encoding_id = self.sdef.DefaultEncodingId
151			self.deps = [field.DataType for field in self.sdef.Fields]
152
153			def __lt__(self, other):
154			if self.desc.NodeId in other.deps:
155			return True
156			return False
157
158			def __str__(self):
159			return f"{self.__class__.__name__}({self.desc.NodeId, self.deps, self.encoding_id})"
160
161			__repr__ = __str__
162
163
164			async def _recursive_parse(server, base_node, dtypes):
165			for desc in await base_node.get_children_descriptions(refs=ua.ObjectIds.HasSubtype):
166			sdef = await _read_data_type_definition(server, desc)
167			if not sdef:
168			continue
169			name = clean_name(desc.BrowseName.Name)
170			dtypes.append(DataTypeSorter(name, desc, sdef))
171			await _recursive_parse(server, server.get_node(desc.NodeId), dtypes)
172
173
174			async def load_data_type_definitions(server, base_node=None):
175			await load_enums(server) # we need all enums to generate structure code
176			if base_node is None:
177			base_node = server.nodes.base_structure_type
178			dtypes = []
179			await _recursive_parse(server, base_node, dtypes)
180			for dts in dtypes:
181			env = await _generate_object(dts.name, dts.sdef)
182			ua.register_extension_object(dts.name, dts.encoding_id, env[dts.name], dts.desc.NodeId)
183
184
185			async def _read_data_type_definition(server, desc):
186			if desc.BrowseName.Name == "FilterOperand":
187			#FIXME: find out why that one is not in ua namespace...
188			return None
189			if hasattr(ua, desc.BrowseName.Name):
190			return None
191			logger.warning("Registring data type %s %s", desc.NodeId, desc.BrowseName)
192			node = server.get_node(desc.NodeId)
193			try:
194			sdef = await node.read_data_type_definition()
195			except ua.uaerrors.BadAttributeIdInvalid:
196			logger.warning("%s has no DataTypeDefinition atttribute", node)
197			return None
198			except Exception:
199			logger.exception("Error getting datatype for node %s", node)
200			return None
201			return sdef
202
203
204			def make_enum_code(name, edef):
205			"""
206			if node has a DataTypeDefinition arttribute, generate enum code
207			"""
208			code = f"""
209
210			class {name}(IntEnum):
211
212			'''
213			{name} EnumInt autogenerated from EnumDefinition
214			'''
215
216			"""
217
218			for field in edef.Fields:
219			name = field.Name
220			value = field.Value
221			code += f" {name} = {value}\n"
222
223			return code
224
225
226			async def load_enums(server, base_node=None):
227			if base_node is None:
228			base_node = server.nodes.enum_data_type
229			for desc in await base_node.get_children_descriptions(refs=ua.ObjectIds.HasSubtype):
230			if hasattr(ua, desc.BrowseName.Name):
231			continue
232			logger.warning("Registring Enum %s %s", desc.NodeId, desc.BrowseName)
233			name = clean_name(desc.BrowseName.Name)
234			print("LOOKING AT ENUM", name)
235			sdef = await _read_data_type_definition(server, desc)
236			if not sdef:
237			continue
238			print("SDEF", sdef)
239			node = server.get_node(desc.NodeId)
240			env = await _generate_object(name, sdef, enum=True)
241			ua.register_enum(name, desc.NodeId, env[name])
242

FreeOpcUa / opcua-asyncio

Pull Request — master (#238)

asyncua.common.structures104.make_structure_code() F

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like