macOS環境のPythonでSwitchBot防水温湿度計の値を自動記録しDiscordへ通知する方法について解説します。
はじめに
「PythonのBLE通信ライブラリ（bleak）」と「Discord Webhook API」、macOSの「LaunchAgent（launchd）」を組み合わせることで、SwitchBot防水温湿度計のデータを10分ごとに自動収集し、時系列グラフとともにDiscordへ定期通知するシステムを構築できます。
SwitchBot防水温湿度計
本システムでは、SwitchBot防水温湿度計（W3400010 / WoIOSensorTH）から発信されるBluetooth Low Energy（BLE）の電波をPythonスクリプトで受信し、ローカルのCSVファイルへ10分間隔で記録します。記録完了と同時に過去24時間・1週間・1ヶ月のグラフ画像を生成し、Discord Webhookへ自動投稿します。
BLEパケット解析
SwitchBot防水温湿度計（内部モデル名: WoIOSensorTH）は、温湿度データを含むManufacturer Dataフレームと、バッテリー残量を含むService Dataフレームを時間差で個別にブロードキャストします。欠損のないデータを取得するには、MACアドレス（またはシステムアドレス）単位で内部キャッシュを保持し、両フレームの受信データをマージする必要があります。
BLEアドバタイズパケットのデータ構造
パケット内の各バイト位置とデコード仕様は以下の通りです。
| 項目 | 参照パケット領域 | バイト位置・デコード仕様 |
|---|---|---|
| メーカー識別子 | Manufacturer Data | Company ID: 2409（0x0969 / Woan Technology） |
| MACアドレス | Manufacturer Data | 0〜5バイト目（ハードウェア固有のMACアドレス） |
| 温度（小数部） | Manufacturer Data | 8バイト目（下位4ビット × 0.1 ℃） |
| 温度（整数部・符号） | Manufacturer Data | 9バイト目（最上位ビットが符号、下位7ビットが整数部） |
| 湿度 | Manufacturer Data | 10バイト目（下位7ビット、単位: %） |
| モデル識別子 | Service Data（UUID: 0xFD3D） | 0バイト目（下位7ビットがASCII文字 ‘w’ または ‘W’） |
| バッテリー残量 | Service Data（UUID: 0xFD3D） | 2バイト目（下位7ビット、単位: %） |
個体識別と近隣電波の混信対策
BLEアドバタイズパケットはペアリングを行わず周囲数十メートルへ暗号化なしでブロードキャストされるため、集合住宅や住宅密集地では近隣住民が所有するSwitchBot製品の電波を受信することがあります。また、SwitchBot防水温湿度計の本体外装にはMACアドレスの印字が存在しません。自機を確実に特定するための手法は以下の3つです。
| 手法 | 手順 | 特徴および適用場面 |
|---|---|---|
| 順次電源投入法 | 絶縁シートを1台ずつ抜き、その都度検出されたMACアドレスを記録してラベリングする | 最も誤認が少なく確実であり、新規導入・初期セットアップ時に適した手法 |
| 環境変化法 | 稼働中の機器のセンサー孔付近に息を吹きかけ、温湿度が急上昇したMACアドレスを特定する | 既に複数台が稼働している環境や近隣電波が混在する環境で物理個体を判別可能 |
| 電波強度（RSSI）近接法 | 受信端末の至近距離に対象機器を置き、RSSIが極めて高い（-40 dBm前後）アドレスを特定する | 機器を物理的に隔離できない場合に有効だが、電波の反射による誤差に留意が必要 |
BLEデータ取得スクリプトの実装
動作に必要なライブラリを定義した requirements.txt を作成します。
bleak
requests
pandas
matplotlib
以下のコマンドで依存ライブラリをインストールします。
pip install -r requirements.txt
パケット集約とデコードを行う switchbot_meter_scanner.py の実装コードは以下の通りです。
from __future__ import annotations
import argparse
import asyncio
import json
import sys
from datetime import datetime
from typing import Any, Optional
from bleak import BleakScanner
from bleak.backends.device import BLEDevice
from bleak.backends.scanner import AdvertisementData
SWITCHBOT_MANUFACTURER_ID = 2409 # 0x0969 (Woan Technology)
SWITCHBOT_SERVICE_UUID = "0000fd3d-0000-1000-8000-00805f9b34fb"
class SwitchBotMeterAggregator:
def __init__(self) -> None:
self.devices: dict[str, dict[str, Any]] = {}
def update(
self,
device: BLEDevice,
advertisement_data: AdvertisementData,
) -> Optional[dict[str, Any]]:
mfr_data = advertisement_data.manufacturer_data.get(SWITCHBOT_MANUFACTURER_ID)
service_data = advertisement_data.service_data
s_data: Optional[bytes] = None
for uuid, val in service_data.items():
if "fd3d" in uuid.lower():
s_data = val
break
if not mfr_data and not s_data:
return None
hw_mac = None
if mfr_data and len(mfr_data) >= 6:
hw_mac = ":".join(f"{b:02X}" for b in mfr_data[:6])
primary_id = device.address.upper()
if primary_id not in self.devices:
self.devices[primary_id] = {
"device_name": device.name or "Unknown",
"mac_address": hw_mac,
"system_address": device.address,
"model": "SwitchBot Device",
"temperature": None,
"humidity": None,
"battery": None,
"rssi": advertisement_data.rssi,
"timestamp": datetime.now().isoformat(),
}
entry = self.devices[primary_id]
if device.name and entry["device_name"] == "Unknown":
entry["device_name"] = device.name
if hw_mac and not entry["mac_address"]:
entry["mac_address"] = hw_mac
entry["rssi"] = advertisement_data.rssi
entry["timestamp"] = datetime.now().isoformat()
if s_data and len(s_data) >= 1:
dev_char = chr(s_data[0] & 0x7F)
if dev_char in ("w", "W"):
entry["model"] = "WoIOSensorTH (Waterproof Meter)"
elif dev_char in ("t", "T"):
entry["model"] = "WoSensorTH (Meter)"
elif dev_char in ("i", "I"):
entry["model"] = "WoSensorTH (Meter Plus)"
if device.name and "WoIOSensorTH" in device.name:
entry["model"] = "WoIOSensorTH (Waterproof Meter)"
if s_data and len(s_data) >= 3:
entry["battery"] = s_data[2] & 0x7F
if mfr_data and len(mfr_data) >= 11:
temp_dec = (mfr_data[8] & 0x0F) * 0.1
is_positive = bool(mfr_data[9] & 0x80)
temp_int = mfr_data[9] & 0x7F
temp = temp_int + temp_dec
entry["temperature"] = round(temp if is_positive else -temp, 1)
entry["humidity"] = mfr_data[10] & 0x7F
elif s_data and len(s_data) >= 6:
is_positive = bool(s_data[3] & 0x80)
temp_int = s_data[3] & 0x7F
temp_dec = (s_data[4] & 0x0F) * 0.1
temp = temp_int + temp_dec
entry["temperature"] = round(temp if is_positive else -temp, 1)
entry["humidity"] = s_data[5] & 0x7F
return dict(entry)
async def scan_once(
target_mac: Optional[str] = None,
timeout: float = 12.0,
) -> Optional[dict[str, Any]]:
aggregator = SwitchBotMeterAggregator()
result: Optional[dict[str, Any]] = None
stop_event = asyncio.Event()
def callback(device: BLEDevice, advertisement_data: AdvertisementData) -> None:
nonlocal result
current = aggregator.update(device, advertisement_data)
if not current:
return
if target_mac:
target = target_mac.upper().replace("-", ":")
mac = (current.get("mac_address") or "").upper()
sys_addr = (current.get("system_address") or "").upper()
if target != mac and target != sys_addr:
return
if (
current.get("temperature") is not None
and current.get("humidity") is not None
and current.get("battery") is not None
):
result = current
stop_event.set()
scanner = BleakScanner(detection_callback=callback)
await scanner.start()
try:
await asyncio.wait_for(stop_event.wait(), timeout=timeout)
except asyncio.TimeoutError:
for entry in aggregator.devices.values():
if target_mac:
target = target_mac.upper().replace("-", ":")
mac = (entry.get("mac_address") or "").upper()
sys_addr = (entry.get("system_address") or "").upper()
if target != mac and target != sys_addr:
continue
if entry.get("temperature") is not None:
result = entry
break
finally:
await scanner.stop()
return result
対象機器のMACアドレスを指定して単発取得を行うコマンドと出力例は以下の通りです。
python switchbot_meter_scanner.py --once --mac ＜MACアドレス＞ --json
CSVへのデータ保存とグラフ可視化・Discord連携
取得したデータのローカル保存方式と、グラフ生成およびDiscord Webhook通知の設計を行います。
温湿度グラフの描画仕様
matplotlib を用いて、温度（℃）と湿度（%）の推移を上下に配置した時系列グラフを描画します。macOS環境での文字化けを防ぐためフォントには Hiragino Sans を指定し、凡例には機器名を含めず設置場所名のみを表記します。また、横軸の表示範囲（xlim）はデータ件数に関わらず最新時刻基準の期間幅に固定し、主目盛（破線）と補助目盛（点線）のグリッドを縦横両軸に描画します。
| 出力ファイル名 | 対象期間（横軸固定幅） | 主目盛（破線） | 補助目盛（点線） | 描画の特徴 |
|---|---|---|---|---|
| 自宅環境グラフ_過去24時間.png | 直近24時間（86,400秒） | 2時間間隔 | 1時間間隔 | 測定点マーカーを表示し微細な変化を可視化 |
| 自宅環境グラフ_過去1週間.png | 直近7日間（168時間） | 1日間隔 | 6時間間隔 | 昼夜の日周変動と温湿度の変化を比較 |
| 自宅環境グラフ_過去1ヶ月.png | 直近30日間（720時間） | 5日間隔 | 1日間隔 | マーカーを省略した細線描画で中長期トレンドを把握 |
Discord Webhookへの3画像一括投稿と30日自動削除
Discord Webhook APIの multipart/form-data 仕様を利用し、1回のPOSTリクエストで最新測定値の埋め込み（Embed）と3枚のグラフ画像（files[0]〜files[2]）を同時に送信します。また、チャンネル内の過去ログ肥大化を防ぐため、投稿から30日（720時間 / 2,592,000秒）以上経過したメッセージを自動削除します。
-
WebhookへのPOSTリクエスト時にURLパラメータ
?wait=trueを付与し、レスポンスJSONから発行されたメッセージID（id）を取得します。 -
取得したメッセージIDと投稿タイムスタンプをローカルの
discord_message_history.jsonに保存します。 -
次回以降の定期実行時、現在時刻から2,592,000秒以上経過しているメッセージIDを抽出し、
DELETE /api/webhooks/{webhook.id}/{webhook.token}/messages/{message.id}エンドポイントへリクエストを送信して削除します。
import json
import requests
def send_multiple_graphs_to_discord(webhook_url: str, embeds: list, image_paths: list[str]) -> str:
wait_url = f"{webhook_url}?wait=true"
payload = {"embeds": embeds}
files = {
"payload_json": (None, json.dumps(payload), "application/json")
}
open_files = []
try:
for idx, path in enumerate(image_paths):
f = open(path, "rb")
open_files.append(f)
files[f"files[{idx}]"] = (f"graph_{idx}.png", f, "image/png")
response = requests.post(wait_url, files=files, timeout=30)
response.raise_for_status()
return response.json().get("id")
finally:
for f in open_files:
f.close()
macOSのlaunchdによる10分ごとの定期実行とTCC権限対策
macOS環境で毎時00, 10, 20, 30, 40, 50分のキリの良い時刻にスクリプトを自動実行するため、LaunchAgent（launchd）を設定します。
macOS TCC（Bluetooth権限）
macOSではBluetoothへのアクセスがTCC（Transparency, Consent, and Control）というプライバシー保護機構によって管理されています。launchd から直接非対話型のPythonバイナリを起動した場合、GUIのアクセス許可ダイアログを表示できないため、CoreBluetoothの初期化処理（kevent待機）でプロセスが永久にハングアップし、次回以降のスケジュール実行が多重起動抑止機能によってすべてスキップされる事象が発生します。
この問題を解決し、プロセス滞留を防ぐための設定手順は以下の通りです。
-
macOSの「システム設定」>「プライバシーとセキュリティ」>「Bluetooth」を開きます。
-
一覧に表示されている
python3（またはアプリケーションバンドル化したHomeAmbientLogger.app）のトグルスイッチをオン（青色）に切り替えます。 -
Pythonスクリプト側では、BLEスキャンのタイムアウトを12秒に設定し、CSV書き込みと通知処理が完了した直後に
os._exit(0)を呼び出してOSレベルでプロセスを即座に終了させます。
LaunchAgent（plist）の登録と動作確認
~/Library/LaunchAgents/com.user.homeambient.logger.plist にスケジュール設定を作成し、StartCalendarInterval で毎時00, 10, 20, 30, 40, 50分を指定します。作成後、以下のコマンドでジョブをロードして稼働状況を確認します。
launchctl load ~/Library/LaunchAgents/com.user.homeambient.logger.plist
cat /Users/＜ユーザー名＞/Library/Logs/HomeAmbient/stdout.log
cat /Users/＜ユーザー名＞/Library/Logs/HomeAmbient/stderr.log
まとめ
本記事では、SwitchBot防水温湿度計のBLEパケット解析、温湿度推移グラフの自動描画、そしてmacOSの launchd とDiscord Webhookを用いた自動通知・ローテーションシステムの構築手順を解説しました。
-
SwitchBot防水温湿度計はManufacturer DataとService Dataをマージして温湿度と電池残量を取得する。
-
データの蓄積には高速かつ堅牢なCSV形式を採用し、期間別の3枚のグラフ（24時間・1週間・1ヶ月）を自動生成する。
-
Discord Webhookの
?wait=trueとDELETEエンドポイントを活用することで、追加権限なしで30日経過後のメッセージを自動削除できる。 -
macOSでBLEスキャンを定期実行する際は、システム設定でBluetooth権限を明示的に許可し、
os._exit(0)でプロセスの滞留を防止する。
この仕組みを導入することで、室内の温湿度変化を常時モニタリングできるようになります。
関連コンテンツ
コメント