much faster load_enums()

asyncua/common/structures.py

@@ -24,6 +24,10 @@ class EnumType(object):
         self.fields = []
         self.typeid = None
 
+    def __str__(self):
+        return f"EnumType({self.name, self.fields})"
+    __repr__ = __str__
+
     def get_code(self):
         code = """
 
@@ -43,7 +47,7 @@ class {0}(IntEnum):
         return code
 
 
-class EnumeratedValue(object):
+class EnumeratedValue:
     def __init__(self, name, value):
         if name == "None":
             name = "None_"
@@ -52,7 +56,7 @@ class EnumeratedValue(object):
         self.Value = value
 
 
-class Struct(object):
+class Struct:
     def __init__(self, name):
         self.name = clean_name(name)
         self.fields = []
@@ -66,7 +70,7 @@ class Struct(object):
     def get_code(self):
         code = f"""
 
-class {self.name}(object):
+class {self.name}:
 
     '''
     {self.name} structure autogenerated from xml
@@ -255,7 +259,7 @@ def _generate_python_class(model, env=None):
     not available and debugging is very hard...
     """
     if env is None:
-        env = {}
+        env = ua.__dict__
     #  Add the required libraries to dict
     if "ua" not in env:
         env['ua'] = ua
@@ -272,41 +276,41 @@ def _generate_python_class(model, env=None):
     return env
 
 
-async def load_enums(server, env=None):
+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 = []
-    nodes = await server.nodes.enum_data_type.get_children()
-    if env is None:
-        env = ua.__dict__
-    for node in nodes:
-        name = (await node.read_browse_name()).Name
-        try:
-            c = await _get_enum_strings(name, node)
-        except ua.UaError as ex:
-            try:
-                c = await _get_enum_values(name, node)
-            except ua.UaError as ex:
-                _logger.warning(f"Node {name}, {node} under DataTypes/Enumeration," f" does not seem to have a child called EnumString or EumValue: {ex}")
-                continue
-        if not hasattr(ua, c.name):
-            _logger.warning("Adding enum %s to ua namespace", c)
+
+    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):
-    def_node = await node.get_child("0:EnumValues")
-    val = await def_node.read_value()
+    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):
-    def_node = await node.get_child("0:EnumStrings")
-    val = await def_node.read_value()
+    val = await node.read_value()
     c = EnumType(name)
     c.fields = [EnumeratedValue(st.Text, idx) for idx, st in enumerate(val)]
     return c

tests/test_server.py

@@ -512,7 +512,7 @@ async def test_get_node_by_ns(server):
 
 async def test_load_enum_strings(server):
     dt = await server.nodes.enum_data_type.add_data_type(0, "MyStringEnum")
-    await dt.add_variable(0, "EnumStrings", [ua.LocalizedText("e1"), ua.LocalizedText("e2"), ua.LocalizedText("e3"),
+    await dt.add_property(0, "EnumStrings", [ua.LocalizedText("e1"), ua.LocalizedText("e2"), ua.LocalizedText("e3"),
                                        ua.LocalizedText("e 4")])
     await server.load_enums()
     e = getattr(ua, "MyStringEnum")
@@ -533,7 +533,7 @@ async def test_load_enum_values(server):
     v3 = ua.EnumValueType()
     v3.DisplayName.Text = "v 3 "
     v3.Value = 4
-    await dt.add_variable(0, "EnumValues", [v1, v2, v3])
+    await dt.add_property(0, "EnumValues", [v1, v2, v3])
     await server.load_enums()
     e = getattr(ua, "MyValuesEnum")
     assert isinstance(e, EnumMeta)