aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorJan200101 <sentrycraft123@gmail.com>2023-07-09 19:37:37 +0200
committerJan200101 <sentrycraft123@gmail.com>2023-07-09 19:37:37 +0200
commitb966f6e4705a813dad13b242e9592e699566d9c9 (patch)
treefba1a57bf3ed01b7826487c98f000ae9f111cf20
downloadShellyPy-b966f6e4705a813dad13b242e9592e699566d9c9.tar.gz
ShellyPy-b966f6e4705a813dad13b242e9592e699566d9c9.zip
Squash rewriterewrite
-rw-r--r--.gitignore7
-rw-r--r--LICENSE22
-rw-r--r--README.md1
-rw-r--r--ShellyPy/__init__.py8
-rw-r--r--ShellyPy/api/__init__.py4
-rw-r--r--ShellyPy/api/device.py55
-rw-r--r--ShellyPy/api/gen1/__init__.py2
-rw-r--r--ShellyPy/api/gen1/backends/__init__.py2
-rw-r--r--ShellyPy/api/gen1/backends/http/__init__.py7
-rw-r--r--ShellyPy/api/gen1/backends/http/light.py104
-rw-r--r--ShellyPy/api/gen1/backends/http/meter.py23
-rw-r--r--ShellyPy/api/gen1/backends/http/relay.py54
-rw-r--r--ShellyPy/api/gen1/backends/http/request.py90
-rw-r--r--ShellyPy/api/gen1/backends/http/roller.py63
-rw-r--r--ShellyPy/api/gen1/backends/http/settings.py39
-rw-r--r--ShellyPy/api/gen1/device.py56
-rw-r--r--ShellyPy/api/gen2/__init__.py2
-rw-r--r--ShellyPy/api/gen2/backends/__init__.py2
-rw-r--r--ShellyPy/api/gen2/backends/json_rpc/__init__.py7
-rw-r--r--ShellyPy/api/gen2/backends/json_rpc/light.py73
-rw-r--r--ShellyPy/api/gen2/backends/json_rpc/meter.py24
-rw-r--r--ShellyPy/api/gen2/backends/json_rpc/relay.py33
-rw-r--r--ShellyPy/api/gen2/backends/json_rpc/request.py136
-rw-r--r--ShellyPy/api/gen2/backends/json_rpc/roller.py45
-rw-r--r--ShellyPy/api/gen2/backends/json_rpc/settings.py25
-rw-r--r--ShellyPy/api/gen2/device.py106
-rw-r--r--ShellyPy/base/__init__.py8
-rw-r--r--ShellyPy/base/device.py167
-rw-r--r--ShellyPy/base/hints.py17
-rw-r--r--ShellyPy/base/light.py248
-rw-r--r--ShellyPy/base/meter.py53
-rw-r--r--ShellyPy/base/relay.py88
-rw-r--r--ShellyPy/base/roller.py128
-rw-r--r--ShellyPy/base/settings.py56
-rw-r--r--ShellyPy/discovery/__init__.py0
-rw-r--r--ShellyPy/exceptions/__init__.py4
-rw-r--r--ShellyPy/exceptions/backend.py6
-rw-r--r--ShellyPy/exceptions/request.py9
-rw-r--r--ShellyPy/exceptions/timer.py3
-rw-r--r--ShellyPy/utils/__init__.py4
-rw-r--r--ShellyPy/utils/attribute_list.py36
-rw-r--r--ShellyPy/utils/clamp.py13
-rw-r--r--ShellyPy/utils/property_fetcher.py16
-rw-r--r--examples/gen.py10
-rw-r--r--examples/light.py27
-rw-r--r--examples/relay.py15
-rw-r--r--examples/roller.py18
-rw-r--r--requirements.txt0
-rwxr-xr-xsetup.py54
49 files changed, 1970 insertions, 0 deletions
diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..6547b63
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,7 @@
+env
+build
+dist
+*.egg-info
+__pycache__
+*.pyc
+.mypy_cache
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..1a47272
--- /dev/null
+++ b/README.md
@@ -0,0 +1 @@
+# ShellyPy
diff --git a/ShellyPy/__init__.py b/ShellyPy/__init__.py
new file mode 100644
index 0000000..21e8dfb
--- /dev/null
+++ b/ShellyPy/__init__.py
@@ -0,0 +1,8 @@
+
+__title__ = "ShellyPy"
+__license__ = "MIT"
+__version__ = "1.0.0"
+
+from . import api
+from . import exceptions
+from . import base
diff --git a/ShellyPy/api/__init__.py b/ShellyPy/api/__init__.py
new file mode 100644
index 0000000..d4b0caa
--- /dev/null
+++ b/ShellyPy/api/__init__.py
@@ -0,0 +1,4 @@
+
+from . import gen1
+from . import gen2
+from .device import Device
diff --git a/ShellyPy/api/device.py b/ShellyPy/api/device.py
new file mode 100644
index 0000000..3193486
--- /dev/null
+++ b/ShellyPy/api/device.py
@@ -0,0 +1,55 @@
+from ..exceptions.backend import InvalidBackend
+from ..exceptions.request import UnauthorizedException, InvalidRequestException
+
+from . import gen1
+from . import gen2
+
+from typing import Optional
+import json
+from urllib.request import urlopen
+
+class Device:
+
+ def __init__(self, hostname: str, port: Optional[int] = None, *args, **kwargs):
+ self._instance = self.__detect__(hostname, port)(hostname, port, *args, **kwargs)
+
+ @staticmethod
+ def __detect__(ip, port):
+ url = f"http://{ip}:{port or 80}/shelly"
+
+ resp = urlopen(url, timeout=5)
+
+ status_code = resp.getcode()
+
+ if status_code == 401:
+ raise UnauthorizedException()
+ elif status_code == 404:
+ raise InvalidRequestException("Endpoint Not Found")
+
+ try:
+ response_data = json.loads(resp.read().decode())
+ except JSONDecodeError:
+ raise InvalidRequestException("Received Invalid Response")
+
+ gen = response_data.get("gen", 1)
+
+ if gen == 1:
+ return gen1.Device
+ elif gen == 2:
+ return gen2.Device
+ else:
+ raise InvalidBackend(f"Generation {gen} not supported")
+
+ def __repr__(self):
+ return self.__getattr__("__repr__")()
+
+ def __str__(self):
+ return self.__getattr__("__str__")()
+
+ def __getattr__(self, name):
+ return getattr(self._instance, name)
+
+ @classmethod
+ def connect(cls, hostname: str, port: Optional[int] = None, *args, **kwargs):
+ instance = cls.__detect__(hostname, port)(hostname, port, *args, **kwargs)
+ return instance
diff --git a/ShellyPy/api/gen1/__init__.py b/ShellyPy/api/gen1/__init__.py
new file mode 100644
index 0000000..e71ed1d
--- /dev/null
+++ b/ShellyPy/api/gen1/__init__.py
@@ -0,0 +1,2 @@
+
+from .device import *
diff --git a/ShellyPy/api/gen1/backends/__init__.py b/ShellyPy/api/gen1/backends/__init__.py
new file mode 100644
index 0000000..5d189ca
--- /dev/null
+++ b/ShellyPy/api/gen1/backends/__init__.py
@@ -0,0 +1,2 @@
+
+from . import http
diff --git a/ShellyPy/api/gen1/backends/http/__init__.py b/ShellyPy/api/gen1/backends/http/__init__.py
new file mode 100644
index 0000000..9859b81
--- /dev/null
+++ b/ShellyPy/api/gen1/backends/http/__init__.py
@@ -0,0 +1,7 @@
+
+from .light import *
+from .meter import *
+from .relay import *
+from .request import *
+from .roller import *
+from .settings import *
diff --git a/ShellyPy/api/gen1/backends/http/light.py b/ShellyPy/api/gen1/backends/http/light.py
new file mode 100644
index 0000000..bd67599
--- /dev/null
+++ b/ShellyPy/api/gen1/backends/http/light.py
@@ -0,0 +1,104 @@
+from datetime import datetime, timedelta
+from typing import Optional, Tuple, Union, Dict, Any
+
+from .request import Request
+from .....base import Light as BaseLight
+from .....base.hints import (
+ rgbw_mode, byte, percent,
+ temperaur, transition
+)
+from .....utils import (
+ clamp, clamp_percent, clamp_byte,
+ clamp_temp, property_fetcher,
+)
+from .....exceptions import InvalidTimer
+
+class Light(BaseLight):
+
+ def toggle(self, timer: Optional[int] = None):
+ parameters: Dict[str, Any] = {
+ "turn": "toggle"
+ }
+ if timer is not None:
+ parameters["timer"] = timer
+
+ self._fetch(f"light/{self.index}", parameters)
+
+ def on(self, timer: Optional[int] = None):
+ parameters: Dict[str, Any] = {
+ "turn": "on"
+ }
+ if timer is not None:
+ parameters["timer"] = timer
+
+ self._fetch(f"light/{self.index}", parameters)
+
+ def off(self, timer: Optional[int] = None):
+ parameters: Dict[str, Any] = {
+ "turn": "off"
+ }
+ if timer is not None:
+ parameters["timer"] = timer
+
+ self._fetch(f"light/{self.index}", parameters)
+
+ def mode_setter(self, mode: rgbw_mode):
+ self._fetch(f"light/{self.index}", {"mode": mode})
+
+ def red_setter(self, val: byte):
+ self._fetch(f"light/{self.index}", {"red": val})
+
+ def green_setter(self, val: byte):
+ self._fetch(f"light/{self.index}", {"green": val})
+
+ def blue_setter(self, val: byte):
+ self._fetch(f"light/{self.index}", {"blue": val})
+
+ def white_setter(self, val: byte):
+ self._fetch(f"light/{self.index}", {"white": val})
+
+ def rgb_setter(self, rgb: Tuple[byte, byte, byte]):
+ self._fetch(f"light/{self.index}", {
+ "red": rgb[0],
+ "green": rgb[1],
+ "blue": rgb[2]
+ })
+
+ def rgbw_setter(self, rgbw: Tuple[byte, byte, byte, byte]):
+ self._fetch(f"light/{self.index}", {
+ "red": rgbw[0],
+ "green": rgbw[1],
+ "blue": rgbw[2],
+ "white": rgbw[3],
+ })
+
+ def brightness_setter(self, val: percent):
+ self._fetch(f"light/{self.index}", {"brightness": val})
+
+ def _fetch(self, *args, **kwargs):
+ data = self._device._request.json_post(*args, *kwargs)
+
+ self._ison = data.get("ison")
+
+ started = data.get("timer_started")
+ duration = data.get("timer_duration")
+ if started and duration:
+ self._timer_start = datetime.utcfromtimestamp(started)
+ self._timer_end = self._timer_start + timedelta(seconds=duration)
+
+ self._mode = data.get("mode")
+ self._red = data.get("red")
+ self._green = data.get("green")
+ self._blue = data.get("blue")
+ self._white = data.get("white")
+ self._gain = data.get("gain")
+ self._temp = data.get("temp")
+ self._brightness = data.get("brightness")
+
+ self._effect = data.get("effect")
+ self._transition = data.get("transition")
+
+ return data
+
+ def update(self) -> None:
+ self._fetch(f"light/{self.index}")
diff --git a/ShellyPy/api/gen1/backends/http/meter.py b/ShellyPy/api/gen1/backends/http/meter.py
new file mode 100644
index 0000000..fe2ec56
--- /dev/null
+++ b/ShellyPy/api/gen1/backends/http/meter.py
@@ -0,0 +1,23 @@
+from datetime import datetime
+from typing import List
+
+from .request import Request
+from .....base import Meter as BaseMeter
+
+from .....exceptions import InvalidTimer
+
+class Meter(BaseMeter):
+
+ def _fetch(self, *args, **kwargs):
+ data = self._device._request.json_post(*args, *kwargs)
+
+ self._power = data["power"]
+ self._is_valid = data["is_valid"]
+ self._timestamp = datetime.fromtimestamp(data["timestamp"])
+ self._counters = data["counters"]
+ self._total = data["total"]
+
+ return data
+
+ def update(self) -> None:
+ self._fetch(f"meter/{self.index}")
diff --git a/ShellyPy/api/gen1/backends/http/relay.py b/ShellyPy/api/gen1/backends/http/relay.py
new file mode 100644
index 0000000..d8d9fd2
--- /dev/null
+++ b/ShellyPy/api/gen1/backends/http/relay.py
@@ -0,0 +1,54 @@
+from datetime import datetime, timedelta
+from typing import Optional, Union, Dict, Any
+
+from .request import Request
+from .....base import Relay as BaseRelay
+from .....utils import property_fetcher
+
+from .....exceptions import InvalidTimer
+
+class Relay(BaseRelay):
+
+ def __init__(self, *args, **kwargs):
+ super().__init__(*args, **kwargs)
+
+ self._ison = kwargs.get("ison")
+
+ def toggle(self, timer: Optional[int] = None):
+ parameters: Dict[str, Any] = {
+ "turn": "toggle"
+ }
+ if timer is not None:
+ parameters["timer"] = timer
+
+ self._fetch(f"relay/{self.index}", parameters)
+
+ def on(self, timer: Optional[int] = None):
+ parameters: Dict[str, Any] = {
+ "turn": "on"
+ }
+ if timer is not None:
+ parameters["timer"] = timer
+
+ self._fetch(f"relay/{self.index}", parameters)
+
+ def off(self, timer: Optional[int] = None):
+ parameters: Dict[str, Any] = {
+ "turn": "off"
+ }
+ if timer is not None:
+ parameters["timer"] = timer
+
+ self._fetch(f"relay/{self.index}", parameters)
+
+ def _fetch(self, *args, **kwargs):
+ data = self._device._request.json_post(*args, *kwargs)
+
+ self._ison = data["ison"]
+ self._timer_started = datetime.fromtimestamp(data["timer_started"])
+ self._timer_end = self._timer_started + timedelta(seconds=data["timer_duration"])
+
+ return data
+
+ def update(self) -> None:
+ self._fetch(f"relay/{self.index}")
diff --git a/ShellyPy/api/gen1/backends/http/request.py b/ShellyPy/api/gen1/backends/http/request.py
new file mode 100644
index 0000000..c9507cf
--- /dev/null
+++ b/ShellyPy/api/gen1/backends/http/request.py
@@ -0,0 +1,90 @@
+from typing import Optional, Dict, Union, List
+from urllib.request import build_opener
+from urllib.request import (
+ HTTPBasicAuthHandler, HTTPPasswordMgrWithDefaultRealm
+)
+from urllib.parse import urlencode
+from urllib.error import HTTPError, URLError
+import json
+
+from .....base import Device
+from .....exceptions import (
+ UnauthorizedException,
+ InvalidRequestException,
+ ConnectionRefusedException,
+)
+
+PARA_TYPE = Optional[
+ Dict[
+ str,
+ Union[
+ str,
+ int,
+ float,
+ bool,
+ List[str]
+ ]
+ ]
+]
+
+EXC_DATA_LIMIT = 0xF
+
+class Request:
+
+ def __init__(self, device: Device):
+ self.device = device
+
+ def post(self, path: str, parameter: PARA_TYPE = None):
+ if parameter is None:
+ parameter = {}
+
+ while path.startswith("/"):
+ path = path[1:]
+
+ hostname = self.device.hostname
+ port = self.device.port or "80"
+ params = urlencode(parameter)
+ url = f"http://{hostname}:{port}/{path}"
+ if params:
+ url += f"?{params}"
+
+ if self.device._debug:
+ print(f"[DEBUG] {url}")
+
+ opener = build_opener()
+ try:
+ if not self.device._credentials:
+ raise ValueError()
+
+ username, password = self.device._credentials
+
+ passman = HTTPPasswordMgrWithDefaultRealm()
+ passman.add_password(None, url, username, password)
+
+ authhandler = HTTPBasicAuthHandler(passman)
+ opener.add_handler(authhandler)
+ except (TypeError, ValueError):
+ # can't unpack
+ pass
+
+ try:
+ with opener.open(url, timeout=self.device.timeout) as f:
+ return f.read().decode()
+ except HTTPError as e:
+ if e.status == 401:
+ raise UnauthorizedException() from None
+
+ raise InvalidRequestException()
+
+ except URLError as e:
+ raise ConnectionRefusedException(e.reason) from None
+
+ def json_post(self, *args, **kwargs):
+ data = self.post(*args, **kwargs)
+ try:
+ return json.loads(data)
+ except Exception as e:
+ data = json.dumps(data)
+ if (len(data) > EXC_DATA_LIMIT):
+ data = f"{data[:EXC_DATA_LIMIT]}..."
+ raise ValueError(f"Expected JSON, received {data}") from None
diff --git a/ShellyPy/api/gen1/backends/http/roller.py b/ShellyPy/api/gen1/backends/http/roller.py
new file mode 100644
index 0000000..fac2064
--- /dev/null
+++ b/ShellyPy/api/gen1/backends/http/roller.py
@@ -0,0 +1,63 @@
+from datetime import datetime, timedelta
+from typing import Optional, Union, Dict, Any
+
+from .request import Request
+from .....base import Roller as BaseRoller
+from .....base.hints import percent
+from .....utils import (
+ clamp,
+ property_fetcher,
+)
+from .....exceptions import InvalidTimer
+
+class Roller(BaseRoller):
+
+ def open(self):
+ parameter = {
+ "go": "open"
+ }
+
+ self._fetch(f"roller/{self.index}", parameter)
+
+ def close(self):
+ parameter = {
+ "go": "close"
+ }
+
+ self._fetch(f"roller/{self.index}", parameter)
+
+ def stop(self):
+ parameter = {
+ "go": "stop"
+ }
+
+ self._fetch(f"roller/{self.index}", parameter)
+
+ def pos_setter(self, pos: percent):
+ parameter = {
+ "go": "to_pos",
+ "roller_pos": pos
+ }
+
+ self._fetch(f"roller/{self.index}", parameter)
+
+ def _fetch(self, *args, **kwargs):
+ data = self._device._request.json_post(*args, *kwargs)
+
+ self._state = data["state"]
+ self._power = data["power"]
+ self._safety_switch = data["safety_switch"]
+ self._stop_reason = data["stop_reason"]
+ self._last_direction = data["last_direction"]
+ self._pos = data["current_pos"]
+
+ self._calibrating = data["calibrating"]
+ self._positioning = data["positioning"]
+
+ return data
+
+ def update(self) -> None:
+ self._fetch(f"roller/{self.index}")
+
+ def calibrate(self) -> None:
+ self._fetch(f"roller/{self.index}/calibrate")
diff --git a/ShellyPy/api/gen1/backends/http/settings.py b/ShellyPy/api/gen1/backends/http/settings.py
new file mode 100644
index 0000000..c9fcd18
--- /dev/null
+++ b/ShellyPy/api/gen1/backends/http/settings.py
@@ -0,0 +1,39 @@
+from .....base import Settings as BaseSettings
+
+class Settings(BaseSettings):
+
+ def _fetch(self, *args, **kwargs):
+ data = self._device._request.json_post(*args, *kwargs)
+
+ device = data.get("device", {})
+ self._device._name = data.get("name")
+ self._device._type = device.get("type")
+ self._device._mac = device.get("mac")
+ self._device._firmware = data.get("fw")
+
+ for index, state in enumerate(data.get("relays", [])):
+ self._device._relays.append(self._device._backend.Relay(self._device, index, **state))
+
+ for index, state in enumerate(data.get("rollers", [])):
+ self._device._rollers.append(self._device._backend.Roller(self._device, index, **state))
+
+ for index, state in enumerate(data.get("lights", [])):
+ self._device._lights.append(self._device._backend.Light(self._device, index, **state))
+
+ self._max_power = not data.get("max_power")
+ self._mode = not data.get("mode")
+ self._led_status = not data.get("led_status_disable", False)
+
+ return data
+
+ def max_power_setter(self, power):
+ self._fetch(f"settings", {"max_power": power})
+
+ def mode_setter(self, mode):
+ self._fetch(f"settings", {"mode": mode})
+
+ def led_status_setter(self, status):
+ self._fetch(f"settings", {"led_status_disable": not power})
+
+ def update(self) -> None:
+ self._fetch(f"settings")
diff --git a/ShellyPy/api/gen1/device.py b/ShellyPy/api/gen1/device.py
new file mode 100644
index 0000000..3d6cf02
--- /dev/null
+++ b/ShellyPy/api/gen1/device.py
@@ -0,0 +1,56 @@
+from typing import Optional
+from importlib import import_module
+
+from ...base import Device as BaseDevice
+from ...exceptions import InvalidBackend
+from ...utils.attribute_list import AttributeList
+from ...utils import property_fetcher
+
+class Device(BaseDevice):
+ __backends__ = [
+ "http",
+ ]
+
+ def _load_any_backend(self):
+ for backend in self.__backends__:
+ try:
+ return self._load_backend(backend)
+ except:
+ continue
+
+ raise Exception("Failed to load any backend")
+
+ def _load_backend(self, backend_name: str):
+ module_name = ".".join(self.__class__.__module__.split(".")[:-1])
+ return import_module(f".backends.{backend_name}", module_name)
+
+ def _create_attributes(self):
+ self._relays = AttributeList()
+ self._rollers = AttributeList()
+ self._lights = AttributeList()
+ self._meters = AttributeList()
+ self._loaded_meters = False
+
+ # settings populate attributes
+ self._settings = self._backend.Settings(self)
+ self._settings.update()
+
+ @property
+ @property_fetcher(update_method="_create_attributes")
+ def meters(self):
+ if not self._loaded_meters:
+ self._loaded_meters = True
+ # there is no way to query meters without brute force
+ meter_index = 0
+ while True:
+
+ meter = self._backend.Meter(self._request, meter_index)
+ try:
+ meter.update()
+ except:
+ break
+
+ self.meter.append(meter)
+ meter_index += 1
+
+ return super().meters
diff --git a/ShellyPy/api/gen2/__init__.py b/ShellyPy/api/gen2/__init__.py
new file mode 100644
index 0000000..e71ed1d
--- /dev/null
+++ b/ShellyPy/api/gen2/__init__.py
@@ -0,0 +1,2 @@
+
+from .device import *
diff --git a/ShellyPy/api/gen2/backends/__init__.py b/ShellyPy/api/gen2/backends/__init__.py
new file mode 100644
index 0000000..44c1fad
--- /dev/null
+++ b/ShellyPy/api/gen2/backends/__init__.py
@@ -0,0 +1,2 @@
+
+from . import json_rpc
diff --git a/ShellyPy/api/gen2/backends/json_rpc/__init__.py b/ShellyPy/api/gen2/backends/json_rpc/__init__.py
new file mode 100644
index 0000000..9859b81
--- /dev/null
+++ b/ShellyPy/api/gen2/backends/json_rpc/__init__.py
@@ -0,0 +1,7 @@
+
+from .light import *
+from .meter import *
+from .relay import *
+from .request import *
+from .roller import *
+from .settings import *
diff --git a/ShellyPy/api/gen2/backends/json_rpc/light.py b/ShellyPy/api/gen2/backends/json_rpc/light.py
new file mode 100644
index 0000000..a508eff
--- /dev/null
+++ b/ShellyPy/api/gen2/backends/json_rpc/light.py
@@ -0,0 +1,73 @@
+from datetime import datetime, timedelta
+from typing import Optional, Tuple, Union, Dict, Any
+
+from .request import Request
+from .....base import Light as BaseLight
+from .....base.hints import (
+ rgbw_mode, byte, percent,
+ temperaur, transition
+)
+from .....utils import (
+ clamp, clamp_percent, clamp_byte,
+ clamp_temp, property_fetcher,
+)
+from .....exceptions import InvalidTimer
+
+class Light(BaseLight):
+
+ def toggle(self, timer: Optional[int] = None):
+ self._fetch("Light.Toggle")
+ self.update()
+
+ def on(self, timer: Optional[int] = None):
+ params: Dict[str, Any] = {"on": True}
+ if timer is not NOne:
+ params["toggle_after"] = timer
+ self._fetch("Light.Set", **params)
+ self._ison = True
+
+ def off(self, timer: Optional[int] = None):
+ params: Dict[str, Any] = {"on": False}
+ if timer is not None:
+ params["toggle_after"] = timer
+ self._fetch("Light.Set", **params)
+ self._ison = False
+
+ def mode_setter(self, mode: rgbw_mode):
+ raise UnimplementedMethod("TODO Gen2 Light::mode_setter()")
+
+ def red_setter(self, val: byte):
+ raise UnimplementedMethod("TODO Gen2 Light::red_setter()")
+
+ def green_setter(self, val: byte):
+ raise UnimplementedMethod("TODO Gen2 Light::green_setter()")
+
+ def blue_setter(self, val: byte):
+ raise UnimplementedMethod("TODO Gen2 Light::blue_setter()")
+
+ def white_setter(self, val: byte):
+ raise UnimplementedMethod("TODO Gen2 Light::white_setter()")
+
+ def rgb_setter(self, rgb: Tuple[byte, byte, byte]):
+ raise UnimplementedMethod("TODO Gen2 Light::rgb_setter()")
+
+ def rgbw_setter(self, rgbw: Tuple[byte, byte, byte, byte]):
+ raise UnimplementedMethod("TODO Gen2 Light::rgbw_setter()")
+
+ def brightness_setter(self, val: percent):
+ self._fetch("Light.Set", brightness=val)
+
+ def _fetch(self, method, **kwargs):
+ kwargs["id"] = self._index
+ return self._device._request.post(method, kwargs)
+
+ def update(self) -> None:
+ result = self._fetch("Light.GetStatus")
+ self._ison = result.get("output")
+ self._brightness = result.get("brightness")
+
+ started = data.get("timer_started")
+ duration = data.get("timer_duration")
+ if started and duration:
+ self._timer_start = datetime.utcfromtimestamp(started)
+ self._timer_end = self._timer_start + timedelta(seconds=duration) \ No newline at end of file
diff --git a/ShellyPy/api/gen2/backends/json_rpc/meter.py b/ShellyPy/api/gen2/backends/json_rpc/meter.py
new file mode 100644
index 0000000..85a0ec8
--- /dev/null
+++ b/ShellyPy/api/gen2/backends/json_rpc/meter.py
@@ -0,0 +1,24 @@
+from datetime import datetime
+from typing import List
+
+from .request import Request
+from .....base import Meter as BaseMeter
+
+from .....exceptions import InvalidTimer
+
+class Meter(BaseMeter):
+
+ def _fetch(self, method, **kwargs):
+ kwargs["id"] = self._index
+ result = self._device._request.post(method, kwargs)
+
+ self._power = result.get("voltage", 0)
+ self._is_valid = True
+ self._timestamp = 0
+ self._counters = 0
+ self._total = 0
+
+ return result
+
+ def update(self) -> None:
+ self._fetch("Voltmeter.GetStatus")
diff --git a/ShellyPy/api/gen2/backends/json_rpc/relay.py b/ShellyPy/api/gen2/backends/json_rpc/relay.py
new file mode 100644
index 0000000..545f308
--- /dev/null
+++ b/ShellyPy/api/gen2/backends/json_rpc/relay.py
@@ -0,0 +1,33 @@
+from datetime import datetime, timedelta
+from typing import Optional, Union, Dict, Any
+
+from .request import Request
+from .....base import Relay as BaseRelay
+from .....utils import property_fetcher
+
+from .....exceptions import InvalidTimer
+
+class Relay(BaseRelay):
+
+ def toggle(self, timer: Optional[int] = None):
+ result = self._fetch("Switch.Toggle")
+ try:
+ self._ison = not result["was_on"]
+ except KeyError:
+ pass
+
+ def on(self, timer: Optional[int] = None):
+ self._fetch("Switch.Set", on=True)
+ self._ison = True
+
+ def off(self, timer: Optional[int] = None):
+ self._fetch("Switch.Set", on=False)
+ self._ison = False
+
+ def _fetch(self, method, **kwargs):
+ kwargs["id"] = self._index
+ return self._device._request.post(method, kwargs)
+
+ def update(self) -> None:
+ result = self._fetch("Switch.GetStatus")
+ self._ison = result.get("output")
diff --git a/ShellyPy/api/gen2/backends/json_rpc/request.py b/ShellyPy/api/gen2/backends/json_rpc/request.py
new file mode 100644
index 0000000..e2f454e
--- /dev/null
+++ b/ShellyPy/api/gen2/backends/json_rpc/request.py
@@ -0,0 +1,136 @@
+from typing import Optional, Dict, Union, List
+from urllib.request import build_opener
+from urllib.request import (
+ HTTPDigestAuthHandler, HTTPPasswordMgrWithDefaultRealm
+)
+from urllib.parse import urlencode
+from urllib.error import HTTPError, URLError
+from hashlib import sha256
+import json
+
+from .....base import Device
+from .....exceptions import (
+ UnauthorizedException,
+ InvalidRequestException,
+ ConnectionRefusedException,
+)
+
+PARA_TYPE = Optional[
+ Dict[
+ str,
+ Union[
+ str,
+ int,
+ float,
+ bool,
+ List[str]
+ ]
+ ]
+]
+
+EXC_DATA_LIMIT = 0xF
+
+_PAYLOAD_COUNTER = 0
+
+# Default urllib digest handler does not support SHA-256
+class ShellyDigestAuthHandler(HTTPDigestAuthHandler):
+ def get_algorithm_impls(self, algorithm):
+ if algorithm == "SHA-256":
+ H = lambda x: sha256(x.encode("ascii")).hexdigest()
+ else:
+ return super().get_algorithm_impls(algorithm)
+ KD = lambda s, d: H("%s:%s" % (s, d))
+ return H, KD
+
+class Request:
+
+ def __init__(self, device: Device):
+ self.auth = None
+ self.device = device
+
+ @staticmethod
+ def get_payload_id():
+ global _PAYLOAD_COUNTER
+
+ _PAYLOAD_COUNTER += 1
+ return _PAYLOAD_COUNTER
+
+ @staticmethod
+ def sha256(inp: str) -> str:
+ return sha256(str.encode()).hexdigest()
+
+ def post(self, method: str, params: PARA_TYPE = None):
+ hostname = self.device.hostname
+ port = self.device.port or "80"
+ url = f"http://{hostname}:{port}/rpc"
+
+ payload_id = self.get_payload_id()
+ payload = {
+ "jsonrpc": "2.0",
+ "id": payload_id,
+ "method": method,
+ }
+
+ if params is not None:
+ payload["params"] = params
+
+ if self.device._debug:
+ str_params = ", ".join([f"{k}={v}" for k,v in params.items()])
+ print(f"[DEBUG] {method}({str_params})")
+
+ opener = build_opener()
+
+ raw_payload = json.dumps(payload)
+ raw_payload_bytes = raw_payload.encode('utf-8') # needs to be bytes
+
+ opener.addheaders = [
+ ('Content-Type', 'application/json; charset=utf-8'),
+ ('Content-Length', len(raw_payload_bytes))
+ ]
+
+ try:
+ if not self.device._credentials:
+ raise ValueError()
+
+ username, password = "admin", self.device._credentials[1]
+
+ passman = HTTPPasswordMgrWithDefaultRealm()
+ passman.add_password(None, url, username, password)
+
+ authhandler = ShellyDigestAuthHandler(passman)
+ opener.add_handler(authhandler)
+ except (TypeError, IndexError):
+ # can't unpack
+ pass
+
+ try:
+ with opener.open(url, data=raw_payload_bytes, timeout=self.device.timeout) as f:
+ result = f.read().decode()
+
+ try:
+ data = json.loads(result)
+ except Exception as e:
+ data = json.dumps(result)
+ if (len(data) > EXC_DATA_LIMIT):
+ data = f"{data[:EXC_DATA_LIMIT]}..."
+ raise ValueError(f"Expected JSON, received {data}") from None
+
+ if (data["id"] != payload_id):
+ raise ValueError("Invalid payload ID")
+
+ try:
+ return data["result"]
+ except KeyError:
+ raise InvalidRequestException(data["error"]["message"])
+
+ except HTTPError as e:
+ if e.status == 401:
+ raise UnauthorizedException() from None
+
+ raise InvalidRequestException()
+
+ except URLError as e:
+ raise ConnectionRefusedException(e.reason) from None
+
+ def json_post(self, *args, **kwargs):
+ return self.post(*args, **kwargs)
diff --git a/ShellyPy/api/gen2/backends/json_rpc/roller.py b/ShellyPy/api/gen2/backends/json_rpc/roller.py
new file mode 100644
index 0000000..3b63aad
--- /dev/null
+++ b/ShellyPy/api/gen2/backends/json_rpc/roller.py
@@ -0,0 +1,45 @@
+from datetime import datetime, timedelta
+from typing import Optional, Union, Dict, Any
+
+from .request import Request
+from .....base import Roller as BaseRoller
+from .....base.hints import percent
+from .....utils import (
+ clamp,
+ property_fetcher,
+)
+from .....exceptions import InvalidTimer
+
+class Roller(BaseRoller):
+
+ def open(self):
+ self._fetch("Cover.Open")
+ self._state = "open"
+
+ def close(self):
+ self._fetch("Cover.Close")
+ self._state = "close"
+
+ def stop(self):
+ self._fetch("Cover.Stop")
+ self._state = "stop"
+
+ def pos_setter(self, pos: percent):
+ self._fetch("Cover.GoToPosition", pos=pos)
+
+ def _fetch(self, method, **kwargs):
+ kwargs["id"] = self._index
+ try:
+ return self._device._request.post(method, kwargs)
+ finally:
+ if method != "Cover.GetStatus":
+ self.update()
+
+ def update(self) -> None:
+ result = self._fetch("Cover.GetStatus")
+ self._state = result.get("state")
+ self._last_direction = result.get("last_direction")
+ self._pos = result.get("current_pos")
+
+ def calibrate(self) -> None:
+ self._fetch("Cover.Calibrate")
diff --git a/ShellyPy/api/gen2/backends/json_rpc/settings.py b/ShellyPy/api/gen2/backends/json_rpc/settings.py
new file mode 100644
index 0000000..2fd9c9d
--- /dev/null
+++ b/ShellyPy/api/gen2/backends/json_rpc/settings.py
@@ -0,0 +1,25 @@
+from .....base import Settings as BaseSettings
+from .....exceptions.backend import UnimplementedMethod
+
+class Settings(BaseSettings):
+
+ def _fetch(self, method, **kwargs):
+ result = self._device._request.post(method, kwargs)
+
+ self._device._name = result.get("name")
+ self._device._type = result.get("model")
+ self._device._mac = result.get("mac")
+ self._device._firmware = result.get("fw_id")
+ return result
+
+ def max_power_setter(self, power):
+ raise UnimplementedMethod("TODO Gen2 Settings::max_power_setter()")
+
+ def mode_setter(self, mode):
+ raise UnimplementedMethod("TODO Gen2 Settings::mode_setter()")
+
+ def led_status_setter(self, status):
+ raise UnimplementedMethod("TODO Gen2 Settings::led_status_setter()")
+
+ def update(self) -> None:
+ self._fetch("Shelly.GetDeviceInfo")
diff --git a/ShellyPy/api/gen2/device.py b/ShellyPy/api/gen2/device.py
new file mode 100644
index 0000000..0026045
--- /dev/null
+++ b/ShellyPy/api/gen2/device.py
@@ -0,0 +1,106 @@
+from typing import Optional
+from importlib import import_module
+
+from ...base import Device as BaseDevice
+from ...exceptions import InvalidBackend, InvalidRequestException
+from ...utils.attribute_list import AttributeList
+from ...utils import property_fetcher
+
+class Device(BaseDevice):
+ __backends__ = [
+ "json_rpc",
+ ]
+
+ def _load_any_backend(self):
+ for backend in self.__backends__:
+ try:
+ return self._load_backend(backend)
+ except Exception as e:
+ continue
+
+ raise Exception("Failed to load any backend")
+
+ def _load_backend(self, backend_name: str):
+ module_name = ".".join(self.__class__.__module__.split(".")[:-1])
+ return import_module(f".backends.{backend_name}", module_name)
+
+ def _create_attributes(self):
+ self._find_relays()
+ self._find_rollers()
+ self._find_lights()
+ self._find_meters()
+
+ self._settings = self._backend.Settings(self)
+ self._settings.update()
+
+ @property
+ @property_fetcher(update_method="_find_relays")
+ def relays(self):
+ return super().relays
+
+ def _find_relays(self):
+ self._relays = AttributeList() # Switch
+
+ relay_index = 0
+ while True:
+ relay = self._backend.Relay(self, relay_index)
+ try:
+ relay.update()
+ except InvalidRequestException:
+ break
+ self._relays.append(relay)
+ relay_index += 1
+
+ @property
+ @property_fetcher(update_method="_find_rollers")
+ def rollers(self):
+ return super().rollers
+
+ def _find_rollers(self):
+ self._rollers = AttributeList() # Cover
+
+ roller_index = 0
+ while True:
+ roller = self._backend.Roller(self, roller_index)
+ try:
+ roller.update()
+ except InvalidRequestException:
+ break
+ self._rollers.append(roller)
+ roller_index += 1
+
+ @property
+ @property_fetcher(update_method="_find_lights")
+ def lights(self):
+ return super().lights
+
+ def _find_lights(self):
+ self._lights = AttributeList()
+
+ light_index = 0
+ while True:
+ light = self._backend.Light(self, light_index)
+ try:
+ light.update()
+ except InvalidRequestException:
+ break
+ self._lights.append(light)
+ light_index += 1
+
+ @property
+ @property_fetcher(update_method="_find_meters")
+ def meters(self):
+ return super().meters
+
+ def _find_meters(self):
+ self._meters = AttributeList()
+
+ meter_index = 0
+ while True:
+ meter = self._backend.Meter(self, meter_index)
+ try:
+ meter.update()
+ except InvalidRequestException as e:
+ break
+ self._meters.append(meter)
+ meter_index += 1
diff --git a/ShellyPy/base/__init__.py b/ShellyPy/base/__init__.py
new file mode 100644
index 0000000..0acce6c
--- /dev/null
+++ b/ShellyPy/base/__init__.py
@@ -0,0 +1,8 @@
+
+from .device import *
+from .hints import *
+from .light import *
+from .meter import *
+from .relay import *
+from .roller import *
+from .settings import *
diff --git a/ShellyPy/base/device.py b/ShellyPy/base/device.py
new file mode 100644
index 0000000..0960af9
--- /dev/null
+++ b/ShellyPy/base/device.py
@@ -0,0 +1,167 @@
+from abc import ABCMeta, abstractmethod
+from typing import TYPE_CHECKING, Optional, Tuple, List
+
+from ..utils import property_fetcher, AttributeList
+
+from .hints import transition
+
+if TYPE_CHECKING:
+ from .settings import Settings
+ from .relay import Relay
+ from .roller import Roller
+ from .light import Light
+ from .meter import Meter
+
+class Device(metaclass=ABCMeta):
+ __backends__: List = []
+
+ _hostname: str
+ _port: Optional[int]
+
+ _name: Optional[str]
+ _type: Optional[str]
+ _mac: Optional[str]
+ _firmware: Optional[str]
+
+ _timeout: transition
+ _credentials: Optional[Tuple[str, str]]
+
+ _debug: bool
+ #_backend: object
+
+ _settings: Optional["Settings"] = None
+ _relays: Optional[List["Relay"]] = None
+ _rollers: Optional[List["Roller"]] = None
+ _lights: Optional[List["Light"]] = None
+ _meters: Optional[List["Meter"]] = None
+
+ def __init__(self, hostname: str, port: Optional[int] = None, backend: Optional[str] = None, *args, **kwargs):
+ self._hostname = hostname
+ self._port = port
+ # to be filled at runtime
+ self._name = None
+ self._type = None
+ self._mac = None
+ self._firmware = None
+
+ self._timeout = kwargs.get("timeout", 5)
+
+ creds = kwargs.get("credentials")
+
+ self.credentials = kwargs.get("credentials", None)
+
+ self._debug = bool(kwargs.get("debug", False))
+
+ if not backend:
+ backend_impl = self._load_any_backend()
+ else:
+ backend_impl = self._load_backend(backend)
+
+ if not backend_impl:
+ raise Exception("No suitable backend found")
+
+ self._backend = backend_impl
+ self._request = self._backend.Request(self)
+
+ if kwargs.get("preload"):
+ self._create_attributes()
+
+ def __repr__(self):
+ class_name = self.__class__.__name__
+ return f"{class_name}({self.name or self.type})"
+
+ @abstractmethod
+ def _load_any_backend(self):
+ pass
+
+ @abstractmethod
+ def _load_backend(self, backend_name: str):
+ pass
+
+ @abstractmethod
+ def _create_attributes(self):
+ pass
+
+ @property
+ def hostname(self):
+ return self._hostname
+
+ @property
+ def port(self):
+ return self._port
+
+ @property
+ @property_fetcher(update_method="_create_attributes")
+ def name(self):
+ return self._name
+
+ @property
+ @property_fetcher(update_method="_create_attributes")
+ def type(self):
+ return self._type
+
+ @property
+ @property_fetcher(update_method="_create_attributes")
+ def mac(self):
+ return self._mac
+
+ @property
+ @property_fetcher(update_method="_create_attributes")
+ def firmware(self):
+ return self._firmware
+
+ @property
+ def timeout(self):
+ return self._timeout
+
+ @timeout.setter
+ def timeout(self, value: int):
+ self._timeout = value
+
+ @property
+ def credentials(self):
+ return self._credentials
+
+ @credentials.setter
+ def credentials(self, val):
+ if not isinstance(val, tuple) or len(val) != 2:
+ val = ()
+
+ self._credentials = val
+
+ @property
+ def backend(self):
+ return self._backend
+
+ @property
+ def request(self):
+ return self._request
+
+ @property
+ @property_fetcher(update_method="_create_attributes")
+ def settings(self):
+ return self._settings
+
+ @property
+ @property_fetcher(update_method="_create_attributes")
+ def relays(self):
+ return self._relays
+
+ @property
+ @property_fetcher(update_method="_create_attributes")
+ def rollers(self):
+ return self._rollers
+
+ @property
+ @property_fetcher(update_method="_create_attributes")
+ def lights(self):
+ return self._lights
+
+ @property
+ @property_fetcher(update_method="_create_attributes")
+ def meters(self):
+ return self._meters
+
+ @classmethod
+ def connect(cls, *args, **kwargs):
+ return cls(*args, **kwargs)
diff --git a/ShellyPy/base/hints.py b/ShellyPy/base/hints.py
new file mode 100644
index 0000000..e4398d1
--- /dev/null
+++ b/ShellyPy/base/hints.py
@@ -0,0 +1,17 @@
+from dataclasses import dataclass
+from typing import Annotated, Literal
+
+@dataclass
+class ValueRange:
+ def __hash__(self):
+ return hash(self.min) + hash(self.max)
+
+ min: float
+ max: float
+
+byte = Annotated[int, ValueRange(0, 255)]
+percent = Annotated[int, ValueRange(0, 100)]
+temperaur = Annotated[int, ValueRange(3000, 6500)]
+transition = Annotated[int, ValueRange(0, 5000)]
+
+rgbw_mode = Literal["color", "white"]
diff --git a/ShellyPy/base/light.py b/ShellyPy/base/light.py
new file mode 100644
index 0000000..afe8096
--- /dev/null
+++ b/ShellyPy/base/light.py
@@ -0,0 +1,248 @@
+from abc import ABCMeta, abstractmethod
+from datetime import datetime, timedelta
+from typing import Union, Tuple, Optional
+
+from ..utils import (
+ clamp, clamp_percent, clamp_byte,
+ clamp_temp, property_fetcher,
+)
+
+from .hints import byte, percent, temperaur, transition, rgbw_mode
+
+class Light(metaclass=ABCMeta):
+ _index: int
+
+ _ison: bool
+ _timer_start: datetime
+ _timer_end: datetime
+
+ _mode: rgbw_mode
+ _red: byte
+ _blue: byte
+ _green: byte
+ _white: byte
+ _gain: percent
+ _temp: temperaur
+ _brightness: percent
+
+ _effect: int
+ _transition: transition
+
+ def __init__(self, device, index, *args, **kwargs):
+ self._device = device
+ self._index = index
+
+ self._ison = None
+ self._timer_started = None
+ self._timer_end = None
+
+ self._mode = None
+ self._red = None
+ self._green = None
+ self._blue = None
+ self._white = None
+ self._gain = None
+ self._temp = None
+ self._brightness = None
+
+ self._effect = None
+ self._transition = None
+
+ def __repr__(self):
+ class_name = self.__class__.__name__
+ return f"{class_name}{self.index}(on={self.ison})"
+
+ @abstractmethod
+ def toggle(self, timer: Optional[int] = None):
+ pass
+
+ @abstractmethod
+ def on(self, timer: Optional[int] = None):
+ pass
+
+ @abstractmethod
+ def off(self, timer: Optional[int] = None):
+ pass
+
+ def turn(self, value: Union[str, bool], timer: Optional[int] = None):
+
+ value_method = {
+ "toggle": self.toggle,
+ "on": self.on,
+ "off": self.off,
+ True: self.on,
+ False: self.off
+ }
+
+ return value_method[value](timer)
+
+ @property
+ def index(self) -> int:
+ return self._index
+
+ @property
+ @property_fetcher()
+ def ison(self) -> bool:
+ return bool(self._ison)
+
+ @property
+ def has_timer(self) -> bool:
+ """
+ return true when the timer is primed and has not happened yet
+ """
+ start = self._timer_start
+ end = self._timer_end
+ if not start or not end:
+ return False
+
+ date_now = datetime.now()
+ return end > date_now and start < date_now
+
+
+ @property
+ @property_fetcher()
+ def timer_started(self) -> Optional[datetime]:
+ if self.has_timer:
+ return self._timer_start
+ return None
+
+ @property
+ def timer_duration(self) -> Optional[timedelta]:
+ # fetch the start before to populate the implementation values
+ if not self.has_timer:
+ return None
+
+ start = self.timer_started
+ end = self._timer_end
+
+ return end - start
+
+ @property
+ @property_fetcher()
+ def mode(self) -> rgbw_mode:
+ return self._mode
+
+ @mode.setter
+ def mode(self, mode: rgbw_mode):
+ if mode != "color" and mode != "white":
+ raise Exception("mode has to be 'color' or 'white'")
+
+ self.mode_setter(mode)
+
+ @abstractmethod
+ def mode_setter(self, mode: rgbw_mode):
+ pass
+
+ # color mode
+ @property
+ @property_fetcher()
+ def red(self) -> byte:
+ return clamp_byte(self._red)
+
+ @red.setter
+ def red(self, *args, **kwargs):
+ return self.red_setter(*args, **kwargs)
+
+ @abstractmethod
+ def red_setter(self, val: byte):
+ pass
+
+ @property
+ @property_fetcher()
+ def green(self) -> byte:
+ return clamp_byte(self._green)
+
+ @green.setter
+ def green(self, *args, **kwargs):
+ return self.green_setter(*args, **kwargs)
+
+ @abstractmethod
+ def green_setter(self, val: byte):
+ pass
+
+ @property
+ @property_fetcher()
+ def blue(self) -> byte:
+ return clamp_byte(self._blue)
+
+ @blue.setter
+ def blue(self, *args, **kwargs):
+ return self.blue_setter(*args, **kwargs)
+
+ @abstractmethod
+ def blue_setter(self, val: byte):
+ pass
+
+ @property
+ @property_fetcher()
+ def white(self) -> byte:
+ return clamp_byte(self._white)
+
+ @white.setter
+ def white(self, *args, **kwargs):
+ return self.white_setter(*args, **kwargs)
+
+ @abstractmethod
+ def white_setter(self, val: byte):
+ pass
+
+ @property
+ def rgb(self) -> Tuple[byte, byte, byte]:
+ return (self.red, self.green, self.blue)
+
+ @rgb.setter
+ def rgb(self, *args, **kwargs):
+ return self.rgb_setter(*args, **kwargs)
+
+ def rgb_setter(self, rgb: Tuple[byte, byte, byte]):
+ self.red, self.green, self.blue = rgb
+
+ @property
+ def rgbw(self) -> Tuple[byte, byte, byte, byte]:
+ return self.rgb + (self.white,)
+
+ @rgbw.setter
+ def rgbw(self, *args, **kwargs):
+ return self.rgbw_setter(*args, **kwargs)
+
+ def rgbw_setter(self, rgbw: Tuple[byte, byte, byte, byte]):
+ self.rgb = rgbw[:3]
+ self.white = rgbw[3]
+
+ @property
+ @property_fetcher()
+ def gain(self) -> percent:
+ return clamp_percent(self._gain)
+
+ # white mode
+ @property
+ @property_fetcher()
+ def temp(self) -> temperaur:
+ return clamp_temp(self._temp)
+
+ @property
+ @property_fetcher()
+ def brightness(self) -> percent:
+ return clamp_percent(self._brightness)
+
+ @brightness.setter
+ def brightness(self, *args, **kwargs):
+ return brightness_setter(*args, **kwargs)
+
+ @abstractmethod
+ def brightness_setter(self, val: percent):
+ pass
+
+ @property
+ @property_fetcher()
+ def effect(self) -> int:
+ return self._effect
+
+ @property
+ @property_fetcher()
+ def transition(self) -> transition:
+ return self._transition
+
+ @abstractmethod
+ def update(self) -> None:
+ pass
diff --git a/ShellyPy/base/meter.py b/ShellyPy/base/meter.py
new file mode 100644
index 0000000..f141fb1
--- /dev/null
+++ b/ShellyPy/base/meter.py
@@ -0,0 +1,53 @@
+from abc import ABCMeta, abstractmethod
+from datetime import datetime
+from typing import List
+
+from ..utils import property_fetcher
+
+class Meter(metaclass=ABCMeta):
+ def __init__(self, device, index, *args, **kwargs):
+ self._device = device
+ self._index = index
+
+ self._power = None
+ self._is_valid = None
+ self._timestamp = None
+ self._counters = None
+ self._total = None
+
+ def __repr__(self):
+ class_name = self.__class__.__name__
+ return f"{class_name}(total={self.total})"
+
+ @property
+ def index(self) -> int:
+ return self._index
+
+ @property
+ @property_fetcher()
+ def power(self):
+ return self._power
+
+ @property
+ @property_fetcher()
+ def is_valid(self) -> bool:
+ return bool(self._is_valid)
+
+ @property
+ @property_fetcher()
+ def timestamp(self) -> datetime:
+ return self._timestamp
+
+ @property
+ @property_fetcher()
+ def counters(self) -> List[float]:
+ return self._counters
+
+ @property
+ @property_fetcher()
+ def total(self) -> float:
+ return self._total
+
+ @abstractmethod
+ def update(self) -> None:
+ pass
diff --git a/ShellyPy/base/relay.py b/ShellyPy/base/relay.py
new file mode 100644
index 0000000..42ae19f
--- /dev/null
+++ b/ShellyPy/base/relay.py
@@ -0,0 +1,88 @@
+from abc import ABCMeta, abstractmethod
+from datetime import datetime, timedelta
+from typing import Optional, Union
+
+from ..utils import property_fetcher
+
+class Relay(metaclass=ABCMeta):
+
+ def __init__(self, device, index, *args, **kwargs):
+ self._device = device
+ self._index = index
+
+ self._ison = None
+ self._timer_start = None
+ self._timer_end = None
+
+ def __repr__(self):
+ class_name = self.__class__.__name__
+ return f"{class_name}{self.index}(on={self.ison})"
+
+ @abstractmethod
+ def toggle(self, timer: Optional[int] = None):
+ pass
+
+ @abstractmethod
+ def on(self, timer: Optional[int] = None):
+ pass
+
+ @abstractmethod
+ def off(self, timer: Optional[int] = None):
+ pass
+
+ def turn(self, value: Union[str, bool], timer: Optional[int] = None):
+
+ value_method = {
+ "toggle": self.toggle,
+ "on": self.on,
+ "off": self.off,
+ True: self.on,
+ False: self.off
+ }
+
+ return value_method[value](timer)
+
+ @property
+ def index(self) -> int:
+ return self._index
+
+ @property
+ @property_fetcher()
+ def ison(self) -> bool:
+ return bool(self._ison)
+
+ @property
+ def has_timer(self) -> bool:
+ """
+ return true when the timer is primed and has not happened yet
+ """
+ start = self._timer_start
+ end = self._timer_end
+ if not start or not end:
+ return False
+
+ date_now = datetime.now()
+ return end > date_now and start < date_now
+
+
+ @property
+ @property_fetcher()
+ def timer_started(self) -> Optional[datetime]:
+ if self.has_timer:
+ return self._timer_start
+ return None
+
+ @property
+ def timer_duration(self) -> Optional[timedelta]:
+ # fetch the start before to populate the implementation values
+ if not self.has_timer:
+ return None
+
+ start = self.timer_started
+ end = self._timer_end
+
+ return end - start
+
+ @abstractmethod
+ def update(self) -> None:
+ pass
diff --git a/ShellyPy/base/roller.py b/ShellyPy/base/roller.py
new file mode 100644
index 0000000..caff8ad
--- /dev/null
+++ b/ShellyPy/base/roller.py
@@ -0,0 +1,128 @@
+from abc import ABCMeta, abstractmethod
+from typing import Literal, Optional
+
+from ..utils import property_fetcher
+
+from .hints import percent
+
+class Roller(metaclass=ABCMeta):
+ _index: int
+
+ _state: Literal["stop", "open", "close"]
+ _power: int
+ _safety_switch: bool
+ _stop_reason: Literal["normal", "safety_switch", "obstacle", "overpower"]
+ _last_direction: Literal["open", "close"] # documented as bool?
+ _pos: percent
+
+ _calibrating: bool
+ _positioning: bool
+
+ def __init__(self, device, index, *args, **kwargs):
+ self._device = device
+ self._index = index
+
+ self._power = None
+ self._safety_switch = None
+ self._stop_reason = None
+ self._last_direction = None
+ self._pos = None
+
+ self._calibrating = None
+ self._positioning = None
+
+ def __repr__(self):
+ class_name = self.__class__.__name__
+ pos = self.pos
+ if pos == "100":
+ pos = "close"
+ elif pos == "0":
+ pos = "open"
+
+ return f"{class_name}{self.index}(pos={pos})"
+
+ @property
+ def index(self) -> int:
+ return self._index
+
+ @property
+ @property_fetcher()
+ def state(self) -> str: # TODO typing
+ return self._state
+
+ @property
+ @property_fetcher()
+ def power(self) -> int:
+ return max(self._power, 0)
+
+ @property
+ @property_fetcher()
+ def safety_switch(self) -> bool:
+ return bool(self._safety_switch)
+
+ @property
+ @property_fetcher()
+ def stop_reason(self) -> str: # TODO typing
+ return self._stop_reason
+
+ @property
+ @property_fetcher()
+ def last_direction(self) -> str: # TODO typing
+ return self._last_direction
+
+ @property
+ @property_fetcher()
+ def pos(self) -> percent:
+ return self._pos
+
+ @pos.setter
+ def pos(self, value: percent):
+ self.pos_setter(value)
+
+ @abstractmethod
+ def pos_setter(self, value: percent):
+ pass
+
+ @abstractmethod
+ def open(self):
+ pass
+
+ @abstractmethod
+ def close(self):
+ pass
+
+ @abstractmethod
+ def stop(self):
+ pass
+
+ def to_pos(self, pos: percent):
+ self.pos = pos
+
+ def go(self, state: Literal["open", "close", "stop", "to_pos"], pos: Optional[percent] = None):
+
+ if state == "to_pos":
+ if pos is None:
+ raise TypeError("pos cannot be None")
+ return self.to_pos(pos)
+
+ state_method = {
+ "open": self.open,
+ "close": self.close,
+ "stop": self.stop,
+ }
+
+ return state_method[state]()
+
+ @property
+ @property_fetcher()
+ def calibrating(self) -> bool:
+ return bool(self._calibrating)
+
+ @property
+ @property_fetcher()
+ def positioning(self) -> bool:
+ return bool(self._positioning)
+
+ @abstractmethod
+ def update(self) -> None:
+ pass
diff --git a/ShellyPy/base/settings.py b/ShellyPy/base/settings.py
new file mode 100644
index 0000000..3e79199
--- /dev/null
+++ b/ShellyPy/base/settings.py
@@ -0,0 +1,56 @@
+from abc import ABCMeta, abstractmethod
+from typing import Dict, List
+
+from ..utils import property_fetcher
+
+class Settings(metaclass=ABCMeta):
+ _max_power: int
+ _actions: Dict # TODO
+ _mode: str
+ _led_status: bool
+
+ def __init__(self, device, *args, **kwargs):
+ self._device = device
+
+ @property
+ @property_fetcher()
+ def max_power(self):
+ return self._max_power
+
+ @max_power.setter
+ def max_power(self, *args, **kwargs):
+ return self.max_power_setter(*args, **kwargs)
+
+ @abstractmethod
+ def max_power_setter(self, power):
+ pass
+
+ @property
+ @property_fetcher()
+ def mode(self):
+ return self._mode
+
+ @mode.setter
+ def mode(self, *args, **kwargs):
+ return self.mode_setter(*args, **kwargs)
+
+ @abstractmethod
+ def mode_setter(self, mode):
+ pass
+
+ @property
+ @property_fetcher()
+ def led_status(self):
+ return self._led_status
+
+ @led_status.setter
+ def led_status(self, *args, **kwargs):
+ return self.led_status_setter(*args, **kwargs)
+
+ @abstractmethod
+ def led_status_setter(self, status):
+ pass
+
+ @abstractmethod
+ def update(self) -> None:
+ pass
diff --git a/ShellyPy/discovery/__init__.py b/ShellyPy/discovery/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/ShellyPy/discovery/__init__.py
diff --git a/ShellyPy/exceptions/__init__.py b/ShellyPy/exceptions/__init__.py
new file mode 100644
index 0000000..09d11ac
--- /dev/null
+++ b/ShellyPy/exceptions/__init__.py
@@ -0,0 +1,4 @@
+
+from .backend import *
+from .request import *
+from .timer import *
diff --git a/ShellyPy/exceptions/backend.py b/ShellyPy/exceptions/backend.py
new file mode 100644
index 0000000..6d64e5d
--- /dev/null
+++ b/ShellyPy/exceptions/backend.py
@@ -0,0 +1,6 @@
+
+class UnimplementedMethod(Exception):
+ pass
+
+class InvalidBackend(Exception):
+ pass \ No newline at end of file
diff --git a/ShellyPy/exceptions/request.py b/ShellyPy/exceptions/request.py
new file mode 100644
index 0000000..ed7b56b
--- /dev/null
+++ b/ShellyPy/exceptions/request.py
@@ -0,0 +1,9 @@
+
+class UnauthorizedException(Exception):
+ pass
+
+class InvalidRequestException(Exception):
+ pass
+
+class ConnectionRefusedException(Exception):
+ pass \ No newline at end of file
diff --git a/ShellyPy/exceptions/timer.py b/ShellyPy/exceptions/timer.py
new file mode 100644
index 0000000..e9f9815
--- /dev/null
+++ b/ShellyPy/exceptions/timer.py
@@ -0,0 +1,3 @@
+
+class InvalidTimer(Exception):
+ pass \ No newline at end of file
diff --git a/ShellyPy/utils/__init__.py b/ShellyPy/utils/__init__.py
new file mode 100644
index 0000000..58d1411
--- /dev/null
+++ b/ShellyPy/utils/__init__.py
@@ -0,0 +1,4 @@
+
+from .attribute_list import *
+from .clamp import *
+from .property_fetcher import *
diff --git a/ShellyPy/utils/attribute_list.py b/ShellyPy/utils/attribute_list.py
new file mode 100644
index 0000000..1b410cc
--- /dev/null
+++ b/ShellyPy/utils/attribute_list.py
@@ -0,0 +1,36 @@
+
+class AttributeList(list):
+
+ def __setattr__(self, name, value):
+ for item in self:
+ setattr(item, name, value)
+
+ def __getattr__(self, name):
+ if not len(self):
+ raise Exception("no values")
+
+ ret = []
+
+ wrap_callable = True
+ for item in self:
+ attr = getattr(item, name)
+ if wrap_callable:
+ wrap_callable = callable(attr)
+ ret.append(attr)
+
+ if wrap_callable:
+ callable_list = ret
+ def _callable_wrapper(*args, **kwargs):
+ ret = []
+ for func in callable_list:
+ ret.append(func(*args, **kwargs))
+
+ if len(ret) == 1:
+ return ret[0]
+ return ret
+
+ return _callable_wrapper
+
+ if len(ret) == 1:
+ return ret[0]
+ return ret
diff --git a/ShellyPy/utils/clamp.py b/ShellyPy/utils/clamp.py
new file mode 100644
index 0000000..00b1df0
--- /dev/null
+++ b/ShellyPy/utils/clamp.py
@@ -0,0 +1,13 @@
+from ..base.hints import byte, temperaur
+
+def clamp(val: int, min_val: int, max_val: int) -> int:
+ return max(min(val, max_val), min_val)
+
+def clamp_percent(val: int) -> int:
+ return clamp(val, 0, 100)
+
+def clamp_byte(val: byte) -> int:
+ return clamp(val, 0, 255)
+
+def clamp_temp(val: temperaur) -> int:
+ return clamp(val, 3000, 6500) \ No newline at end of file
diff --git a/ShellyPy/utils/property_fetcher.py b/ShellyPy/utils/property_fetcher.py
new file mode 100644
index 0000000..fa1d588
--- /dev/null
+++ b/ShellyPy/utils/property_fetcher.py
@@ -0,0 +1,16 @@
+from functools import wraps
+from typing import Optional
+
+def property_fetcher(attr_name: Optional[str] = None, update_method: str = "update"):
+ def decorator(func):
+ nonlocal attr_name
+ if attr_name is None:
+ attr_name = f"_{func.__name__}"
+ @wraps(func)
+ def wrapper(self, *args, **kwargs):
+ if getattr(self, attr_name) is None:
+ update = getattr(self, update_method)
+ if update: update()
+ return func(self, *args, **kwargs)
+ return wrapper
+ return decorator
diff --git a/examples/gen.py b/examples/gen.py
new file mode 100644
index 0000000..84c5316
--- /dev/null
+++ b/examples/gen.py
@@ -0,0 +1,10 @@
+import ShellyPy
+
+# Automatically detect device generation and return the right class
+device = ShellyPy.api.Device.connect("192.0.0.20")
+
+# Explicitly connect a Generation 1 Device
+device_gen1 = ShellyPy.api.gen1.Device.connect("192.0.0.21")
+
+# Explicitly connect a Generation 2 Device
+device_gen2 = ShellyPy.api.gen2.Device.connect("192.0.0.22")
diff --git a/examples/light.py b/examples/light.py
new file mode 100644
index 0000000..3b679ab
--- /dev/null
+++ b/examples/light.py
@@ -0,0 +1,27 @@
+import ShellyPy
+
+device = ShellyPy.api.Device.connect("192.0.0.20")
+
+# turn all lights off
+device.lights.off()
+
+# turn all lights on
+device.lights.on()
+
+# toggle all lights
+device.lights.toggle()
+
+# return light 0 on
+device.lights[0].on()
+
+# Set Brightness in percent
+device.lights[0].brightness = 50 # 50%
+
+# Set RGB colors (Gen 1 only)
+device.lights[0].rgb = (255, 255, 255)
+
+# Set RGBW colors (Gen 1 only)
+device.lights[0].rgbw = (128, 128, 128, 0)
+
+# Set only one color (Gen 1 only)
+device.lights[0].red = 64
diff --git a/examples/relay.py b/examples/relay.py
new file mode 100644
index 0000000..36bbe45
--- /dev/null
+++ b/examples/relay.py
@@ -0,0 +1,15 @@
+import ShellyPy
+
+device = ShellyPy.api.Device.connect("192.0.0.20")
+
+# turn all relays off
+device.relays.off()
+
+# turn all relays on
+device.relays.on()
+
+# toggle all relays
+device.relays.toggle()
+
+# return relay 0 on
+device.relays[0].on()
diff --git a/examples/roller.py b/examples/roller.py
new file mode 100644
index 0000000..70a6446
--- /dev/null
+++ b/examples/roller.py
@@ -0,0 +1,18 @@
+import ShellyPy
+
+device = ShellyPy.api.Device.connect("192.0.0.20")
+
+# open all rollers
+device.rollers.open()
+
+# close all rollers
+device.rollers.close()
+
+# stop all rollers
+device.rollers.stop()
+
+# set roller 1 position to 50% (requires calibration)
+device.rollers[1].pos = 50 # 50%
+
+# calibrate roller 1 (will find new min and max position)
+device.rollers[1].calibrate()
diff --git a/requirements.txt b/requirements.txt
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/requirements.txt
diff --git a/setup.py b/setup.py
new file mode 100755
index 0000000..ba2fd2f
--- /dev/null
+++ b/setup.py
@@ -0,0 +1,54 @@
+# -*- coding: utf-8 -*-
+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()
+
+extras_require = {
+ "discovery": ["zeroconf"]
+}
+
+setuptools.setup(
+ name="ShellyPy",
+ version=version,
+ author="Jan Drögehoff",
+ author_email="jandroegehoff@gmail.com",
+ long_description=long_description,
+ description="Library to interfacewith Shelly Smart Home Devices",
+ long_description_content_type="text/markdown",
+ url="https://github.com/Jan200101/ShellyPy",
+ packages=[
+ "ShellyPy",
+ "ShellyPy.api",
+
+ "ShellyPy.api.gen1",
+ "ShellyPy.api.gen1.backends.http",
+
+ "ShellyPy.api.gen2",
+ "ShellyPy.api.gen2.backends.json_rpc",
+
+ "ShellyPy.discovery",
+ ],
+ license="MIT",
+ install_requires=requirements,
+ include_package_data=True,
+ extras_require=extras_require,
+ classifiers=[
+ "Programming Language :: Python :: 3",
+ "License :: OSI Approved :: MIT License",
+ "Operating System :: OS Independent",
+ ],
+)