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
|
import importlib.resources
import json
import logging
import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.poolmanager import PoolManager
from .exceptions import SHCSessionError
logger = logging.getLogger("boschshcpy")
class JSONRPCError(Exception):
def __init__(self, code, message):
super().__init__()
self._code = code
self._message = message
@property
def code(self):
return self._code
@property
def message(self):
return self._message
def __str__(self):
return f"JSONRPCError (code: {self.code}, message: {self.message})"
class HostNameIgnoringAdapter(HTTPAdapter):
def init_poolmanager(self, connections, maxsize, block=False):
self.poolmanager = PoolManager(
num_pools=connections, maxsize=maxsize, block=block, assert_hostname=False
)
class SHCAPI:
def __init__(self, controller_ip: str, certificate, key):
self._certificate = certificate
self._key = key
self._controller_ip = controller_ip
self._api_root = f"https://{self._controller_ip}:8444/smarthome"
self._public_root = f"https://{self._controller_ip}:8446/smarthome/public"
self._rpc_root = f"https://{self._controller_ip}:8444/remote/json-rpc"
# Settings for all API calls
self._requests_session = requests.Session()
self._requests_session.mount("https://", HostNameIgnoringAdapter())
self._requests_session.cert = (self._certificate, self._key)
self._requests_session.headers.update(
{"api-version": "3.2", "Content-Type": "application/json"}
)
self._requests_session.verify = str(
importlib.resources.files("boschshcpy") / "tls_ca_chain.pem"
)
import urllib3
urllib3.disable_warnings()
@property
def controller_ip(self):
return self._controller_ip
def _get_api_result_or_fail(
self,
api_url,
expected_type=None,
expected_element_type=None,
headers=None,
timeout=30,
):
try:
result = self._requests_session.get(
api_url, headers=headers, timeout=timeout
)
if not result.ok:
self._process_nok_result(result)
else:
if len(result.content) > 0:
result = json.loads(result.content)
if expected_type is not None:
assert result["@type"] == expected_type
if expected_element_type is not None:
for result_ in result:
assert result_["@type"] == expected_element_type
return result
else:
return {}
except requests.exceptions.SSLError as e:
raise Exception(f"API call returned SSLError: {e}.")
def _put_api_or_fail(self, api_url, body, timeout=30):
result = self._requests_session.put(
api_url, data=json.dumps(body), timeout=timeout
)
if not result.ok:
self._process_nok_result(result)
if len(result.content) > 0:
return json.loads(result.content)
else:
return {}
def _post_api_or_fail(self, api_url, body, timeout=30):
result = self._requests_session.post(
api_url, data=json.dumps(body), timeout=timeout
)
if not result.ok:
self._process_nok_result(result)
if len(result.content) > 0:
return json.loads(result.content)
else:
return {}
def _process_nok_result(self, result):
logging.error(f"Body: {result.request.body}")
logging.error(f"Headers: {result.request.headers}")
logging.error(f"URL: {result.request.url}")
raise SHCSessionError(
f"API call returned non-OK result (code {result.status_code})!: {result.content}"
)
# API calls here
def get_information(self):
api_url = f"{self._api_root}/information"
try:
result = self._get_api_result_or_fail(api_url)
except Exception as e:
logging.error(f"Failed to get information from SHC controller: {e}")
return None
return result
def get_public_information(self):
api_url = f"{self._public_root}/information"
try:
result = self._get_api_result_or_fail(api_url, headers={})
except Exception as e:
logging.error(f"Failed to get public information from SHC controller: {e}")
return None
return result
def get_rooms(self):
api_url = f"{self._api_root}/rooms"
return self._get_api_result_or_fail(api_url, expected_element_type="room")
def get_scenarios(self):
api_url = f"{self._api_root}/scenarios"
return self._get_api_result_or_fail(api_url, expected_element_type="scenario")
def get_userdefinedstates(self):
api_url = f"{self._api_root}/userdefinedstates"
return self._get_api_result_or_fail(
api_url, expected_element_type="userDefinedState"
)
def get_devices(self):
api_url = f"{self._api_root}/devices"
return self._get_api_result_or_fail(api_url, expected_element_type="device")
def get_device(self, device_id):
api_url = f"{self._api_root}/devices/{device_id}"
return self._get_api_result_or_fail(api_url, expected_type="device")
def get_services(self):
api_url = f"{self._api_root}/services"
return self._get_api_result_or_fail(
api_url, expected_element_type="DeviceServiceData"
)
def get_device_services(self, device_id):
api_url = f"{self._api_root}/devices/{device_id}/services"
return self._get_api_result_or_fail(api_url)
def get_device_service(self, device_id, service_id):
api_url = f"{self._api_root}/devices/{device_id}/services/{service_id}"
return self._get_api_result_or_fail(api_url, expected_type="DeviceServiceData")
def put_device_service_state(self, device_id, service_id, state_update):
api_url = f"{self._api_root}/devices/{device_id}/services/{service_id}/state"
self._put_api_or_fail(api_url, state_update)
def put_shading_shutters_stop(self, device_id):
api_url = f"{self._api_root}/shading/shutters/{device_id}/stop"
self._put_api_or_fail(api_url, body=None)
def get_domain_intrusion_detection(self):
api_url = f"{self._api_root}/intrusion/states/system"
return self._get_api_result_or_fail(api_url, expected_type="systemState")
def post_domain_action(self, path, data=None):
api_url = f"{self._api_root}/{path}"
self._post_api_or_fail(api_url, body=data)
def long_polling_subscribe(self):
data = [
{
"jsonrpc": "2.0",
"method": "RE/subscribe",
"params": ["com/bosch/sh/remote/*", None],
}
]
result = self._post_api_or_fail(self._rpc_root, data)
assert result[0]["jsonrpc"] == "2.0"
if "error" in result[0].keys():
raise JSONRPCError(
result[0]["error"]["code"], result[0]["error"]["message"]
)
else:
return result[0]["result"]
def long_polling_poll(self, poll_id, wait_seconds=30):
data = [
{
"jsonrpc": "2.0",
"method": "RE/longPoll",
"params": [poll_id, wait_seconds],
}
]
result = self._post_api_or_fail(self._rpc_root, data, wait_seconds + 5)
assert result[0]["jsonrpc"] == "2.0"
if "error" in result[0].keys():
raise JSONRPCError(
result[0]["error"]["code"], result[0]["error"]["message"]
)
else:
return result[0]["result"]
def long_polling_unsubscribe(self, poll_id):
data = [{"jsonrpc": "2.0", "method": "RE/unsubscribe", "params": [poll_id]}]
result = self._post_api_or_fail(self._rpc_root, data)
assert result[0]["jsonrpc"] == "2.0"
if "error" in result[0].keys():
raise JSONRPCError(
result[0]["error"]["code"], result[0]["error"]["message"]
)
else:
return result[0]["result"]
|
