ทดลองทำระบบแจ้งเตือนแผ่นดินไหว ผ่าน Telegram ด้วย Python

หลายๆคนน่าจะได้รับ SMS แจ้งเตือน หรือแจ้งข้อมูลจาก DDPM (Department of Disaster Prevention and Mitigation. Ministry of Interior. Thailand) ซึ่งภาษาไทยก็คือ กรมป้องกันและบรรเทาสาธารณภัย (ปภ) ไปแล้ว จนกลายเป็นมีมยกใหญ่ รวมถึงคำถามถึงระบบการแจ้งเตือนภัยของประเทศไทย

นอกเหนือจากการสื่อสารที่ทำให้การเข้าถึงข้อมูลและสื่อสังคมออนไลน์ไร้พรมแดนแล้ว ดูเหมือนว่าวิกฤตทางธรรมชาติก็ไม่มีพรมแดนเหมือนกัน จากทั้งอุทกภัย ฝุ่น PM2.5 และล่าสุดแผ่นดินไหวที่พม่าสะเทือนจนตึก สตง. ถล่มที่จตุจักร กรุงเทพฯ

ในส่วนของการแจ้งเตือนแผ่นดินไหวนอกจุดเกิดเหตุ อาจจะทำได้ไม่มากเพราะเมื่อเหตุเกิด คลื่นแผ่นดินไหวที่ส่งผ่านมานั้นเดินทางด้วยความเร็ว 6-8km/s สำหรับ Primary Wave และ 3.5-4.5km/s สำหรับ Secondary Wave ซึ่งในเหตุการณ์ล่าสุด เทียบระยะประมาณ 1000km มาถึงกรุงเทพฯ ก็ใช้เวลาเดินทางประมาณ 3-4 นาทีเท่านั้น แต่ถึงแม้ 3-4 นาทีก็ถือว่ายังเป็นข้อมูลที่มีประโยชน์เพื่อเตรียมตัว หรือกรณีของสึนามินั้น ไม่ว่าเวลาล่วงหน้าจะมากน้อยขนาดไหน ก็ถือว่าช่วยชีวิตคนได้แน่ๆ

วันนี้คงไม่พูดถึงเหตุการณ์ที่ผ่านมามากนัก แต่มาคุยในมุมของ Data ที่สามารถนำมาใช้ได้มากกว่า โดยข้อมูลแผ่นดินไหวเราสามารถติดตามได้จากหลายช่องทางมาก ซึ่งแหล่งข้อมูลหนึ่งที่เชื่อถือได้ก็คือ USGS (United States Geological Survey) ซึ่งมี API ให้ดึงข้อมูลมาใช้ได้เลย

สำหรับรายละเอียดของ API Document – Earthquake Catalog สามารถคลิกที่ลิงค์นี้ได้เลย คราวนี้เรามาทดลองใช้งาน API ผ่าน Python กันดู ซึ่งเราจะทดสอบด้วยกัน 3 รูปแบบด้วยกันคือ ข้อมูลของแผ่นดินไหวย้อนหลัง โดยกำหนดตามกรอบของพื้นที่ การเรียกดู Alert หรือ Pager และสุดท้ายการ Download รายงาน PDF

โดยกรอบของพื้นที่ ที่เราจะใช้ในการส่งข้อมูลเพื่อ Query นั้นผมเลือกกรอบตามรูปด้านบนเลย ซึ่งครอบคลุมจุดที่เคยเป็นต้นตอของสึนามิในปี 2004 ด้วย

ซึ่ง Format Parameter จะเป็นลักษณะดังนี้

region_bbox = {
    "minlatitude": -15,
    "maxlatitude": 30,
    "minlongitude": 80,   
    "maxlongitude": 150
}

สำหรับ Code ที่ใช้ในการเรียก Endpoint ของทาง USGS สำหรับดึงข้อมูลล่าสุดย้อนหลัง 10 รายการ ที่ Magnitude ตั้งแต่ 3.5 ขึ้นไปใครใช้ Jupyter Notebook หรือ Colab ก็สามารถนำไปทดสอบกันได้เลย

#retrive data from USGS

import requests
import pandas as pd
from datetime import datetime, timezone
from zoneinfo import ZoneInfo 
import time

# Define the bounding box for the region
# Format: minlatitude, maxlatitude, minlongitude, maxlongitude
region_bbox = {
    "minlatitude": -15,   # South of Indonesia
    "maxlatitude": 30,    # North of Myanmar
    "minlongitude": 80,   # Andaman Sea
    "maxlongitude": 150   # Includes part of the Pacific
}

# USGS API endpoint
url = "https://earthquake.usgs.gov/fdsnws/event/1/query"

