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
276
277
278
279
280
|
from sys import version_info
if version_info.major == 3:
from json.decoder import JSONDecodeError
else:
JSONDecodeError = ValueError
from requests import post
from requests.auth import HTTPBasicAuth
from .error import *
def confirm_ip(ip):
"""
@brief Confirm IPv4 adress
@param ip IP given as either a string or List containing strings or integers
@return returns True if ip is valid
"""
if isinstance(ip, str):
ip = ip.split(".")
if len(ip) != 4:
return False
if isinstance(ip, list):
for value in ip:
if isinstance(value, str):
try:
value = int(value)
except ValueError:
value = 256
if value > 255 or value < 0:
return False
return True
return False
class Shelly:
def __init__(self, ip, port = "80", *args, **kwargs):
"""
@param ip the target IP of the shelly device. Can be a string, list of strings or list of integers
@param port target port, may be useful for non Shelly devices that have the same HTTP Api
@param login dict of login credentials. Keys needed are "username" and "password"
@param timeout specify the amount of time until requests are aborted.
@param debug enable debug printing
"""
# TODO add domain support
self.__debugging__ = kwargs.get("debug", None)
self.__PROTOCOL__ = "http"
if not confirm_ip(ip):
raise MalformedIP("IP is is malformed or not IPv4")
login = kwargs.get("login", {})
if isinstance(ip, list):
self.__ip__ = ".".join([str(val) for val in ip])
else:
self.__ip__ = ip
self.__port__ = port
self.__timeout__ = kwargs.get("timeout", 5)
self.__credentials__ = HTTPBasicAuth(
login.get("username", ""), login.get("password", ""))
self.update()
def __repr__(self):
return "<{} {} ({})>".format(self.__name__, self.__type__, self.__ip__)
def __str__(self):
return str(self.__name__)
def update(self):
"""
@brief update the Shelly attributes
"""
status = self.settings()
self.__type__ = status['device']['type']
self.__name__ = status['device']['hostname']
self.__relay__ = status.get("relays", [])
self.__roller__ = status.get("rollers", [])
self.__light__ = status.get("lights", [])
# documentation for the Shelly Sense is very weird
# FIXME
self.__ir__ = status.get("ir", [])
# There isn't even an example of the response for the RGBW
# FIXME
self.__color__ = status.get("color", [])
self.__emeter__ = status.get("emeter", [])
def post(self, page, values = None):
"""
@brief returns settings
@param page page to be accesed. Use the Shelly HTTP API Reference to see whats possible
@return returns json response
"""
url = "{}://{}:{}/{}?".format(self.__PROTOCOL__, self.__ip__, self.__port__, page)
if values:
url += "&".join(["{}={}".format(key, value) for key, value in values.items()])
if self.__debugging__:
print("Target Adress: {}\n"
"Authentication: {}\n"
"Timeout: {}"
"".format(url, any(self.__credentials__.username + self.__credentials__.password), self.__timeout__))
response = post(url, auth=self.__credentials__,
timeout=self.__timeout__)
if response.status_code == 401:
raise BadLogin()
elif response.status_code == 404:
raise NotFound("Not Found")
try:
return response.json()
except JSONDecodeError:
raise BadResponse("Bad JSON")
def settings(self, subpage = None):
"""
@brief returns settings
@param page page to be accesed. Use the Shelly HTTP API Reference to see whats possible
@return returns settings as a dict
"""
page = "settings"
if subpage:
page += "/" + subpage
return self.post(page)
def relay(self, index, *args, **kwargs):
"""
@brief Interacts with a relay at the given index
@param self The object
@param index index of the relay
@param turn Will turn the relay on or off
@param timer a one-shot flip-back timer in seconds
"""
values = {}
turn = kwargs.get("turn", None)
timer = kwargs.get("timer", None)
if turn is not None:
if turn:
values["turn"] = "on"
else:
values["turn"] = "off"
if timer:
values["timer"] = timer
return self.post("relay/{}".format(index), values)
def roller(self, index, *args, **kwargs):
"""
@brief Interacts with a roller at a given index
@param self The object
@param index index of the roller. When in doubt use 0
@param go way of the roller to go. Accepted are "open", "close", "stop", "to_pos"
@param roller_pos the wanted position in percent
@param duration how long it will take to get to that position
"""
go = kwargs.get("go", None)
roller_pos = kwargs.get("roller_pos", None)
duration = kwargs.get("duration", None)
def clamp_percentage(val):
"""
clamp given percentage to a range from 0 to 100
"""
if val > 100:
val = 100
elif val < 0:
val = 0
return val
values = {}
if go:
values["go"] = go
if roller_pos is not None:
values["roller_pos"] = clamp_percentage(roller_pos)
if duration is not None:
values["duration"] = duration
return self.post("roller/{index}".format(index), values)
def light(self, index, *args, **kwargs):
mode = kwargs.get("mode", None)
timer = kwargs.get("timer", None)
turn = kwargs.get("turn", None)
red = kwargs.get("red", None)
green = kwargs.get("green", None)
blue = kwargs.get("blue", None)
white = kwargs.get("white", None)
gain = kwargs.get("gain", None)
temp = kwargs.get("temp", None)
brightness = kwargs.get("brightness", None)
def clamp(val):
"""clamp any number to 8 bit"""
if val > 255:
val = 255
elif val < 0:
val = 0
return val
values = {}
if mode:
values["mode"] = mode
if timer is not None:
values["timer"] = timer
if turn is not None:
if turn:
values["turn"] = "on"
else:
values["turn"] = "off"
if red is not None:
values["red"] = clamp(red)
if green is not None:
values["green"] = clamp(green)
if blue is not None:
values["blue"] = clamp(blue)
if white is not None:
values["white"] = clamp(white)
if gain is not None:
values["gain"] = clamp(gain)
if temp is not None:
values["temp"] = temp
if brightness is not None:
values["brightness"] = brightness
return self.post("light/{index}".format(index), values)
def emeter(self, i, *args, **kwargs):
#TODO
return
|
