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
|
"""Authenticates with the Control4 API, retrieves account and registered
controller info, and retrieves a bearer token for connecting to a Control4 Director.
"""
import aiohttp
import async_timeout
import json
import logging
import datetime
from .error_handling import checkResponseForError
AUTHENTICATION_ENDPOINT = "https://apis.control4.com/authentication/v1/rest"
CONTROLLER_AUTHORIZATION_ENDPOINT = (
"https://apis.control4.com/authentication/v1/rest/authorization"
)
GET_CONTROLLERS_ENDPOINT = "https://apis.control4.com/account/v3/rest/accounts"
APPLICATION_KEY = "78f6791373d61bea49fdb9fb8897f1f3af193f11"
_LOGGER = logging.getLogger(__name__)
class C4Account:
def __init__(
self,
username,
password,
session: aiohttp.ClientSession = None,
):
"""Creates a Control4 account object.
Parameters:
`username` - Control4 account username/email.
`password` - Control4 account password.
`session` - (Optional) Allows the use of an `aiohttp.ClientSession` object for all network requests. This session will not be closed by the library.
If not provided, the library will open and close its own `ClientSession`s as needed.
"""
self.username = username
self.password = password
self.session = session
async def __sendAccountAuthRequest(self):
"""Used internally to retrieve an account bearer token. Returns the entire
JSON response from the Control4 auth API.
"""
dataDictionary = {
"clientInfo": {
"device": {
"deviceName": "pyControl4",
"deviceUUID": "0000000000000000",
"make": "pyControl4",
"model": "pyControl4",
"os": "Android",
"osVersion": "10",
},
"userInfo": {
"applicationKey": APPLICATION_KEY,
"password": self.password,
"userName": self.username,
},
}
}
if self.session is None:
async with aiohttp.ClientSession() as session:
with async_timeout.timeout(10):
async with session.post(
AUTHENTICATION_ENDPOINT, json=dataDictionary
) as resp:
await checkResponseForError(await resp.text())
return await resp.text()
else:
with async_timeout.timeout(10):
async with self.session.post(
AUTHENTICATION_ENDPOINT, json=dataDictionary
) as resp:
await checkResponseForError(await resp.text())
return await resp.text()
async def __sendAccountGetRequest(self, uri):
"""Used internally to send GET requests to the Control4 API,
authenticated with the account bearer token. Returns the entire JSON
response from the Control4 auth API.
Parameters:
`uri` - Full URI to send GET request to.
"""
try:
headers = {"Authorization": "Bearer {}".format(self.account_bearer_token)}
except AttributeError:
msg = "The account bearer token is missing - was your username/password correct? "
_LOGGER.error(msg)
raise
if self.session is None:
async with aiohttp.ClientSession() as session:
with async_timeout.timeout(10):
async with session.get(uri, headers=headers) as resp:
await checkResponseForError(await resp.text())
return await resp.text()
else:
with async_timeout.timeout(10):
async with self.session.get(uri, headers=headers) as resp:
await checkResponseForError(await resp.text())
return await resp.text()
async def __sendControllerAuthRequest(self, controller_common_name):
"""Used internally to retrieve an director bearer token. Returns the
entire JSON response from the Control4 auth API.
Parameters:
`controller_common_name`: Common name of the controller. See `getAccountControllers()` for details.
"""
try:
headers = {"Authorization": "Bearer {}".format(self.account_bearer_token)}
except AttributeError:
msg = "The account bearer token is missing - was your username/password correct? "
_LOGGER.error(msg)
raise
dataDictionary = {
"serviceInfo": {
"commonName": controller_common_name,
"services": "director",
}
}
if self.session is None:
async with aiohttp.ClientSession() as session:
with async_timeout.timeout(10):
async with session.post(
CONTROLLER_AUTHORIZATION_ENDPOINT,
headers=headers,
json=dataDictionary,
) as resp:
await checkResponseForError(await resp.text())
return await resp.text()
else:
with async_timeout.timeout(10):
async with self.session.post(
CONTROLLER_AUTHORIZATION_ENDPOINT,
headers=headers,
json=dataDictionary,
) as resp:
await checkResponseForError(await resp.text())
return await resp.text()
async def getAccountBearerToken(self):
"""Gets an account bearer token for making Control4 online API requests."""
data = await self.__sendAccountAuthRequest()
jsonDictionary = json.loads(data)
try:
self.account_bearer_token = jsonDictionary["authToken"]["token"]
return self.account_bearer_token
except KeyError:
msg = "Did not recieve an account bearer token. Is your username/password correct? "
_LOGGER.error(msg + data)
raise
async def getAccountControllers(self):
"""Returns a dictionary of the information for all controllers registered to an account.
Returns:
```
{
"controllerCommonName": "control4_MODEL_MACADDRESS",
"href": "https://apis.control4.com/account/v3/rest/accounts/000000",
"name": "Name"
}
```
"""
data = await self.__sendAccountGetRequest(GET_CONTROLLERS_ENDPOINT)
jsonDictionary = json.loads(data)
return jsonDictionary["account"]
async def getControllerInfo(self, controller_href):
"""Returns a dictionary of the information of a specific controller.
Parameters:
`controller_href` - The API `href` of the controller (get this from the output of `getAccountControllers()`)
Returns:
```
{
'allowsPatching': True,
'allowsSupport': False,
'blockNotifications': False,
'controllerCommonName': 'control4_MODEL_MACADDRESS',
'controller': {
'href': 'https://apis.control4.com/account/v3/rest/accounts/000000/controller'
},
'created': '2017-08-26T18:33:31Z',
'dealer': {
'href': 'https://apis.control4.com/account/v3/rest/dealers/12345'
},
'enabled': True,
'hasLoggedIn': True,
'href': 'https://apis.control4.com/account/v3/rest/accounts/000000',
'id': 000000,
'lastCheckIn': '2020-06-13T21:52:34Z',
'licenses': {
'href': 'https://apis.control4.com/account/v3/rest/accounts/000000/licenses'
},
'modified': '2020-06-13T21:52:34Z',
'name': 'Name',
'provisionDate': '2017-08-26T18:35:11Z',
'storage': {
'href': 'https://apis.control4.com/storage/v1/rest/accounts/000000'
},
'type': 'Consumer',
'users': {
'href': 'https://apis.control4.com/account/v3/rest/accounts/000000/users'
}
}
```
"""
data = await self.__sendAccountGetRequest(controller_href)
jsonDictionary = json.loads(data)
return jsonDictionary
async def getControllerOSVersion(self, controller_href):
"""Returns the OS version of a controller as a string.
Parameters:
`controller_href` - The API `href` of the controller (get this from the output of `getAccountControllers()`)
"""
data = await self.__sendAccountGetRequest(controller_href + "/controller")
jsonDictionary = json.loads(data)
return jsonDictionary["osVersion"]
async def getDirectorBearerToken(self, controller_common_name):
"""Returns a dictionary with a director bearer token for making Control4 Director API requests, and its time valid in seconds (usually 86400 seconds)
Parameters:
`controller_common_name`: Common name of the controller. See `getAccountControllers()` for details.
"""
data = await self.__sendControllerAuthRequest(controller_common_name)
jsonDictionary = json.loads(data)
return {
"token": jsonDictionary["authToken"]["token"],
"validSeconds": jsonDictionary["authToken"]["validSeconds"],
}
|
