Python API - How to get Earthquake data into a Geo SCADA Map

sbeadle · ‎2025-11-26

The objective of this sample is to retrieve and display earthquake data onto a map. A basic query is used to the USGS web site to get all earthquakes globally for the past week - those above a certain magnitude. This data is placed into a table, and then a query is used to list location data to be displayed on a Geo SCADA Map Set item.

The code is listed here. An .sde file is attached to this article, which includes the code in a Python item, the Data Table and associated Map Set.

It would be simple to adapt and extend this code to:

Filter the locations to a specific area, and focus the map on that area
Scan the data for earthquakes over a specific magnitude and in a specific area, then raise an alarm if those criteria are met

Sample image - click to enlarge:

# Example: Get Earthquake Map Data
import geo_scada_types
from geoscada.lib.variant import *
from datetime import datetime, timedelta
from geoscada.client.interface_scx import InterfaceScx
from geoscada.client.types import QueryStatus
from typing import Optional
import unicodedata
# Allow program to write to its StdErr property
import sys

################################################################
# Web request library - needs you to run "pip install requests"
################################################################
import requests

def execute(program_path: str, program_name: str, trigger_type: geo_scada_types.ProgramTrigger, geo_scada_args: dict, connection: Optional[InterfaceScx], *args, **kwargs):
    # Find this program item
    ppath = connection.find_object(program_path)

    #############################################################
    # RUN SECTION - READ FROM WEB WRITE TO GEO SCADA
    #############################################################

    # Make the web request
    starttime = (datetime.now() - timedelta(days=7)).strftime("%Y-%m-%d%%20%H:%M:%S")
    endtime = datetime.now().strftime("%Y-%m-%d%%20%H:%M:%S")
    minmagnitude = 4.5
    base_url = f"https://earthquake.usgs.gov/fdsnws/event/1/query.geojson"
    # Parameters
    final_url = base_url + f"?starttime={starttime}&endtime={endtime}&minmagnitude={minmagnitude}&orderby=time"
    # Constrain by bounding box, e.g. this is the USA
    # &maxlatitude=50&minlatitude=24.6&maxlongitude=-65&minlongitude=-125
    
    # Get data in JSON
    earthquake_response = requests.get(final_url)
    earthquake_data = earthquake_response.json()

    # The following would be more efficient to keep existing records, purge old and insert new

    # Remove all previous records
    sqltext = f"DELETE FROM Earthquakes"
    print(sqltext)
    q = connection.prepare_query(sqltext, False)
    result = q.execute_sync()
    if result.status == QueryStatus.Succeeded:
        print(result.rows_affected)
    else:
        print(result.status, file=sys.stderr)

    # Get all records and insert each one
    index = 0
    while (index < len(earthquake_data["features"])):
        # Get parameters
        time = earthquake_data["features"][index]["properties"]["time"] #milliseconds
        timevalue = datetime.fromtimestamp(time/1000.0)
        mag = earthquake_data["features"][index]["properties"]["mag"]
        title = earthquake_data["features"][index]["properties"]["title"]
        title = unicodedata.normalize('NFC', title).encode('ascii', 'ignore')
        title = title.decode()
        lat = earthquake_data["features"][index]["geometry"]["coordinates"][1]
        long = earthquake_data["features"][index]["geometry"]["coordinates"][0]
        print(f"{mag}, {timevalue.strftime('%Y-%m-%d %H:%M:%S')}, {title}, {lat}, {long}")
    
        sqltext = f"INSERT INTO Earthquakes (timevalue, mag, title, lat, long) VALUES (TIMESTAMP('{timevalue.strftime("%Y-%m-%d %H:%M:%S")}'), {mag}, '{title}', {lat}, {long})"
        print(sqltext)
        q = connection.prepare_query(sqltext, False)
        result = q.execute_sync()
        if result.status == QueryStatus.Succeeded:
            print(result.rows_affected)
        else:
            print(result.status, file=sys.stderr)
        index += 1

    print("Success")

See below for the .sde file.

📖 Home

📖 Python article index