OPCUA-Python实例

潘小伟

OPCUA-Python

1. pip install opcua

复制代码

Server

1. from threading import Thread
   
2. import copy
   
3. import logging
   
4. from datetime import datetime
   
5. import time
   
6. from math import sin
   
7. import sys
   
8. sys.path.insert(0, "..")
   
9. 
   
10. try:
    
11.     from IPython import embed
    
12. except ImportError:
    
13.     import code
    
14. 
    
15.     def embed():
    
16.         myvars = globals()
    
17.         myvars.update(locals())
    
18.         shell = code.InteractiveConsole(myvars)
    
19.         shell.interact()
    
20. 
    
21. 
    
22. from opcua import ua, uamethod, Server
    
23. 
    
24. 
    
25. class SubHandler(object):
    
26. 
    
27.     """
    
28.     Subscription Handler. To receive events from server for a subscription
    
29.     """
    
30. 
    
31.     def datachange_notification(self, node, val, data):
    
32.         print("Python: New data change event", node, val)
    
33. 
    
34.     def event_notification(self, event):
    
35.         print("Python: New event", event)
    
36. 
    
37. 
    
38. # method to be exposed through server
    
39. 
    
40. def func(parent, variant):
    
41.     ret = False
    
42.     if variant.Value % 2 == 0:
    
43.         ret = True
    
44.     return [ua.Variant(ret, ua.VariantType.Boolean)]
    
45. 
    
46. 
    
47. # method to be exposed through server
    
48. # uses a decorator to automatically convert to and from variants
    
49. 
    
50. @uamethod
    
51. def multiply(parent, x, y):
    
52.     print("multiply method call with parameters: ", x, y)
    
53.     return x * y
    
54. 
    
55. 
    
56. class VarUpdater(Thread):
    
57.     def __init__(self, var):
    
58.         Thread.__init__(self)
    
59.         self._stopev = False
    
60.         self.var = var
    
61. 
    
62.     def stop(self):
    
63.         self._stopev = True
    
64. 
    
65.     def run(self):
    
66.         while not self._stopev:
    
67.             v = sin(time.time() / 10)
    
68.             self.var.set_value(v)
    
69.             time.sleep(0.1)
    
70. 
    
71. 
    
72. 
    
73. if __name__ == "__main__":
    
74.     # optional: setup logging
    
75.     logging.basicConfig(level=logging.WARN)
    
76.     #logger = logging.getLogger("opcua.address_space")
    
77.     # logger.setLevel(logging.DEBUG)
    
78.     #logger = logging.getLogger("opcua.internal_server")
    
79.     # logger.setLevel(logging.DEBUG)
    
80.     #logger = logging.getLogger("opcua.binary_server_asyncio")
    
81.     # logger.setLevel(logging.DEBUG)
    
82.     #logger = logging.getLogger("opcua.uaprocessor")
    
83.     # logger.setLevel(logging.DEBUG)
    
84. 
    
85.     # now setup our server
    
86.     server = Server()
    
87.     #server.disable_clock()
    
88.     #server.set_endpoint("opc.tcp://localhost:4840/freeopcua/server/")
    
89.     server.set_endpoint("opc.tcp://0.0.0.0:4841/freeopcua/server/")
    
90.     server.set_server_name("FreeOpcUa Example Server")
    
91.     # set all possible endpoint policies for clients to connect through
    
92.     server.set_security_policy([
    
93.         ua.SecurityPolicyType.NoSecurity,
    
94.         # ua.SecurityPolicyType.Basic256Sha256_SignAndEncrypt,
    
95.         # ua.SecurityPolicyType.Basic256Sha256_Sign
    
96.     ])
    
97. 
    
98.     # setup our own namespace
    
99.     uri = "http://examples.freeopcua.github.io"
    
100.     idx = server.register_namespace(uri)
     
101. 
     
102.     # create a new node type we can instantiate in our address space
     
103.     dev = server.nodes.base_object_type.add_object_type(0, "MyDevice")
     
104.     dev.add_variable(0, "sensor1", 1.0).set_modelling_rule(True)
     
105.     dev.add_property(0, "device_id", "0340").set_modelling_rule(True)
     
106.     ctrl = dev.add_object(0, "controller")
     
107.     ctrl.set_modelling_rule(True)
     
108.     ctrl.add_property(0, "state", "Idle").set_modelling_rule(True)
     
109. 
     
110.     # populating our address space
     
111. 
     
112.     # First a folder to organise our nodes
     
113.     myfolder = server.nodes.objects.add_folder(idx, "myEmptyFolder")
     
114.     # instanciate one instance of our device
     
115.     mydevice = server.nodes.objects.add_object(idx, "Device0001", dev)
     
116.     mydevice_var = mydevice.get_child(["0:controller", "0:state"])  # get proxy to our device state variable
     
