Databricks にて Azure IoT Hub に対してdevice-to-cloud メッセージ(テレメトリー)を送受信する方法

## 概要

Databricks にて Azure IoT SDK により Azure IoT Hub に対してdevice-to-cloud メッセージ(テレメトリー)を送受信する方法を紹介します。本記事で詳細する方法は、検証目的で実行する方法です。

## 事前準備

### 環境構築

- Azure IoT Hub の構築 （本手順は Free tier で実施）
- Databricks （本手順は Serveless ではなく汎用コンピュートで実行）

### デバイスの作成と接続文字列の取得

Azure IoT Hub のリソース画面にて、`デバイス`タブを選択後、`+ デバイスの追加`を選択します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/32708e89-0264-4b64-b5f2-494be974a7d4.png)

`デバイス ID`に任意の値を入力し、`保存`を選択してデバイスを作成します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/37ac238e-cbd3-48d6-82f5-68cf9038354e.png)

作成したデバイスを表示して、`プライマリ接続文字列`の`コピー済み`を選択して値をコピーします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/4f8d5549-6a72-4ff4-9e10-0e23214201f6.png)

### 組み込みのエンドポイントの接続文字列を取得

Azure IoT Hub のリソース画面にて、`組み込みのエンドポイント`タブを選択して、`イベントハブ互換エンドポイント`の値をコピーします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0d7b0ebb-998d-473d-8ee6-cbee3af3bbed.png)

## device-to-cloud メッセージ(テレメトリー)を送信する方法

### 同期処理により送信する方法

Databricks にて azure-iot-device をインストールします。

```python
%pip install azure-iot-device -q
dbutils.library.restartPython()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/60d08c38-421b-4437-b86a-294846eac223.png)

デバイスの接続文字列を変数にセットします。

```python
device_conn_str = "HostName=iot-hub-test-001.azure-devices.net;DeviceId=qiita_test_01;SharedAccessKey=r2fxXUdn/4w9iNJdhqnIJZ+h0qxxxxx="
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0018e890-76b4-49d8-8f00-38d0ffeea171.png)

Azure IoT Hub にテレメトリーを送信するコードを記述します。コードの実行は、次の手順である Azure にてテレメトリーの監視のコマンドを実行してから実行してください。

```python
import os, json, random, time, uuid
from datetime import datetime, timezone
from azure.iot.device import IoTHubDeviceClient, Message


def get_sample_date():
    date = {
        "temperature": round(random.uniform(20, 25), 2),
        "timestamp": datetime.now(timezone.utc).isoformat(),
    }

    return json.dumps(date)

# クライアントのインスタンス化
device_client = IoTHubDeviceClient.create_from_connection_string(
    device_conn_str,
    websockets=True,
)

# データの準備
payload = get_sample_date()
msg = Message(payload)
msg.message_id = uuid.uuid4()
msg.content_encoding = "utf-8"
msg.content_type = "application/json"
msg.custom_properties["sensor-location"] = "jp"

# クライアントに接続
device_client.connect()

# テレメトリーの送信
device_client.send_message(msg)
print("✓ sent:", payload)
time.sleep(5)

# クライアントの終了
device_client.shutdown()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/e18b64d1-978e-4be1-bca3-a6271b2ccced.png)


Azure ポータルにて下記のコマンドを実行します。Databricks に戻り、テレメトリーの送信を実施して、テレメトリーが受信できることを確認します。

```bash
IOTHUB_NAME="iot-hub-test-001"
az iot hub monitor-events --hub-name "$IOTHUB_NAME" --props all
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/a0fc2b21-46a6-4434-9657-4aaff2c1de34.png)

### 非同期処理により送信する方法

Databricks にて azure-iot-device をインストールします。

```python
%pip install azure-iot-device -q
dbutils.library.restartPython()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/9fe5346a-d22c-47c1-925e-187822cfd97d.png)

デバイスの接続文字列を変数にセットします。

```python
device_conn_str = "HostName=iot-hub-test-001.azure-devices.net;DeviceId=qiita_test_01;SharedAccessKey=r2fxXUdn/4w9iNJdhqnIJZ+h0qxxxxx="
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/7f346929-a46b-44d4-abfc-e091cf40ce70.png)

Azure IoT Hub にテレメトリーを送信するコードを記述します。コードの実行は、次の手順である Azure にてテレメトリーの監視のコマンドを実行してから実行してください。

```python
import os

os.environ["DEVICE_CONNECTION_STRING"] = device_conn_str
```

