BACnetには1度のリクエストで複数の情報を読み取る「ReadPropertyMultipleRequest」に対して1度のリクエストで複数の書き込みを行う「WritePropertyMultipleRequest」があるようなので使用できるか試してみました。
「WritePropertyMultipleRequest」はBAC0、bacpypesライブラリにて使用できるようで、Pythonのバージョンは3.10以降に対応しているようです。
結論から言うと「WritePropertyMultipleRequest」による1度のリクエストで複数書き込みすることはできませんでした。どうやらCoolMasterが「WritePropertyMultiple」に対応しておらずできないようです。他の方法も調べてみましたが「WritePropertyMultiple」を使う以外にBACnetで複数同時書き込みはできないようでした。
疑似的に行う
そこで今回は1度のリクエストに対し1つの書き込みを行うものをディレイをかけることで複数のリクエストをひとつの処理で行うものを作成しました。
Pythonコードはこちらです。
import threading
from bacpypes.app import BIPSimpleApplication
from bacpypes.object import AnalogValueObject
from bacpypes.apdu import ReadPropertyRequest, WritePropertyRequest
from bacpypes.primitivedata import ObjectIdentifier, Real, Unsigned
from bacpypes.constructeddata import Any
from bacpypes.pdu import Address
from bacpypes.core import run, enable_sleeping, stop
from bacpypes.iocb import IOCB
import time
# BACnet 設定
BACNET_DEVICE_IP = "192.168.1.100"
DEVICE_ID = 64
SETPOINT_OBJECT_ID = "analogValue:256"
MODE_OBJECT_ID = "multiStateValue:257"
FAN_SPEED_OBJECT_ID = "multiStateValue:256"
LOCAL_BACNET_IP = "192.168.1.168" #自分のローカルIP
class BACnetClient(BIPSimpleApplication):
def __init__(self, local_address):
super().__init__(None, local_address)
def read_setpoint(self):
print("[INFO] Setpoint を取得中...", flush=True)
request = ReadPropertyRequest(
objectIdentifier=ObjectIdentifier(SETPOINT_OBJECT_ID),
propertyIdentifier="presentValue"
)
request.pduDestination = Address(BACNET_DEVICE_IP)
iocb = IOCB(request)
self.request_io(iocb)
def on_response(iocb):
if iocb.ioError:
print(f"[ERROR] 読み取りエラー: {iocb.ioError}", flush=True)
else:
value = iocb.ioResponse.propertyValue.cast_out(Real)
print(f"[INFO] 現在のセットポイント: {value}°C", flush=True)
iocb.add_callback(on_response)
def write_setpoint(self, new_value):
print(f"[INFO] Setpoint を {new_value}°C に変更中...", flush=True)
request = WritePropertyRequest(
objectIdentifier=ObjectIdentifier(SETPOINT_OBJECT_ID),
propertyIdentifier="presentValue",
propertyValue=Any(Real(new_value))
)
request.pduDestination = Address(BACNET_DEVICE_IP)
iocb = IOCB(request)
self.request_io(iocb)
def on_response(iocb):
if iocb.ioError:
print(f"[ERROR] 書き込みエラー: {iocb.ioError}", flush=True)
else:
print(f"[INFO] Setpoint を {new_value}°C に変更しました", flush=True)
iocb.add_callback(on_response)
def write_mode(self, mode_value):
print(f"[INFO] モードを {mode_value} に変更中...", flush=True)
request = WritePropertyRequest(
objectIdentifier=ObjectIdentifier(MODE_OBJECT_ID),
propertyIdentifier="presentValue",
propertyValue=Any(Unsigned(mode_value))
)
request.pduDestination = Address(BACNET_DEVICE_IP)
iocb = IOCB(request)
self.request_io(iocb)
def on_response(iocb):
if iocb.ioError:
print(f"[ERROR] モード変更エラー: {iocb.ioError}", flush=True)
else:
print(f"[INFO] モードを {mode_value} に変更しました", flush=True)
iocb.add_callback(on_response)
def read_mode(self):
print("[INFO] モードを取得中...", flush=True)
request = ReadPropertyRequest(
objectIdentifier=ObjectIdentifier(MODE_OBJECT_ID),
propertyIdentifier="presentValue"
)
request.pduDestination = Address(BACNET_DEVICE_IP)
iocb = IOCB(request)
self.request_io(iocb)
def on_response(iocb):
if iocb.ioError:
print(f"[ERROR] 読み取りエラー: {iocb.ioError}", flush=True)
else:
value = iocb.ioResponse.propertyValue.cast_out(Unsigned)
print(f"[INFO] 現在のモード値: {value}", flush=True)
iocb.add_callback(on_response)
def write_fan_speed(self, fan_speed_value):
print(f"[INFO] 風量を {fan_speed_value} に変更中...", flush=True)
request = WritePropertyRequest(
objectIdentifier=ObjectIdentifier(FAN_SPEED_OBJECT_ID),
propertyIdentifier="presentValue",
propertyValue=Any(Unsigned(fan_speed_value))
)
request.pduDestination = Address(BACNET_DEVICE_IP)
iocb = IOCB(request)
self.request_io(iocb)
def on_response(iocb):
if iocb.ioError:
print(f"[ERROR] 風量変更エラー: {iocb.ioError}", flush=True)
else:
print(f"[INFO] 風量を {fan_speed_value} に変更しました", flush=True)
iocb.add_callback(on_response)
def read_fan_speed(self):
print("[INFO] 現在の風量を取得中...", flush=True)
request = ReadPropertyRequest(
objectIdentifier=ObjectIdentifier(FAN_SPEED_OBJECT_ID),
propertyIdentifier="presentValue"
)
request.pduDestination = Address(BACNET_DEVICE_IP)
iocb = IOCB(request)
self.request_io(iocb)
def on_response(iocb):
if iocb.ioError:
print(f"[ERROR] 読み取りエラー: {iocb.ioError}", flush=True)
else:
value = iocb.ioResponse.propertyValue.cast_out(Unsigned)
print(f"[INFO] 現在の風量モード値: {value}", flush=True)
iocb.add_callback(on_response)
# イベントループ
client = BACnetClient(Address(LOCAL_BACNET_IP))
def start_bacnet_loop():
enable_sleeping()
print("[INFO] BACnet イベントループ開始...", flush=True)
run()
thread = threading.Thread(target=start_bacnet_loop, daemon=True)
thread.start()
# メイン処理
try:
time.sleep(2)
client.read_mode()
time.sleep(2)
client.write_mode(1) # モード
time.sleep(2)
client.write_setpoint(27.0) #設定温度
time.sleep(2)
client.read_setpoint()
time.sleep(2)
client.read_fan_speed()
time.sleep(2)
client.write_fan_speed(1) # 風量変更
time.sleep(2)
client.read_fan_speed()
while True:
time.sleep(1)
except KeyboardInterrupt:
print("\n[INFO] ユーザーが停止しました。BACnetを終了します。", flush=True)
stop()BACnet設定の箇所で下記のように設定しておりこちらはyabeよりアドレスを調べ入力を行っています。
SETPOINT_OBJECT_ID = “analogValue:256”
MODE_OBJECT_ID = “multiStateValue:257”
FAN_SPEED_OBJECT_ID = “multiStateValue:256”
上から温度設定, モード設定, 風量設定になっています。
client.read_mode()
time.sleep(2)
client.write_mode(1)
time.sleep(2)
client.write_setpoint(27.0)
time.sleep(2)このコードではBACnetデバイスが応答を返すまでの時間を確保し、リクエストが混雑しないようにするためにリクエストごとに2秒のディレイをかけています。
こちらがコードの実行結果です。うまく３つの項目を変更することが出来ました。
まとめ
今回は「WritePropertyMultipleRequest」が使えるのか試してみましたができなかったため、別の方法で似たような仕様のコードの作成を行いました。デバイスの仕様を理解することが大切であることを学びました。また、リクエストを送る際のディレイの必要性に関しても学びました。
