关于HikiMu-EEW
HikiMu-EEW地震预警系统,依托于福建地震局提供的API,致力于为用户提供及时、准确的地震预警服务。该系统通过实时监测地震局发布的数据,一旦检测到地震活动,便会立即启动预警机制
由于我们目前没有比较靠谱的算法,仅根据假设地震波在地壳中的平均速度为6.8公里/秒
利用哈弗辛公式(Haversine formula)来计算震波抵达的剩余时间。由于计算本地预估烈度的算法比较复杂,故预览图的"[颜色]"仅代表"震源
"级别[蓝色]、[黄色]、[橙色]、[红色]。且不参考深度信息(因为没有提供)
由于沿海地震速报主要用于中国台湾和大陆华南沿海地带,速报平台由CWA(台湾中央气象署)和FJEA(福建地震局)提供不同地区的地震速报内容;目前大陆地区虽然可以采用地牛Wake Up!、DPIP、TREMV等一系列由台湾地区制作的速报系统;且对于台湾地区的地震速报比大陆地区快
福建地震局是自主研发的一个地震速报系统;在以往的参考中,虽然他们的观测站也有部署在台湾地区;但对于台湾的地震有些许的漏报等...
也因此,我们所使用的API是福建地震局,虽然说对于台湾地区有些许的漏报,但对于大陆方面是没有影响的。但同时,我们的API也同时接入了台湾地震局的API,以便于提供更加准确的地震速报信息。
需要说明的是,虽然福建地震局的观测站也部署在台湾地区,并对台湾地区的地震进行监测,但由于地震活动的复杂性和不可预测性,对于台湾地区的地震可能存在些许的漏报情况。然而,对于大陆地区的地震预警,FJEA(福建地震局)具有高度的准确性和可靠性。
此项目开始的时间是2024年03月04日
钉钉和飞书地震预警推送工作原理
预览图(图片仅供参考)
后端相关代码(Python),仅小范围使用。
import requests
import json
import time
import threading
from datetime import datetime
import math
# 飞书Webhook URL
FEISHU_WEBHOOK = "https://open.feishu.cn/open-apis/bot/v2/hook/"
# 钉钉Webhook URL
DINGTALK_WEBHOOK = "https://oapi.dingtalk.com/robot/send?access_token="
# 本地文件路径
API_FILE_PATH = "api\\api1.json"
LOCAL_FILE_PATH = "api\\api2.json"
# 地球半径,单位:公里
EARTH_RADIUS = 6371
# 地震波在地壳中的平均速度,单位:公里/秒
WAVE_SPEED = 6.8
# 目标地点的经纬度
TARGET_LONGITUDE =
TARGET_LATITUDE =
last_api1_data = None
last_api2_data = None
def fetch_api1_data():
global last_api1_data
try:
with open(API_FILE_PATH, 'r', encoding='utf-8') as f:
data = json.load(f)
if data and isinstance(data, dict):
return data
except (FileNotFoundError, json.JSONDecodeError) as e:
print(f"读取错误: {e}")
return None
def fetch_api2_data():
global last_api_data
try:
with open(LOCAL_FILE_PATH, 'r', encoding='utf-8') as f:
data = json.load(f)
if data and isinstance(data, list) and len(data) > 0:
return data[0]
except (FileNotFoundError, json.JSONDecodeError) as e:
print(f"读取错误: {e}")
return None
def calculate_intensity_and_time(longitude, latitude, magnitude):
dlon = math.radians(TARGET_LONGITUDE - longitude)
dlat = math.radians(TARGET_LATITUDE - latitude)
a = math.sin(dlat / 2) ** 2 + math.cos(math.radians(TARGET_LATITUDE)) * math.cos(math.radians(latitude)) * math.sin(dlon / 2) ** 2
c = 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))
distance = EARTH_RADIUS * c
arrival_time_seconds = distance / WAVE_SPEED
return magnitude, arrival_time_seconds
def send_webhook(message):
headers = {'Content-Type': 'application/json'}
# 输出发送消息到控制台
print(f"发送消息到飞书: {message}") # 添加打印
response = requests.post(FEISHU_WEBHOOK, json=message, headers=headers)
if response.status_code != 200:
print(f"发送失败到飞书,状态码:{response.status_code}")
def send_dingtalk_webhook(message):
headers = {'Content-Type': 'application/json'}
dingtalk_message = {
"msgtype": "text",
"text": {
"content": message['content']['text']
}
}
print(f"发送消息到钉钉: {dingtalk_message}")
response = requests.post(DINGTALK_WEBHOOK, json=dingtalk_message, headers=headers)
if response.status_code != 200:
print(f"发送失败到钉钉,状态码:{response.status_code}")
def update_and_send_api1(earthquake):
global last_api1_data
if earthquake == last_api1_data:
return
try:
magnitude = earthquake.get('magnitude', 0)
shock_time_str = earthquake.get('shockTime', '1970-01-01 00:00:00.0')
shock_time = datetime.strptime(shock_time_str[:-2], "%Y-%m-%d %H:%M:%S")
place_name = earthquake.get('placeName', '未知地点')
longitude = earthquake.get('longitude', 0)
latitude = earthquake.get('latitude', 0)
magnitude, arrival_time_seconds = calculate_intensity_and_time(longitude, latitude, magnitude)
arrival_time_string = f"{arrival_time_seconds:.2f}秒"
color_message = ""
if magnitude >= 7.0:
color_message = "[🟥红色🟥]"
elif magnitude >= 6.0:
color_message = "[🟧橙色🟧]"
elif magnitude >= 5.0:
color_message = "[🟨黄色🟨]"
elif magnitude >= 3.0:
color_message = "[🟦蓝色🟦]"
message = {
"msg_type": "text",
"content": {
"text": f"地震速报{color_message} \n发震时间:{shock_time}\n{arrival_time_string}后抵达\n震源地:{place_name}\n震源震级:{magnitude}"
}
}
send_webhook(message)
send_dingtalk_webhook(message)
last_api1_data = earthquake
except Exception as e:
print(f"{e}")
def update_and_send_api2(earthquake):
global last_api2_data
if earthquake == last_api2_data:
return
try:
eq = earthquake.get('eq', {})
magnitude = eq.get('mag', 0)
eq_time = eq.get('time', 0)
eq_time_str = datetime.fromtimestamp(eq_time / 1000).strftime("%Y-%m-%d %H:%M:%S") if eq_time else None
place_name = eq.get('loc')
longitude = eq.get('lon', 0)
latitude = eq.get('lat', 0)
magnitude, arrival_time_seconds = calculate_intensity_and_time(longitude, latitude, magnitude)
arrival_time_string = f"{arrival_time_seconds:.2f}秒"
color_message = ""
if magnitude >= 7.0:
color_message = "[🟥红色🟥]"
elif magnitude >= 6.0:
color_message = "[🟧橙色🟧]"
elif magnitude >= 5.0:
color_message = "[🟨黄色🟨]"
elif magnitude >= 3.0:
color_message = "[🟦蓝色🟦]"
message = {
"msg_type": "text",
"content": {
"text": f"地震速报{color_message} \n发震时间:{eq_time_str}\n{arrival_time_string}后抵达\n震源地:{place_name}\n震源震级:{magnitude}"
}
}
send_webhook(message)
send_dingtalk_webhook(message)
last_api2_data = earthquake
except Exception as e:
print(f"更新和发送时发生错误: {e}")
def fetch_and_update_ap1():
while True:
earthquake_data = fetch_api1_data()
if earthquake_data:
update_and_send_api1(earthquake_data)
time.sleep(2)
def fetch_and_update_api2():
while True:
earthquake_data = fetch_api2_data()
if earthquake_data:
update_and_send_api2(earthquake_data)
time.sleep(2)
thread_api1 = threading.Thread(target=fetch_and_update_api1)
thread_api2 = threading.Thread(target=fetch_and_update_api2)
thread_api1.daemon = True
thread_api2.daemon = True
thread_api1.start()
thread_api2.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
print("程序已停止")
ESP32地震观测站
图中含加速度监测器和地震预警系统
HikiMu地震预警系统 | HikiMu-EEW (福建地震预警)(硬件版)
说明:基于福建地震预警API打造出 地震预警系统,仅用于个人使用,不提供技术支持。
引脚设定
灯类
RED_LED(红灯) = PIN 22
ORANGE_LED(橙色) = PIN 23
YELLOW_LED(黄灯) = PIN 21
BLUE_LED(蓝灯) = PIN 18
WIFI_LEN(白色) = 15
BUZZER_PIN(有源蜂鸣器) = 5
FJEEW API (HikiMu-EEW中转) = "
目前只能提供硬件的信息,不能提供代码(程序不稳定-持续开发中-不开源)
http://119.29.227.6:8266/fjea_list.json
"
元器件:
ESP32 (ESP-WROOM-32) 23.84
ESP32 30Pin扩展板 6.10
有源蜂鸣器 2.62
若干跳线
若干LED(具有红、黄、橙、蓝)
开发成本 35(根据需求)
预计成本:(618淘宝价格) 67.56 CNY
关于HikiMu-EEW网站
HikiMu-EEW WEB是综合地震速报和地震观测站一个网页。波形图仅供参考;没有声音; 此项目持续开发中。
如何获取速报信息
1.钉钉平台(个人开发)
2.飞书平台(个人开发)
3.福建地震局官方微信推送小程序(推荐)