Google Homeで緊急地震速報！テレビより数秒早く家族に知らせるスマートホーム防災システムの作り方

pip install websockets

{
    "home": {
        "name": "XX市XX区",
        "latitude": 35.xxx,
        "longitude": 139.xxx
    },
    "notify_conditions": [
        {"distance_km":  50, "min_magnitude": 5.0, "min_intensity": 40},
        {"distance_km": 100, "min_magnitude": 5.5, "min_intensity": 45},
        {"distance_km": 200, "min_magnitude": 6.5, "min_intensity": 45},
        {"distance_km": 300, "min_magnitude": 7.0, "min_intensity": 50}
    ]
}

# -*- coding: utf-8 -*-
__group__ = '安心安全'
__description__ = '緊急地震速報'
# ------ 全Pythonプログラム共通の宣言 ----------------------------
from pathlib import Path
import sys
PROJECT_ROOT = Path(__file__).resolve()
while PROJECT_ROOT.name != "smarthome":
    if PROJECT_ROOT.parent == PROJECT_ROOT:
        raise RuntimeError("smarthome project root not found")
    PROJECT_ROOT = PROJECT_ROOT.parent
sys.path.insert(0, str(PROJECT_ROOT))
from libs.bootstrap import Bootstrap
bootstrap = Bootstrap(__file__)
settings = bootstrap.settings
# ------------------------------------------------------------
import asyncio
import json
import math
import websockets
from influx.v2 import client as influxclient
from influx.v2 import writer as influxwriter
import googlehome
import smarthomelog

# P2P地震情報 WebSocket エンドポイント
P2PQUAKE_WS_URL = 'wss://api.p2pquake.net/v2/ws'

# 受信するEEWコード
CODE_EEW_FORECAST = 556  # 緊急地震速報（予報）
CODE_EEW_WARNING  = 554  # 緊急地震速報（警報）

# 緊急地震速報チャイム音ファイル名
QUAKE_ALERT_AUDIO_FILE = 'quake_alert.mp3'

# WebSocket切断後の再接続待機時間（秒）
RECONNECT_WAIT_SEC = 5

# 通知条件をJSONから読み込む
config = bootstrap.load_local_settings('config/conditions.json')


def _haversine_km(lat1, lon1, lat2, lon2):
    """2点間の大円距離をkmで返す（Haversine公式）"""
    R = 6371
    dlat = math.radians(lat2 - lat1)
    dlon = math.radians(lon2 - lon1)
    a = (math.sin(dlat / 2) ** 2
         + math.cos(math.radians(lat1)) * math.cos(math.radians(lat2))
         * math.sin(dlon / 2) ** 2)
    return 2 * R * math.asin(math.sqrt(a))


# P2P地震情報APIのmaxScale値 → 震度文字列の対応表
_SCALE_LABEL = {-1: '不明', 10: '1', 20: '2', 30: '3', 40: '4',
                45: '5弱', 50: '5強', 55: '6弱', 60: '6強', 70: '7'}

def _scale_str(max_scale):
    return _SCALE_LABEL.get(max_scale, f'不明({max_scale})')


def _should_notify(dist_km, magnitude, max_scale, conditions):
    """距離内で マグニチュード OR 震度 のいずれかが閾値を超えたら通知"""
    for cond in conditions:
        if dist_km <= cond['distance_km']:
            if magnitude >= cond['min_magnitude']:
                return True
            if max_scale != -1 and max_scale >= cond['min_intensity']:
                return True
    return False