117.     # create directly some objects and variables
     
118.     myobj = server.nodes.objects.add_object(idx, "MyObject")
     
119.     myvar = myobj.add_variable(idx, "MyVariable", 6.7)
     
120.     mysin = myobj.add_variable(idx, "MySin", 0, ua.VariantType.Float)
     
121.     myvar.set_writable()    # Set MyVariable to be writable by clients
     
122.     mystringvar = myobj.add_variable(idx, "MyStringVariable", "Really nice string")
     
123.     mystringvar.set_writable()    # Set MyVariable to be writable by clients
     
124.     mydtvar = myobj.add_variable(idx, "MyDateTimeVar", datetime.utcnow())
     
125.     mydtvar.set_writable()    # Set MyVariable to be writable by clients
     
126.     myarrayvar = myobj.add_variable(idx, "myarrayvar", [6.7, 7.9])
     
127.     myarrayvar = myobj.add_variable(idx, "myStronglytTypedVariable", ua.Variant([], ua.VariantType.UInt32))
     
128.     myprop = myobj.add_property(idx, "myproperty", "I am a property")
     
129.     mymethod = myobj.add_method(idx, "mymethod", func, [ua.VariantType.Int64], [ua.VariantType.Boolean])
     
130.     multiply_node = myobj.add_method(idx, "multiply", multiply, [ua.VariantType.Int64, ua.VariantType.Int64], [ua.VariantType.Int64])
     
131. 
     
132.     # import some nodes from xml
     
133.     # server.import_xml("custom_nodes.xml")
     
134. 
     
135.     # creating a default event object
     
136.     # The event object automatically will have members for all events properties
     
137.     # you probably want to create a custom event type, see other examples
     
138.     myevgen = server.get_event_generator()
     
139.     myevgen.event.Severity = 300
     
140. 
     
141.     # starting!
     
142.     server.start()
     
143.     print("Available loggers are: ", logging.Logger.manager.loggerDict.keys())
     
144.     vup = VarUpdater(mysin)  # just  a stupide class update a variable
     
145.     vup.start()
     
146.     try:
     
147.         # enable following if you want to subscribe to nodes on server side
     
148.         #handler = SubHandler()
     
149.         #sub = server.create_subscription(500, handler)
     
150.         #handle = sub.subscribe_data_change(myvar)
     
151.         # trigger event, all subscribed clients wil receive it
     
152.         var = myarrayvar.get_value()  # return a ref to value in db server side! not a copy!
     
153.         var = copy.copy(var)  # WARNING: we need to copy before writting again otherwise no data change event will be generated
     
154.         var.append(9.3)
     
155.         myarrayvar.set_value(var)
     
156.         mydevice_var.set_value("Running")
     
157.         myevgen.trigger(message="This is BaseEvent")
     
158.         server.set_attribute_value(myvar.nodeid, ua.DataValue(9.9))  # Server side write method which is a but faster than using set_value
     
159. 
     
160.         embed()
     
161.     finally:
     
162.         vup.stop()
     
163.         server.stop()

复制代码

这个服务程序演示了opcua服务端，几乎所有的功能，其中Event部分没有不断发送，所以仅供参考。

下图展示了server端的对象结构

客户端

1. from IPython import embed
   
2. from opcua import Client
   
3. 
   
4. class SubHandler(object):
   
5.     def event_notification(self, event):
   
6.         print("Event:", event.EventId, event.Time, event.proper_random, event.Message.Text)
   
7. 
   
8. def main_c():
   
9.     url = "opc.tcp://127.0.0.1:4841/freeopcua/server/"
   
10.     c = Client(url)
    
11.     try:
    
12.         c.connect()
    
13.         root = c.get_root_node()
    
14.         embed()
    
15.     except Exception as e:
    
16.         print("Client Exception:", e)
    
17.     finally:
    
18.         c.disconnect()
    
19.         
    
20. if __name__ == "__main__":
    
21.     main_c()

复制代码

客户端遍历流程

• 1.获取
  1. root

  复制代码
  下的
  1. Objects

  复制代码
  节点的所有子节点，一般类型都为
  1. Object

  复制代码
  ,取一代表为A
  
• 2.获取A的
  1. NodeClass

  复制代码
  (一般为Object)和
  1. TypeDefinition

  复制代码
  ,其中NodeId需要与路径
  1. ["0:Types","0:ObjectTypes","0:BaseObjectType"]

  复制代码
  下的类型对照(自定义类型也在内)。
  
• 3.继续以A为根，遍历子节点，如果时Object继续2，如果是
  1. Variable

  复制代码
  ,NodeId对照
  1. ["0:Types","0:VariableTypes","0:BaseVariableType"]

  复制代码
  ,可以判断是变量(variable,63)还是属性(property,65)。
  
