[oBIX包使用教程] 使用 Python 通過 oBIX 協議訪問 Niagara 資料
oBIX 全稱是 Open Building Information Exchange,它是基於 RESTful Web Service 的介面的標準,用於構建控制系統。oBIX是在專為樓宇自動化設計的框架內,使用XML和URI在裝置網路上讀寫資料的。
因專案需要使用 Python 對 Niagara 軟體中的資料進行讀寫和控制,所以寫了一個該協議的Python版本包,釋出在這裡:https://pypi.org/project/oBIX/
使用 pip 安裝使用即可:
pip install oBIX
本文主要介紹使用 Python 通過 oBIX 協議對 Niagara 軟體中的點進行讀、寫操作。
一、準備工作
1. 在 Niagara 軟體中配置好 oBIX 協議,確保已經可以正常訪問;
(1)Palette 搜 oBIX, 新增一個 ObixNetwork 到 Drivers中
(2)Palette 搜 baja, 將 AuthenticationSchemes/WebServicesSchemes/的 HTTPBasicScheme 拖拽到 Services/AuthenticationService/Authentication Schemes/
(3)UserServices 右鍵 View, AX User Manager下新建一個使用者,配置如下:
* 使用者名稱:oBIX * 密碼:oBIX.12345 * Authentication Schemes Name 選:HTTPBasicScheme * Admin 許可權2. Niagara 中新建一個數值型別的可讀寫的點,命名為:temp1,完整路徑是:/config/AHU/temp1/,後面以此為例進行訪問
3. 安裝python的oBIX包:pip install oBIX
二、快速開始
from oBIX.common import Point, DataType from oBIX import Client if __name__ == '__main__': # ip, userName, password # 可選項: # port: 埠號,如:8080 # https: 是否使用 https,預設:True client = Client("127.0.0.1", "oBIX", "oBIX.12345") # 點的路徑 point_path = "/config/AHU/temp1/" # 讀取一個點的值 point_value = client.read_point_value(point_path) print("point value is {0}".format(point_value))
三、基本例項
3.1 讀取點
# 點的路徑 point_path = "/config/AHU/temp1/" # 讀取一個點的值 point_value = client.read_point_value(point_path) print("point value is {0}".format(point_value)) # 讀取一個點例項 # 然後就能獲取到這個點所包含的常用屬性 # 例如:name, val, status, display, href, in1, in2 ... in16, fallback, out point_obj = client.read_point(point_path) print("name is {0}".format(point_obj.name)) print("fallback is {0}".format(point_obj.fallback)) print("in10 is {0}".format(point_obj.in10)) # 也可以使用下面程式碼直接獲取 point_in10_value = client.read_point_slot(point_path, "in10") print("in10 is {0}".format(point_in10_value))
3.2 寫入點
# 點的路徑 point_path = "/config/AHU/temp1/" # set 一個點的值 client.write_point(point_path, 15.2, DataType.real) # set point auto client.set_point_auto(point_path, DataType.real) # override a point client.override_point(point_path, 14, DataType.real) # emergency override a point client.emergency_override_point(point_path, 15, DataType.real) # set a point emergency auto client.set_point_emergency_auto(point_path, DataType.real)
四、高階應用
4.1 讀取歷史資料
# 起始時間 start_time = datetime.now(tz=timezone(timedelta(hours=8))) - timedelta(minutes=10) # 結束時間 end_time = datetime.now(tz=timezone(timedelta(hours=8))) # 讀取該斷時間內的歷史資料 history = client.read_history("Station01", "OutDoorTemp", start_time, end_time) # 取起始時間往後指定個數的歷史資料 limit_num = 1 history = client.read_history("Station01", "OutDoorTemp", start_time=start_time, limit=limit_num)
4.2 讀取報警資料
# 起始時間 start_time = datetime.now(tz=timezone(timedelta(hours=8))) - timedelta(minutes=10) # 結束時間 end_time = datetime.now(tz=timezone(timedelta(hours=8))) # 讀取該段時間內的報警資料 alarms = client.read_alarms("Station01", "OutDoorTemp", start_time, end_time) # 取起始時間往後指定個數的報警資料 limit_num = 1 alarms = client.read_alarms("Station01", "OutDoorTemp", start_time=start_time, limit=limit_num)
4.3 監控點的資料變化
監控點的資料變化時 oBIX 協議的一部分。新增想要監控的點,然後當 Niagara 中點的值發生變化後,會自動觸發相應的函式。
from oBIX.common import Point, DataType from oBIX import Client def init_watch(): global client, point_path # 新增監控 point_path_list = [point_path] # 這裡可以是多個點 result = client.add_watch_points(point_path_list) client.watch_changed_handler.on_change += on_watch_changed # Niagara 裡改點的值發生變化時,會自動觸發改函式 def on_watch_changed(points: [Point]): for point in points: val = point.val print(f"on_watch_changed: {val}") if __name__ == '__main__': # ip, userName, password # 可選項: # port: 埠號,如:8080 # https: 是否使用 https,預設:True client = Client("127.0.0.1", "oBIX", "oBIX.12345") # 點的路徑 point_path = "/config/AHU/temp1/" init_watch() client.start_watch() while True: i = 0
4.5 匯出所有點的資訊
如果一個專案中有大量的目錄和點,手動挨個去寫比較麻煩,所以這裡提供了一個匯出點資訊的函式。將點的資訊儲存檔案後,再直接從檔案中讀取點的資訊就會方便很多。
# 匯出所有點的資訊 export_result = client.export_points() # folder_path [optional]: 想要匯出的目錄,如: "/config/xxx/",預設會匯出所有點的資訊 # export_file_name [optional]: 匯出檔案的名稱,預設: "all_points.json" # export_type [optional]: # 0: JSON格式,巢狀格式並保留目錄資訊 # 1: JSON格式, 只保留點的資訊,不保留目錄資訊 # 2: 字串列表格式, 只輸出點的路徑資訊 export_result = client.export_points(folder_path="/config/AHU/", export_file_name="output.json", export_type=1)
&n