やりたかったこと
自宅にHome Assistant環境を構築しているものの、緊急地震速報（EEW）は結局スマホの通知頼りでした。せっかくGoogle Nest HubやChromecastが家中にあるので、地震が来る前に家中のデバイスに警告を出したいと考え、Home Assistant用のカスタムコンポーネントを自作しました。
- P2P地震情報のWebSocketから緊急地震速報（コード556）を受信
- しきい値以上の震度を検知したら
- 警告画像を動的生成してChromecast/Google Nest系デバイスへキャスト
- 指定した照明をON
- （任意）玄関の鍵を解錠
- 誰も家にいない時は反応しない（無人宅で解錠だけ走るのを防止）
作ったものはこちらに公開しています。
なぜMQTTブローカーを使わなかったか
同じような自作事例では、Node-RED経由やMQTTブローカーを別途立ててP2P地震情報のクライアントを動かす構成をよく見かけます。ただ、うちの環境ではEEW専用に常時稼働のコンテナやブローカーを増やしたくなかったので、Home AssistantのカスタムコンポーネントとしてWebSocketクライアントを直接持たせる方式にしました。aiohttpのWebSocketクライアントをHome Assistantのイベントループ上で動かすだけなので、追加のミドルウェアは不要です。
全体構成
WebSocket受信部分
Home AssistantのConfigEntryのライフサイクルに合わせてaiohttp.ClientSessionを張り、切断時は指数バックオフで再接続します。
async def _run(self) -> None:
backoff = 5
while not self._stopped:
try:
async with self._session.ws_connect(WS_URL, heartbeat=30) as ws:
_LOGGER.info("Connected to P2P Quake WebSocket")
backoff = 5
async for msg in ws:
if msg.type == aiohttp.WSMsgType.TEXT:
self._handle_message(msg.data)
elif msg.type in (aiohttp.WSMsgType.CLOSED, aiohttp.WSMsgType.ERROR):
break
except asyncio.CancelledError:
raise
except Exception as err:
_LOGGER.warning("P2P Quake WebSocket error: %s; retrying in %ss", err, backoff)
if self._stopped:
return
await asyncio.sleep(backoff)
backoff = min(backoff * 2, 60)
受信したメッセージのうちcode == 556（緊急地震速報の発表）のみを処理し、testフラグが立っている訓練・試験配信は設定でオン・オフできるようにしています。これは後述する「本番で動くか検証したい」ときに役立ちます。
キャストはcattをサブプロセスで呼ぶ
Chromecastへのキャストには catt（Cast All The Things）を使っています。当初はPython内で直接pychromecastを叩いていましたが、Home Assistant本体のイベントループ内からpychromecastを直接呼ぶとハングする問題に当たったため、確実に別OSプロセスとして実行されるcattサブプロセス呼び出し方式に切り替えました。
def _run_catt(device_name: str, *args: str) -> "subprocess.CompletedProcess[str]":
cmd = ["catt", "-d", device_name, *args]
return subprocess.run(cmd, capture_output=True, text=True, timeout=CATT_TIMEOUT_SECONDS)
また、Chromecast系デバイスの検出はmDNSベースのため、稀に一時的に見つからないことがあります。これに対処するため、失敗時は最大5回、3秒間隔でリトライする設計にしています。
MAX_CAST_ATTEMPTS = 5
RETRY_WAIT_SECONDS = 3
for attempt in range(1, MAX_CAST_ATTEMPTS + 1):
result = _run_catt(device_name, "cast", image_url)
if result.returncode == 0:
_LOGGER.info("cast_alert: catt cast succeeded on attempt %d", attempt)
return
_time.sleep(RETRY_WAIT_SECONDS)
無人宅での誤作動防止
地震で照明が点くだけなら実害はありませんが、鍵の解錠は「誰もいないのに玄関が開けっぱなしになる」リスクがあります。そのため在宅検知用エンティティ（person.*やdevice_tracker.*など）を設定でき、指定した場合は誰も"home"状態でなければ何もしないようにしています。
def _is_anyone_home(self) -> bool:
presence_entities = self.entry.options.get(CONF_PRESENCE_ENTITIES, DEFAULT_PRESENCE_ENTITIES)
if not presence_entities:
return True
return any(self.hass.states.is_state(entity_id, "home") for entity_id in presence_entities)
「本番で実際に動くか」の検証方法
作って満足するだけでなく、地震という滅多に起きないイベントに対してどう動作確認するかが一番悩ましいポイントでした。整理すると3段階の検証があります。
-
ロジックの単体テスト: 動作確認用の「テスト送信」ボタンを用意し、WebSocket受信をスキップしてダミーデータで画像生成〜キャスト〜照明/解錠までを直接実行できるようにしました
-
WebSocket接続の生存確認: Home Assistantのlogger設定で該当コンポーネントのログレベルをINFOにしておくと、接続確立・再接続・キャスト成功/失敗のログが追えます
logger: default: warning logs: custom_components.eew_alert: info -
受信パイプライン全体を本番と同じ経路で検証: P2P地震情報は本物と同じ形式の「訓練・試験配信」を定期的に配信しています。「訓練・試験配信を無視する」設定を一時的にOFFにしておけば、次にテスト配信が来たタイミングで、WebSocket受信→パース→センサー更新→キャストという実際のコードパスをそのまま検証できます。テストボタンでは再現できない、本番同等の確認ができるのはこの方法だけです
使ってみた所感
- P2P地震情報のWebSocketは想像以上に安定していて、数時間〜十数時間単位で接続維持できています
- Chromecastのキャストはネットワーク環境によってはmDNS探索が不安定になることがあるので、リトライ処理は必須でした
- 「無人宅では解錠しない」のようなガードは、緊急時アクションを自動化する上で忘れずに入れておきたいポイントでした
まとめ
MQTTブローカーなど追加のミドルウェアを増やさず、Home Assistantのカスタムコンポーネントだけで緊急地震速報の受信〜家中のデバイスへの警告表示までを完結させました。コードは以下で公開しています。フィードバックや改善提案があれば、IssueやPull Requestをお待ちしています。