PythonライブラリBACpypes3でBACnetデバイスと通信する

## BACnetとは

BACnetとは、ビルディングオートメーションで広く利用されている、産業用プロトコルです。
例えば、空調・衛生（水道）・防犯（ドアロック）・防災（火災や消火）などで利用されています。

ASHRAE（アメリカ暖房冷凍空調学会）が中心となって規格化していますが、日本の市場の特徴に合わせるため、電気設備学会が日本向けの仕様も作っています。

## BACpypes3

BACnetをPythonで利用するには、BACpypes3というPythonライブラリを導入します。

BACpypesという「3」が末尾につかないライブラリも存在しますが、こちらはasyncioに対応しておらず、非同期IOが一般化した現在においては使いにくいため、`async`  `await`により並行処理を記述しやすいBACpypes3を今回利用します。

非同期IOに対応したBACpypes3の開発は開始したばかりのため、まだAPIなどは安定していませんし、未実装のメソッドがあります。
しかし、今更非同期IOに対応しないBACpypesを使うのは大変なので、BACpypes3を選択します。

## デモサーバを起動

BACnetに対応したデバイスとして、電気設備のコントローラを構築するのは大変です。
そのため、パソコン上で動作するデモサーバを導入します。

なお、デモサーバを起動するパソコンは、BACpypes3を実行するパソコンとは別にしてください。

今回は、bacnet-toolsに同梱されている、bacservを利用します。
Windows向けの実行ファイルは下記のURLから入手できます。

https://sourceforge.net/projects/bacnet/files/bacnet-tools/

ZIPファイルをダウンロードして解凍し、bacserv.exeを起動してください。

ただし、コマンドラインオプションを指定するため、PowerShellのようなターミナルを用います。
例えばPowerShellの場合、まずは以下のように、デモサーバで使うIPアドレスを指定します。
複数のネットワークインターフェースを有しているパソコンの場合、どのインターフェース上にデモサーバを起動すればよいか分からないためです。

```
> $Env:BACNET_IFACE = "192.168.0.184"
```

次に、device idを引数にしてデモサーバを起動します。
以下の例では、device idを123としています。
```
> .\bacserv.exe 123
BACnet Server Demo
BACnet Stack Version 1.3.4
BACnet Device ID: 123
Max APDU: 1476
BACnet Device Name: SimpleServer
```

# BACpypes3をインストール
BACpypes3はpipで簡単にインストールできます。

```
> pip install bacpypes3
```

Pythonプログラムでインポートして実行すると、正常にインストールできたか分かります。

```python
import bacpypes3
print('bacpypes3', bacpypes3.__version__)
```

```:実行結果
bacpypes3 0.0.86
```

ちなみに、私はVSCode上でJupyterNotebookを開いてプログラムを実行しています。
インタラクティブに実行結果を確認できるため、大変便利です。

## BACnetデバイスを見つける

まずは、BACpypes3関連や非同期IOのパッケージをインポートします。

```python
import asyncio

from bacpypes3.pdu import Address
from bacpypes3.primitivedata import ObjectIdentifier
from bacpypes3.basetypes import PropertyIdentifier
from bacpypes3.local.device import DeviceObject
from bacpypes3.ipv4.app import NormalApplication
from bacpypes3.pdu import Address, IPv4Address
```

次に、自分自身のデバイスを定義し、アプリケーションを起動します。

```python
# make a device object
local_device = DeviceObject(
        objectIdentifier=("device", 1),
        objectName="test",
        vendorIdentifier=999,
    )
ipv4_address = IPv4Address("192.168.0.184/24", 47808)
app = NormalApplication(local_device, ipv4_address)
```

`IPv4Address("192.168.0.184/24", 47808)`で指定している`"192.168.0.184/24"`は、自身のパソコンのIPアドレスを指定しています。
BACnetデバイスを探す際にブロードキャストをするため、必ず`/24`のようなサブネットマスクの情報を追加しておいてください。

最後に、同じネットワーク内のBACnetデバイスを探す、who-isサービスを実行します。

```python
# デバイスを探す
i_ams = await app.who_is()
for i_am in i_ams:
    device_address: Address = i_am.pduSource
    device_identifier: ObjectIdentifier = i_am.iAmDeviceIdentifier
    print(f"{device_identifier} @ {device_address}")
```

