บทนำ การแจ้งเตือนเหตุแผ่นดินไหว

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

Telegram Bot https://t.me/DisasterwarningBot

หรือ กลุ่ม https://t.me/disasterThaiAleart

วัตถุประสงค์ของแอปพลิเคชัน

• ดึงข้อมูลแผ่นดินไหวล่าสุดจาก USGS Earthquake API
• กรองข้อมูลตามช่วงเวลา, ขนาดแผ่นดินไหว (แมกนิจูด), และจำนวนเหตุการณ์ที่ต้องการ
• แสดงผลข้อมูลในรูปแบบที่อ่านง่าย เพื่อให้ผู้ใช้สามารถรับทราบข้อมูลแผ่นดินไหวได้อย่างรวดเร็ว

เทคโนโลยีและเครื่องมือที่ใช้

• ภาษา Python – ใช้สำหรับเขียนโค้ดหลักของแอปพลิเคชัน
• Requests Library – สำหรับส่ง HTTP Request ไปยัง USGS API
• USGS Earthquake API – ให้ข้อมูลแผ่นดินไหวในรูปแบบ GeoJSON
• Datetime Module – สำหรับจัดการและจัดรูปแบบเวลา

ขั้นตอนการทำงาน

1. การเชื่อมต่อกับ USGS API

import requests
from datetime import datetime

def fetch_earthquake_data(start_time=None, end_time=None, min_magnitude=None, max_magnitude=None, limit=10):
    base_url = "https://earthquake.usgs.gov/fdsnws/event/1/query"
    params = {
        "format": "geojson",
        "starttime": start_time,
        "endtime": end_time,
        "minmagnitude": min_magnitude,
        "maxmagnitude": max_magnitude,
        "limit": limit
    }
    response = requests.get(base_url, params=params)
    if response.status_code == 200:
        return response.json()
    else:
        print(f"Error: Unable to fetch data. Status code: {response.status_code}")
        return None

2. การแสดงผลข้อมูลแผ่นดินไหว

def display_earthquake_info(data):
    if not data or "features" not in data:
        print("No earthquake data available.")
        return

    print("Earthquake Information:")
    print("-" * 40)
    for feature in data["features"]:
        properties = feature.get("properties", {})
        geometry = feature.get("geometry", {})
        
        place = properties.get("place", "N/A")
        magnitude = properties.get("mag", "N/A")
        time = properties.get("time", "N/A")
        coordinates = geometry.get("coordinates", [None, None, None])

        if time and isinstance(time, int):
            time_formatted = datetime.utcfromtimestamp(time / 1000).strftime('%Y-%m-%d %H:%M:%S')
        else:
            time_formatted = "N/A"

        print(f"Place: {place}")
        print(f"Magnitude: {magnitude}")
        print(f"Time: {time_formatted}")
        print(f"Coordinates: Latitude={coordinates[1]}, Longitude={coordinates[0]}, Depth={coordinates[2]} km")
        print("-" * 40)

3. ตัวอย่างการใช้งาน

if __name__ == "__main__":
    start_time = "2025-03-28T00:00:00"
    end_time = "2025-03-28T20:59:59"
    min_magnitude = 5.0  # แสดงเฉพาะแผ่นดินไหวขนาด 5.0 ขึ้นไป
    limit = 5  # แสดงผลล่าสุด 5 เหตุการณ์

    earthquake_data = fetch_earthquake_data(
        start_time=start_time,
        end_time=end_time,
        min_magnitude=min_magnitude,
        limit=limit
    )

    if earthquake_data:
        display_earthquake_info(earthquake_data)

ประโยชน์ของแอปพลิเคชัน

• รับข้อมูลแบบ Real-time – สามารถตรวจสอบแผ่นดินไหวล่าสุดได้ทันที
• ปรับแต่งการค้นหาได้ – สามารถกำหนดช่วงเวลาและขนาดแผ่นดินไหวที่ต้องการ
• ใช้งานง่าย – แสดงผลข้อมูลในรูปแบบที่เข้าใจง่าย

แนวทางการพัฒนาในอนาคต

• เพิ่มการแจ้งเตือนแบบ Real-time (ผ่าน Email, LINE, หรือ SMS)
• แสดงผลบนแผนที่ โดยใช้ไลบรารีเช่น Folium หรือ Google Maps API
• พัฒนาเป็นเว็บแอปหรือโมบายแอป เพื่อให้เข้าถึงได้สะดวกยิ่งขึ้น

สรุป

แอปพลิเคชันตรวจสอบการแจ้งเตือนเหตุแผ่นดินไหวนี้เป็นตัวอย่างที่ดีของการนำ API ข้อมูลเปิด (Open Data API) มาใช้ประโยชน์ โดยใช้ Python ซึ่งเป็นภาษาที่มีความยืดหยุ่นสูง หากคุณสนใจสามารถนำโค้ดนี้ไปพัฒนาต่อยอดได้ตามต้องการ เพื่อเพิ่มประสิทธิภาพในการรับมือกับภัยแผ่นดินไหวได้ดียิ่งขึ้น

Micropython ESP32

import urequests
import utime
import ujson

def get_current_utc_time():
    """
    Get current UTC time in ISO format (YYYY-MM-DDTHH:MM:SS) using utime
    """
    now = utime.gmtime()
    return "{:04d}-{:02d}-{:02d}T{:02d}:{:02d}:{:02d}".format(
        now[0], now[1], now[2], now[3], now[4], now[5]
    )

