"""
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'])
Comments