dotnet-learn-vs/WebMVC/WebMVCApi/md/python 运行 opcua.md

icon	date	category	tag	headerDepth
edit	2022-12-26	python	opcua	5

python 运行 opcua

pip 安装 opcua

pip install opcua

opcua 客户端

from opcua import Client
if __name__ == "__main__":
 client = Client("opc.tcp://localhost:4840/freeopcua/server/")
 # client = Client("opc.tcp://admin@localhost:4840/freeopcua/server/") #connect using a user
 try:
     client.connect()
     # Client has a few methods to get proxy to UA nodes that should always be in address space such as Root or Objects
     root = client.get_root_node()
     print("Objects node is: ", root)
     # Node objects have methods to read and write node attributes as well as browse or populate address space
     print("Children of root are: ", root.get_children())
     # get a specific node knowing its node id
     #var = client.get_node(ua.NodeId(1002, 2))
     #var = client.get_node("ns=3;i=2002")
     #print(var)
     #var.get_data_value() # get value of node as a DataValue object
     #var.get_value() # get value of node as a python builtin
     #var.set_value(ua.Variant([23], ua.VariantType.Int64)) #set node value using explicit data type
     #var.set_value(3.9) # set node value using implicit data type
     # Now getting a variable node using its browse path
     myvar = root.get_child(["0:Objects", "2:MyObject", "2:MyVariable"])
     obj = root.get_child(["0:Objects", "2:MyObject"])
     print("myvar is: ", myvar)
     print("myobj is: ", obj)
     # Stacked myvar access
     # print("myvar is: ", root.get_children()[0].get_children()[1].get_variables()[0].get_value())
 finally:
    client.disconnect()

opcua 服务端

import sys
sys.path.insert(0, "..")
import time
from opcua import ua, Server
if __name__ == "__main__":
 # setup our server
 server = Server()
 server.set_endpoint("opc.tcp://0.0.0.0:4840/freeopcua/server/")
 # setup our own namespace, not really necessary but should as spec
 uri = "http://examples.freeopcua.github.io"
 idx = server.register_namespace(uri)
 # get Objects node, this is where we should put our nodes
 objects = server.get_objects_node()
 print(objects)
 # populating our address space
 myobj = objects.add_object(idx, "MyObject")
 myvar = myobj.add_variable(idx, "MyVariable", 6.7)
 myvar.set_writable() # Set MyVariable to be writable by clients
 # starting!
 server.start()
 try:
     count = 0
     while True:
         time.sleep(1)
         count += 0.1
         myvar.set_value(count)
 finally:
 #close connection, remove subcsriptions, etc
    server.stop()