def format_timestamp(timestamp_ms):
    """
    Format UNIX timestamp in milliseconds to readable string
    """
    if not timestamp_ms:
        return "N/A"
    timestamp_sec = timestamp_ms // 1000
    t = utime.localtime(timestamp_sec)
    return "{:04d}-{:02d}-{:02d} {:02d}:{:02d}:{:02d}".format(
        t[0], t[1], t[2], t[3], t[4], t[5]
    )

def fetch_earthquake_data(start_time=None, end_time=None, min_magnitude=None, max_magnitude=None, limit=10):
    """
    Fetch earthquake data from USGS API
    """
    base_url = "https://earthquake.usgs.gov/fdsnws/event/1/query"
    params = {
        "format": "geojson",
        "limit": limit
    }
    
    if start_time:
        params["starttime"] = start_time
    if end_time:
        params["endtime"] = end_time
    if min_magnitude:
        params["minmagnitude"] = min_magnitude
    if max_magnitude:
        params["maxmagnitude"] = max_magnitude
    
    try:
        response = urequests.get(base_url, params=params)
        if response.status_code == 200:
            data = ujson.loads(response.text)
            response.close()
            return data
        else:
            print("Error:", response.status_code)
            response.close()
            return None
    except Exception as e:
        print("Request failed:", e)
        return None

def display_earthquake_info(data):
    """
    Display earthquake information
    """
    if not data or "features" not in data or not data["features"]:
        print("No earthquake data available")
        return
    
    print("\nEarthquake Information:")
    print("-" * 40)
    for feature in data["features"]:
        props = feature.get("properties", {})
        geom = feature.get("geometry", {})
        coords = geom.get("coordinates", [])
        
        print("ID:", feature.get("id", "N/A"))
        print("Place:", props.get("place", "N/A"))
        print("Magnitude:", props.get("mag", "N/A"))
        print("Time:", format_timestamp(props.get("time")))
        if len(coords) >= 2:
            print("Location: {:.2f}°N, {:.2f}°E".format(coords[1], coords[0]))
        if len(coords) >= 3:
            print("Depth: {:.1f} km".format(coords[2]))
        print("-" * 40)

# Main loop
print("Starting earthquake monitor...")
while True:
    print("\nChecking for earthquakes...")
    data = fetch_earthquake_data(
        start_time=get_current_utc_time(),
        min_magnitude=5.0,
        limit=5
    )
    
    if data:
        display_earthquake_info(data)
    else:
        print("Failed to fetch data")
    
    print("Waiting 30 seconds...")
    utime.sleep(30)

ของแถม . สำหรับท่านที่สนใจและใช้ Arduino

#include <WiFi.h>
#include <WiFiClientSecure.h>
#include <ArduinoJson.h>

// WiFi Credentials
const char* ssid = "YOUR_WIFI_SSID";
const char* password = "YOUR_WIFI_PASSWORD";

// USGS API URL
const char* usgsApiUrl = "https://earthquake.usgs.gov/fdsnws/event/1/query?format=geojson&limit=1&minmagnitude=5";

void setup() {
  // Initialize Serial Monitor
  Serial.begin(115200);

  // Connect to WiFi
  WiFi.begin(ssid, password);
  Serial.println("Connecting to WiFi...");
  while (WiFi.status() != WL_CONNECTED) {
    delay(500);
    Serial.print(".");
  }
  Serial.println("\nWiFi connected");
}

void loop() {
  // Fetch Earthquake Data
  WiFiClientSecure client;
  client.setInsecure(); // Disable SSL certificate verification (not recommended for production)
  if (!client.connect("earthquake.usgs.gov", 443)) {
    Serial.println("Connection to API failed");
    return;
  }

  // Send GET Request
  client.print(String("GET ") + usgsApiUrl + " HTTP/1.1\r\n" +
               "Host: earthquake.usgs.gov\r\n" +
               "Connection: close\r\n\r\n");

  // Read Response
  String payload = "";
  while (client.connected() || client.available()) {
    if (client.available()) {
      String line = client.readStringUntil('\n');
      payload += line;
    }
  }
  client.stop();

  // Parse JSON
  DynamicJsonDocument doc(1024);
  DeserializationError error = deserializeJson(doc, payload);
  if (error) {
    Serial.println("JSON parsing failed");
    return;
  }

  // Extract Earthquake Data
  JsonObject properties = doc["features"][0]["properties"];
  const char* place = properties["place"];
  float magnitude = properties["mag"];
  long time = properties["time"];

  // Convert Time
  time_t epochTime = time / 1000;
  struct tm* timeInfo = gmtime(&epochTime);
  char timeString[20];
  strftime(timeString, sizeof(timeString), "%Y-%m-%d %H:%M:%S", timeInfo);

  // Display on Serial Monitor
  Serial.println("Earthquake Alert!");
  Serial.println("----------------");
  Serial.print("Place: ");
  Serial.println(place);
  Serial.print("Magnitude: ");
  Serial.println(magnitude);
  Serial.print("Time: ");
  Serial.println(timeString);
  Serial.println("");

  // Wait before next check
  delay(60000); // Check every 60 seconds
}