• 4.对于
  1. Method

  复制代码
  使用
  1. a_root.call_method(a, arg1)

  复制代码
  调用。对于
  1. Variable

  复制代码
  使用
  1. get_value/get_data_value()

  复制代码
  获取存储的值。如果是
  1. Object

  复制代码
  继续2.
  
• 5.对于
  1. Variable

  复制代码
  使用
  1. access_level

  复制代码
  获取是否有写权限，如果有
  1. set_value

  复制代码
  可以设置值。
  

一个比较完美的遍历客户端

1. from opcua import Client, ua
   
2. 
   
3. 
   
4. def brower_child(root):
   
5.     """
   
6.     递归调用遍历，格式化不好做，有深度问题
   
7.     """
   
8.     name = root.get_node_class().name
   
9.     # print(name)
   
10.     if name == "Object":
    
11.         brower_obj(root)
    
12.         for c in root.get_children():
    
13.             print("  ", end='')
    
14.             brower_child(c)
    
15.     elif name == 'Variable':
    
16.         brower_var(root)
    
17.     else:
    
18.         brower_method(root)
    
19. 
    
20. 
    
21. class CurState():
    
22.     def __init__(self, parent=None, p=None, d=0):
    
23.         self.parent = parent  # unused
    
24.         self.p = p
    
25.         self.d = d
    
26. 
    
27. 
    
28. def brower_child2(root, max_d=-1, ignore=[]):
    
29.     """
    
30.     栈+循环遍历，非常好用
    
31.     """
    
32.     stack = [CurState(None, root, 0)]
    
33.     while len(stack):
    
34.         cur = stack.pop()
    
35.         name = cur.p.get_node_class().name
    
36. 
    
37.         print(''.join(['  ' for i in range(cur.d)]), end="")
    
38. 
    
39.         if cur.p.get_browse_name().Name in ignore:
    
40.             continue
    
41. 
    
42.         if name == "Object":
    
43.             brower_obj(cur.p)
    
44.             if max_d > 0 and cur.d >= max_d:
    
45.                 continue
    
46.             for c in cur.p.get_children():
    
47.                 stack.append(CurState(cur.p, c, cur.d+1))
    
48. 
    
49.         elif name == 'Variable':
    
50.             brower_var(cur.p)
    
51.         else:
    
52.             brower_method(cur.p)
    
53. 
    
54. 
    
55. def brower_obj(v):
    
56.     # print(v.get_browse_name())
    
57.     rw = 'R '
    
58.     bname = v.get_browse_name()
    
59.     print("*%2d:%-30s (%-2s, %-23s)" %
    
60.           (bname.NamespaceIndex, bname.Name, rw, "Object"))
    
61. 
    
62. 
    
63. def brower_var(v):
    
64.     # print(v.get_browse_name())
    
65.     rw = 'R '
    
66.     if ua.AccessLevel.CurrentWrite in v.get_access_level():
    
67.         rw = "RW"
    
68.     bname = v.get_browse_name()
    
69.     tv = v.get_data_value().Value
    
70.     v_show = tv.Value
    
71.     if len(str(v_show)) > 1024:
    
72.         v_show = str(v_show[:56]) + "..."
    
73.     print("-%2d:%-30s (%-2s, %-23s) =>" %
    
74.           (bname.NamespaceIndex, bname.Name, rw, tv.VariantType), v_show)
    
75. 
    
76. 
    
77. def brower_method(v):
    
78.     # print(v.get_description())
    
79.     rw = 'C '
    
80.     bname = v.get_browse_name()
    
81.     # args = []
    
82.     # for a in v.get_properties():
    
83.     #     dt = a.get_data_type().NodeIdType.name
    
84.     #     args.append(dt)
    
85.     print("@%2d:%-30s (%-2s, %-23s)" %
    
86.           (bname.NamespaceIndex, bname.Name, rw, "Method"))
    
87. 
    
88. 
    
89. def main_c():
    
90.     url = "opc.tcp://127.0.0.1:4841/freeopcua/server/"
    
91.     c = Client(url)
    
92.     try:
    
93.         c.connect()
    
94.         root = c.get_root_node()
    
95.         print("\r\nBrower:")
    
96.         brower_child2(root.get_child(["0:Objects"]), -1, ["Server"])
    
97.     except Exception as e:
    
98.         print("Client Exception:", e)
    
99.     finally:
    
100.         c.disconnect()
     
101. 
     
102. 
     
103. if __name__ == "__main__":
     
104.     main_c()

复制代码

总结

以上为个人经验，希望能给大家一个参考，也希望大家多多支持脚本之家。

来源:https://www.jb51.net/python/31619408f.htm
免责声明：由于采集信息均来自互联网，如果侵犯了您的权益，请联系我们【E-Mail:cb@itdo.tech】 我们会及时删除侵权内容，谢谢合作！