Taras Kravchuk
Created November 24, 2016

Earthquake Tracker

Keep track of recent earthquakes with Earthquake Tracker.

30
Earthquake Tracker

Things used in this project

Story

Read more

Code

earthquake.py

Python
AWS Lambda function that runs this python code.
"""
Author: Taras Kravchuk
App: Earthquake Tracker
Description: An Alexa skill that uses the API at earthquake.usgs.gov to return
            info on recent earthquakes
"""

from __future__ import print_function
import requests
import json
from datetime import datetime, timedelta


# --------------- Helpers that build all of the responses ----------------------

def build_speechlet_response(title, output, reprompt_text, should_end_session):
    return {
        'outputSpeech': {
            'type': 'PlainText',
            'text': output
        },
        'card': {
            'type': 'Simple',
            'title': title,
            'content': output
        },
        'reprompt': {
            'outputSpeech': {
                'type': 'PlainText',
                'text': reprompt_text
            }
        },
        'shouldEndSession': should_end_session
    }


def build_response(session_attributes, speechlet_response):
    return {
        'version': '1.0',
        'sessionAttributes': session_attributes,
        'response': speechlet_response
    }


# --------------- Functions that control the skill's behavior ------------------

def get_welcome_response():
    """ If we wanted to initialize the session to have some attributes we could
    add those here
    """

    session_attributes = {}
    card_title = "Welcome to Earthquake Tracker"
    speech_output = "Welcome to Earthquake Tracker. " \
                    "Earthquake Tracker can return earthquake information recorded in the last twenty days.  " \
                    "Try asking Earthquake Tracker for info on the most recent earthquake by saying, " \
                    "ask earthquake tracker about the most recent earthquake."
    # If the user either does not reply to the welcome message or says something
    # that is not understood, they will be prompted again with this text.
    reprompt_text = "Try asking Earthquake Tracker for info on the most recent earthquake by saying, " \
                    "ask earthquake tracker what was the most recent earthquake."
    should_end_session = False
    return build_response(session_attributes, build_speechlet_response(
        card_title, speech_output, reprompt_text, should_end_session))


def handle_session_end_request():
    card_title = "Session Ended"
    speech_output = "Thank you for trying Earthquake Tracker. " \
                    "Stay safe and have a nice day! "
    # Setting this to true ends the session and exits the skill.
    should_end_session = True
    return build_response({}, build_speechlet_response(
        card_title, speech_output, None, should_end_session))

def wrong_intent():
    session_attributes = {}
    title = "Intent Question Error"
    output = "I'm sorry, I did not understand that. " \
             "Try asking Earthquake Tracker for info on a recent large earthquake by saying, " \
             "ask earthquake tracker what was the largest earthquake in the last ten days. "
    reprompt = "Try asking Earthquake Tracker for info on the most recent earthquake by saying, " \
                "ask earthquake tracker what was the most recent earthquake."
    should_end_session = False
    return build_response(session_attributes, build_speechlet_response(
        title, output, reprompt, should_end_session))
        
def help_intent():
    session_attributes = {}
    title = "Earthquake Tracker Help"
    output = "Try asking Earthquake Tracker for info on a large earthquake by saying, " \
             "ask earthquake tracker about the largest earthquake in the last ten days."
    reprompt = "Try asking Earthquake Tracker for info on the most recent earthquake by saying, " \
                "ask earthquake tracker what was the most recent earthquake."
    should_end_session = False
    return build_response(session_attributes, build_speechlet_response(
        title, output, reprompt, should_end_session))

def get_day_range_warning():
    session_attributes = {}
    title = "Day Range Error"
    output = "Earthquake Tracker can only return earthquakes recorded in the last twenty days.  " \
             "Try asking Earthquake Tracker for info on a recent large earthquake by saying, " \
             "ask earthquake tracker what was the largest earthquake in the last ten days."
    reprompt = "Try asking Earthquake Tracker for info on the most recent earthquake by saying, " \
                "ask earthquake tracker what was the most recent earthquake."
    should_end_session = False
    return build_response(session_attributes, build_speechlet_response(
        title, output, reprompt, should_end_session))

def get_largest_index(eq_list):
    largest_index = 0
    largest_mag = eq_list[0]['properties']['mag']
    for i, item in enumerate(eq_list):
        if eq_list[i]['properties']['mag'] > largest_mag:
            largest_mag = eq_list[i]['properties']['mag']
            largest_index = i
    return largest_index
        

