はじめに
以前、上記内容を記事を書いたのですが、こちらの内容にNode-REDも組み合わせた内容となっています。今回は、pyファイルの実行結果と同じ内容をNode-REDのDebugログにも出力させるというのをやっていきます。
coolmaster_bacnet-multiread to Node-RED.py
まずはこちらのソースファイルを用意します。
# --- 最新のNode-RED環境ではUTF-8の文字列の方がスムーズ ---
import sys
import io
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
from bacpypes.app import BIPSimpleApplication
from bacpypes.object import get_datatype, DeviceObject
from bacpypes.apdu import ReadPropertyMultipleRequest, ReadAccessSpecification, PropertyReference
from bacpypes.pdu import Address
from bacpypes.primitivedata import ObjectIdentifier
from bacpypes.core import run, stop
from bacpypes.iocb import IOCB
# --- LocalDeviceObject を定義（2系では自作） ---
class MyLocalDeviceObject(DeviceObject):
def __init__(self, **kwargs):
super().__init__(**kwargs)
# --- アドレス設定 ---
LOCAL_ADDRESS = Address("192.168.1.26")
TARGET_ADDRESS = Address("192.168.1.100")
# --- 読み取り対象 ---
OBJECTS = {
("binaryValue", 256): "電源 (on/off)",
("analogValue", 256): "温度 SetPoint",
("multiStateValue", 256): "風量 fan speed",
("multiStateValue", 257): "運転モード mode"
}
# --- デバイス定義 ---
local_device = MyLocalDeviceObject(
objectIdentifier=("device", 599),
objectName="MyBACnetClient",
vendorIdentifier=15,
maxApduLengthAccepted=1024,
segmentationSupported='noSegmentation'
)
# --- アプリケーション作成 ---
app = BIPSimpleApplication(local_device, LOCAL_ADDRESS)
# --- 読み取り要求作成 ---
request = ReadPropertyMultipleRequest(destination=TARGET_ADDRESS)
request.listOfReadAccessSpecs = []
for (obj_type, obj_inst), label in OBJECTS.items():
obj_id = ObjectIdentifier((obj_type, obj_inst))
prop_ref = PropertyReference(propertyIdentifier="presentValue")
access_spec = ReadAccessSpecification(
objectIdentifier=obj_id,
listOfPropertyReferences=[prop_ref]
)
request.listOfReadAccessSpecs.append(access_spec)
# --- リクエスト送信 ---
iocb = IOCB(request)
app.request_io(iocb)
# --- 応答処理と自動停止 ---
def on_response(iocb):
if iocb.ioResponse:
apdu = iocb.ioResponse
for result in apdu.listOfReadAccessResults:
obj_id = result.objectIdentifier
label = OBJECTS.get(obj_id, str(obj_id))
for prop_result in result.listOfResults:
if prop_result.readResult.propertyAccessError is not None:
print(f"[CoolMaster][ERROR] {label}: {prop_result.readResult.propertyAccessError}", flush=True)
else:
datatype = get_datatype(obj_id[0], prop_result.propertyIdentifier)
value = prop_result.readResult.propertyValue.cast_out(datatype)
if label == "電源 (on/off)":
status_map = {
0: "off", 1: "on",
"active": "on", "inactive": "off"
}
status_str = status_map.get(value, str(value))
print(f"[CoolMaster][INFO] {label}: {value} ({status_str})", flush=True)
elif label == "温度 SetPoint":
print(f"[CoolMaster][INFO] {label}: {value}°C", flush=True)
elif label == "風量 fan speed":
speed_map = {
1: "弱", 2: "中", 3: "強", 4: "自動"
}
try:
speed_str = speed_map.get(int(value), "不明")
except Exception:
speed_str = str(value)
print(f"[CoolMaster][INFO] {label}: {value} ({speed_str})", flush=True)
elif label == "運転モード mode":
mode_map = {
1: "冷房", 2: "暖房", 3: "冷暖自動", 4: "ドライ"
}
try:
mode_str = mode_map.get(int(value), "不明")
except Exception:
mode_str = str(value)
print(f"[CoolMaster][INFO] {label}: {value} ({mode_str})", flush=True)
else:
print(f"[CoolMaster][INFO] {label}: {value}", flush=True)
else:
print("[CoolMaster][ERROR] No response received.", flush=True)
stop()
iocb.add_callback(on_response)
# --- メイン ---
if __name__ == "__main__":
run()実行結果
ターミナルから実行させても実行結果は至って普通です。
Node-REDのフローを用意
実行トリガーが押されるとexecで指定されているローカル上のpyファイルが実行され、取得した内容をUTF-8変換をかけてデバッグノードに出力させるといったものです。
以下コピペ可能な全体フローです。pyファイルの場所は指定してあげてください。
[{"id":"35c657ec8fc64896","type":"tab","label":"CoolMaster出力","disabled":false,"info":""},{"id":"eeac2f456a3f37ff","type":"inject","z":"35c657ec8fc64896","name":"実行トリガー","props":[],"repeat":"","crontab":"","once":false,"onceDelay":0.1,"topic":"","payloadType":"date","x":140,"y":100,"wires":[["dd19a842505fd303"]]},{"id":"dd19a842505fd303","type":"exec","z":"35c657ec8fc64896","command":"\"C:\\VisualStudioCode_ProjectFile\\CoolMaster\\.venv\\Scripts\\python.exe\" \"C:\\VisualStudioCode_ProjectFile\\CoolMaster\\coolmaster_bacnet-multiread to Node-RED.py\"","addpay":false,"append":"","useSpawn":"false","timer":"","winHide":false,"oldrc":false,"name":"Python実行","x":360,"y":100,"wires":[["5b785f0e42f73cf5"],[],[]]},{"id":"5b785f0e42f73cf5","type":"function","z":"35c657ec8fc64896","name":"Shift_JIS → UTF-8文字列","func":"if (Buffer.isBuffer(msg.payload)) {\n msg.payload = msg.payload.toString(\"utf8\");\n} else if (Array.isArray(msg.payload)) {\n msg.payload = Buffer.from(msg.payload).toString(\"utf8\");\n}\nreturn msg;","outputs":1,"timeout":"","noerr":0,"initialize":"","finalize":"","libs":[],"x":590,"y":80,"wires":[["388e2a81c4e2e4d4"]]},{"id":"388e2a81c4e2e4d4","type":"debug","z":"35c657ec8fc64896","name":"出力結果","active":true,"tosidebar":true,"console":false,"tostatus":false,"complete":"payload","targetType":"msg","statusVal":"","statusType":"auto","x":780,"y":80,"wires":[]}]実行結果
Visual Studio Codeのターミナルで実行されていたものと同じ内容がNode-REDのDebugから確認することが出来ました。
まとめ
Node-REDからもPythonを実行してBACnetのデータポイントをreadするということができるので、条件式やreadとwriteの組み合わせ方によっては、人為的な操作を必要としない完全フルオートのコントロールも実現できそうということですね。