Commit faa4f3b9 authored by olivier R-D's avatar olivier R-D

added shortcuts to common nodes

parent 785f1852
...@@ -13,6 +13,7 @@ from opcua.common.manage_nodes import delete_nodes ...@@ -13,6 +13,7 @@ from opcua.common.manage_nodes import delete_nodes
from opcua.common.subscription import Subscription from opcua.common.subscription import Subscription
from opcua.common import utils from opcua.common import utils
from opcua.crypto import security_policies from opcua.crypto import security_policies
from opcua.common.shortcuts import Shortcuts
use_crypto = True use_crypto = True
try: try:
from opcua.crypto import uacrypto from opcua.crypto import uacrypto
...@@ -97,6 +98,7 @@ class Client(object): ...@@ -97,6 +98,7 @@ class Client(object):
self.user_private_key = None self.user_private_key = None
self._session_counter = 1 self._session_counter = 1
self.keepalive = None self.keepalive = None
self.nodes = Shortcuts(self.uaclient)
def __enter__(self): def __enter__(self):
self.connect() self.connect()
......
from opcua.ua import ObjectIds
from opcua import Node
class Shortcuts(object):
"""
This object contains Node objects to some commonly used nodes
"""
def __init__(self, server):
self.root = Node(server, ObjectIds.RootFolder)
self.objects = Node(server, ObjectIds.ObjectsFolder)
self.server = Node(server, ObjectIds.Server)
self.types = Node(server, ObjectIds.TypesFolder)
self.base_object_type = Node(server, ObjectIds.BaseObjectType)
self.base_data_type = Node(server, ObjectIds.BaseDataType)
...@@ -21,6 +21,7 @@ from opcua.common.manage_nodes import delete_nodes ...@@ -21,6 +21,7 @@ from opcua.common.manage_nodes import delete_nodes
from opcua.client.client import Client from opcua.client.client import Client
from opcua.crypto import security_policies from opcua.crypto import security_policies
from opcua.common.event_objects import BaseEvent from opcua.common.event_objects import BaseEvent
from opcua.common.shortcuts import Shortcuts
use_crypto = True use_crypto = True
try: try:
from opcua.crypto import uacrypto from opcua.crypto import uacrypto
...@@ -66,6 +67,8 @@ class Server(object): ...@@ -66,6 +67,8 @@ class Server(object):
:vartype default_timeout: InternalServer :vartype default_timeout: InternalServer
:ivar bserver: binary protocol server :ivar bserver: binary protocol server
:vartype bserver: BinaryServer :vartype bserver: BinaryServer
:ivar nodes: shortcuts to common nodes
:vartype nodes: Shortcuts
""" """
...@@ -84,6 +87,7 @@ class Server(object): ...@@ -84,6 +87,7 @@ class Server(object):
self.certificate = None self.certificate = None
self.private_key = None self.private_key = None
self._policies = [] self._policies = []
self.nodes = Shortcuts(self.iserver.isession)
# setup some expected values # setup some expected values
self.register_namespace(self.application_uri) self.register_namespace(self.application_uri)
......
...@@ -545,16 +545,13 @@ class CommonTests(object): ...@@ -545,16 +545,13 @@ class CommonTests(object):
def test_instantiate_1(self): def test_instantiate_1(self):
base_type = self.opc.get_root_node().get_child(["0:Types", "0:ObjectTypes", "0:BaseObjectType"]) dev_t = self.opc.nodes.base_data_type.add_object_type(0, "MyDevice")
dev_t = base_type.add_object_type(0, "MyDevice")
v_t = dev_t.add_variable(0, "sensor", 1.0) v_t = dev_t.add_variable(0, "sensor", 1.0)
p_t = dev_t.add_property(0, "sensor_id", "0340") p_t = dev_t.add_property(0, "sensor_id", "0340")
ctrl_t = dev_t.add_object(0, "controller") ctrl_t = dev_t.add_object(0, "controller")
prop_t = ctrl_t.add_property(0, "state", "Running") prop_t = ctrl_t.add_property(0, "state", "Running")
objects = self.opc.get_objects_node()
mydevice = instantiate(objects, dev_t, bname="2:Device0001") mydevice = instantiate(self.opc.nodes.objects, dev_t, bname="2:Device0001")
self.assertEqual(mydevice.get_type_definition(), dev_t.nodeid) self.assertEqual(mydevice.get_type_definition(), dev_t.nodeid)
obj = mydevice.get_child(["0:controller"]) obj = mydevice.get_child(["0:controller"])
...@@ -565,12 +562,11 @@ class CommonTests(object): ...@@ -565,12 +562,11 @@ class CommonTests(object):
def test_variable_with_datatype(self): def test_variable_with_datatype(self):
o = self.opc.get_objects_node() v1 = self.opc.nodes.objects.add_variable(3, 'VariableEnumType1', ua.ApplicationType.ClientAndServer, datatype=ua.NodeId(ua.ObjectIds.ApplicationType))
v1 = o.add_variable(3, 'VariableEnumType1', ua.ApplicationType.ClientAndServer, datatype=ua.NodeId(ua.ObjectIds.ApplicationType))
tp1 = v1.get_data_type() tp1 = v1.get_data_type()
self.assertEqual(ua.NodeId(ua.ObjectIds.ApplicationType), tp1) self.assertEqual(ua.NodeId(ua.ObjectIds.ApplicationType), tp1)
v2 = o.add_variable(3, 'VariableEnumType2', ua.ApplicationType.ClientAndServer, datatype=ua.NodeId(ua.ObjectIds.ApplicationType) ) v2 = self.opc.nodes.objects.add_variable(3, 'VariableEnumType2', ua.ApplicationType.ClientAndServer, datatype=ua.NodeId(ua.ObjectIds.ApplicationType) )
tp2 = v2.get_data_type() tp2 = v2.get_data_type()
self.assertEqual( ua.NodeId(ua.ObjectIds.ApplicationType), tp2) self.assertEqual( ua.NodeId(ua.ObjectIds.ApplicationType), tp2)
......
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