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
|
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 HTTPDigestAuth
from .error import BadLogin, NotFound, BadResponse
from .base import ShellyBase
class ShellyGen2(ShellyBase):
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
@param init calls the update method on init
"""
super().__init__(ip, port, *args, **kwargs)
self.__generation__ = 2
self.payload_id = 1
def update(self):
status = self.settings()
self.__name__ = status["device"].get("name", self.__name__)
self.__type__ = status["device"].get("mac", self.__type__)
def post(self, page, values = None):
url = "{}://{}:{}/rpc".format(self.__PROTOCOL__, self.__ip__, self.__port__)
# increment payload id globally
self.payload_id += 1
# but keep a local copy around so we face no race conditions
payload_id = self.payload_id
payload = {
"jsonrpc": "2.0",
"id": payload_id,
"method": page,
}
if values:
payload["params"] = values
credentials = None
try:
credentials = auth=HTTPDigestAuth('admin', self.__credentials__[1])
except IndexError:
pass
response = post(url, auth=credentials,
json=payload,
timeout=self.__timeout__)
if response.status_code == 401:
raise BadLogin()
elif response.status_code == 404:
raise NotFound("Not Found")
try:
response_data = response.json()
except JSONDecodeError:
raise BadResponse("Bad JSON")
if "error" in response_data:
error_code = response_data["error"].get("code", None)
error_message = response_data["error"].get("message", "")
if error_code == 401:
raise BadLogin(error_message)
elif error_code == 404:
raise NotFound(error_message)
else:
raise BadResponse("{}: {}".format(error_code, error_message))
if response_data["id"] != payload_id:
raise BadResponse("invalid payload id was returned")
return response_data.get("result", {})
def status(self):
return self.post("Sys.GetStatus")
def settings(self, subpage = None):
return self.post("Sys.GetConfig")
def meter(self, index):
raise NotImplementedError("Unavailable")
def relay(self, index, *args, **kwargs):
values = {
"id": index
}
turn = kwargs.get("turn", None)
timer = kwargs.get("timer", None)
if turn is not None:
method = "Switch.Set"
if turn:
values["on"] = True
else:
values["on"] = False
if timer:
values["toggle_after"] = timer
else:
method = "Switch.GetStatus"
return self.post(method, values)
def roller(self, index, *args, **kwargs):
go = kwargs.get("go", None)
roller_pos = kwargs.get("roller_pos", None)
duration = kwargs.get("duration", None)
values = {
"id": index
}
if go:
if go == "open":
method = "Cover.Open"
elif go == "close":
method = "Cover.Close"
elif go == "stop":
method = "Cover.Stop"
else:
raise ValueError("Method is not open, close or stop")
if roller_pos is not None:
method = "Cover.GoToPosition"
values["pos"] = self.__clamp_percentage__(roller_pos)
if duration is not None:
values["duration"] = duration
return self.post(method, values)
def light(self, index, *args, **kwargs):
raise NotImplementedError("Unavailable")
def emeter(self, index, *args, **kwargs):
raise NotImplementedError("Unavailable")
|
