-
Notifications
You must be signed in to change notification settings - Fork 1
/
danabridge-python.py
236 lines (161 loc) · 8.74 KB
/
danabridge-python.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
#!/usr/bin/env python3
from oauthlib.oauth2 import LegacyApplicationClient
from requests_oauthlib import OAuth2Session
import requests
import json
import time
my_username = "[username]"
my_password = "[password]"
def danalock_initialise(user_id, secret):
client_id = "danalock-web"
BASE_URL = "https://api.danalock.com/oauth2/token"
oauth = OAuth2Session(client=LegacyApplicationClient(client_id=client_id))
token = oauth.fetch_token(token_url=BASE_URL,
username=user_id, password=secret, client_id=client_id,
client_secret="")
if "access_token" not in token:
print(f"error getting danalock token - response was {token}")
exit()
return token["access_token"], token["refresh_token"]
# returns a dict of each lock's name and its serial number (needed for commands)
def get_all_danalocks(token):
url = "https://api.danalock.com/locks/v1"
r = requests.get(url, headers={'Authorization': 'Bearer ' + token, 'Accept': 'application/json', 'content-type': 'application/json'})
if r.status_code != 200:
print(f"error getting list of danalocks - error {r.status_code} and response {r.text}")
exit()
results = json.loads(r.text)
danalocks = {}
for x in results:
danalocks[x["name"].lower()] = x["afi"]["serial_number"]
return danalocks
def get_danalock_status(token, serial):
execute_url = "https://bridge.danalockservices.com/bridge/v1/execute"
payload = json.dumps({"device": serial, "operation": "afi.lock.get-state"})
r = requests.post(execute_url, headers={'Authorization': 'Bearer ' + token, 'Accept': 'application/json', 'content-type': 'application/json'}, data=payload)
if r.status_code != 200:
print(f"error getting job ID - error {r.status_code} and response {r.text}")
return None
job_id = json.loads(r.text)["id"]
# print(f"Got job ID {job_id}. Will now poll for status")
# try ten times
for i in range(0, 10):
time.sleep(2)
poll_url = "https://bridge.danalockservices.com/bridge/v1/poll"
payload = json.dumps({"id": job_id})
# print(f"Attempt {i}")
r = requests.post(poll_url, headers={'Authorization': 'Bearer ' + token, 'Accept': 'application/json', 'content-type': 'application/json'}, data=payload)
if r.status_code == 404:
print(f"Lock either not on danabridge, lock out of battery, danabridge disconnected, or lock out of danabridge bluetooth range")
return None
elif r.status_code != 200:
print(f"error getting lock status - error {r.status_code} and response {r.text}")
return None
if "Succeeded" in r.text:
return json.loads(r.text)["result"]["state"]
print(f"did not succeed getting lock status after {i*2}s - final response was {r.text}")
return None
# resturns a list of pincodes, each item being a dict with the following keys
# [{'identifier': [the pin code identifier, 1 thru 20],
# 'status': [something to do with the timer function - haven't investigated[],
# 'digits': 'CODE'},
def get_pin_codes(token, serial):
execute_url = "https://bridge.danalockservices.com/bridge/v1/execute"
payload = json.dumps({"device": serial, "operation": "afi.pin-codes.get-pin-codes", "arguments": ["20"]})
r = requests.post(execute_url, headers={'Authorization': 'Bearer ' + token, 'Accept': 'application/json', 'content-type': 'application/json'}, data=payload)
if r.status_code != 200:
print(f"error getting job ID - error {r.status_code} and response {r.text}")
return None
job_id = json.loads(r.text)["id"]
print(f"Got job ID {job_id}. Will now poll for status")
# try ten times
for i in range(0, 30):
time.sleep(2)
poll_url = "https://bridge.danalockservices.com/bridge/v1/poll"
payload = json.dumps({"id": job_id})
r = requests.post(poll_url, headers={'Authorization': 'Bearer ' + token, 'Accept': 'application/json', 'content-type': 'application/json'}, data=payload)
if r.status_code == 404:
print(f"Lock either not on danabridge, lock out of battery, danabridge disconnected, or lock out of danabridge bluetooth range")
return None
elif r.status_code != 200:
print(f"error getting lock PINs - error {r.status_code} and response {r.text}")
return None
if "Succeeded" in r.text:
pin_codes = json.loads(r.text)["result"]["pin_codes"]
# this returns many blank pins - remove all blanks from list
for x in range(len(pin_codes), 0, -1):
if pin_codes[x - 1]["digits"] == "":
pin_codes.pop(x - 1)
return pin_codes
else:
print(f"Attempt {i}: {json.loads(r.text)}")
print(f"did not succeed getting lock PINs after {i*2}s - final response was {r.text}")
return None
# sets pin code #identifier for lock #serial to #new_code
# if new_code is "" then deletees that PIN
def set_pin_code(token, serial, identifier, new_code):
execute_url = "https://bridge.danalockservices.com/bridge/v1/execute"
if new_code == "":
payload = json.dumps({"device": serial, "operation": "afi.pin-codes.set-pin-code", "arguments": [str(identifier), "Cleared", ""]})
else:
payload = json.dumps({"device": serial, "operation": "afi.pin-codes.set-pin-code", "arguments": [str(identifier), "Enabled", new_code]})
r = requests.post(execute_url, headers={'Authorization': 'Bearer ' + token, 'Accept': 'application/json', 'content-type': 'application/json'}, data=payload)
if r.status_code != 200:
print(f"error getting job ID - error {r.status_code} and response {r.text}")
return None
job_id = json.loads(r.text)["id"]
print(f"Got job ID {job_id}. Will now poll for status")
# try ten times
for i in range(0, 30):
time.sleep(2)
poll_url = "https://bridge.danalockservices.com/bridge/v1/poll"
payload = json.dumps({"id": job_id})
r = requests.post(poll_url, headers={'Authorization': 'Bearer ' + token, 'Accept': 'application/json', 'content-type': 'application/json'}, data=payload)
if r.status_code == 404:
print(f"Lock either not on danabridge, lock out of battery, danabridge disconnected, or lock out of danabridge bluetooth range")
return False
elif r.status_code != 200:
print(f"error setting lock status - error {r.status_code} and response {r.text}")
return False
if "Succeeded" in r.text:
result = json.loads(r.text)["result"]
if result["afi_status_text"] == "Ok" and result["dmi_status_text"] == "Ok":
if new_code == "":
print(f"Successfully deleted lock {serial} pin #{identifier}")
else:
print(f"Successfully changed lock {serial} pin #{identifier} to {new_code}")
return True
else:
print(f"Error changing lock {serial} pin #{identifier} to {new_code} - message was '{r.text}'")
return False
else:
print(f"Attempt {i}: {json.loads(r.text)}")
print(f"did not succeed setting lock status after {i*2}s - final response was {r.text}")
return False
def operate_danalock(token, serial, command):
execute_url = "https://bridge.danalockservices.com/bridge/v1/execute"
payload = json.dumps({"device": serial, "operation": "afi.lock.operate", "arguments": [command]})
r = requests.post(execute_url, headers={'Authorization': 'Bearer ' + token, 'Accept': 'application/json', 'content-type': 'application/json'}, data=payload)
if r.status_code != 200:
print(f"error getting job ID - error {r.status_code} and response {r.text}")
return None
elif "id" in r.text:
print(f"Succesfully sent command {command} to lock {serial}")
else:
print(f"Unexpected response to command {command} was {r.text}")
# start here
access_token, refresh_token = danalock_initialise(my_username, my_password)
print(f"got access token {access_token}")
danalocks = get_all_danalocks(access_token)
print(f"Dict with all danalocks on the account: {danalocks}")
# gets all the pincodes for the danalock named "front door"
pin_codes = get_pin_codes(access_token, danalocks["front door"])
print(f"Front door pin codes: {pin_codes}")
# to create a new PIN #2 or change existing PIN #2
# set_pin_code(access_token, danalocks["front door"], 2, "1234")
# to delete existing PIN #2
# set_pin_code(access_token, danalocks["front door"], 2, "")
# to lock the front door
# operate_danalock(access_token, danalocks["front door"], "lock")
# to unlock the front door
# operate_danalock(access_token, danalocks["front door"], "unlock")