```python
import asyncio, json, random, uuid, os
from datetime import datetime, timezone
from azure.iot.device.aio import IoTHubDeviceClient
from azure.iot.device import Message


def get_sample_data() -> str:
    return json.dumps(
        {
            "temperature": round(random.uniform(20, 25), 2),
            "timestamp": datetime.now(timezone.utc).isoformat(),
        }
    )


async def main() -> None:
    device_conn_str = os.getenv("DEVICE_CONNECTION_STRING")
    device_client = IoTHubDeviceClient.create_from_connection_string(
        device_conn_str, websockets=True
    )

    await device_client.connect()

    payload = get_sample_data()
    msg = Message(payload)
    msg.message_id = str(uuid.uuid4())
    msg.content_encoding = "utf-8"
    msg.content_type = "application/json"
    msg.custom_properties["sensor-location"] = "jp"

    await device_client.send_message(msg)
    print("✓ sent:", payload)

    await asyncio.sleep(5)
    await device_client.shutdown()
```

```python
await main()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/00fa0280-a70e-4322-88c0-1baff385771f.png)

Azure IoT Hub にテレメトリーを送信するコードを記述します。コードの実行は、次の手順である Azure にてテレメトリーの監視のコマンドを実行してから実行してください。

```bash
IOTHUB_NAME="iot-hub-test-001"
az iot hub monitor-events --hub-name "$IOTHUB_NAME" --props all
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ca1f75ce-a448-4c26-b842-a56ba14104cc.png)

:::note info

複数回の送信を実施したい場合には下記のようなコードを実行します。

```python
import os

os.environ["DEVICE_CONNECTION_STRING"] = device_conn_str

import asyncio, json, random, uuid, os, sys
from datetime import datetime, timezone
from azure.iot.device.aio import IoTHubDeviceClient
from azure.iot.device import Message


def get_sample_data() -> str:
    return json.dumps(
        {
            "temperature": round(random.uniform(20, 25), 2),
            "timestamp": datetime.now(timezone.utc).isoformat(),
        }
    )


async def send_messages(
    client: IoTHubDeviceClient,
    num_messages: int,
    interval_sec: float = 0.0,
) -> None:
    """指定回数メッセージを送信するユーティリティ"""
    for i in range(num_messages):
        payload = get_sample_data()
        msg = Message(payload)
        msg.message_id = str(uuid.uuid4())
        msg.content_encoding = "utf-8"
        msg.content_type = "application/json"
        msg.custom_properties["sensor-location"] = "jp"

        await client.send_message(msg)
        print(f"{i + 1}/{num_messages} ✓ sent:", payload)

        if i < num_messages - 1 and interval_sec > 0:
            await asyncio.sleep(interval_sec)


async def main(
    num_messages: int = 1,
    interval_sec: float = 0.0,
) -> None:
    device_conn_str = (
        os.getenv("DEVICE_CONNECTION_STRING")
    )
    device_client = IoTHubDeviceClient.create_from_connection_string(
        device_conn_str, websockets=True
    )

    await device_client.connect()
    await send_messages(device_client, num_messages, interval_sec)
    await device_client.shutdown()

await main(
    num_messages=3,
    interval_sec=0.2,
)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/088f0d38-6a0f-4ab4-9786-39b84d5a5925.png)

:::

## テレメトリーを受信する方法

### 組み込みのエンドポイントから受信する方法

Databricks にて azure-eventhub をインストールします。

```python
%pip install azure-eventhub -q
dbutils.library.restartPython()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5f04097c-6444-4e68-9b13-603e8c7673fb.png)

組み込みのエンドポイントの接続文字列を変数にセットします。また、コンシューマーグループを`$Default`で指定します。

```python
eh_conn_str = "Endpoint=sb://ihsuprodblres095dednamespace.servicebus.windows.net/;SharedAccessKeyName=iothubowner;SharedAccessKey=FUum/tiOgEuhyeXXX=;EntityPath=iothub-ehub-iot-hub-te-65629521-b811712f03"
consumer_group = "$Default"
```

下記のコードを実行して、送信済みのテレメトリーを表示できることを確認します。処理が継続するため、セルの実行を停止することを忘れないようにしてください。

```python
import os

os.environ["EH_CONN"] = eh_conn_str
os.environ["CONSUMER_GROUP"] = consumer_group
```

```python
import asyncio, os, json
from azure.eventhub.aio import EventHubConsumerClient



async def on_event(partition_context, event):
    print(f"{event.body_as_str()}")
    await partition_context.update_checkpoint(event)


async def main():
    eh_conn_str = os.getenv("EH_CONN")
    group = os.getenv("CONSUMER_GROUP")
    client = EventHubConsumerClient.from_connection_string(
        eh_conn_str,
        consumer_group=group,
    )
    async with client:
        await client.receive(
            on_event,
            starting_position="-1",   # 検証目的であるため -1 を指定
        )
```

```python
await main()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/4e934aa8-9bb3-456d-8da3-1e10cde8e521.png)