私の環境での実行結果は以下です。

```:実行例
device,123 @ 192.168.0.223
```

## デバイスの情報を取得
最後に見つけたデバイスのアドレスとIDが、それぞれdevice_address、device_identifierに格納されています。
これらに加え、DeviceObjectのプロパティ名を引数で指定することで、プロパティの値を取得できます。
今回は`"object-name"`を取得します。

```python
# デバイスの情報をもらう
device_description: str = await app.read_property(
    device_address, device_identifier, "object-name")
device_description
```

```:出力例
'SimpleServer'
```

## データを取得

BACnetデバイスのオブジェクトをすべて列挙します。
```python
object_list = await app.read_property(
            device_address, device_identifier, "object-list"
        )
object_list
```

```:出力例
[(<ObjectType: device>, 123),
 (<ObjectType: network-port>, 1),
 (<ObjectType: analog-input>, 0),
 (<ObjectType: analog-input>, 1),
 (<ObjectType: analog-input>, 2),
...
 (<ObjectType: accumulator>, 59),
 (<ObjectType: accumulator>, 60),
 (<ObjectType: accumulator>, 61),
 (<ObjectType: accumulator>, 62),
 (<ObjectType: accumulator>, 63)]
```

様々なオブジェクトをこのデバイスは持っていますが、今回はアナログ入力の0番を取得してみます。

以下のプログラムは、複数の値を読み込むサービスを実行する例です。
`analog-input:0`から、現在の値と、値の信頼性を取得します。

```python
object_identifier = ObjectIdentifier("analog-input:0")
response = await app.read_property_multiple(
                device_address,
                [object_identifier, [PropertyIdentifier.presentValue, PropertyIdentifier.reliability]]
            )
response
```

```:実行例
[((<ObjectType: analog-input>, 0),
  <PropertyIdentifier: present-value>,
  None,
  0.0),
 ((<ObjectType: analog-input>, 0),
  <PropertyIdentifier: reliability>,
  None,
  <Reliability: no-fault-detected>)]
```

この実行結果から、値は「0.0」で、信頼性は「問題なし」ということが分かります。

## データを設定

write_propertyメソッドを使うことで、オブジェクトに対して値を設定することもできます。
値を設定するため、`analog-output:1`を対象のオブジェクトとして指定します。
今回は、`10`を書き込んでみます。

まずは、書き込む前に、現在の値を取得します。

```python
present_value = await app.read_property(
    device_address,
    ObjectIdentifier("analog-output:1"),
    PropertyIdentifier.presentValue
)
present_value
```

```:出力例
0.0
```
次に、値を設定します。

```python
await app.write_property(
    device_address,
    ObjectIdentifier("analog-output:1"),
    PropertyIdentifier.presentValue,
    10.0
)
```

設定後に、再度、現在値を取得します。

```python
present_value = await app.read_property(
    device_address,
    ObjectIdentifier("analog-output:1"),
    PropertyIdentifier.presentValue
)
present_value
```

```:出力例
10.0
```

以上の結果から、正しく値の設定ができていることが分かりました。

## データの変化を通知(COV)

BACnetとは、値を監視する仕組みとしてCOV(Change Of Value)があります。

以下のプログラムは、`analog-value:1`オブジェクトの値を30秒間の中で変更があれば通知する設定です。
通知があるまで処理がブロックされるため、デモサーバへの値の書き込みは別のツールから行います。
`count < 10`とループ条件を指定することで、10回変化が発生すれば、停止するようにしています。

```python
async with app.change_of_value(
                device_address,
                ObjectIdentifier("analog-value:1"),
                lifetime=30
            ) as scm:
    count = 0
    while count < 10:
        property_identifier, property_value = await scm.get_value()
        print(f"{property_identifier} {property_value}")
        count = count + 1
```

```:実行例
present-value 10.0
status-flags 
present-value 1.0
status-flags 
present-value 10.0
status-flags 
present-value 1.0
status-flags 
present-value 10.0
status-flags 
```

Yet Another BACnet Explorerなどの別のツールから`analog-value:1`オブジェクトの値を変更したところ、COVの仕組みにより値の変更が通知されました。

## アプリケーションを終了する

忘れずに終了する処理も実行しましょう。

```python
app.close()
```