async def _process_eew(data, notified_events, config, google_living, google_child, influx, success_log):
    event_id   = data.get('issue', {}).get('eventId', '')
    is_final   = data.get('isFinal', False)
    cancelled  = data.get('cancelled', False)
    earthquake = data.get('earthquake') or {}
    hypocenter = earthquake.get('hypocenter') or {}

    # キャンセルされたEEWはログだけ残して終了
    if cancelled:
        if event_id in notified_events:
            influx.insert_execution_log(__file__, f'EEW キャンセル: eventId={event_id}')
        return

    magnitude   = hypocenter.get('magnitude', 0.0)
    depth       = hypocenter.get('depth', -1)
    region_name = hypocenter.get('name', '不明')
    latitude    = hypocenter.get('latitude', -200)
    longitude   = hypocenter.get('longitude', -200)
    max_scale   = earthquake.get('maxScale', -1)
    if max_scale is None:
        max_scale = -1
    depth_str   = f'{depth}km' if depth >= 0 else '不明'

    # 震源座標が不明な報は処理しない
    if latitude <= -100 or longitude <= -100:
        return

    home    = config['home']
    dist_km = _haversine_km(home['latitude'], home['longitude'], latitude, longitude)
    is_match = _should_notify(dist_km, magnitude, max_scale, config['notify_conditions'])

    log_prefix = (
        f'eventId:{event_id}-isFinal:{is_final}'
        f'-震源地:{region_name}-M:{magnitude}-深さ:{depth_str}'
        f'-最大震度:{_scale_str(max_scale)}-震源距離:{dist_km:.0f}km-通知:{is_match}'
    )

    # 通知条件に合致し、かつ同一eventIdで初回受信の場合に速報を通知
    if is_match and event_id not in notified_events:
        # まず緊急地震速報チャイム音を即時再生（TTSより高速）
        google_living.play_audio(QUAKE_ALERT_AUDIO_FILE)
        google_child.play_audio(QUAKE_ALERT_AUDIO_FILE)
        # 再生が終わるまで待機してから詳細をTTSで読み上げ
        await asyncio.sleep(20)
        message = (
            f'緊急地震速報です。'
            f'{home["name"]}から約{dist_km:.0f}kmの地点で大きな地震が発生しました。'
            f'身の安全を確保してください。'
            f'震源地は{region_name}、マグニチュードは{magnitude}、震源の深さは{depth_str}です。'
        )
        google_living.speak(message)
        google_child.speak(message)
        influx.insert_execution_log(__file__, log_prefix)
        influx.insert_execution_log(__file__, message)
        notified_events[event_id] = {'final_sent': False}
    else:
        influx.insert_execution_log(__file__, log_prefix)

    # 通知済みのeventIdでisFinalがTrueになった場合、確定情報を追加通知
    if event_id in notified_events and not notified_events[event_id]['final_sent'] and is_final:
        message = (
            f'先ほどの地震の情報が確定しました。'
            f'震源地は{region_name}、マグニチュードは{magnitude}、震源の深さは{depth_str}です。'
        )
        influx.insert_execution_log(__file__, message)
        google_living.speak(message)
        notified_events[event_id]['final_sent'] = True

    success_log.write('処理が正常に終了しました。', __file__)


async def _listen_eew(google_living, google_child, influx, success_log, app_err_log):
    # 同一地震イベントの重複通知を防ぐための管理辞書 {eventId: {'final_sent': bool}}
    notified_events = {}

    while True:
        try:
            async with websockets.connect(P2PQUAKE_WS_URL) as ws:
                influx.insert_execution_log(__file__, 'P2P地震情報WebSocket接続完了')
                async for raw_message in ws:
                    try:
                        data = json.loads(raw_message)
                        code = data.get('code')
                        if code in (CODE_EEW_FORECAST, CODE_EEW_WARNING):
                            await _process_eew(
                                data, notified_events, config,
                                google_living, google_child,
                                influx, success_log
                            )
                    except Exception:
                        app_err_log.write_error(sys.exc_info())

        except Exception as e:
            influx.insert_execution_log(
                __file__,
                f'WebSocket接続エラー({e})。{RECONNECT_WAIT_SEC}秒後に再接続します。'
            )

        await asyncio.sleep(RECONNECT_WAIT_SEC)


try:
    influx_client = influxclient
    influx = influxwriter.Writer()
    success_log = smarthomelog.Log(__file__, smarthomelog.LogType.SUCCESS)
    app_err_log = smarthomelog.Log(__file__, smarthomelog.LogType.APPLICATION_ERROR)

    influx.insert_execution_log(__file__, 'Start Earth Quake Alert!')

    # Google Homeを最高優先度（priority=10）で初期化
    google_living = googlehome.Notify(googlehome.GoogleHome.LIVING, priority=10)
    google_child  = googlehome.Notify(googlehome.GoogleHome.CHILD_ROOM, priority=10)

    # P2P地震情報WebSocketをリッスン（無限ループ・自動再接続）
    asyncio.run(_listen_eew(google_living, google_child, influx, success_log, app_err_log))

except Exception as e:
    log = smarthomelog.Log(__file__, smarthomelog.LogType.SYSTEM_ERROR)
    log.write_error(sys.exc_info())

python earth_quake_alert.py

[Unit]
Description=Earth Quake Alert
After=network.target

[Service]
Type=simple
ExecStart=/usr/bin/python3 /home/<username>/projects/smarthome/core/alert/earth_quake/earth_quake_alert.py
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target

sudo systemctl daemon-reload
sudo systemctl enable earth-quake-alert
sudo systemctl start earth-quake-alert