CoolMasterのBACnetデータポイントをPythonを通して操作してみた

YABEから行っていた動作のように、Visual Studio CodeのPythonからプログラムを実行することでSet Pointの確認と変更をやってみます。

環境構築として、[BACpypes]というプラグインがインストールされている必要がありますので、[pip install bacpypes]のようなコマンドからインストールします。

[coolmaster_bacnet.py]というファイルを生成し、下記のようなソースコードを用意します。

import threading
from bacpypes.app import BIPSimpleApplication
from bacpypes.object import AnalogValueObject
from bacpypes.apdu import ReadPropertyRequest, WritePropertyRequest
from bacpypes.primitivedata import ObjectIdentifier, Real
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"  # CoolMaster の IP
DEVICE_ID = 64  # CoolMaster の BACnet Device ID
SETPOINT_OBJECT_ID = "analogValue:256"  # YABE で確認したオブジェクトID

# PC 側のローカル BACnet アドレス（YABEと同じものにする）
LOCAL_BACNET_IP = "192.168.1.4"  # PC のローカルIP

# BACnet クライアント
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)  # 修正: `self.request()` → `self.request_io()`
        
        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)  # 修正: `self.request()` → `self.request_io()`

        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)  # 修正: 非同期で処理

# BACnet クライアントの作成
client = BACnetClient(Address(LOCAL_BACNET_IP))

# BACnet イベントループを別スレッドで実行
def start_bacnet_loop():
    enable_sleeping()
    print("[INFO] BACnet イベントループ開始...", flush=True)
    run()  # イベントループの開始

# バックグラウンドスレッドで `run()` を実行
thread = threading.Thread(target=start_bacnet_loop, daemon=True)
thread.start()

# メイン処理（セットポイントの取得 & 変更）
try:
    time.sleep(2)  # BACnet の準備を待つ
    client.read_setpoint()  # 現在のセットポイントを取得
    time.sleep(2)
    client.write_setpoint(27.0)  # 設定したい温度を入力
    time.sleep(2)
    client.read_setpoint()  # 変更後のセットポイントを取得

    while True:
        time.sleep(1)  # メインスレッドを維持（Ctrl+Cで抜けられるようにする）

except KeyboardInterrupt:
    print("\n[INFO] ユーザーが停止しました。BACnetを終了します。", flush=True)
    stop()  # BACnet を安全に停止