def get_recent_earthquake(intent, session, magnitude, day_range, largest):
    if not day_range:
        day_range = 2
    if day_range > 20:
        return get_day_range_warning()
    start_date = (datetime.today() - timedelta(int(day_range))).strftime("%Y-%m-%d")
    session_attributes = {}
    reprompt = None
    title = "Earthquake Tracker"
    end_sess = False
    
    if not magnitude:
        eq_magnitude = ''
    else:
        eq_magnitude = '&minmagnitude=%s'%magnitude
    url = "http://earthquake.usgs.gov/fdsnws/event/1/query?format=geojson&eventtype=earthquake&starttime=%s%s"%(start_date, eq_magnitude)
    r = requests.get(url)
    earthquakes = json.loads(r.text)
    try:
        count = earthquakes['metadata']['count']
        if count <= 0:
            context = "There are no recent earthquakes to report"
        else:
            if largest:
                largest_index = get_largest_index(earthquakes['features'])
                context_fill = "largest earthquake in the last %s days"%day_range
            else:
                largest_index = 0
                context_fill = "most recent earthquake"
            mag = earthquakes['features'][largest_index]['properties']['mag']
            ts_date = earthquakes['features'][largest_index]['properties']['time']
            eq_date = datetime.fromtimestamp(ts_date/1000.0).strftime("%B %d %Y at %I:%M %p GMT")
            tsunami = earthquakes['features'][largest_index]['properties']['tsunami']
            if tsunami == 0:
                tsunami = ""
            else:
                tsunami = "The earthquake may have triggered a tsunami which could affect the surrounding areas!"
            location = earthquakes['features'][largest_index]['properties']['place']
            loc_map = {"N":"North", "E":"East", "S":"South", "W":"West"}
            loc_list = location.split(' ')
            loc_string = ''
            for i in loc_list[1]:
                loc_string += loc_map[i]
            final_location = loc_list[0] + " " + loc_string + " " + " ".join(loc_list[2:])
            context = "The %s with a magnitude of %s, occured %s on %s. %s"%(context_fill, mag, final_location,eq_date, tsunami)
    except:
        if magnitude:
            context = "There are no recent earthquakes with a magnitude of %s and greater"%magnitude
        else:
            context = "There are no recent earthquakes to report"
    return build_response(session_attributes, build_speechlet_response(
        title, context, reprompt, end_sess))

# --------------- Events ------------------

def on_session_started(session_started_request, event, session):
    """ Called when the session starts """

    print("on_session_started requestId=" + session_started_request['requestId']
          + ", sessionId=" + session['sessionId'])


def on_launch(launch_request, session):
    """ Called when the user launches the skill without specifying what they
    want
    """

    print("on_launch requestId=" + launch_request['requestId'] +
          ", sessionId=" + session['sessionId'])
    # Dispatch to your skill's launch
    return get_welcome_response()


def on_intent(intent_request, session):
    """ Called when the user specifies an intent for this skill """

    print("on_intent Intent=" + intent_request['type'] +
          ", MORE INFO=" + str(intent_request['intent']))

    intent = intent_request['intent']
    intent_name = intent_request['intent']['name']
    try:
        magnitude = float(intent_request['intent']['slots']['Magnitude']['value'])
    except:
        magnitude = None
    
    try:
        day_range = int(intent_request['intent']['slots']['DayRange']['value'])
    except:
        day_range = None

    try:
        largest = intent['slots']['Largest']['value']
    except:
        largest = False
    else:
        if largest == "largest":
            largest = True

    if intent_name == "RecentEarthquakeTracker" and (magnitude or day_range or largest):
        return get_recent_earthquake(intent, session, magnitude, day_range, largest)
    elif intent_name == "AMAZON.HelpIntent":
        return help_intent()
    elif intent_name == "AMAZON.CancelIntent" or intent_name == "AMAZON.StopIntent":
        return handle_session_end_request()
    else:
        return wrong_intent()


def on_session_ended(session_ended_request, session):
    """ Called when the user ends the session.
    """
    print("on_session_ended requestId=" + session_ended_request['requestId'] +
          ", sessionId=" + session['sessionId'])
    # add cleanup logic here
    return handle_session_end_request()


# --------------- Main handler ------------------

def lambda_handler(event, context):
    """ Route the incoming request based on type (LaunchRequest, IntentRequest,
    etc.) The JSON body of the request is provided in the event parameter.
    """
    print("event.session.application.applicationId=" +
          event['session']['application']['applicationId'])

    if (event['session']['application']['applicationId'] != "amzn1.ask.skill.9583873c-bbfb-4822-abeb-ec841f3fc7eb"):
        raise ValueError("Invalid Application ID")

    if event['request']['type'] == "LaunchRequest":
        return on_launch(event['request'], event['session'])
    elif event['request']['type'] == "IntentRequest":
        return on_intent(event['request'], event['session'])
    elif event['request']['type'] == "SessionEndedRequest":
        return on_session_ended(event['request'], event['session'])

Credits

Taras Kravchuk

Taras Kravchuk

1 project • 0 followers

Comments