Azure Functionsをつかう その３：Azure IoT Hubと連動する

実際のところ

Azure

from azure.iot.device import IoTHubDeviceClient, Message
import json
import time
import random
import os

# Read the config file
with open('config.json', 'r') as f:
    config = json.load(f)

CONNECTION_STRING = config['CONNECTION_STRING']  # Azure IoT Hubから取得した接続文字列
INTERVAL = config['INTERVAL']  # Interval in seconds

def iothub_client_init():
    client = IoTHubDeviceClient.create_from_connection_string(CONNECTION_STRING)
    return client

def send_to_hub(client):
    while True:
        payload = {"timestamp": time.time(), "value": random.randint(0, 254)}  # タイムスタンプとランダムな整数値を持つJSONを作成
        print(f"Sending message: {json.dumps(payload)}")
        client.send_message(json.dumps(payload))  # メッセージを送信
        time.sleep(INTERVAL)  # Pause for the interval specified

if __name__ == '__main__':
    client = iothub_client_init()
    send_to_hub(client)

Azure Functions

"__init__.py"

import logging
import azure.functions as func
import json

def main(event: func.EventHubEvent):
    logging.info('Python EventHub trigger function processed an event.')
    
    message = json.loads(event.get_body().decode())
    logging.info(f"Processed message: {message}")

参考もと

Azure Functions の Python 開発者向けリファレンス | Microsoft Learn