把地震速報接進自家推播:我用 WebSocket 補上延遲那 8 秒

这篇内容暂时只有 繁體中文 版本,以下是原文。

上個月幫一個小型社區管委會做地震告警,環境是 Raspberry Pi 4、Raspbian Bookworm、Python 3.11,接收端是十幾支 Android 手機跟兩顆藍牙喇叭。現象很具體:官方 App 的推播常常比我先收到的資料慢 8 到 15 秒,尖峰時段更久。我要的不是取代官方,而是讓社區廣播早幾秒響,資料量不大,一天幾十則,但延遲很敏感。

做法是抓公開的 WebSocket 地震資訊流,自己算距離再決定要不要推。先用 websockets 接,收到訊息只做兩件事:解析時間與規模、比對社區座標。

import asyncio, json, websockets, math

def dist_km(lat1, lon1, lat2, lon2):
    r = 6371.0
    p1, p2 = math.radians(lat1), math.radians(lat2)
    dp = p2 - p1
    dl = math.radians(lon2 - lon1)
    a = math.sin(dp/2)**2 + math.cos(p1)*math.cos(p2)*math.sin(dl/2)**2
    return 2*r*math.asin(math.sqrt(a))

async def main():
    async with websockets.connect(URL, ping_interval=20) as ws:
        async for raw in ws:
            m = json.loads(raw)
            if m.get("mag", 0) < 3.5:
                continue
            d = dist_km(HOME_LAT, HOME_LON, m["lat"], m["lon"])
            if d < 80:
                await push(f"M{m['mag']} 距 {d:.0f} km")

asyncio.run(main())

為什麼這樣切:門檻設在 M3.5、80 公里內,是因為再小的震動社區廣播只會變成噪音,代價是近距離的極淺層小震會被漏掉,我接受這個取捨。延遲主要來自兩段:資料源本身彙整時間,以及我這端的重連等待。我用 ping_interval=20 搭配外層 while True 重連,斷線後最多等 5 秒再連,實測整體從事件到喇叭出聲約 12 到 20 秒。

會失效的地方不少。第一,WebSocket 斷了不會報錯,只是安靜,我曾有一晚完全沒收到訊息,隔天看 log 才發現卡在握手;後來加了一個每 10 分鐘檢查最後收訊時間的 watchdog。第二,mag 欄位在小震時常常延後補上,先到的是定位而非規模,直接拿它過濾會漏推。第三,社區座標我用的是管委會地址,跟實際建築中心差了幾百公尺,對 80 公里門檻影響不大,但對「最近測站」的判斷就會偏。

你們在自建告警時,是接受固定門檻,還是會依時段調靈敏度?我現在晚上把門檻降到 M3.0,早上再調回來,還在觀察誤報率。

0
赞同度 100% · 0 条评论 · 2 阅读

0 条评论

支持 Markdown · 回复会实时推送给对方

还没有评论,写下第一条。