# Define parameters
params = {
    "format": "geojson",
    "starttime": datetime.now(timezone.utc).strftime("%Y-%m-%d"),  # Adjust this as needed
    "orderby": "time",
    "minmagnitude": 3.5,  # Filter smaller ones
    **region_bbox,
    "limit": 10  # Number of earthquakes to retrieve
}

headers = {
    'User-Agent': 'TH_Earthquake_Alert/1.0 (dev@emal.com)' #for info on request
}

def get_earthquake_data():
    last_event_id = None



    # Make request
    response = requests.get(url, params=params, headers=headers)
    data = response.json()

    # Parse data into DataFrame
    earthquakes = []
    for feature in data['features']:
        props = feature['properties']
        geometry = feature['geometry']['coordinates']
        timestamp = props['time'] / 1000
        bkk_time = datetime.fromtimestamp(timestamp, tz=ZoneInfo("Asia/Bangkok"))
        formatted_time = bkk_time.strftime('%Y-%m-%d %H:%M:%S')
        

        earthquakes.append({
        "Time": formatted_time,
        "Place": props['place'],
        "Magnitude": props['mag'],
        "MMI": props.get('mmi'),
        "CDI": props.get('cdi'),
        "Felt Reports": props.get('felt'),
        "Alert Level": props.get('alert'),
        "Tsunami Warning": "Yes" if props.get('tsunami') == 1 else "No",
        "Longitude": geometry[0],
        "Latitude": geometry[1],
        "Depth (km)": geometry[2],
        "Status": props.get('status'),
        "Event Type": props.get('type'),
        "USGS URL": props['url'],
        "Event ID": props['net']+props['code']
    })

    

    df = pd.DataFrame(earthquakes)
    

    print('\n-------------------------------------------')
    print('Latest Earthquake:')
    selected_columns = ['Time', 'Place', 'Magnitude', 'MMI', 'Alert Level', 'Tsunami Warning', 'Longitude', 'Latitude', 'Depth (km)', 'Event ID']
    last_three_rows = df[selected_columns].iloc[0:]
    print(f'{last_three_rows.to_string(index=False)}\n')
    print('-------------------------------------------\n\n')
    


get_earthquake_data()

ผลลัพท์ที่ได้ก็จะออกมาดังนี้

ซึ่งถ้าย้อนไปดูเมื่อสองวันก่อน ตัว Event ID: “us7000pn9s” นี้คือตัวที่ส่งผลกระทบกับประเทศไทยเลย ในส่วนของ Alert Level ที่มีค่าเป็น Red ตรงนี้จะเป็น Point ให้เราสามารถไปดึงข้อมูลในส่วนของ Alert ว่าจะกระทบกับเมืองไหน และลิงค์สำหรับโหลดรายงานเป็น PDF ออกมาได้ ซึ่งจะอ้างถึงโดยใช้ Event ID ในส่วนถัดไป

เมื่อเรารู้ Event ID แล้วในหน้ารายงานหลัก หรือตัวข้อมูลชุดแรกที่ตอบกลับมาจะมี Link URL เพื่อให้สามารถเข้าไปดูรายละเอียดต่างๆได้ เช่นตัว Event ID: “us7000pn9s” Response ที่ตอบในส่วนของ URL กลับมาก็คือ

https://earthquake.usgs.gov/earthquakes/eventpage/us7000pn9s

ซึ่งจะมีข้อมูลให้เราเข้าไปดูรายละเอียดต่างๆ ไม่ว่าจะเป็นส่วนของ แผนที่ , Aftershock Forecast และที่ผมสนใจก็คือในส่วนของ Pager ซึ่งจะใช้สำหรับการส่งเตือนภัยต่างๆ

ในส่วนของ Code ที่ใช้ในการดึง Pager โดยอาศัย Event ID เป็นดังนี้

#access to the cities expose to quake
import requests

event_id = 'us7000pn9s' #<-- bkk cas#


url = f'https://earthquake.usgs.gov/fdsnws/event/1/query?eventid={event_id}&format=geojson'


response1 = requests.get(url)
response1_data = response1.json()


# Assuming 'data' is the JSON you received
cities_url = response1_data['properties']['products']['losspager'][0]['contents']['json/cities.json']['url']
print("URL to cities.json:", cities_url)



response = requests.get(cities_url)
cities_data = response.json()

for city in cities_data.get('onepager_cities', []):
    name = city.get('name')
    country = city.get('ccode')
    pop = city.get('pop', 0)
    mmi = city.get('mmi')

    print(f"{name}, {country} | Pop: {pop:,} | MMI: {mmi} ")

