Commit 417c0ac3 authored by oroulet's avatar oroulet

some typing add tests for spaces in custom enums and structs names and members

parent 124fe686
......@@ -4,15 +4,18 @@ import uuid
from enum import IntEnum
import logging
import re
from typing import Union, List, TYPE_CHECKING, Tuple
from asyncua import ua
from asyncua import Node
from asyncua.common.manage_nodes import create_encoding, create_data_type
if TYPE_CHECKING:
from asyncua import Client, Server
logger = logging.getLogger(__name__)
def new_struct_field(name, dtype, array=False, optional=False, description=""):
def new_struct_field(name: str, dtype: Union[ua.NodeId, Node, ua.VariantType], array: bool = False, optional: bool = False, description: str = "") -> ua.StructureField:
"""
simple way to create a StructureField
"""
......@@ -40,12 +43,16 @@ def new_struct_field(name, dtype, array=False, optional=False, description=""):
return field
async def new_struct(server, idx, name, fields):
async def new_struct(server: Union["Server", "Client"], idx: Union[int, ua.NodeId], name: Union[int, ua.QualifiedName], fields: List[ua.StructureField]) -> Tuple[Node, List[Node]]:
"""
simple way to create a new structure
return the created data type node and the list of encoding nodes
"""
dtype = await create_data_type(server.nodes.base_structure_type, idx, name)
if isinstance(idx, ua.NodeId):
# user has provided a node id, we cannot reuse it
idx = idx.NamespaceIndex
enc = await create_encoding(dtype, idx, "Default Binary")
# TODO: add other encoding the day we support them
......@@ -63,7 +70,7 @@ async def new_struct(server, idx, name, fields):
return dtype, [enc]
async def new_enum(server, idx, name, values):
async def new_enum(server: Union["Server", "Client"], idx: Union[int, ua.NodeId], name: Union[int, ua.QualifiedName], values: List[str]) -> Node:
edef = ua.EnumDefinition()
counter = 0
for val_name in values:
......@@ -258,7 +265,7 @@ async def _recursive_parse(server, base_node, dtypes, parent_sdef=None):
await _recursive_parse(server, server.get_node(desc.NodeId), dtypes, parent_sdef=sdef)
async def load_data_type_definitions(server, base_node=None):
async def load_data_type_definitions(server: Union["Server", "Client"], base_node: Node = None) -> 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
......@@ -315,7 +322,7 @@ class {name}(IntEnum):
return code
async def load_enums(server, base_node=None):
async def load_enums(server: Union["Server", "Client"], base_node: Node = None) -> 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):
......
......@@ -4,23 +4,21 @@ import sys
sys.path.insert(0, "..")
from IPython import embed
from asyncua import ua, uamethod, Server
async def main():
logging.basicConfig(level=logging.INFO)
server = Server()
await server.init()
# import some nodes from xml
await server.import_xml("../schemas/UA-Nodeset/DI/Opc.Ua.Di.NodeSet2.xml")
await server.import_xml("../schemas/UA-Nodeset/Robotics/Opc.Ua.Robotics.NodeSet2.xml")
await server.import_xml("../schemas/UA-Nodeset-master/DI/Opc.Ua.Di.NodeSet2.xml")
await server.import_xml("../schemas/UA-Nodeset-master/Robotics/Opc.Ua.Robotics.NodeSet2.xml")
# starting!
async with server:
embed()
while True:
await asyncio.sleep(1)
if __name__ == "__main__":
......
......@@ -1248,14 +1248,14 @@ async def test_custom_struct_with_enum(opc):
async def test_two_times_enum(opc):
idx = 4
dtype = await new_enum(opc.opc, idx, "MyCustEnum5", [
await new_enum(opc.opc, idx, "MyCustEnum5", [
"titi",
"toto",
"tutu",
])
with pytest.raises(ua.uaerrors.BadBrowseNameDuplicated):
dtype = await new_enum(opc.opc, idx, "MyCustEnum5", [
await new_enum(opc.opc, idx, "MyCustEnum5", [
"titi",
])
......@@ -1301,3 +1301,45 @@ async def test_custom_struct_import(opc):
assert sdef.Fields[0].Name == "MyBool"
await opc.opc.export_xml(nodes, "tests/custom_struct_v2.xml")
async def test_enum_string_identifier_and_spaces(opc):
idx = 4
nodeid = ua.NodeId("My Identifier", idx)
qname = ua.QualifiedName("My Enum", idx)
await new_enum(opc.opc, nodeid, qname, [
"my name with hole",
"toto",
"tutu",
])
await opc.opc.load_data_type_definitions()
var = await opc.opc.nodes.objects.add_variable(idx, "my enum", ua.My_Enum.my_name_with_hole)
val = await var.read_value()
assert val == 0
async def test_custom_struct_of_struct_with_spaces(opc):
idx = 6
nodeid = ua.NodeId("toto.My Identifier", idx)
qname = ua.QualifiedName("My Sub Struct 1", idx)
dtype, encs = await new_struct(opc.opc, nodeid, qname, [
new_struct_field("My Bool", ua.VariantType.Boolean),
new_struct_field("My UInt32", ua.VariantType.UInt32),
])
await new_struct(opc.opc, idx, "My Mother Struct", [
new_struct_field("My Bool", ua.VariantType.Boolean),
new_struct_field("My Sub Struct", dtype),
])
await opc.opc.load_data_type_definitions()
mystruct = ua.My_Mother_Struct()
mystruct.My_Sub_Struct = ua.My_Sub_Struct_1()
mystruct.My_Sub_Struct.My_UInt32 = 78
var = await opc.opc.nodes.objects.add_variable(idx, "my mother struct", ua.Variant(mystruct, ua.VariantType.ExtensionObject))
val = await var.read_value()
assert val.My_Sub_Struct.My_UInt32 == 78
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment