Python訪問OPCUA服務器,訂閱一個變量標簽方式
更新時間：2024年02月23日 10:32:16 作者：weloveut
這篇文章主要介紹了Python訪問OPCUA服務器,訂閱一個變量標簽方式,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教
Python訪問OPCUA服務器,訂閱一個變量標簽
from opcua import Client
from opcua import ua
import time
#By Weloveut:Python訪問OPCUA服務器，展示get變量的方法和注冊逢變觸發(fā)的回調(diào)函數(shù)
#自定義回調(diào)函數(shù)
class SubHandler(object):
def data_change(self, handle, node, val, attr):
print("Python: New data change event", handle, node, val, attr)
#創(chuàng)建UA的客戶端，包含IP地址和端口號
client = Client("opc.tcp://172.32.1.82:4840/")
#建立到服務器連接
client.connect()
#獲取object對象
objects = client.get_objects_node()
#獲取根對象
root = client.get_root_node()
#依據(jù)變量地址創(chuàng)建變量對象
myvar = client.get_node('ns=3;s="clocl0.5hz"."tags"[0]')
#獲取變量當前值
valuetmp=myvar.get_value()
print(valuetmp)
#注冊到服務器的變量訂閱，變量逢變觸發(fā)
handler = SubHandler()
sub = client.create_subscription(500, handler)
sub.subscribe_data_change(myvar)
time.sleep(100000)
client.disconnect()
opcua-python學習踩坑記錄
OPCUA通信協(xié)議介紹
OPC UA（OPC 統(tǒng)一架構）是一種用于工業(yè)控制系統(tǒng)的通信協(xié)議。它是一種開放的、平臺無關的協(xié)議，可用于連接各種工業(yè)設備和軟件應用程序，使它們能夠互相通信并共享數(shù)據(jù)。
OPC UA 的主要優(yōu)點在于，它能夠提供一種統(tǒng)一的方式來連接不同廠商的設備和應用程序。這意味著，使用 OPC UA 的軟件應用程序可以與使用其他協(xié)議的設備和應用程序進行通信，而無需考慮底層的通信協(xié)議。
其實這類似于物聯(lián)網(wǎng)中的mqtt協(xié)議，只不過mqtt較為輕量級，opcua則是用于工業(yè)相關的控制系統(tǒng)
OPCUA的兩種工作方式
- 客戶端/服務器模式：在這種模式下，OPC UA 客戶端向 OPC UA 服務器請求數(shù)據(jù)，服務器接收請求并向客戶端發(fā)送數(shù)據(jù)。
- 發(fā)布/訂閱模式：在這種模式下，OPC UA 服務器將數(shù)據(jù)“發(fā)布”到網(wǎng)絡上，而客戶端可以“訂閱”特定的數(shù)據(jù)。當服務器更新數(shù)據(jù)時，客戶端會收到更新的數(shù)據(jù)。
這兩種模式可以根據(jù)應用場景的需要進行選擇。例如，客戶端/服務器模式適用于客戶端需要向服務器請求單個數(shù)據(jù)項的情況，而發(fā)布/訂閱模式則適用于客戶端需要定期接收大量數(shù)據(jù)的情況。
個人覺得客戶端/服務器模式類似于TCP通訊協(xié)議，是點對點的，對實時性要求比較高，而發(fā)布訂閱模式類似于UDP通訊協(xié)議，對實時性要求并沒有那么高，但接收數(shù)據(jù)較多
其中發(fā)布訂閱模式對小的嵌入式設備（stm32系列）比較友好，可以融合物聯(lián)網(wǎng)的mqtt協(xié)議，真正將opcua納入網(wǎng)絡，而不只是自己的工業(yè)網(wǎng)絡
OPCUA-python是什么？
顧名思義，那就是利用python實現(xiàn)這種通信協(xié)議的庫唄，這個庫目前已經(jīng)停止更新，開發(fā)者們開發(fā)了一個新的庫，名為opcua-asyncio，可通過如下鏈接訪問：
https://github.com/FreeOpcUa/opcua-asyncio
不過做一些比較基本的功能，只用OPCUA-python這個庫還是夠用的
服務器常用函數(shù)
endpoint = "opc.tcp://{}:{}".format(url, port)
myserver.set_endpoint(endpoint)設置endpoint(其實endpoint就像一個服務器鏈接）
myserver = opcua.Server()
開啟opcua服務器
objects = myserver.get_objects_node()
設置服務器節(jié)點（使客戶端得以訪問）
addspace = myserver.register_namespace(name)
設置服務器命名空間
root = myserver.get_root_node()
獲取服務器根節(jié)點
objects.add_method(1, "Conveyor", Start_Conveyor_prog)
為節(jié)點（objects) 添加方法Start_Conveyor_prog
客戶端常用函數(shù)
myclient = Client(endpoint)
在客戶端引用endpoint
myclient.get_node("ns=2;i=3")在客戶端里使用ns=2;i=3這個節(jié)點
Start_Lathe_Prog1 = method[4]
調(diào)用服務器里設置的方法
Workpiece = objects_node.call_method(Start_Conveyor_prog,[WorkpieceID.get_value()])
call_method函數(shù)，第一個參數(shù)是節(jié)點名稱，第二個參數(shù)是值，是這個節(jié)點里的值，屬于數(shù)組，或者傳遞字典也是可以的
return_value_kuka_prog1 = objects_node.call_method(Start_Kuka_Prog1)
返回某個方法
示例程序：溫度實時檢測，每隔五秒傳到客戶端
服務器程序：
import opcua
from time import sleep
from random import randint
from datetime import datetime
# Assign endpoint URL
url = "localhost"
port = 5001
endpoint = "opc.tcp://{}:{}".format(url, port)
# create instance/object of type opcua.Server
myserver = opcua.Server()
# Assign endpoint url on the OPC UA server address space
myserver.set_endpoint(endpoint)
# Create a name for OPC UA namespace
name = "Temperature Sensor"
# Register the name on OPC UA namespace
addspace = myserver.register_namespace(name)
# Get reference to the Objects node of the OPC UA server
objects = myserver.get_objects_node()
# Create objects on the object node
param = objects.add_object(addspace, "parameters")
# Create variables
Sensor_name = param.add_variable(addspace, "Sensor Name", "Temperature_Sensor_SF12")
Temperature = param.add_variable(addspace, "Temperature Value", 'NA')
# Set variable as writable
Sensor_name.set_writable()
Temperature.set_writable()
#########################################################################################
# Present the data structure of the OPC UA server
root = myserver.get_root_node()
print("Root Node Id: %s" % root)
print("Children of root are: %s" % root.get_children())
print()
my_objects = myserver.get_objects_node()
print("Defined object Node Id: %s" % my_objects)
print("Children of the defined object are: %s" % my_objects.get_children())
print()
sensor_name_node = my_objects.get_children()[1].get_children()[0]
print("Sensor name node Id: %s" % sensor_name_node)
print("Sensor name node browse name: %s" % sensor_name_node.get_browse_name())
print("Sensor name default value: %s" % sensor_name_node.get_value())
print()
temperature_node = my_objects.get_children()[1].get_children()[1]
print("Temperature node Id: %s" % temperature_node)
print("Temperature node browse name: %s" % temperature_node.get_browse_name())
print("Temperature default value: %s" % temperature_node.get_value())
print()
########################################################################################
# starting! The server will continue running
myserver.start()
current_time = str(datetime.now().time())[:-7]
print("{} - Server is initialising...".format(current_time))
while True:
sleep(5)
current_time = str(datetime.now().time())[:-7]
current_temp = randint(10, 25)
Temperature.set_value(current_temp) # publish temperature value
print("{} - Current temperature: {} Celsius".format(current_time, current_temp))客戶端程序：
from opcua import Client
from datetime import datetime
# Assign endpoint URL
url = "localhost"
port = 5001
endpoint = "opc.tcp://{}:{}".format(url, port)
# Assign endpoint url on the OPC UA client address space
myclient = Client(endpoint)
# Connect to server
myclient.connect()
# Assign nodes
Temperature_node = myclient.get_node("ns=2;i=3")
# Subhandler Class from OPC UA
class SubHandler(object):
"""
Subscription Handler. To receive events from server for a subscription
"""
def datachange_notification(self, node, val, data):
"""
called for every datachange notification from server
"""
global DataChange # Global variable
current_time = str(datetime.now().time())[:-7]
DataChange = val # Assigning value to global variable
print("{} - Received temperature: {} Celsius".format(current_time, val))
# Initailise variable
nodes = myclient.get_objects_node()
DataChange = "null"
# Create Sub handler
handler = SubHandler()
# create subscription by passing period in milliseconds and handler
subscribe = myclient.create_subscription(0, handler)
# pass variable node to subscribe data change method
handler = subscribe.subscribe_data_change(Temperature_node)程序運行效果：服務器
程序運行效果：客戶端
總結
以上為個人經(jīng)驗，希望能給大家一個參考，也希望大家多多支持腳本之家。
相關文章
本文將以實戰(zhàn)的形式基于scipy模塊使用Python實現(xiàn)簡單濾波處理。這篇文章主要介紹了Python基于scipy實現(xiàn)信號濾波功能,需要的朋友可以參考下2019-05-05
這篇文章主要介紹了使用Pytorch搭建模型的步驟，幫助大家更好的理解和使用Pytorch，進行機器學習，感興趣的朋友可以了解下2020-11-11
Opencv中的cv2.calcHist()函數(shù)的作用及返回值說明
這篇文章主要介紹了Opencv中的cv2.calcHist()函數(shù)的作用及返回值說明，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2022-11-11
pdf2docx作為第三方包,提供了非常優(yōu)秀的功能,僅僅幾行代碼就可以完成PDF轉換為DOCX的工作,所以本文就來利用pdf2docx實現(xiàn)PDF轉DOCX文檔功能吧2023-12-12
這篇文章主要介紹了python實現(xiàn)的socket端口重定向示例,實現(xiàn)功能是在端口12345開啟文件共享，此時我也想通過8000端口同時訪問此共享2014-02-02
這篇文章主要介紹了Django權限控制的使用，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2021-01-01
matplotlib源碼解析標題實現(xiàn)(窗口標題，標題，子圖標題不同之間的差異)
這篇文章主要介紹了matplotlib源碼解析標題實現(xiàn)(窗口標題，標題，子圖標題不同之間的差異)，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2021-02-02
這篇文章主要介紹了python刪除某個字符,現(xiàn)在發(fā)布的是一個比較簡單易行的方法，就是遇到該字符便跳過去，不對其進行操作，完美呈現(xiàn)出刪除該字符的功能,需要的朋友可以參考下2018-03-03
這篇文章主要介紹了Django Auth用戶認證組件實現(xiàn)代碼,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2020-10-10
MarkupPy是Python模塊用于生成HTML和XML格式的字符串，它的主要作用是提供了一種比原生HTML/XML更加易讀和易寫的編寫方式，通過Python代碼來生成HTML或XML代碼，這篇文章主要介紹了Python模塊MarkupPy&自定義html報告的相關知識,需要的朋友可以參考下2023-07-07