From 6dcd8edc69fd405164f09589d8316cf302da571a Mon Sep 17 00:00:00 2001 From: Jan200101 Date: Mon, 23 Sep 2019 08:14:50 +0200 Subject: release 0.1.2 add UTF-8 encoding to setup header rename shelly.py to wrapper.py to get rid of any confusion between Shelly and shelly --- ShellyPy/__init__.py | 4 +- ShellyPy/shelly.py | 280 --------------------------------------------------- ShellyPy/wrapper.py | 280 +++++++++++++++++++++++++++++++++++++++++++++++++++ setup.py | 1 + 4 files changed, 283 insertions(+), 282 deletions(-) delete mode 100644 ShellyPy/shelly.py create mode 100644 ShellyPy/wrapper.py diff --git a/ShellyPy/__init__.py b/ShellyPy/__init__.py index 802ff5d..63908d4 100644 --- a/ShellyPy/__init__.py +++ b/ShellyPy/__init__.py @@ -1,3 +1,3 @@ -__version__ = "0.1.1" +__version__ = "0.1.2" -from .shelly import Shelly +from .wrapper import Shelly diff --git a/ShellyPy/shelly.py b/ShellyPy/shelly.py deleted file mode 100644 index adb34e4..0000000 --- a/ShellyPy/shelly.py +++ /dev/null @@ -1,280 +0,0 @@ -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 diff --git a/ShellyPy/wrapper.py b/ShellyPy/wrapper.py new file mode 100644 index 0000000..adb34e4 --- /dev/null +++ b/ShellyPy/wrapper.py @@ -0,0 +1,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 diff --git a/setup.py b/setup.py index 1d60440..8a49edf 100755 --- a/setup.py +++ b/setup.py @@ -1,3 +1,4 @@ +# -*- coding: utf-8 -*- import setuptools import re -- cgit v1.2.3