สำหรับ Output ที่ได้ก็จะเป็นลิสต์ เมืองที่ได้รับผลกระทบ จำนวนประชากร และค่า MMI (Modified Mercalli Intensity Scale) ซึ่งเป็น มาตรวัด “ความรุนแรงของแผ่นดินไหวที่รู้สึกได้” โดยอิงจากผลกระทบที่เกิดขึ้นจริงกับผู้คน สิ่งปลูกสร้าง และสิ่งแวดล้อม (ไม่ใช่พลังงานของแผ่นดินไหวแบบ Richter)

โดยค่า MMI มีทั้งหมด 12 ระดับ (I ถึง XII):

X-XII: อาคารถล่ม ดินแตก น้ำพุ่ง รุนแรงมาก

I: แทบไม่รู้สึก

II-III: รู้สึกได้เบาๆ โดยเฉพาะคนอยู่เฉยๆ

IV-V: คนทั่วไปเริ่มรู้สึกได้ ของแขวนไหว

VI-VII: ของตก หล่น ผนังแตกร้าวเล็กน้อย

VIII-IX: อาคารเสียหายรุนแรงบางส่วน

ในส่วนรายงาน PDF สำหรับ Event ID นี้ซึ่งจะเป็น One Page Report ก็สามารถ Download ได้โดยใช้ Event ID นี้เช่นเดียวกัน สำหรับ Code ที่ใช้ในการ Download นั้นก็ตามด้านล่างนี้เลยครับ

#download report
import requests
import os

# Set event_id and URL
event_id = 'us7000pn9s'
url = f'https://earthquake.usgs.gov/fdsnws/event/1/query?eventid={event_id}&format=geojson'

# Step 1: Request event JSON
response1 = requests.get(url)
response1_data = response1.json()

# Step 2: Extract PDF URL
pdf_url = response1_data['properties']['products']['losspager'][0]['contents']['onepager.pdf']['url']

# Step 3: Download PDF
response_pdf = requests.get(pdf_url)

# Step 4: Save to file
filename = f"{event_id}_onepager.pdf"
with open(filename, 'wb') as f:
    f.write(response_pdf.content)

# Step 5: Print saved filename
print("PDF saved as:", os.path.abspath(filename))

รูปร่างหน้าตาของรายงานที่ Download เป็น PDF File มาก็จะมีหน้าตาลักษณะนี้

จริงๆโปรแกรมอื่นๆ ที่แสดงรายการแผ่นดินไหวล่าสุด หรือแจ้งเตือนแผ่นดินไหวในมือถือมีอีกหลายตัวเลย ซึ่งส่วนมากก็จะใช้อ้างอิงจาก USGS โดยในเรื่องของ Delay นั้นขึ้นกับเหตุการณ์ ถ้าเป็นเหตุการณ์ที่มีความรุนแรงและส่งผลกระทบ ก็จะ Update เร็วหน่อยประมาณ 2-3 นาที ถ้าไม่มีผลกระทบก็จะใช้ระยะเวลานานขึ้นมากว่านั้นหน่อย

จากสามส่วนด้านบน ใครจะนำไปต่อยอดสำหรับเก็บข้อมูล ทำการวิเคราะห์ ทำ Visualize หรือสถิติต่อก็ไม่ยากละ ส่วนผมก็เอามาทำแจ้งเตือนบน Telegram ซึ่งจริงๆ ก็อยากจะแจ้งเตือนผ่าน Line Notify เพียงแต่ว่า Line Notify แบบใช้ฟรีกำลังจะหยุดให้บริการภายในสิ้นเดือนมีนาคมนี้ ก็เลยมีอันต้องย้ายระบบ Report ต่างๆที่ทำอยู่มาบน Telegram ตัวข้อมูลแผ่นดินไหวนี้ก็เลยทดลองทำด้วยเช่นกัน ออกมาเป็นแบบด้านล่างนี้ หรือใครสนใจจะเข้ามา join Private Telegram Channel นี้ก็ตามลิงค์นี้ มาได้เลยครับ รับรองไม่มีโฆษณา ชวนลงทุน ขายประกัน เล่นการพนัน แน่นอน 😀

ตัดจบดื้อๆเลยดีกว่า เพราะตาเริ่มปรือและใกล้ได้เวลาที่ต้องเตรียมปลุกลูกกินนมละ แต่ก่อนจบมีแอบแว้บไปดูข้อมูลระบบ DART ที่ใช้เตือนสึนามิของไทย DART Station 23461 – ANDAMAN SEA of Phuket, Thailand ก็ยังดูเอ๋อๆ ข้อมูลมาๆหายๆเหมือนเดิม วันนี้ก็พอแต่เพียงเท่านี้ ราตรีสวัสดิ์ครับ