はじめに
Rasberry Pi からAzure IoT Hub にデータを送信する方法を紹介していきます。
前回の記事の続きになります。
前回の記事では、Raspberry Pi 4+ BMS280 + MH-Z19B でディスプレイでモニタリングする方法を紹介しましたが、これではデータをリアルタイムに表示させるだけでオンラインでも確認できません。
そこでAzure IoT Hub を利用してデータを送信していきます。また、次回は可視化を行います。
IoT Hubの作成
IoT Hubの作成は下記を参考に行っていきます。デバイスの登録まで手順通り行っていきます。
サンプルプログラム
サンプルプログラムはPythonを利用します。実際のRasberry PiではPythonでデータの取得、表示を行っているからです。
git clone https://github.com/Azure/azure-iot-sdk-python
ダウンロードしたら azure-iot-sdk-python/samples 移動します。
simple_send_message.py を利用して、既存のプログラムに追加していきます。
main関数を変更し、async def connect_iot():として定義して追加しています。
sendmg = {'messageId':num,'deviceId':'Raspberry Pi Web Client','temperature':format(float(data.temperature), '.1f'),'humidity':format(float(data.humidity), '.1f')}
は送信するフォーマットです。
Azure IoT HubではJSON形式になります。
あとは、データ取得するwhile内で実行します。
IOTHUB_DEVICE_CONNECTION_STRING は"デバイスの接続文字列を入力します。これはセキュリティリスクがあります。検証のみにしてください。
#bme280_scd30_sample_TFT_csv.py
#coding: utf-8
IOTHUB_DEVICE_CONNECTION_STRING ="接続文字列を入力"
n=1
#---TFT-ili9341 init-----rgb_display_pillow_stats.py---
import time
import subprocess
import digitalio
import board
import mh_z19
import json
import os
import asyncio
from azure.iot.device.aio import IoTHubDeviceClient
from PIL import Image, ImageDraw, ImageFont
from adafruit_rgb_display import ili9341
# Configuration for CS and DC pins (these are PiTFT defaults):
cs_pin = digitalio.DigitalInOut(board.CE0)
dc_pin = digitalio.DigitalInOut(board.D25)
reset_pin = digitalio.DigitalInOut(board.D24)
# Config for display baudrate (default max is 24mhz):
BAUDRATE = 24000000
# Setup SPI bus using hardware SPI:
spi = board.SPI()
# pylint: disable=line-too-long
# Create the display:
disp = ili9341.ILI9341(
spi,
rotation=90, # 2.2", 2.4", 2.8", 3.2" ILI9341
cs=cs_pin,
dc=dc_pin,
rst=reset_pin,
baudrate=BAUDRATE,
)
# pylint: enable=line-too-long
# Create blank image for drawing.
# Make sure to create image with mode 'RGB' for full color.
if disp.rotation % 180 == 90:
height = disp.width # we swap height/width to rotate it to landscape!
width = disp.height
else:
width = disp.width # we swap height/width to rotate it to landscape!
height = disp.height
image = Image.new("RGB", (width, height))
# Get drawing object to draw on image.
draw = ImageDraw.Draw(image)
# Draw a black filled box to clear the image.
draw.rectangle((0, 0, width, height), outline=0, fill=(0, 0, 0))
disp.image(image)
# First define some constants to allow easy positioning of text.
padding = -2
x = 0
# Load a ipaexfont-gothic font.
font1 = ImageFont.truetype("/usr/share/fonts/opentype/ipaexfont-gothic/ipaexg.ttf", 50)
font2 = ImageFont.truetype("/usr/share/fonts/opentype/ipaexfont-gothic/ipaexg.ttf", 20)
font3 = ImageFont.truetype("/usr/share/fonts/opentype/ipaexfont-gothic/ipaexg.ttf", 15)
#---BME280 init---------------bme280_sample.py------------
import smbus2
import bme280
port = 1
address = 0x76
bus = smbus2.SMBus(port)
calibration_params = bme280.load_calibration_params(bus, address)
# the sample method will take a single reading and return a
# compensated_reading object
data = bme280.sample(bus, address, calibration_params)
time.sleep(2)
#Azure IoT Connect and Send Message
async def connect_iot():
# Fetch the connection string from an environment variable
conn_str =IOTHUB_DEVICE_CONNECTION_STRING
# Create instance of the device client using the connection string
device_client = IoTHubDeviceClient.create_from_connection_string(conn_str)
# Connect the device client.
await device_client.connect()
# Send a single message
print("Sending message...")
sendmg = {'messageId':num,'deviceId':'Raspberry Pi Web Client','temperature':format(float(data.temperature), '.1f'),'humidity':format(float(data.humidity), '.1f')}
await device_client.send_message(json.dumps(sendmg))
print("Message successfully sent!")
# Finally, shut down the client
await device_client.shutdown()
num = 0
while True:
outco2 = mh_z19.read_co2valueonly()
num = num + 1
#---csv out ------------------------------------------
dtd = time.strftime('%Y年%m月%d日', time.localtime())
dtt = time.strftime('%H時%M分%S秒', time.localtime())
dta = time.strftime('%a', time.localtime())
dta = time.strftime('%a', time.localtime())
if dta == "Mon":
jdta = dta.replace("Mon", "(月)")
elif dta == "Tue":
jdta = dta.replace("Tue", "(火)")
elif dta == "Wed":
jdta = dta.replace("Wed", "(水)")
elif dta == "Thu":
jdta = dta.replace("Thu", "(木)")
elif dta == "Fri":
jdta = dta.replace("Fri", "(金)")
elif dta == "Sat":
jdta = dta.replace("Sat", "(土)")
elif dta == "Sun":
jdta = dta.replace("Sun", "(日)")
else:
jdta="ー"
data = bme280.sample(bus, address, calibration_params)
pbme280 = f"{data.pressure:.1f},{data.temperature:.1f},{data.humidity:.1f}"
pcsv = dtd + jdta + " " + dtt + "," + str(outco2) + "," + pbme280
#Run connect_iot()
asyncio.run(connect_iot())
# Draw a black filled box to clear the image.
draw.rectangle((0, 0, width, height), outline=0, fill=(0, 0, 0))
draw.text((50, 192), dtd, font=font2, fill="#00ff00")
draw.text((50, 212), dtt, font=font2, fill="#00ff00")
draw.text((220, 192), jdta, font=font2, fill="#00ff00")
draw.text((20, 3), "温度:", font=font2, fill="#ffffff")
draw.text((24, 25), "( ℃ )", font=font3, fill="#ffffff")
temp =format(float(data.temperature), '.1f')
draw.text((90, 0), temp, font=font1, fill="#ffffff")
draw.text((14, 50), "CO :", font=font2, fill="#FFFF00")
draw.text((47, 58), "2", font=font3, fill="#FFFF00")
draw.text((14, 70), "( ppm )", font=font3, fill="#FFFF00")
co2 =format(float(outco2), '.1f')
draw.text((90, 47), co2, font=font1, fill="#FFFF00")
draw.text((20, 97), "気圧:", font=font2, fill="#ffffff")
draw.text((18, 117), "( hpa )", font=font3, fill="#ffffff")
pres =format(float(data.pressure), '.1f')
draw.text((90, 93), pres, font=font1, fill="#ffffff")
draw.text((20, 143), "湿度:", font=font2, fill="#ffffff")
draw.text((24, 165), "( ％ )", font=font3, fill="#ffffff")
humi =format(float(data.humidity), '.1f')
draw.text((90, 140), humi, font=font1, fill="#ffffff")
disp.image(image)
time.sleep(2)
Azure IoT Python SDK をインストールします。
pip3 install azure-iot-device
対象のスクリプトを実行します。
あとはAzure Iot Explorerで確認します。
以上でデータの送信は完了です。
まとめ
データの送信はいたってシンプルに行えます。ただし、スクリプトに直接接続文字列を書き込むことは基本的にセキュリティのリスクでしかないので検証目的以外では行わないでください。他の手段を検討してください。