mirror/opcua-asyncio
1
0
Fork 0
mirror of https://github.com/FreeOpcUa/opcua-asyncio synced 2025-10-29 17:07:18 +08:00
Code Issues Releases Wiki Activity

Fix custom data type creation from amonsch and other cleanups

Browse Source
This commit is contained in:
oroulet 2019-10-01 09:27:46 +02:00 committed by oroulet
parent d7ed62e7cb
commit b65a1fa4b7
2 changed files with 37 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
asyncua/server/server.py
View File

@ -28,6 +28,14 @@ from ..crypto import security_policies, uacrypto
_logger = logging.getLogger(__name__)
def _get_node(isession, whatever):
if isinstance(whatever, Node):
return whatever
if isinstance(whatever, ua.NodeId):
return Node(isession, whatever)
return Node(isession, ua.NodeId(whatever))
class Server:
"""
High level Server class
@ -412,17 +420,25 @@ class Server:
await ev_gen.init(etype, emitting_node=emitting_node)
return ev_gen
def create_custom_data_type(self, idx, name, basetype=ua.ObjectIds.BaseDataType, properties=None) -> Coroutine:
async def create_custom_data_type(self, idx, name, basetype=ua.ObjectIds.BaseDataType, properties=None, description=None) -> Coroutine:
if properties is None:
properties = []
return self._create_custom_type(idx, name, basetype, properties, [], [])
base_t = _get_node(self.iserver.isession, basetype)
def create_custom_event_type(self, idx, name, basetype=ua.ObjectIds.BaseEventType, properties=None) -> Coroutine:
custom_t = await base_t.add_data_type(idx, name, description)
for prop in properties:
datatype = None
if len(prop) > 2:
datatype = prop[2]
await custom_t.add_property(idx, prop[0], ua.get_default_value(prop[1]), varianttype=prop[1], datatype=datatype)
return custom_t
async def create_custom_event_type(self, idx, name, basetype=ua.ObjectIds.BaseEventType, properties=None) -> Coroutine:
if properties is None:
properties = []
return self._create_custom_type(idx, name, basetype, properties, [], [])
return await self._create_custom_type(idx, name, basetype, properties, [], [])
def create_custom_object_type(self,
async def create_custom_object_type(self,
idx,
name,
basetype=ua.ObjectIds.BaseObjectType,
@ -435,12 +451,12 @@ class Server:
variables = []
if methods is None:
methods = []
return self._create_custom_type(idx, name, basetype, properties, variables, methods)
return await self._create_custom_type(idx, name, basetype, properties, variables, methods)
# def create_custom_reference_type(self, idx, name, basetype=ua.ObjectIds.BaseReferenceType, properties=[]):
# return self._create_custom_type(idx, name, basetype, properties)
def create_custom_variable_type(self,
async def create_custom_variable_type(self,
idx,
name,
basetype=ua.ObjectIds.BaseVariableType,
@ -453,15 +469,10 @@ class Server:
variables = []
if methods is None:
methods = []
return self._create_custom_type(idx, name, basetype, properties, variables, methods)
return await self._create_custom_type(idx, name, basetype, properties, variables, methods)
async def _create_custom_type(self, idx, name, basetype, properties, variables, methods):
if isinstance(basetype, Node):
base_t = basetype
elif isinstance(basetype, ua.NodeId):
base_t = Node(self.iserver.isession, basetype)
else:
base_t = Node(self.iserver.isession, ua.NodeId(basetype))
base_t = _get_node(self.iserver.isession, basetype)
custom_t = await base_t.add_object_type(idx, name)
for prop in properties:
datatype = None

22
tests/test_server.py
View File

@ -173,7 +173,7 @@ async def test_references_for_added_nodes_method(server):
async def test_get_event_from_type_node_BaseEvent(server):
"""
This should work for following BaseEvent tests to work
(maybe to write it a bit differentlly since they are not independent)
(maybe to write it a bit differentlly since they are not independent)
"""
ev = await asyncua.common.events.get_event_obj_from_type_node(
asyncua.Node(server.iserver.isession, ua.NodeId(ua.ObjectIds.BaseEventType))
@ -333,7 +333,7 @@ async def test_create_custom_data_type_object_id(server):
type = await server.create_custom_data_type(2, 'MyDataType', ua.ObjectIds.BaseDataType,
[('PropertyNum', ua.VariantType.Int32),
('PropertyString', ua.VariantType.String)])
await check_custom_type(type, ua.ObjectIds.BaseDataType, server)
await check_custom_type(type, ua.ObjectIds.BaseDataType, server, ua.NodeClass.DataType)
async def test_create_custom_event_type_object_id(server):
@ -596,19 +596,21 @@ def check_custom_event(ev, etype):
assert ev.Severity == 1
async def check_custom_type(type, base_type, server: Server):
async def check_custom_type(ntype, base_type, server: Server, node_class=None):
base = asyncua.Node(server.iserver.isession, ua.NodeId(base_type))
assert type in await base.get_children()
nodes = await type.get_referenced_nodes(refs=ua.ObjectIds.HasSubtype, direction=ua.BrowseDirection.Inverse,
assert ntype in await base.get_children()
nodes = await ntype.get_referenced_nodes(refs=ua.ObjectIds.HasSubtype, direction=ua.BrowseDirection.Inverse,
includesubtypes=True)
assert base == nodes[0]
properties = await type.get_properties()
if node_class:
assert node_class == await ntype.get_node_class()
properties = await ntype.get_properties()
assert properties is not None
assert len(properties) == 2
assert await type.get_child("2:PropertyNum") in properties
assert (await(await type.get_child("2:PropertyNum")).get_data_value()).Value.VariantType == ua.VariantType.Int32
assert await type.get_child("2:PropertyString") in properties
assert (await(await type.get_child("2:PropertyString")).get_data_value()).Value.VariantType == ua.VariantType.String
assert await ntype.get_child("2:PropertyNum") in properties
assert (await(await ntype.get_child("2:PropertyNum")).get_data_value()).Value.VariantType == ua.VariantType.Int32
assert await ntype.get_child("2:PropertyString") in properties
assert (await(await ntype.get_child("2:PropertyString")).get_data_value()).Value.VariantType == ua.VariantType.String
"""