Initial Commit0.1.0

9 files changed, 359 insertions, 0 deletions

diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..da91870
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,4 @@
+build
+dist
+*.egg-info
+__pycache__
\ No newline at end of file
diff --git a/LICENSE b/LICENSE
new file mode 100644
index 0000000..f7174f2
--- /dev/null
+++ b/LICENSE
@@ -0,0 +1,22 @@
+MIT License
+
+Copyright (c) 2019 Jan Drögehoff
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
+
diff --git a/README.md b/README.md
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/README.md
diff --git a/ShellyPy/__init__.py b/ShellyPy/__init__.py
new file mode 100644
index 0000000..d14ca66
--- /dev/null
+++ b/ShellyPy/__init__.py
@@ -0,0 +1,3 @@
+__version__ = "0.1.0"
+
+from .shelly import Shelly
diff --git a/ShellyPy/error.py b/ShellyPy/error.py
new file mode 100644
index 0000000..e595bed
--- /dev/null
+++ b/ShellyPy/error.py
@@ -0,0 +1,25 @@
+
+class MalformedIP(Exception):
+    """
+    Exception for malformed IPv4 adresses
+    """
+    pass
+
+
+class BadLogin(Exception):
+    """
+    Exception for bad login details
+    """
+    pass
+
+class BadResponse(Exception):
+    """
+    Exception for bad responses from the target
+    """
+    pass
+
+class NotFound(Exception):
+    """
+    Exception for 404 Not Found
+    """
+    pass
diff --git a/ShellyPy/shelly.py b/ShellyPy/shelly.py
new file mode 100644
index 0000000..f167c22
--- /dev/null
+++ b/ShellyPy/shelly.py
@@ -0,0 +1,261 @@
+from typing import Union, List, Dict
+from json.decoder import JSONDecodeError
+
+from requests import post
+from requests.auth import HTTPBasicAuth
+
+from .error import *
+
+
+def confirm_ip(ip: Union[str, List[str], List[int]]) -> bool:
+    """
+    @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: Union[str, List[str], List[int]], port: Union[str, int] = "80", *, login: Dict[str, str]=None, **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")
+
+        if login is None:
+            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 f"<{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: str, values: Dict[str, Union[str, int]]=None) -> dict:
+        """
+        @brief      returns settings
+
+        @param      page   page to be accesed. Use the Shelly HTTP API Reference to see whats possible
+
+        @return     returns json response
+        """
+
+        url = f"{self.__PROTOCOL__}://{self.__ip__}:{self.__port__}/{page}?"
+
+        if values:
+            url += "&".join([f"{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: str = None) -> dict:
+        """
+        @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: int, *, turn: bool = None, timer: float = None) -> dict:
+        """
+        @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 = {}
+
+        if turn is not None:
+            if turn:
+                values["turn"] = "on"
+            else:
+                values["turn"] = "off"
+
+        if timer:
+            values["timer"] = timer
+
+        return self.post(f"relay/{index}", values)
+
+    def roller(self, index: int, *, go: str = None,
+               roller_pos: float = None, duration: float = None) -> dict:
+        """
+        @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
+        """
+
+        def clamp_percentage(val: int):
+            """
+            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(f"roller/{index}", values)
+
+    def light(self, index: int, *, mode: str = None, timer: int = None, turn: bool = None,
+              red: int = None, green: int = None, blue: int = None, white: int = None,
+              gain: int = None, temp: int = None, brightness: int = None) -> dict:
+
+        def clamp(val: int) -> int:
+            """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(f"light/{index}", values)
+
+    def emeter(self, i: int, *args, **kwargs):
+        #TODO
+        return
diff --git a/examples/toggle_lights.py b/examples/toggle_lights.py
new file mode 100644
index 0000000..191cf36
--- /dev/null
+++ b/examples/toggle_lights.py
@@ -0,0 +1,6 @@
+import ShellyPy
+
+device = ShellyPy.Shelly("192.168.0.5") # try connecting to the Shelly device under that ip
+
+device.relay(0, turn=True) # turn the relay at index 0 on
+device.relay(0, turn=False) # same as bove but turn it off
diff --git a/requirements.txt b/requirements.txt
new file mode 100644
index 0000000..f229360
--- /dev/null
+++ b/requirements.txt
@@ -0,0 +1 @@
+requests
diff --git a/setup.py b/setup.py
new file mode 100755
index 0000000..f62ed46
--- /dev/null
+++ b/setup.py
@@ -0,0 +1,37 @@
+import setuptools
+import re
+
+requirements = []
+with open('requirements.txt') as f:
+  requirements = f.read().splitlines()
+
+version = ''
+with open('ShellyPy/__init__.py') as f:
+    version = re.search(r'^__version__\s*=\s*[\'"]([^\'"]*)[\'"]', f.read(), re.MULTILINE).group(1)
+
+if not version:
+    raise RuntimeError('version is not set')
+
+long_description = ""
+with open("README.md", "r") as fh:
+    long_description = fh.read()
+
+
+setuptools.setup(
+    name="ShellyPy",
+    version=version,
+    author="Jan Drögehoff",
+    description="Wrapper around the Shelly HTTP api",
+    long_description=long_description,
+    long_description_content_type="text/markdown",
+    url="https://github.com/Jan200101/ShellyPy",
+    packages=["ShellyPy"],
+    license="MIT",
+    install_requires=requirements,
+    classifiers=[
+        "Programming Language :: Python :: 2",
+        "Programming Language :: Python :: 3",
+        "License :: OSI Approved :: MIT License",
+        "Operating System :: OS Independent",
+    ],
+)
\ No newline at end of file