-
Notifications
You must be signed in to change notification settings - Fork 1
/
myq.py
178 lines (160 loc) · 7.58 KB
/
myq.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
'''
Copyright 2020 Amazon.com, Inc. or its affiliates. All Rights Reserved.
SPDX-License-Identifier: MIT-0
Disclaimer: The code here is based off of an unsupported API from Chamberlain
and is subject to change without notice. The authors claim no responsibility
for damages to your garage door or property by use of the code within.
'''
import requests
import boto3
import time
import os
import json
class myqapi:
# These constants can vary depends on the atual hardware
app_id = os.environ['APP_ID']
device_list_endpoint = os.environ['DEVICE_LIST_ENDPOINT']
device_set_endpoint = os.environ['DEVICE_SET_ENDPOINT']
uri = os.environ['BASE_URI']
endpoint = os.environ['BASE_ENDPOINT']
#retrive MyQ username and password from aws secret manager
secret_name = os.environ['SECRET_NAME']
# Create a Secrets Manager client
session = boto3.session.Session()
region_name = session.region_name
client = session.client(
service_name='secretsmanager',
region_name=region_name
)
try:
get_secret_value_response = client.get_secret_value(
SecretId=secret_name
)
print(get_secret_value_response)
except ClientError as e:
print(e)
if e.response['Error']['Code'] == 'DecryptionFailureException':
# Secrets Manager can't decrypt the protected secret text using the provided KMS key.
# Deal with the exception here, and/or rethrow at your discretion.
raise e
elif e.response['Error']['Code'] == 'InternalServiceErrorException':
# An error occurred on the server side.
# Deal with the exception here, and/or rethrow at your discretion.
raise e
elif e.response['Error']['Code'] == 'InvalidParameterException':
# You provided an invalid value for a parameter.
# Deal with the exception here, and/or rethrow at your discretion.
raise e
elif e.response['Error']['Code'] == 'InvalidRequestException':
# You provided a parameter value that is not valid for the current state of the resource.
# Deal with the exception here, and/or rethrow at your discretion.
raise e
elif e.response['Error']['Code'] == 'ResourceNotFoundException':
# We can't find the resource that you asked for.
# Deal with the exception here, and/or rethrow at your discretion.
raise e
else:
# Decrypts secret using the associated KMS CMK.
# Depending on whether the secret is a string or binary, one of these fields will be populated.
if 'SecretString' in get_secret_value_response:
secret = get_secret_value_response['SecretString']
j = json.loads(secret)
password = j['myq_password']
username = j['myq_username']
else:
decoded_binary_secret = base64.b64decode(get_secret_value_response['SecretBinary'])
username = decoded_binary_secret.myq_username
password = decoded_binary_secret.myq_password
def __init__(self):
# Get the security token used to authenticate to myq API
params = {
'username': self.username,
'password': self.password
}
login = requests.post(
'https://{host_uri}/{login_endpoint}'.format(
host_uri=self.uri,
login_endpoint=self.endpoint),
json=params,
headers={
'MyQApplicationId': self.app_id
}
)
auth = login.json()
self.security_token = auth['SecurityToken']
print (auth['SecurityToken'])
def get_devices(self):
# Gets all devices within your myQ account
devices = requests.get(
'https://{host_uri}/{device_list_endpoint}'.format(
host_uri=self.uri,
device_list_endpoint=self.device_list_endpoint),
headers={
'MyQApplicationId': self.app_id,
'SecurityToken': self.security_token
}
)
devices = devices.json()['Devices']
return devices
def get_garagedeviceid(self, description):
# This gets the device id for the garage door state will be set on
devices = self.get_devices()
deviceid = None
garagedoors = [x for x in devices if x['MyQDeviceTypeName'] == 'Garage Door Opener WGDO']
for garagedoor in garagedoors:
for attrib in garagedoor['Attributes']:
if attrib['Value'] == description:
deviceid = garagedoor['MyQDeviceId']
if deviceid == None:
return "Liftmaster device name not found"
else:
return deviceid
def get_state(self, description):
# This gets the state of a garage door
deviceid = self.get_garagedeviceid(description)
devices = self.get_devices()
garagedoors = [x for x in devices if x['MyQDeviceTypeName'] == 'Garage Door Opener WGDO']
# print (garagedoors)
garagedoor = garagedoors[0]['Attributes']
state = [x['Value'] for x in garagedoor if x['AttributeDisplayName'] == "doorstate"]
return state[0]
def set_state(self, description):
# state = 2 (close) and state = 1 (open)
state = self.get_state(description)
device_id = self.get_garagedeviceid(description)
if state == "2":
new_state = "1"
payload = {
'attributeName': 'desireddoorstate',
'myQDeviceId': device_id,
'AttributeValue': new_state,
}
device_action = requests.put(
'https://{host_uri}/{device_set_endpoint}'.format(
host_uri=self.uri,
device_set_endpoint=self.device_set_endpoint),
data=payload,
headers={
'MyQApplicationId': self.app_id,
'SecurityToken': self.security_token
}
)
# wait 60 sec and close the gate
time.sleep(60)
new_state = "2"
payload = {
'attributeName': 'desireddoorstate',
'myQDeviceId': device_id,
'AttributeValue': new_state,
}
device_action = requests.put(
'https://{host_uri}/{device_set_endpoint}'.format(
host_uri=self.uri,
device_set_endpoint=self.device_set_endpoint),
data=payload,
headers={
'MyQApplicationId': self.app_id,
'SecurityToken': self.security_token
}
)
return device_action.status_code == 200