opcua/InteractionServerClient

# load cvb.opcua module
import cvb
import cvb.opcua
import time
import threading
 
test_node_name = "TestNode"
port = 4840
stop = False
ready = False
 
 
def read_callback(data):
    print("callback: reading")
 
 
def write_callback(data):
    print("callback: writing")
 
# ------------------------------------------------------------------------------
# Server
# ------------------------------------------------------------------------------
 
 
def start_server():
    global stop
    global ready
    global port
    global test_node_name
 
    server = cvb.opcua.Server.create(port)
 
    # Server add namespace
    ns_index = server.add_namespace("StemmerNamespace")
 
    # Server add folder
    folder_type = cvb.opcua.NodeID.create_namespace0(
        cvb.opcua.Namespace0NodeID.FOLDERTYPE)
    loc = cvb.opcua.NodeID.create_namespace0(
        cvb.opcua.Namespace0NodeID.OBJECTSFOLDER)
    folder = cvb.opcua.ObjectNode.create_from_type(
        ns_index, "StemmerFolder", loc, folder_type)
    server.add_node(folder)
 
    # Server Add Node
    var_node = cvb.opcua.FloatNode.create(
        ns_index, test_node_name, folder.node_id(), 1.111)
    server.add_node(var_node)
 
    # register callbacks
    readCookie = var_node.register_read_callback(read_callback)
    writeCookie = var_node.register_write_callback(write_callback)
 
    server.start()
    ready = True
 
    while not stop:
        time.sleep(1)
 
    var_node.deregister_callback(readCookie)
    var_node.deregister_callback(writeCookie)
    server.stop()
 
# ------------------------------------------------------------------------------
# Client Helper
# ------------------------------------------------------------------------------
 
 
def make_browse_filter():
    bd = cvb.opcua.ReferenceDirection.Forward  # how to iterate through the server
    relationship = cvb.opcua.ReferenceType.Hierarchical  # relationship to explore
    max_ref = 5000  # max number of references to follow
    # allow subtype of search type
    allow_subtype = cvb.opcua.BrowseSubType.SubTypesIncluded
    # masked based filter for node classes based on bitwise AND of cvb.opcua.NodeClass
    node_class_mask = 0
 
    # combine them in an object
    browse_filter = cvb.opcua.BrowseFilter.create(
        bd, relationship, max_ref, allow_subtype, node_class_mask)
 
    return browse_filter
 
# ------------------------------------------------------------------------------
# Client
# ------------------------------------------------------------------------------
 
 
def start_client():
    global stop
    global ready
    global port
    global test_node_name
 
    while not ready:
        time.sleep(1)
 
    # create client
    client = cvb.opcua.Client.create("opc.tcp://127.0.0.1:"+str(port))
 
    # get node id of object folder
    root_folder = client.root_node()
 
    # browse
    result = root_folder.browse(make_browse_filter())
 
    for nid in result:
         
        node_extracted = client.node(nid)
        # print display name
        print("found: " + node_extracted.display_name)
 
        # our custom node
        if(node_extracted.display_name == test_node_name):
            # change values (trigger callbacks)
            print("value access: "+str(node_extracted.value()))
 
            node_extracted.set_value(2.2)
            print("value access: "+str(node_extracted.value()))
    stop = True
 
 
# start server in independent thread
server_thread = threading.Thread(target=start_server)
server_thread.start()
 
# start client in current thread
start_client()
 
# collect server thread
server_thread.join()
 
print("\nfinished\n")