opc ua
是一种应用层协议，基于tcp之上，其url通常为opc.tcp://127.0.0.1:4840/abc，在opc ua中常被称为endpoint
两种模式
opc ua支持c/s模式，同时也支持类似mqtt的发布订阅模式，通常各种设备作为opc ua的服务端提供各种服务。
信息模型
opc ua采用面向对象的设计思路， 使用了对象（objects）作为过程系统表示数据和活动的基础。对象包含了变量，事件和方法，它们通过引用（reference）来互相连接。
OPC UA 信息模型是节点的网络（Network of Node,），或者称为结构化图（graph），由节点(node)和引用(References)组成，这种结构图称之为OPC UA 的地址空间。这种图形结构可以描述各种各样的结构化信息(对象)。
注意⚠️：opc ua中所说的节点是在一个opc ua服务器中，不要理解为一个服务器对应一个node
节点
opc ua定义了8种类型的节点
对象(Object)
对象类型(Object Type)
变量(Variable)
变量类型(Variable Type)
方法(Method)
视图(View)
引用(Reference)
数据类型(Data Type)
每种节点都包含一些公共属性，如下：
| 属性 | 数据类型 | 说明 |
|---|---|---|
| NodeId | NodeId | 在OPC UA服务器内唯一确定的一个节点，并且在OPC UA服务器中定位该节点 |
| NodeClass | Int32 | 该节点的类型（上面列出的8种之一） |
| BrowseName | QualifiedName | 浏览OPC UA服务器事定义的节点。它是非本地化的 |
| DisplayName | LocalizedText | 包含节点的名字，用来在用户接口中显示名字，本地化 |
| Description | LocalizedText | 本地化的描述（可选） |
| WriteMask | Uint32 | 指示哪个节点属性是可写的，即可被OPC UA客户端修改（可选） |
| UserWriteMask | Uint32 | 指示哪个节点属性可以被当前连接到服务器上的用户修改（可选） |
除了数据类型节点之外，其他各个节点都有额外的专属属性
引用
引用描述了两个节点之间的关系，用来连接多个节点。OPC UA预定义了多种引用，常见的引用有：
hasTypeDefinition
描述对象、变量和类型之间的关系
ObjectNode的hasTypeDefinition引用，指向了一个ObjectTypeNode，表示该ObjectNode的类型;
VariableNode的hasTypeDefinition引用，指向一个VariableTypeNode，表示该 VariableNode的类型。
hasSubType
描述对象的挤成关系，当子类从父类继承后，子类拥有一个hasSubType引用指向父类。
hasComponents
描述一种组合关系
ObjectNode一般都由多个VariableNode组成，ObjectNode包含某个VariableNode时，ObjectNode拥有一个hasComponents引用，指向该VariableNode；
VariableNode也可以包含子VariableNode，此时也用hasComponents描述它们的关系。
Organizes
指明两个节点的层次结构，通过organizes可以把多个节点组织到同一个父节点下。
完整引用如下
服务
服务可以看成是OPC UA服务器提供的API集合，OPC UA与定义了37个标准服务，常用的服务有：
读写服务
可以获取和修改服务器指定节点指定属性的值
调用服务
执行服务器上指定节点的方法
订阅数据变化和订阅事件
可以监控服务器数据的变化
opc ua编程
Sdk
python（支持客户端和服务端）
https://github.com/FreeOpcUa/python-opcua
golang（支持客户端，服务端尚不完善）
https://github.com/gopcua/opcua
客户端
使用python（pyqt5）开发使用pip可以安装，跨平台
sudo pip3 install pyqt5 -i https://pypi.mirrors.ustc.edu.cn/simple/
sudo pip3 install numpy -i https://pypi.mirrors.ustc.edu.cn/simple/
sudo pip3 install pyqtgraph -i https://pypi.mirrors.ustc.edu.cn/simple/
sudo pip3 install cryptography -i https://pypi.mirrors.ustc.edu.cn/simple/
sudo pip3 install opcua-client -i https://pypi.mirrors.ustc.edu.cn/simple/
模拟设备
可利用sdk自己开发 见下面的python demo
golang Demo
读取服务器数据
package main
import (
"context"
"log"
"github.com/gopcua/opcua"
"github.com/gopcua/opcua/ua"
)
func main() {
endpoint := "opc.tcp://milo.digitalpetri.com:62541/milo"
nodeID := "ns=2;s=Dynamic/RandomFloat"
ctx := context.Background()
c := opcua.NewClient(endpoint, opcua.SecurityMode(ua.MessageSecurityModeNone))
if err := c.Connect(ctx); err != nil {
log.Fatal(err)
}
defer c.Close()
id, err := ua.ParseNodeID(nodeID)
if err != nil {
log.Fatalf("invalid node id: %v", err)
}
req := &ua.ReadRequest{
MaxAge: 2000,
NodesToRead: []*ua.ReadValueID{{NodeID: id}},
TimestampsToReturn: ua.TimestampsToReturnBoth,
}
resp, err := c.Read(req)
if err != nil {
log.Fatalf("Read failed: %s", err)
}
if resp.Results[0].Status != ua.StatusOK {
log.Fatalf("Status not OK: %v", resp.Results[0].Status)
}
log.Printf("%#v", resp.Results[0].Value.Value())
}
向服务器写数据
package main
import (
"context"
"github.com/gopcua/opcua"
"github.com/gopcua/opcua/ua"
"log"
)
func main() {
endpoint := "opc.tcp://milo.digitalpetri.com:62541/milo"
nodeID := "ns=2;s=Dynamic/RandomFloat"
ctx := context.Background()
c := opcua.NewClient(endpoint, opcua.SecurityMode(ua.MessageSecurityModeNone))
if err := c.Connect(ctx); err != nil {
log.Fatal(err)
}
defer c.Close()
id, err := ua.ParseNodeID(nodeID)
if err != nil {
log.Fatalf("invalid node id: %v", err)
}
v, err := ua.NewVariant(10.0)
if err != nil {
log.Fatalf("invalid value: %v", err)
}
req := &ua.WriteRequest{
NodesToWrite: []*ua.WriteValue{
{
NodeID: id,
AttributeID: ua.AttributeIDValue,
Value: &ua.DataValue{
EncodingMask: ua.DataValueValue,
Value: v,
},
},
},
}
resp, err := c.Write(req)
if err != nil {
log.Fatalf("Read failed: %s", err)
}
log.Printf("%v", resp.Results[0])
}
监听服务器数据变化
package main
import (
"context"
"github.com/gopcua/opcua/monitor"
"log"
"os"
"os/signal"
"sync"
"time"
"github.com/gopcua/opcua"
"github.com/gopcua/opcua/ua"
)
func cleanup(sub *monitor.Subscription, wg *sync.WaitGroup) {
log.Printf("stats: sub=%d delivered=%d dropped=%d", sub.SubscriptionID(), sub.Delivered(), sub.Dropped())
sub.Unsubscribe()
wg.Done()
}
func startCallbackSub(ctx context.Context, m *monitor.NodeMonitor, interval, lag time.Duration, wg *sync.WaitGroup, nodes ...string) {
sub, err := m.Subscribe(
ctx,
&opcua.SubscriptionParameters{
Interval: interval,
},
func(s *monitor.Subscription, msg *monitor.DataChangeMessage) {
if msg.Error != nil {
log.Printf("[callback] error=%s", msg.Error)
} else {
log.Printf("[callback] node=%s value=%v", msg.NodeID, msg.Value.Value())
}
time.Sleep(lag)
},
nodes...)
if err != nil {
log.Fatal(err)
}
defer cleanup(sub, wg)
<-ctx.Done()
}
func main() {
endpoint := "opc.tcp://milo.digitalpetri.com:62541/milo"
nodeID := "ns=2;s=Dynamic/RandomFloat"
signalCh := make(chan os.Signal, 1)
signal.Notify(signalCh, os.Interrupt)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go func() {
<-signalCh
println()
cancel()
}()
c := opcua.NewClient(endpoint, opcua.SecurityMode(ua.MessageSecurityModeNone))
if err := c.Connect(ctx); err != nil {
log.Fatal(err)
}
defer c.Close()
m, err := monitor.NewNodeMonitor(c)
if err != nil {
log.Fatal(err)
}
m.SetErrorHandler(func(_ *opcua.Client, sub *monitor.Subscription, err error) {
log.Printf("error: sub=%d err=%s", sub.SubscriptionID(), err.Error())
})
wg := &sync.WaitGroup{}
// start callback-based subscription
wg.Add(1)
go startCallbackSub(ctx, m, time.Second, 0, wg, nodeID)
<-ctx.Done()
wg.Wait()
}
python opcua server demo
#!/usr/bin/env python3
from threading import Thread
import random
import time
from opcua import ua, uamethod, Server
@uamethod
def set_temperature(parent, variant):
print(f"set_temperature {variant.Value}")
temperature_thread.temperature.set_value(variant.Value)
@uamethod
def set_onoff(parent, variant):
print(f"set_onoff {variant.Value}")
temperature_thread.temperature.set_value(variant.Value)
# 这个类用于后台定时随机修改值
class Temperature(Thread):
def __init__(self, temperature, onoff):
Thread.__init__(self)
self._stop = False
self.temperature = temperature
self.onoff = onoff
def stop(self):
self._stop = True
def run(self):
count = 1
while not self._stop:
value = random.randint(-20, 100)
self.temperature.set_value(value)
print(f"random set temperature {value}")
value = bool(random.randint(0, 1))
self.onoff.set_value(value)
print(f"random set onoff {value}")
led_event.event.Message = ua.LocalizedText("high_temperature %d" % count)
led_event.event.Severity = count
#led_event.event.temperature = random.randint(60, 100)
led_event.event.onoff = bool(random.randint(0, 1))
led_event.trigger()
count += 1
time.sleep(10)
if __name__ == "__main__":
# now setup our server
server = Server()
server.set_endpoint("opc.tcp://0.0.0.0:40840/tuyaopcua/server/")
server.set_server_name("TuyaOpcUa Driver Demo Device")
# set all possible endpoint policies for clients to connect through
server.set_security_policy([
ua.SecurityPolicyType.NoSecurity,
ua.SecurityPolicyType.Basic128Rsa15_SignAndEncrypt,
ua.SecurityPolicyType.Basic128Rsa15_Sign,
ua.SecurityPolicyType.Basic256_SignAndEncrypt,
ua.SecurityPolicyType.Basic256_Sign])
# setup our own namespace
uri = "http://tuya.com"
idx = server.register_namespace(uri)
# 添加一个 `空调` 对象
air_conditioner = server.nodes.objects.add_object(idx, "AirConditioner")
temperature = air_conditioner.add_variable(idx, "temperature", 20)
temperature.set_writable()
onoff = air_conditioner.add_variable(idx, "onoff", True)
onoff.set_writable()
air_conditioner.add_method(idx, "set_temperature", set_temperature, [ua.VariantType.UInt32])
air_conditioner.add_method(idx, "set_onoff", set_onoff, [ua.VariantType.Boolean])
# creating a default event object, the event object automatically will have members for all events properties
led_event_type = server.create_custom_event_type(idx,
'high_temperature',
ua.ObjectIds.BaseEventType,
[('temperature', ua.VariantType.UInt32), ('onoff', ua.VariantType.Boolean)])
led_event = server.get_event_generator(led_event_type, air_conditioner)
led_event.event.Severity = 300
# start opcua server
server.start()
print("Start opcua server...")
temperature_thread = Temperature(temperature, onoff)
temperature_thread.start()
try:
led_event.trigger(message="This is BaseEvent")
while True:
time.sleep(5)
finally:
print("Exit opcua server...")
temperature_thread.stop()
server.stop()
