[Azure] IoT Hubへの通信をローカル実行Functions(Python)で受けて表示する方法

前回はIoT HubへのメッセージをAzure Stream Analyticsで受ける方法を調べていきましたが、今回はAzure Functionsで受ける方法を調べていきます。

前回記事
[Azure] IoT Hub と Azure SQL を Stream Analytics を介して接続する #Azure - Qiita

環境

Windows11, WSL2

❯ cat /etc/lsb-release
DISTRIB_ID=Ubuntu
DISTRIB_RELEASE=22.04
DISTRIB_CODENAME=jammy
DISTRIB_DESCRIPTION="Ubuntu 22.04.2 LTS"

おおまかな流れ

1. Azure Functions Core Toolsのインストール
2. Azure Functions リソースの用意
3. funcコマンドでFunctionプロジェクトの作成
4. 出来上がったプロジェクトの設定
5. Functionsのローカル起動

Azure Functions Core Toolsのインストール

まずはAzure Functions Core Toolsが必要なので、公式サイトに従ってインストールしましょう。

私の環境の場合はWSL2/Ubuntu なので、以下のリンクの手順でインストールしました。
（マイクロソフトのgpgキーとリポジトリをaptに追加したあとにパッケージインストール）

これでシステム上で func コマンドが使えるようになります。

❯ func --version
4.0.5030

Azure Functions リソースの用意

以前も紹介しました以下の記事の抜粋になります

AzureのFunctionsをCLIで使う（ローカルリソースの生成からデプロイまで）with Python #Azure - Qiita

下記コードでFunctionsリソースとそれに対応するストレージを作成します。
コマンドを実行する前に az login しておくことを忘れずに

main.sh

#!/bin/sh

STORAGE_NAME="storagename"
RESOURCE_GROUP_NAME="reouce-group-name"
LOCATION="japaneast"
APP_NAME="func-name"

az storage account create --name $STORAGE_NAME --resource-group $RESOURCE_GROUP_NAME
az functionapp create --consumption-plan-location $LOCATION --runtime python --runtime-version 3.10 --functions-version 4 --name $APP_NAME --os-type linux --storage-account $STORAGE_NAME --resource-group $RESOURCE_GROUP_NAME

すると、Functionsリソースが指定のリソースグループに作成されています。
概要を見て、これからAzure Event Hub triggerに対応した関数を作っていくのですが、これはコマンドとエディターで作っていくということで、やり方を確認するのは以下の画面でできます。

funcコマンドでFunctionプロジェクトの作成

上で確認した方法をなぞっていくことになりますが、私の場合はすこしやり方が違う場合もあります（やり方はいくつもあると思いますので、ご自分のやりやすい方法で解決されてください）

適当なディレクトリで func init を実行します。
実際には、func init test-func --python というコマンドを使いました。
これによって、PythonをつかったFunctionsプロジェクトを test-funcというディレクトリに作成することができます。

次に cd test-func した後に、func newします。

すると、下図のように、templateを選ぶ選択肢が出てきますので、Azure Event Hub trigger(8)を選択します。

次に名前の入力を求められるのでiot-hub-conと名付けてみます。

これで、Functionプロジェクトの作成が完了しました。

加えて、pip install azure-functionsでPython側に Azure functions core tools のSDKをインストールしておきましょう。

私の場合はPythonプロジェクトをpoetryで管理していますので、poetry add azure-functionsでインストールを完了しました。

出来上がったプロジェクトの設定

次に作成したプロジェクトの設定ファイルを編集し、実際にIoT Hubと接続してゆきます。

test-func/local.settings.jsonの編集

まずはtest-func/local.settings.jsonを編集していきます

local.settings.json

{
  "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": <ストレージ接続文字列>,
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "IoTHubConnectionString": <イベントハブ互換エンドポイント>
  }
}

上記の通り、ここではストレージ接続文字列とイベントハブ互換エンドポイントが必要です。
これらを取得する方法を以下に説明します。

ストレージ接続文字列の取得

ストレージ接続文字列は、Functionsリソースを作るときにFunctionsリソースに紐付けるために作ったあのストレージの接続文字列です。
以下の場所で取得できます。

イベントハブ互換エンドポイントの取得

イベントハブ互換エンドポイントは、IoT Hubの組み込みエンドポイントから取得できます。
下図の通りです。

test-func/iot-hub-con/function.jsonの編集

function.json

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "type": "eventHubTrigger",
      "name": "events",
      "direction": "in",
      "eventHubName": <Iot Hubリソースの名前>,
      "connection": "IoTHubConnectionString",
      "cardinality": "many",
      "consumerGroup": "$Default"
    }
  ]
}

ここで変更が必要なものはeventHubNameとconnectionの2つです。

eventHubNameは作成したIoT Hubのリソースの名前をそのまま入力します。
connectionの"IoTHubConnectionString"という文字列は、local.settings.jsonの
"IoTHubConnectionString": <イベントハブ互換エンドポイント>　部分を設定項目名を使って参照する形になっています。

Functionsのローカル起動

ここまで終わったら func start で作成したアプリを実行します。
poetryの場合はpoetry run func startです。

するとローカルのアプリがIoT Hubと接続されて、IoT Hubデータの待ち受けが開始されます。

この状態で、IoT Hubに対してmessageを送信すると、__init__.pyの内容どおり、それを出力する内容が表示されるはずです。

Azure IoT Hubにエッジデバイスからデータを送信から確認まで #Azure - Qiita

__init__.py

import logging

from azure.functions import EventHubEvent
from typing import List


def main(events: List[EventHubEvent]):
    for event in events:
        logging.info('Python EventHub trigger processed an event: %s',
                     event.get_body().decode('utf-8'))

無事メッセージを受け取ると下記のような表示が得られます。