HikiMu慕慕

关于HikiMu-EEW

HikiMu-EEW地震预警系统,依托于福建地震局提供的API,致力于为用户提供及时、准确的地震预警服务。该系统通过实时监测地震局发布的数据,一旦检测到地震活动,便会立即启动预警机制
由于我们目前没有比较靠谱的算法,仅根据假设地震波在地壳中的平均速度为6.8公里/秒
利用哈弗辛公式(Haversine formula)来计算震波抵达的剩余时间。由于计算本地预估烈度的算法比较复杂,故预览图的"[颜色]"仅代表"震源"级别[蓝色]、[黄色]、[橙色]、[红色]。且不参考深度信息(因为没有提供)
由于沿海地震速报主要用于中国台湾和大陆华南沿海地带,速报平台由CWA(台湾中央气象署)和FJEA(福建地震局)提供不同地区的地震速报内容;目前大陆地区虽然可以采用地牛Wake Up!DPIPTREMV等一系列由台湾地区制作的速报系统;且对于台湾地区的地震速报比大陆地区快
福建地震局是自主研发的一个地震速报系统;在以往的参考中,虽然他们的观测站也有部署在台湾地区;但对于台湾的地震有些许的漏报等...
也因此,我们所使用的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是综合地震速报和地震观测站一个网页。波形图仅供参考;没有声音; 此项目持续开发中。