Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
lambda-function.py for Hackerster.io (Subject to Licence Agreement)
#!/usr/bin/python
# -*- coding: utf-8 -*-
# you need to install these libraries, firebase.py is included in the repo
import firebase
import requests
import json
##################################################################################
# Initialize this with your own firebase, url for example - "room-service-abc123"#
firebase_base_url = '{your firebase url}'
##################################################################################
# You can extend these dicts by any choice of device you want, just remember to add the appropriate HEXCODE for your model
device_on_dict = {'tata sky': ['C0000C'], 'samsung tv': [
'E0E040BF'], 'air conditioner': ['B24D7B84', 'B24DBF40']}
device_volume_up_dict = {'tata sky': ['C00010'], 'samsung tv': ['E0E0E01F', 'E0E0E01F']}
device_volume_down_dict = {'tata sky': ['C00011'], 'samsung tv': ['E0E0D02F', 'E0E0D02F']}
dth_device_channel_dict = {'0': 'C00000', '1': 'C00001', '2': 'C00002',
'3': 'C00003', '4': 'C00004', '5': 'C00005', '6': 'C00006', '7': 'C00007', '8': 'C00008', '9': 'C00009'}
# Writing data to firebase
def add_to_firebase(
user,
task,
status
):
firebase.put(firebase_base_url+"/users/"+user, {'task': { 'status': status, 'taskid': task } })
# Reading data from firebase
def get_firebase_data(user, task):
result = firebase.get('/users')
return result
# Tata sky api to fetch channel numbers
def call_tatasky():
response = \
requests.get('http://mobileapp.tatasky.com/hma/api/1.0a/epg/1/genres.json'
)
result = json.loads(response.text)
d = dict()
for x in result:
for y in x['l']:
for z in y['m']:
d[str(z['t'])] = int(z['c'])
return [d]
# Tata sky api to fetch channel numbers by category
def call_tatasky_by_category(user_category, user_language):
response = \
requests.get('http://mobileapp.tatasky.com/hma/api/1.0a/epg/1/genres.json'
)
result = json.loads(response.text)
for categories in result:
for channels in categories['l']:
category = categories['n']
if category == user_category:
for channel in channels['m']:
language = channels['e']
if language == user_language or language == '':
d = str(channel['c'])
return [d]
return [str(100)]
def build_speechlet_response(
title,
output,
reprompt_text,
should_end_session,
):
return {
'outputSpeech': {'type': 'PlainText', 'text': output},
'card': {'type': 'Simple', 'title': 'Room Service',
'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}
def get_user_info(access_token):
amazonProfileURL = \
'https://api.amazon.com/user/profile?access_token='
r = requests.get(url=amazonProfileURL + access_token)
if r.status_code == 200:
return r.json()
else:
return False
def handle_end():
return build_response({}, build_speechlet_response('Session ended',
'Goodbye!', '', True))
def launch_request(access_token):
session_attributes = {}
card_title = 'Welcome'
if access_token is None:
speech_output = \
'Your user details are not available at this time. Have you completed account linking via the Alexa app?'
reprompt_text = 'Please link your account in the Alexa app'
should_end_session = True
else:
user_details = get_user_info(access_token)
if user_details is None:
speech_output = \
'There was a problem getting your user details.'
should_end_sesion = True
else:
# Add user info to firebase
firebase.put(firebase_base_url + "/users/" + user_details['user_id'].split('.')[2],
{'task': {'status': False, 'taskid': ['']}})
speech_output = 'Hello ' + user_details['name'].split(' '
)[0] + '! How can I help you?'
reprompt_text = 'What can I do, turn on your TV or AC?'
should_end_session = False
return build_response(session_attributes,
build_speechlet_response(card_title,
speech_output, reprompt_text,
should_end_session))
def intent_request(intent_request, access_token):
intent = intent_request['intent']
intent_name = intent['name']
# CATEGORY INTENT
if intent_name == 'CategoryIntent':
if access_token is not None:
user_details = get_user_info(access_token)
if user_details is None:
device_type = False
category = False
language = False
api_function = False
else:
try:
device_type = intent['slots']['DthDevice'
]['resolutions']['resolutionsPerAuthority'
][0]['values'][0]['value']['name']
category = intent['slots']['Category']['resolutions'
]['resolutionsPerAuthority'][0]['values'
][0]['value']['name']
try:
language = intent['slots']['Language'
]['resolutions'
]['resolutionsPerAuthority'
][0]['values'][0]['value']['name']
# This api_funtion can be changed to your prefered dth operator, but we stick to tata sky, due to api limitations
api_function = \
call_tatasky_by_category(category, language)
except:
language = "Hindi"
# This api_funtion can be changed to your prefered dth operator, but we stick to tata sky, due to api limitations
api_function = \
call_tatasky_by_category(category, language)
except:
device_type = False
category = False
language = False
api_function = False
else:
return handle_end()
if device_type == False or category == False or language \
== False or api_function == False:
speech_output = \
'Sorry, I was not able to process that command, please give me clear instructions, I am not a human after all'
card_title = speech_output
reprompt_text = 'What can I do, turn on your TV or AC?'
should_end_session = False
return build_response({},
build_speechlet_response(card_title,
speech_output, reprompt_text, should_end_session))
else:
# We are checking just for tata sky due to api limitations, but this can be extended in future by creating a dict of all dth operators
if device_type == 'tata sky':
if category != False:
channel_list = list(api_function[0])
task = list()
for x in channel_list:
task.append(dth_device_channel_dict[x])
status = True
# Add task to firebase
add_to_firebase(user_details['user_id'].split('.')[
2], task, status)
speech_output = 'Ok changing channel to ' \
+ api_function[0] + ". Anything else?"
else:
speech_output = \
'Sorry, I can only change channel for tatasky right now'
card_title = speech_output
reprompt_text = 'What can I do, turn on your TV or AC?'
should_end_session = False
return build_response({},
build_speechlet_response(card_title,
speech_output, reprompt_text, should_end_session))
# HELP INTENT
if intent_name == 'AMAZON.HelpIntent':
card_title = \
'Try saying like - Alexa, play STAR PLUS on my tata sky'
speech_output = \
'Try saying like - Alexa, play STAR PLUS on my tata sky'
should_end_session = False
reprompt_text = 'Try saying like - Alexa, play STAR PLUS on my tata sky'
return build_response({}, build_speechlet_response(card_title,
speech_output, should_end_session, should_end_session))
# CANCEL AND STOP INTENTS
if intent_name == 'AMAZON.CancelIntent' or intent_name \
== 'AMAZON.StopIntent':
card_title = 'Session Ended'
speech_output = 'Thank you for using room service. Good bye!'
return build_response({}, build_speechlet_response(card_title,
speech_output, None, True))
def lambda_handler(event, context):
try:
access_token = event['context']['System']['user']['accessToken']
except:
access_token = None
if event['request']['type'] == 'LaunchRequest':
return launch_request(access_token)
elif event['request']['type'] == 'IntentRequest':
return intent_request(event['request'], access_token)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.