写在最前面

Python、工业4.0、大数据等如此火热，难得闲下来数小时，用Python实现下OPC DA Client玩儿玩儿吧。

工具

1.kepserver 作为OPC DA 服务端

2.Python 3.7 

3.OPC DA COM组件即opcdaauto.dll

开始

1.创建keyserver测试工程，这里就不多讲了直接上图。测试而已，所以只加了一个变量。

 2.如果机器没注册opcdaauto.dll Com组件，注册COM。方法为：

1）以管理员方式打开命令行

2）输入regsvr32 + 路径+opcdaauto.dll

3.上面工作准备完成后开始测试喽，很简单的。

1）首先引入必要包

2）拿到OPC DA组件

3）获取OPCServer实例

opcserver = dll.OPCServer()

4）测试一下枚举检测到的OPC DA Server

for svr in opcserver.GetOPCServers():
    print (svr)

因为我电脑只装了一个kepserver 6，所以结果如下图

5）连接

opcserver.Connect('KEPware.KEPServerEx.V6')

6）拿到组实例

groups=opcserver.OPCGroups
groups.DefaultGroupIsActive = True
groups.DefaultGroupDeadband = 0
groups.DefaultGroupUpdateRate = 200

7）添加组，并订阅回调事件

group = DispatchWithEvents(groups.Add('Group1'),GroupEvent)
group.IsActive = True
group.IsSubscribed = True
group.UpdateRate = 100

8）添加需要操作的条目

item1 = "T1.T1.Tag1"
item=items.AddItem(item1,1)
item.IsActive = True
myItems[1]=item

9）测试读写

a=0
b=0
c=0
d=0
item.Read(win32com.client.constants.OPCDevice,b,c,d)

# write a new value
item.Write(100)

item.Write(111)

item.Write(120)

#read item value
item.Read(win32com.client.constants.OPCDevice,b,c,d)

10）搞一个消息循环，不然订阅无法正常触发。这个问题卡了我挺长时间

root = Tk()

#root.withdraw()   # 隐藏窗口
root.mainloop()  # 消息循环

11）断开连接

opcserver.Disconnect()

12）别忘记回调函数注册

                def OnDataChange(unknow,TransactionID,NumItems,ClientHandles,ItemValues,Qualities,TimeStamps):               
                    i=0
                    print("callback")
                    while(i<NumItems):                       
                        handle = ClientHandles[i]
                        value = ItemValues[i]
                        quality = Qualities[i]
                        time = TimeStamps[i]
                        print("item:%s , value:%s , quality:%s , time:%s ",myItems[handle],value,quality,time)
                        i=i+1

结果：

经测试，读、写、订阅均没问题。

简单封装下就可以项目中使用了呦。

虽然现在UA是趋势，但是DA也得会用哒。

看看测试截图吧