-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
275 lines (230 loc) · 11.7 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
import requests
import os
import json
import time
from geopy.geocoders import Nominatim
EMAIL = ""
PASSWORD = ""
geolocator = Nominatim()
def lambda_handler(event, context):
if event['session']['application']['applicationId'] != "amzn1.ask.skill.c8fd59e4-a9df-4bf4-9c5d-b0971baebfbf":
print ("Invalid Application ID")
raise
else:
#Not using session currently
sessionAttributes = {}
headers = getAccessToken()
vehicle_id = getVehicleId(headers)
if event['session']['new']:
onSessionStarted(event['request']['requestId'], event['session'])
if event['request']['type'] == "LaunchRequest":
speechlet = onLaunch(event['request'], event['session'])
response = buildResponse(sessionAttributes, speechlet)
elif event['request']['type'] == "IntentRequest":
speechlet = onIntent(event['request'], event['session'], headers, str(vehicle_id))
response = buildResponse(sessionAttributes, speechlet)
elif event['request']['type'] == "SessionEndedRequest":
speechlet = onSessionEnded(event['request'], event['session'])
response = buildResponse(sessionAttributes, speechlet)
return (response)
def getAccessToken():
params = { "grant_type" : "password", "client_id" : "81527cff06843c8634fdc09e8ac0abefb46ac849f38fe1e431c2ef2106796384", "client_secret" : "c7257eb71a564034f9419ee651c7d0e5f7aa6bfbd18bafb5c5c033b093bb2fa3", "email" : EMAIL, "password" : PASSWORD }
response = requests.post("https://owner-api.teslamotors.com/oauth/token/", params = params).json()
headers = {"Authorization" : "Bearer " + response["access_token"]}
return(headers)
def getVehicleId(headers):
vehicles = requests.get("https://owner-api.teslamotors.com/api/1/vehicles", headers= headers).json()
vehicle_id = vehicles["response"][0]["id"]
return vehicle_id
def onSessionStarted(requestId, session):
print("onSessionStarted requestId=" + requestId + ", sessionId=" + session['sessionId'])
def onSessionEnded(sessionEndedRequest, session):
# Add cleanup logic here
print ("Session ended")
def onLaunch(launchRequest, session):
# Dispatch to your skill's launch.
getWelcomeResponse()
def getWelcomeResponse():
cardTitle = "Welcome to myTesla"
speechOutput = """By using this skill it is possible to control many functions of your Tesla Vehicle."""
# 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.
repromptText = 'Ask me commands that relate to the vehicle'
shouldEndSession = True
return (buildSpeechletResponse(cardTitle, speechOutput, repromptText, shouldEndSession))
def onIntent(intentRequest, session,headers,vehicle_id):
intent = intentRequest['intent']
intentName = intentRequest['intent']['name']
if intentName == "GetCharging":
return chargingResponse(headers, vehicle_id)
elif intentName == "HonkHorn":
return honkingResponse(headers, vehicle_id)
elif intentName == "Lock_Unlock":
return changeLockState(headers, vehicle_id, intent)
elif intentName == "GetCarStatus":
return carStatus(headers,vehicle_id, intent)
elif intentName == "AC_State":
return changeACState(headers, vehicle_id, intent)
elif intentName == "FlashLight":
return flashLights(headers, vehicle_id)
elif intentName == "Location":
return getLocation(headers,vehicle_id)
elif intentName == "startCar":
return startCar(headers,vehicle_id)
elif intentName == "AMAZON.HelpIntent":
return getWelcomeResponse()
elif intentName == "AMAZON.StopIntent":
return onSessionEnded()
else:
print ("Invalid Intent: " + intentName)
raise
def chargingResponse(headers,vehicle_id):
res = requests.get("https://owner-api.teslamotors.com/api/1/vehicles/" + vehicle_id + "/data_request/charge_state", headers = headers)
if(res.status_code == 200):
res = res.json()
if res["response"]["charging_state"] == "Complete":
speechOutput = "Your car has completed charging. With a range of " + str(res["response"]["battery_range"]) + "miles."
cardTitle = "Your car has completed charging. With a range of " + str(res["response"]["battery_range"]) + "miles."
else:
speechOutput = "Your car is " + str(res["response"]["battery_level"]) + " percent charged. With a range of " + str(res["response"]["battery_range"]) + "miles."
cardTitle = "Your car is " + str(res["response"]["battery_level"]) + " percent charged. With a range of " + str(res["response"]["battery_range"]) + "miles."
else:
speechOutput = "Error connecting to Tesla Servers. Please try again"
cardTitle = "Error connecting to Tesla Servers. Please try again"
repromptText = "I didn't understand that. Please try again"
shouldEndSession = True
return(buildSpeechletResponse(cardTitle,speechOutput,repromptText,shouldEndSession))
def honkingResponse(headers,vehicle_id):
res = requests.post("https://owner-api.teslamotors.com/api/1/vehicles/" + str(vehicle_id) + "/command/honk_horn", headers = headers)
if(res.status_code == 200):
speechOutput = "Car Horn Honked."
cardTitle = "Car Horn Honked."
else:
speechOutput = "Error connecting to Tesla Servers. Please try again"
cardTitle = "Error connecting to Tesla Servers. Please try again"
repromptText = "I didnt understand that. Please try again"
shouldEndSession = True
return(buildSpeechletResponse(cardTitle,speechOutput,repromptText,shouldEndSession))
def changeLockState(headers, vehicle_id, intent):
if(intent['slots']['lockstate']['value'] == "unlock"):
unlock(headers,vehicle_id)
speechOutput = "Ok, I'm unlocking your car"
cardTitle = "Ok, I'm unlocking your car"
elif(intent['slots']['lockstate']['value'] == "lock"):
lock(headers,vehicle_id)
speechOutput = "Ok, I'm locking your car"
cardTitle = "Ok, I'm locking your car"
else:
speechOutput = "Error connceting to Tesla Servers. Please try again"
cardTitle = speechOutput
repromptText = "I didnt understand that. Please try again"
shouldEndSession = True
return(buildSpeechletResponse(cardTitle, speechOutput,repromptText,shouldEndSession))
def unlock(headers,vehicle_id):
requests.post("https://owner-api.teslamotors.com/api/1/vehicles/"+str(vehicle_id) + "/command/door_unlock", headers = headers)
def lock(headers, vehicle_id):
requests.post("https://owner-api.teslamotors.com/api/1/vehicles/" + str(vehicle_id) + "/command/door_lock", headers = headers)
def carStatus(headers,vehicle_id, intent):
res = requests.get("https://owner-api.teslamotors.com/api/1/vehicles/"+ str(vehicle_id) + "/data_request/vehicle_state", headers = headers)
if(res.status_code == 200):
res = res.json()
if(intent['slots']['status']['value'] == "miles" or intent['slots']['status']['value'] == "odometer"):
speechOutput = "Your car has " + str(res["response"]["odometer"]) + " miles on it."
cardTitle = speechOutput
elif(intent['slots']['status']['value'] == "version" or intent['slots']['status']['value'] == "software"):
speechOutput = "Your car is running version " + str(res["response"]["car_version"])
cardTitle = speechOutput
else:
speechOutput = "Error connecting to Tesla Servers. Please try again"
cardTitle = speechOutput
repromptText = "I didnt understand that. Please try again"
shouldEndSession = True
return(buildSpeechletResponse(cardTitle,speechOutput,repromptText,shouldEndSession))
def changeACState(headers, vehicle_id, intent):
if(intent['slots']['power']['value'] == "on"):
res = requests.post("https://owner-api.teslamotors.com/api/1/vehicles/" + str(vehicle_id) + "/command/auto_conditioning_start", headers = headers)
if(res.status_code == 200):
speechOutput = "Ok, HVAC system turned on"
cardTitle = speechOutput
else:
speechOutput = "Error connceting to Tesla Server. Please try again"
cardTitle = speechOutput
elif(intent['slots']['power']['value'] == "off"):
res = requests.post("https://owner-api.teslamotors.com/api/1/vehicles/" + str(vehicle_id) + "/command/auto_conditioning_stop", headers = headers)
if(res.status_code == 200):
speechOutput = "Ok, HVAC system turned off"
cardTitle = speechOutput
else:
speechOutput = "Error connceting to Tesla Server. Please try again"
cardTitle = speechOutput
else:
speechOutput = "Error connecting to Tesla Server. Please try again"
cardTitle = speechOutput
repromptText = "I didnt understand that. Please try again"
shouldEndSession = True
return(buildSpeechletResponse(cardTitle,speechOutput,repromptText,shouldEndSession))
def flashLights(headers, vehicle_id):
res = requests.post("https://owner-api.teslamotors.com/api/1/vehicles/" + str(vehicle_id) + "/command/flash_lights", headers = headers)
if(res.status_code == 200):
speechOutput = "Ok, Lights Flashed"
cardTitle = speechOutput
else:
speechOutput = "Error connecting to Tesla Server. Please try again"
cardTitle = speechOutput
repromptText = "I didnt understand that. Please try again"
shouldEndSession = True
return(buildSpeechletResponse(cardTitle,speechOutput,repromptText,shouldEndSession))
def getLocation(headers,vehicle_id):
res = requests.get("https://owner-api.teslamotors.com/api/1/vehicles/"+str(vehicle_id)+"/data_request/drive_state",headers=headers)
if(res.status_code == 200):
res = res.json()
latitude = str(res["response"]["latitude"])
longitude = str(res["response"]["longitude"])
latLong = latitude + "," + longitude
location = geolocator.reverse(latLong)
speechOutput = "Your Tesla is at " + location.address
cardTitle = speechOutput
else:
speechOutput = "Error connecting to Tesla Server. Please try again"
cardTitle = speechOutput
repromptText = "I didnt understand that. Please try again"
shouldEndSession = True
return(buildSpeechletResponse(cardTitle,speechOutput,repromptText,shouldEndSession))
def startCar(headers,vehicle_id):
params = {"password":PASSWORD}
res = requests.post("https://owner-api.teslamotors.com/api/1/vehicles/"+str(vehicle_id)+"/command/remote_start_drive", params = params, headers = headers)
if(res.status_code == 200):
speechOutput = "Ok, your Tesla has been started"
cardTitle = speechOutput
else:
speechOutput = "Error connecting to Tesla Server. Please try again"
cardTitle = speechOutput
repromptText = "I didnt understand that. Please try again"
shouldEndSession = True
return(buildSpeechletResponse(cardTitle,speechOutput,repromptText,shouldEndSession))
# --------------- Helpers that build all of the responses -----------------------
def buildSpeechletResponse(title, output, repromptText, shouldEndSession):
return ({
"outputSpeech": {
"type": "PlainText",
"text": output
},
"card": {
"type": "Simple",
"title": "myTesla - " + title,
"content": "myTesla - " + output
},
"reprompt": {
"outputSpeech": {
"type": "PlainText",
"text": repromptText
}
},
"shouldEndSession": shouldEndSession
})
def buildResponse(sessionAttributes, speechletResponse):
return ({
"version": "1.0",
"sessionAttributes": sessionAttributes,
"response": speechletResponse
})