asyncua.common.structures104.make_enum_code() - Code Metrics - FreeOpcUa/opcua-asyncio - Measure and Improve Code Quality continuously with Scrutinizer

asyncua.common.structures104.make_enum_code() A
last analyzed 2020-12-22 17:52 UTC

↳ Parent: asyncua.common.structures104

Complexity

Conditions

Size

Total Lines	20
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	7
nop	2
dl	0
loc	20
rs	10
c	0
b	0
f	0

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(data_type, 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
    '''

    data_type = ua.NodeId({data_type.Identifier}, {data_type.NamespaceIndex})

"""
    counter = 0
    # FIXME: with subscturutre weprobably need to add all fields from parents
    # this requires network call etc...
    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'
    if sdef.StructureType == ua.StructureType.StructureWithOptionalFields:
        code += "        ('Encoding', 'Byte'),\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:
            # FIXME: we are probably missing many custom tyes here based on builtin types
            # maybe we can use ua_utils.get_base_data_type()
            raise RuntimeError(f"Unknown datatype for field: {field} in structure:{name}, please report")
        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 = [f"{{name}}:{{val}}" for name, val in self.__dict__.items()]
        return f"{name}({{','.join(vals)}})"

    __repr__ = __str__

    def __init__(self):
"""
    if not sdef.Fields:
        code += "      pass"
    if sdef.StructureType == ua.StructureType.StructureWithOptionalFields:
        code += "        self.Encoding = 0\n"
    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"
    return code


async def _generate_object(name, sdef, data_type=None, 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(data_type, name, sdef)
    logger.debug("Executing code: %s", code)
    print("CODE", code)
    exec(code, env)
    return env


class DataTypeSorter:
    def __init__(self, data_type, name, desc, sdef):
        self.data_type = data_type
        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, parent_sdef=None):
    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)
        if parent_sdef:
            for field in reversed(parent_sdef.Fields):
                sdef.Fields.insert(0, field)
        dtypes.append(DataTypeSorter(desc.NodeId, name, desc, sdef))
        await _recursive_parse(server, server.get_node(desc.NodeId), dtypes, parent_sdef=sdef)


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)
    dtypes.sort()
    for dts in dtypes:
        try:
            env = await _generate_object(dts.name, dts.sdef, data_type=dts.data_type)
            ua.register_extension_object(dts.name, dts.encoding_id, env[dts.name], dts.desc.NodeId)
        except NotImplementedError:
            logger.exception("Structure type %s not implemented", dts.sdef)


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
    # FIXME: this is fishy, we may have same name in different Namespaces
    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)
        edef = await _read_data_type_definition(server, desc)
        if not edef:
            continue
        env = await _generate_object(name, edef, 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(data_type, 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			data_type = ua.NodeId({data_type.Identifier}, {data_type.NamespaceIndex})
68
69			"""
70			counter = 0
71			# FIXME: with subscturutre weprobably need to add all fields from parents
72			# this requires network call etc...
73			if sdef.StructureType == ua.StructureType.StructureWithOptionalFields:
74			code += ' ua_switches = {\n'
75			for field in sdef.Fields:
76
77			if field.IsOptional:
78			code += f" '{field.Name}': ('Encoding', {counter}),\n"
79			counter += 1
80			code += " }\n\n"
81
82			code += ' ua_types = [\n'
83			if sdef.StructureType == ua.StructureType.StructureWithOptionalFields:
84			code += " ('Encoding', 'Byte'),\n"
85			uatypes = []
86			for field in sdef.Fields:
87			prefix = 'ListOf' if field.ValueRank >= 1 else ''
88			if field.DataType.NamespaceIndex == 0 and field.DataType.Identifier in ua.ObjectIdNames:
89			uatype = ua.ObjectIdNames[field.DataType.Identifier]
90			elif field.DataType in ua.extension_objects_by_datatype:
91			uatype = ua.extension_objects_by_datatype[field.DataType].__name__
92			elif field.DataType in ua.enums_by_datatype:
93			uatype = ua.enums_by_datatype[field.DataType].__name__
94			else:
95			# FIXME: we are probably missing many custom tyes here based on builtin types
96			# maybe we can use ua_utils.get_base_data_type()
97			raise RuntimeError(f"Unknown datatype for field: {field} in structure:{name}, please report")
98			if field.ValueRank >= 1 and uatype == 'Char':
99			uatype = 'String'
100			uatypes.append((field, uatype))
101			code += f" ('{field.Name}', '{prefix + uatype}'),\n"
102			code += " ]\n"
103			code += f"""
104			def __str__(self):
105			vals = [f"{{name}}:{{val}}" for name, val in self.__dict__.items()]
106			return f"{name}({{','.join(vals)}})"
107
108			__repr__ = __str__
109
110			def __init__(self):
111			"""
112			if not sdef.Fields:
113			code += " pass"
114			if sdef.StructureType == ua.StructureType.StructureWithOptionalFields:
115			code += " self.Encoding = 0\n"
116			for field, uatype in uatypes:
117			if field.ValueRank >= 1:
118			default_value = "[]"
119			else:
120			default_value = get_default_value(uatype)
121			code += f" self.{field.Name} = {default_value}\n"
122			return code
123
124
125			async def _generate_object(name, sdef, data_type=None, env=None, enum=False):
126			"""
127			generate Python code and execute in a new environment
128			return a dict of structures {name: class}
129			Rmw: Since the code is generated on the fly, in case of error the stack trace is
130			not available and debugging is very hard...
131			"""
132			if env is None:
133			env = {}
134			# Add the required libraries to dict
135			if "ua" not in env:
136			env['ua'] = ua
137			if "datetime" not in env:
138			env['datetime'] = datetime
139			if "uuid" not in env:
140			env['uuid'] = uuid
141			if "enum" not in env:
142			env['IntEnum'] = IntEnum
143			# generate classe add it to env dict
144			if enum:
145			code = make_enum_code(name, sdef)
146			else:
147			code = make_structure_code(data_type, name, sdef)
148			logger.debug("Executing code: %s", code)
149			print("CODE", code)
150			exec(code, env)
151			return env
152
153
154			class DataTypeSorter:
155			def __init__(self, data_type, name, desc, sdef):
156			self.data_type = data_type
157			self.name = name
158			self.desc = desc
159			self.sdef = sdef
160			self.encoding_id = self.sdef.DefaultEncodingId
161			self.deps = [field.DataType for field in self.sdef.Fields]
162
163			def __lt__(self, other):
164			if self.desc.NodeId in other.deps:
165			return True
166			return False
167
168			def __str__(self):
169			return f"{self.__class__.__name__}({self.desc.NodeId, self.deps, self.encoding_id})"
170
171			__repr__ = __str__
172
173
174			async def _recursive_parse(server, base_node, dtypes, parent_sdef=None):
175			for desc in await base_node.get_children_descriptions(refs=ua.ObjectIds.HasSubtype):
176			sdef = await _read_data_type_definition(server, desc)
177			if not sdef:
178			continue
179			name = clean_name(desc.BrowseName.Name)
180			if parent_sdef:
181			for field in reversed(parent_sdef.Fields):
182			sdef.Fields.insert(0, field)
183			dtypes.append(DataTypeSorter(desc.NodeId, name, desc, sdef))
184			await _recursive_parse(server, server.get_node(desc.NodeId), dtypes, parent_sdef=sdef)
185
186
187			async def load_data_type_definitions(server, base_node=None):
188			await load_enums(server) # we need all enums to generate structure code
189			if base_node is None:
190			base_node = server.nodes.base_structure_type
191			dtypes = []
192			await _recursive_parse(server, base_node, dtypes)
193			dtypes.sort()
194			for dts in dtypes:
195			try:
196			env = await _generate_object(dts.name, dts.sdef, data_type=dts.data_type)
197			ua.register_extension_object(dts.name, dts.encoding_id, env[dts.name], dts.desc.NodeId)
198			except NotImplementedError:
199			logger.exception("Structure type %s not implemented", dts.sdef)
200
201
202			async def _read_data_type_definition(server, desc):
203			if desc.BrowseName.Name == "FilterOperand":
204			#FIXME: find out why that one is not in ua namespace...
205			return None
206			# FIXME: this is fishy, we may have same name in different Namespaces
207			if hasattr(ua, desc.BrowseName.Name):
208			return None
209			logger.warning("Registring data type %s %s", desc.NodeId, desc.BrowseName)
210			node = server.get_node(desc.NodeId)
211			try:
212			sdef = await node.read_data_type_definition()
213			except ua.uaerrors.BadAttributeIdInvalid:
214			logger.warning("%s has no DataTypeDefinition atttribute", node)
215			return None
216			except Exception:
217			logger.exception("Error getting datatype for node %s", node)
218			return None
219			return sdef
220
221
222			def make_enum_code(name, edef):
223			"""
224			if node has a DataTypeDefinition arttribute, generate enum code
225			"""
226			code = f"""
227
228			class {name}(IntEnum):
229
230			'''
231			{name} EnumInt autogenerated from EnumDefinition
232			'''
233
234			"""
235
236			for field in edef.Fields:
237			name = field.Name
238			value = field.Value
239			code += f" {name} = {value}\n"
240
241			return code
242
243
244			async def load_enums(server, base_node=None):
245			if base_node is None:
246			base_node = server.nodes.enum_data_type
247			for desc in await base_node.get_children_descriptions(refs=ua.ObjectIds.HasSubtype):
248			if hasattr(ua, desc.BrowseName.Name):
249			continue
250			logger.warning("Registring Enum %s %s", desc.NodeId, desc.BrowseName)
251			name = clean_name(desc.BrowseName.Name)
252			edef = await _read_data_type_definition(server, desc)
253			if not edef:
254			continue
255			env = await _generate_object(name, edef, enum=True)
256			ua.register_enum(name, desc.NodeId, env[name])
257

FreeOpcUa / opcua-asyncio

asyncua.common.structures104.make_enum_code() A last analyzed 2020-12-22 17:52 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

asyncua.common.structures104.make_enum_code() A
last analyzed 2020-12-22 17:52 UTC