diff options
| -rw-r--r-- | .gitignore | 1 | ||||
| -rw-r--r-- | ShellyPy/__init__.py | 4 | ||||
| -rw-r--r-- | ShellyPy/base.py | 107 | ||||
| -rw-r--r-- | ShellyPy/gen1.py | 227 | ||||
| -rw-r--r-- | ShellyPy/gen2.py | 136 | ||||
| -rw-r--r-- | ShellyPy/wrapper.py | 284 |
6 files changed, 506 insertions, 253 deletions
@@ -1,3 +1,4 @@ +env build dist *.egg-info diff --git a/ShellyPy/__init__.py b/ShellyPy/__init__.py index 145cd2a..9a357f7 100644 --- a/ShellyPy/__init__.py +++ b/ShellyPy/__init__.py @@ -1,3 +1,5 @@ -__version__ = "0.1.4" +__version__ = "0.2.0" from .wrapper import Shelly +from .gen1 import ShellyGen1 +from .gen2 import ShellyGen2 diff --git a/ShellyPy/base.py b/ShellyPy/base.py new file mode 100644 index 0000000..6c64785 --- /dev/null +++ b/ShellyPy/base.py @@ -0,0 +1,107 @@ +from sys import version_info + +if version_info.major == 3: + from json.decoder import JSONDecodeError +else: + JSONDecodeError = ValueError + +from requests.auth import HTTPBasicAuth + +class 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 + """ + + self.__name__ = "Unknown" + self.__type__ = "Unknown" + self.__generation__ = 0 + + self.__debugging__ = kwargs.get("debug", None) + + self.__PROTOCOL__ = "http" + + login = kwargs.get("login", {}) + + # hostname would be more fitting, + # but backwards compatibility + self.__ip__ = ip + + self.__port__ = port + + self.__timeout__ = kwargs.get("timeout", 5) + + self.__credentials__ = HTTPBasicAuth( + login.get("username", ""), login.get("password", "") + ) + + if kwargs.get("init"): + self.update() + + def __repr__(self): + return "<{} {} Gen {} ({})>".format(self.__name__, self.__type__, self.__generation__, self.__ip__) + + def __str__(self): + return str(self.__name__) + + @staticmethod + def __clamp__(val): + """clamp any number to 8 bit""" + if val > 255: + val = 255 + elif val < 0: + val = 0 + + return val + + @staticmethod + 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 + + @staticmethod + def __clamp_kalvin__(val): + """ + clamp given kalvin values for a range from 3000..6500 + """ + if val > 6500: + val = 6500 + elif val < 3000: + val = 3000 + return val + + def update(self): + raise NotImplementedError("Base Class") + + def post(self, page, values = None): + raise NotImplementedError("Base Class") + + def status(self): + raise NotImplementedError("Base Class") + + def settings(self, subpage = None): + raise NotImplementedError("Base Class") + + def relay(self, index, *args, **kwargs): + raise NotImplementedError("Base Class") + + def roller(self, index, *args, **kwargs): + raise NotImplementedError("Base Class") + + def light(self, index, *args, **kwargs): + raise NotImplementedError("Base Class") + + def emeter(self, index, *args, **kwargs): + raise NotImplementedError("Base Class") diff --git a/ShellyPy/gen1.py b/ShellyPy/gen1.py new file mode 100644 index 0000000..ea8c975 --- /dev/null +++ b/ShellyPy/gen1.py @@ -0,0 +1,227 @@ +from sys import version_info + +if version_info.major == 3: + from json.decoder import JSONDecodeError +else: + JSONDecodeError = ValueError + + +from requests import post + +from .error import BadLogin, NotFound, BadResponse + +from .base import ShellyBase + +class ShellyGen1(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__ = 1 + + def update(self): + """ + @brief update the Shelly attributes + """ + status = self.settings() + + self.__type__ = status['device'].get("type", self.__type__) + self.__name__ = status['device'].get("hostname", self.__name__) + + # Settings are already fetched to get device information might as well put the list of things the device has somewhere + self.relays = status.get("relays", []) + self.rollers = status.get("rollers", []) + # RGBW reuses the same lights array + self.lights = status.get("lights", []) + + self.irs = status.get("light_sensor", None) + + self.emeters = 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 status(self): + """ + @brief returns status response + + @return status dict + """ + return self.post("status") + + 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 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) + + values = {} + + if go: + values["go"] = go + + if roller_pos is not None: + values["roller_pos"] = self.__clamp_percentage__(roller_pos) + + if duration is not None: + values["duration"] = duration + + return self.post("roller/{}".format(index), values) + + def light(self, index, *args, **kwargs): + """ + @brief Interacts with lights at a given index + + @param mode Accepts "white" and "color" as possible modes + @param index index of the light. When in doubt use 0 + @param timer a one-shot flip-back timer in seconds + @param turn Will turn the lights on or off + @param red Red brightness, 0..255, only works if mode="color" + @param green Green brightness, 0..255, only works if mode="color" + @param blue Blue brightness, 0..255, only works if mode="color" + @param white White brightness, 0..255, only works if mode="color" + @param gain Gain for all channels, 0...100, only works if mode="color" + @param temp Color temperature in K, 3000..6500, only works if mode="white" + @param brightness Brightness, 0..100, only works if mode="white" + """ + 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) + + 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"] = self.__clamp__(red) + + if green is not None: + values["green"] = self.__clamp__(green) + + if blue is not None: + values["blue"] = self.__clamp__(blue) + + if white is not None: + values["white"] = self.__clamp__(white) + + if gain is not None: + values["gain"] = self.__clamp_percentage__(gain) + + if temp is not None: + values["temp"] = self.__clamp_kalvin__(temp) + + if brightness is not None: + values["brightness"] = self.__clamp_percentage__(brightness) + + return self.post("light/{}".format(index), values) + + def emeter(self, index, *args, **kwargs): + + return self.post("emeter/{}".format(index)) + +# backwards compatibility with old code +Shelly = ShellyGen1
\ No newline at end of file diff --git a/ShellyPy/gen2.py b/ShellyPy/gen2.py new file mode 100644 index 0000000..38824a6 --- /dev/null +++ b/ShellyPy/gen2.py @@ -0,0 +1,136 @@ +from sys import version_info + +if version_info.major == 3: + from json.decoder import JSONDecodeError +else: + JSONDecodeError = ValueError + +from requests import post + +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 + + 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__) + + payload = { + "id": 1, + "method": page, + } + + if values: + payload["params"] = values + + response = post(url, auth=self.__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)) + + return response_data.get("result", {}) + + def status(self): + return self.post("Sys.GetStatus") + + def settings(self, subpage = None): + return self.post("Sys.GetConfig") + + 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")
\ No newline at end of file diff --git a/ShellyPy/wrapper.py b/ShellyPy/wrapper.py index 2415063..021a446 100644 --- a/ShellyPy/wrapper.py +++ b/ShellyPy/wrapper.py @@ -5,276 +5,56 @@ if version_info.major == 3: else: JSONDecodeError = ValueError - from requests import post -from requests.auth import HTTPBasicAuth -import error +from .error import BadLogin, NotFound, BadResponse +from .gen1 import ShellyGen1 +from .gen2 import ShellyGen2 -class Shelly: +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 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" - - 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__) - - @staticmethod - def __clamp__(val): - """clamp any number to 8 bit""" - if val > 255: - val = 255 - elif val < 0: - val = 0 - - return val - - @staticmethod - 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 - - @staticmethod - def __clamp_kalvin__(val): - """ - clamp given kalvin values for a range from 3000..6500 - """ - if val > 6500: - val = 6500 - elif val < 3000: - val = 3000 - return val - - def update(self): - """ - @brief update the Shelly attributes + @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 """ - status = self.settings() - - self.__type__ = status['device']['type'] - self.__name__ = status['device']['hostname'] - # Settings are already fetched to get device information might as well put the list of things the device has somewhere - self.relays = status.get("relays", []) - self.rollers = status.get("rollers", []) - # RGBW reuses the same lights array - self.lights = status.get("lights", []) + self._instance = self.__detect__(ip, port)(ip, port, *args, **kwargs) - self.irs = status.get("light_sensor", None) + def __detect__(self, ip, port): + url = "{}://{}:{}/shelly".format("http", ip, port) - self.emeters = 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__) + response = post(url, timeout=5) if response.status_code == 401: - raise error.BadLogin() + raise BadLogin() elif response.status_code == 404: - raise error.NotFound("Not Found") + raise NotFound("Not Found") try: - return response.json() + response_data = response.json() except JSONDecodeError: - raise error.BadResponse("Bad JSON") - - def status(self): - """ - @brief returns status response - - @return status dict - """ - return self.post("status") - - 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 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) - - values = {} - - if go: - values["go"] = go - - if roller_pos is not None: - values["roller_pos"] = self.__clamp_percentage__(roller_pos) - - if duration is not None: - values["duration"] = duration - - return self.post("roller/{}".format(index), values) - - def light(self, index, *args, **kwargs): - """ - @brief Interacts with lights at a given index - - @param mode Accepts "white" and "color" as possible modes - @param index index of the light. When in doubt use 0 - @param timer a one-shot flip-back timer in seconds - @param turn Will turn the lights on or off - @param red Red brightness, 0..255, only works if mode="color" - @param green Green brightness, 0..255, only works if mode="color" - @param blue Blue brightness, 0..255, only works if mode="color" - @param white White brightness, 0..255, only works if mode="color" - @param gain Gain for all channels, 0...100, only works if mode="color" - @param temp Color temperature in K, 3000..6500, only works if mode="white" - @param brightness Brightness, 0..100, only works if mode="white" - """ - 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) - - 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"] = self.__clamp__(red) - - if green is not None: - values["green"] = self.__clamp__(green) - - if blue is not None: - values["blue"] = self.__clamp__(blue) - - if white is not None: - values["white"] = self.__clamp__(white) - - if gain is not None: - values["gain"] = self.__clamp_percentage__(gain) - - if temp is not None: - values["temp"] = self.__clamp_kalvin__(temp) - - if brightness is not None: - values["brightness"] = self.__clamp_percentage__(brightness) + raise BadResponse("Bad JSON") + + gen = response_data.get("gen", 1) + + if gen == 1: + return ShellyGen1 + elif gen == 2: + return ShellyGen2 + else: + raise ValueError("Generation {} not supported".format(gen)) - return self.post("light/{}".format(index), values) + def __repr__(self): + return self.__getattr__("__repr__")() - def emeter(self, index, *args, **kwargs): + def __str__(self): + return self.__getattr__("__str__")() - return self.post("emeter/{}".format(index)) + def __getattr__(self, name): + return getattr(self._instance, name) |