Address review feedback: fix field duplication and refactor type resolution

Author: RinZ27

asyncua/common/structures104.py

@@ -196,7 +196,9 @@ def setter(self, value):
     return property(getter, setter)
 
 
-def make_structure(data_type: ua.NodeId, struct_name: str, sdef: ua.StructureDefinition, log_error: bool = True) -> dict[str, type]:
+def make_structure(
+    data_type: ua.NodeId, struct_name: str, sdef: ua.StructureDefinition, log_error: bool = True
+) -> dict[str, type]:
     """
     given a StructureDefinition object, generate Python class
     """
@@ -235,9 +237,6 @@ def make_structure(data_type: ua.NodeId, struct_name: str, sdef: ua.StructureDef
 
     for idx, sfield in enumerate(sdef.Fields, start=1):
         fname = clean_name(sfield.Name)
-        if fname in seen_names:
-            _logger.warning("Field name %s is duplicated in structure %s, renaming to %s_%s", fname, struct_name, fname, idx)
-            fname = f"{fname}_{idx}"
         seen_names.add(fname)
         if sfield.DataType.NamespaceIndex == 0 and sfield.DataType.Identifier in ua.ObjectIdNames:
             if sfield.DataType.Identifier == 24:
@@ -409,18 +408,20 @@ def __str__(self):
         return f"<{self.__class__.__name__}: {self.name!r}>"
 
 
-async def _recursive_parse(server, base_node, dtypes, parent_sdef=None, add_existing=False):
+async def _recursive_parse(server, base_node, dtypes, parent_sdef=None, add_existing=False) -> None:
     ch = await base_node.get_children_descriptions(refs=ua.ObjectIds.HasSubtype)
 
     requests = [_read_data_type_definition(server, desc, read_existing=add_existing) for desc in ch]
     results = await asyncio.gather(*requests)
 
-    def __add_recursion(sdef, desc):
+    def __add_recursion(sdef, desc) -> Any:
         name = clean_name(desc.BrowseName.Name)
         if sdef:
             if parent_sdef:
+                names = [f.Name for f in sdef.Fields]
                 for sfield in reversed(parent_sdef.Fields):
-                    sdef.Fields.insert(0, sfield)
+                    if sfield.Name not in names:
+                        sdef.Fields.insert(0, sfield)
             if isinstance(sdef, ua.StructureDefinition):
                 dtypes.append(DataTypeSorter(desc.NodeId, name, desc, sdef))
             return _recursive_parse(
@@ -443,7 +444,7 @@ def __add_recursion(sdef, desc):
     await asyncio.gather(*requests)
 
 
-async def _get_parent_types(node: Node):
+async def _get_parent_types(node: Node) -> list[Node]:
     parents = []
     tmp_node = node
     for _ in range(10):
@@ -463,16 +464,18 @@ async def load_custom_struct(node: Node) -> Any:
     name = (await node.read_browse_name()).Name
     for parent in await _get_parent_types(node):
         parent_sdef = await parent.read_data_type_definition()
+        names = [f.Name for f in sdef.Fields]
         for f in reversed(parent_sdef.Fields):
-            sdef.Fields.insert(0, f)
+            if f.Name not in names:
+                sdef.Fields.insert(0, f)
     env = _generate_object(name, sdef, data_type=node.nodeid)
     struct = env[name]
     setattr(ua, name, struct)
     ua.register_extension_object(name, sdef.DefaultEncodingId, struct, node.nodeid)
     return struct
 
 
-async def load_custom_struct_xml_import(node_id: ua.NodeId, attrs: ua.DataTypeAttributes):
+async def load_custom_struct_xml_import(node_id: ua.NodeId, attrs: ua.DataTypeAttributes) -> Any:
     """
     This function is used to load custom structs from xmlimporter
     """
@@ -488,7 +491,7 @@ async def load_custom_struct_xml_import(node_id: ua.NodeId, attrs: ua.DataTypeAt
     return struct
 
 
-async def _recursive_parse_basedatatypes(server, base_node, parent_datatype, new_alias) -> Any:
+async def _recursive_parse_basedatatypes(server, base_node, parent_datatype, new_alias) -> None:
     for desc in await base_node.get_children_descriptions(refs=ua.ObjectIds.HasSubtype):
         name = clean_name(desc.BrowseName.Name)
         if parent_datatype not in "Number":
@@ -500,7 +503,7 @@ async def _recursive_parse_basedatatypes(server, base_node, parent_datatype, new
         await _recursive_parse_basedatatypes(server, server.get_node(desc.NodeId), name, new_alias)
 
 
-async def load_basetype_alias_xml_import(server, name, nodeid, parent_datatype_nid):
+async def load_basetype_alias_xml_import(server, name, nodeid, parent_datatype_nid) -> Any:
     """
     Insert alias for a datatype used for xml import
     """
@@ -521,7 +524,7 @@ def make_basetype(name: str, parent_datatype: str) -> dict[str, Any]:
     return {name: getattr(ua, parent_datatype)}
 
 
-async def _load_base_datatypes(server: Server | Client) -> Any:
+async def _load_base_datatypes(server: Server | Client) -> dict[str, Any]:
     new_alias = {}
     descriptions = await server.nodes.base_data_type.get_children_descriptions()
     for desc in descriptions:

asyncua/ua/ua_binary.py

@@ -285,26 +285,33 @@ def deserialize(data):
 def resolve_uatype(ftype: Any) -> tuple[Any, bool]:
     if isinstance(ftype, str):
         # Resolve string hint
-        try:
-            # This is a bit hacky, we assume ftype was derived from a field's hint
-            # But for Unions, we might need a better way.
-            # Actually, let's just use eval if it looks like a simple name
-            if ftype.startswith("ua."):
-                ftype = getattr(ua, ftype[3:])
-            elif hasattr(ua, ftype):
-                ftype = getattr(ua, ftype)
-            else:
-                ftype = eval(ftype, {"ua": ua, "typing": typing, "list": list, "List": list, "Union": typing.Union, "Optional": typing.Optional, "Dict": dict})  # type: ignore[arg-type]
-        except Exception:
-            _logger.exception("Failed to resolve type hint: %s", ftype)
+        if ftype.startswith("ua."):
+            ftype = getattr(ua, ftype[3:])
+        elif hasattr(ua, ftype):
+            ftype = getattr(ua, ftype)
+        else:
+            try:
+                ftype = eval(
+                    ftype,
+                    {
+                        "ua": ua,
+                        "typing": typing,
+                        "list": list,
+                        "List": list,
+                        "Union": typing.Union,
+                        "Optional": typing.Optional,
+                        "Dict": dict,
+                    },
+                )  # type: ignore[arg-type]
+            except Exception:
+                _logger.exception("Failed to resolve type hint: %s", ftype)
     is_optional = type_is_optional(ftype)
     if is_optional:
         ftype = type_from_optional(ftype)
     return ftype, is_optional
 
 
-def field_serializer(ftype, dataclazz) -> Callable[[Any], bytes]:
-    uatype, is_optional = resolve_uatype(ftype)
+def field_serializer(uatype: Any, is_optional: bool, dataclazz: type) -> Callable[[Any], bytes]:
     if type_is_list(uatype):
         ft = type_from_list(uatype)
         if is_optional:
@@ -334,7 +341,7 @@ def create_dataclass_serializer(dataclazz):
     if issubclass(dataclazz, ua.UaUnion):
         # Union is a class with Encoding and Value field
         # the value depends on encoding
-        encoding_funcs = [field_serializer(t, dataclazz) for t in dataclazz._union_types]
+        encoding_funcs = [field_serializer(*resolve_uatype(t), dataclazz) for t in dataclazz._union_types]
 
         def union_serialize(obj):
             bin = Primitives.UInt32.pack(obj.Encoding)
@@ -358,7 +365,9 @@ def enc_value(obj):
                 enc |= enc_val
         return enc
 
-    encoding_functions = [(f.name, field_serializer(resolved_fieldtypes[f.name], dataclazz)) for f in data_fields]
+    encoding_functions = [
+        (f.name, field_serializer(*resolve_uatype(resolved_fieldtypes[f.name]), dataclazz)) for f in data_fields
+    ]
 
     def serialize(obj):
         return b"".join(