AWSのサービスで自席の快適度を測定できるようにしてみた話
2025年09月04日 木曜日
CONTENTS
はじめに
クラウドソリューション部 ソリューション１課の瀬川です。2023年にIIJに入社し、クラウドソリューションの開発や、クラウド導入力の向上に関する活動などに携わっています。
2024年後半からAWS IoT Coreなどに関わることが増えてきました。業務的にはAWS側のシステムのみを設計・構築することが主であり、物理機器本体や通信に直接関わることはないためいまいちAWS IoT Coreについて理解できていないのではないかと思っていました。手軽に入手可能な実環境のデータとして自席付近の室温・湿度・無線LANの電波強度を収集し可視化することでサービスへの理解を深めていきたいと思います。
今回は普段業務で触れることの無い物理機器の設定を含めて以下を記事にしました。
- Raspberry Pi Pico WHにプログラムを書き込んでMQTT通信を行う
- AWS IoT SiteWise Monitorで自席付近の室温などを可視化する
構成図
Raspberry Pi Picoからの通信をAWS IoT Coreで受けてAWS IoT SiteWiseに流す一般的な構成です。
アラームについては、CloudWatchなどを利用して外部アラーム処理を手組で設定しています。
（2026年にAWS IoT Events アラームのサポート終了が予告されており、新規アカウントでは既に当該機能が利用できなくなっているためアラーム機能を使用したい場合は外部アラームを設定する必要があります。）
準備したもの
物理機器
ファームウェア/ライブラリ/開発環境など
- MicroPython
- MicroPython remote control: mpremote
- micropython-lib/micropython/umqtt.robust at master · micropython/micropython-lib · GitHub
- micropython-lib/micropython/umqtt.simple at master · micropython/micropython-lib · GitHub
- Thonny, Python IDE for beginners
AWS IoT Core設定
エンドポイントの設定
まず初めにデータ送信先となるエンドポイントを作成します。「ドメイン設定」画面にて作成します。
認証タイプのみ「X.509 証明書」に変更しそれ以外はすべてデフォルト設定とします。
EXAMPLE-ats.iot.<リージョン>.amazonaws.com
上記のような形式でエンドポイントが作成されるのでメモします。
モノ（Thing）の作成
次にモノを作成します。デフォルト値で先に進みます。
Raspberry Pi Pico用のポリシーをアタッチします。
今回はAWSアカウントに対して1つのモノしか存在しないため、かなり緩いポリシーにしています。
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"iot:Publish",
"iot:Receive",
"iot:PublishRetain"
],
"Resource": [
"arn:aws:iot:ap-northeast-1:123456789012:*"
]
},
{
"Effect": "Allow",
"Action": "iot:Subscribe",
"Resource": [
"arn:aws:iot:ap-northeast-1:123456789012:*"
]
},
{
"Effect": "Allow",
"Action": "iot:Connect",
"Resource": [
"arn:aws:iot:ap-northeast-1:123456789012:*"
]
}
]
}
デバイス証明書の生成・ダウンロード
モノを作成するとモノに紐づく証明書とAWSのルートCA証明書のダウンロード画面がポップアップします。
以下4つをダウンロードします。
- デバイス証明書（*-certificate.pem.crt）
- プライベートキー（*-private.pem.key）
- パブリックキー（*-public.pem.key）
- ルートCA証明書（AmazonRootCA1.pem）
証明書ファイルの変換
MicroPython環境ではAWSからダウンロードした証明書をそのまま使うことが出来ないため、OpenSSLで利用可能な形式に変換します。
（改行コードなどRaspberry Pi Pico側がシビアなため、WSL2などLinux環境で実行する方が無難です）
openssl pkey -inform PEM -in *-private.pem.key -outform DER -out private.key.der openssl x509 -outform DER -in *-certificate.pem.crt -out certificate.crt.der openssl x509 -outform DER -in AmazonRootCA1.pem -out AmazonRootCA1.der
送信プログラムなどの配置
以下のPythonコードと設定ファイルおよび変換後の証明書ファイルを配置します。
(MicroPythonの実行環境、ライブラリの配置については割愛します。)
import network
import utime
import ujson
import ntptime
import ssl as tls
from machine import Pin, reset
from umqtt.robust import MQTTClient
import dht
# --- デバッグ設定 ---
# テスト中はTrueに、それ以外にはFalseに変更する
DEBUG_MODE = True
def debug_print(message):
"""デバッグモードが有効な場合のみメッセージを出力する"""
if DEBUG_MODE:
print(message)
# --- メインクラス ---
class IoTDevice:
"""
センサーデータの収集とAWS IoTへの送信を管理するデバイスクラス。
"""
# --- クラス定数 ---
LED_PIN_NAME = "LED"
PATTERN = {
"SUCCESS": [50, 50],
"WARNING": [50, 50] * 3,
"RETRY": [200, 200] * 2
}
MQTT_PORT = 8883
MQTT_KEEPALIVE = 3600
CERT_BASE_PATH = '123456789012/'
MAX_INIT_RETRIES = 5
MAX_LOOP_RETRIES = 5
def __init__(self, config_path='config.json'):
"""デバイスを初期化する。"""
# スタンドアロン実行時の安定性向上のため、ハードウェアが安定するのを待つ
utime.sleep(3)
self.led = Pin(self.LED_PIN_NAME, Pin.OUT)
self.led.on() # 処理開始を通知
self.config = self._load_config(config_path)
self.wlan = network.WLAN(network.STA_IF)
self.dht_sensor = dht.DHT22(Pin(self.config['DHT_PIN']))
self.mqtt_client = None
self.error_count = 0
ntptime.host = "ntp.nict.jp"
def _load_config(self, path):
"""設定ファイルを読み込み、必須キーを検証する。"""
try:
with open(path) as f:
config = ujson.load(f)
required_keys = ['SSID', 'PASSWORD', 'CLIENT_ID_BASE', 'AWS_ENDPOINT',
'DEVICE_ID', 'DHT_PIN', 'LOOP_INTERVAL_SEC', 'RETRY_INTERVAL_SEC']
if not all(key in config for key in required_keys):
raise ValueError("設定ファイルに必須キーが不足しています。")
return config
except (OSError, ValueError) as e:
debug_print("設定ファイルエラー: {}".format(e))
self._handle_fatal_error()
def _signal_led(self, pattern, repetitions=1):
"""指定されたパターンでLEDを点滅させる。"""
for _ in range(repetitions):
for i, duration in enumerate(pattern):
self.led.value(1 if i % 2 == 0 else 0)
utime.sleep_ms(duration)
self.led.off()
def _handle_fatal_error(self):
debug_print("復旧不可能なエラー。5秒後にリセットします。")
error_pattern = [100, 100] * 10
for _ in range(5):
self._signal_led(error_pattern)
utime.sleep_ms(800)
reset()
def _connect_wifi(self):
debug_print("Wi-Fiに接続中...")
self.wlan.active(True)
self.wlan.connect(self.config['SSID'], self.config['PASSWORD'])
for _ in range(20):
if self.wlan.isconnected():
debug_print("Wi-Fi接続成功。 IP: {}".format(self.wlan.ifconfig()[0]))
return
utime.sleep(1)
debug_print("Wi-Fi接続に失敗しました。")
self._handle_fatal_error()
def _sync_time(self):
debug_print("NTPと時刻同期中...")
for _ in range(self.MAX_INIT_RETRIES):
try:
ntptime.settime()
now = utime.localtime()
debug_print("NTP同期成功: {}/{}/{} {}:{}:{}".format(now[0], now[1], now[2], now[3], now[4], now[5]))
return
except Exception as e:
debug_print("NTP同期リトライ... ({})".format(e))
utime.sleep(2)
debug_print("NTP時刻同期に失敗しました。")
self._handle_fatal_error()
def _init_mqtt_client(self):
debug_print("MQTTクライアントを初期化中...")
client_id = "{}_{}".format(self.config['CLIENT_ID_BASE'], self.config['DEVICE_ID'])
try:
context = tls.SSLContext(tls.PROTOCOL_TLS_CLIENT)
context.load_cert_chain(
certfile=self.CERT_BASE_PATH + 'certificate.crt.der',
keyfile=self.CERT_BASE_PATH + 'private.key.der'
)
context.load_verify_locations(cafile=self.CERT_BASE_PATH + 'AmazonRootCA1.der')
self.mqtt_client = MQTTClient(
client_id=client_id,
server=self.config['AWS_ENDPOINT'],
port=self.MQTT_PORT,
keepalive=self.MQTT_KEEPALIVE,
ssl=context
)
self.mqtt_client.connect()
debug_print("MQTT接続成功。")
except Exception as e:
debug_print("MQTT初期接続エラー: {}".format(e))
self._handle_fatal_error()
def _read_sensor_data(self):
"""DHTセンサーから温湿度を読み取る。失敗した場合はNoneを返す。"""
for _ in range(3):
try:
self.dht_sensor.measure()
temp = self.dht_sensor.temperature()
hum = self.dht_sensor.humidity()
if -40 <= temp <= 80 and 0 <= hum self.MAX_LOOP_RETRIES:
self._handle_fatal_error()
utime.sleep(self.config['RETRY_INTERVAL_SEC'])
# --- プログラムのエントリーポイント ---
if __name__ == "__main__":
try:
device = IoTDevice()
device.run()
except Exception as e:
debug_print("致命的な初期化エラーが発生しました: {}".format(e))
utime.sleep(5)
reset()
{
"SSID": "適切なSSID",
"PASSWORD": "パスワード",
"CLIENT_ID_BASE": "クライアント識別子のプレフィックスを入力します",
"AWS_ENDPOINT": "<適切なエンドポイント>-ats.iot.<適切なリージョン>.amazonaws.com",
"DEVICE_ID": "適切なデバイスIDを入力します",
"DHT_PIN": 5,
"LOOP_INTERVAL_SEC": 90,
"RETRY_INTERVAL_SEC": 30
}
Raspberry Pi Picoに各種資材を配置した後に必要に応じてMQTTクライアントを使用して実際に通信できているか確認します。
LEDの発光パターンの場合、以下で判断できます。
- LEDが点灯している = 接続できていない
- LEDが点滅している = 接続中 or 一部でエラーが発生している
- LEDが消灯している = エラーなくプログラムが実行されている
参考：MQTT クライアントで AWS IoT MQTT メッセージを表示する – AWS IoT Core
AWS IoT Core , AWS IoT SiteWise連携設定
IoT Coreルールの作成
Raspberry Pi Pico Wから送信されてくるデータのうち必要な情報をSQL形式でクエリし、SiteWiseに転送するルールを作成します。
SQL ステートメント
SELECT * FROM 'device/+/data'
ルールアクション
「AWS IoT SiteWise のアセットプロパティにメッセージデータを送信」を選択して以下を設定します。
- エントリ1(プロパティエイリアス別)
- プロパティエイリアス：/pico/${topic(2)}/temperature
- プロパティの値
- 時間 (秒)：${timestamp}
- データ型：DOUBLE
- 値：${external_temperature}
- エントリ2(プロパティエイリアス別)
- プロパティエイリアス：/pico/${topic(2)}/humidity
- プロパティの値
- 時間 (秒)：${timestamp}
- データ型：DOUBLE
- 値：${humidity}
- エントリ3(プロパティエイリアス別)
- プロパティエイリアス：/pico/${topic(2)}/rssi
- プロパティの値
- 時間 (秒)：${timestamp}
- データ型：DOUBLE
- 値：${rssi}
- IAMロール：新しいロール（「AWS IoT CoreからメッセージをIoT SiteWiseに転送」、「Lambdaの呼び出し」が出来るものを設定します。）
AWS IoT SiteWiseの設定
モデル
IoT Core側のデータ転送設定に合わせてSiteWise側でもデータを受信するための設定を行います。
今回は3つのエントリを設定しているので、それに合わせた測定値の設定を作成します。（データ型については今回はすべてDOUBLEとなります。）
ここで設定する名前、単位はSiteWiseモニターの表示値として使用されます。
アセット
モデルでデータの名前とデータ型を定義した後にアセット上でIoT Coreから送信されるデータとモデルを紐づけます。
（IoT Coreで設定したプロパティエイリアスと対応するように入力する必要があります。）
IoT Coreの設定値：/pico/${topic(2)}/temperature
SiteWiseの設定値：/pico/e66430a64b9f5b35/temperature
今回のケースでは${topic(2)}が変数値になり、当該部分にソースコードと共にRaspberry Pi Picoに書き込んだConfigに含まれるDEVICE_IDが設定されます。
ポータルを作成
上記まで実施出来たら「ポータルを作成」からポータルを作成し生成されるリンクにアクセスします。ここで生成されるリンクは「.aws」ドメインとなり変更不可です。
作成されたポータルでプロジェクトとダッシュボードを作成した結果が以下になります。事前に設定したアセットやモデルの状態を元に設定を行います。
データがある程度蓄積されると以下のように室温の推移などを知ることが出来るようになりました。
おわりに
自席付近の温度湿度などを計測する環境を構築してみて一番感じたことは、AWS IoT SiteWiseは癖があるもののセンサー用のデータソースとしてかなり使いやすいということでした。
時系列データベースとしては、Amazon Timestream for InfluxDBなどもありますが記事作成時点ではAWS IoT Coreとの接続が若干面倒であったり、RDSのようなパフォーマンスチューニングの必要性がありそうに見えたりとハードルが若干あります。AWS IoT SiteWiseはその点でフルマネージドなサービスであるため考慮点が少ないところが非常に良く感じました。
Apache Icebergなどもセンサーデータの保存先として有力と言われていますので、マネージドサービス・セルフホスト含め今後そちらも検証したいと思います。
