From b966f6e4705a813dad13b242e9592e699566d9c9 Mon Sep 17 00:00:00 2001 From: Jan200101 Date: Sun, 9 Jul 2023 19:37:37 +0200 Subject: Squash rewrite --- .gitignore | 7 + LICENSE | 22 +++ README.md | 1 + ShellyPy/__init__.py | 8 + ShellyPy/api/__init__.py | 4 + ShellyPy/api/device.py | 55 ++++++ ShellyPy/api/gen1/__init__.py | 2 + ShellyPy/api/gen1/backends/__init__.py | 2 + ShellyPy/api/gen1/backends/http/__init__.py | 7 + ShellyPy/api/gen1/backends/http/light.py | 104 ++++++++++ ShellyPy/api/gen1/backends/http/meter.py | 23 +++ ShellyPy/api/gen1/backends/http/relay.py | 54 ++++++ ShellyPy/api/gen1/backends/http/request.py | 90 +++++++++ ShellyPy/api/gen1/backends/http/roller.py | 63 ++++++ ShellyPy/api/gen1/backends/http/settings.py | 39 ++++ ShellyPy/api/gen1/device.py | 56 ++++++ ShellyPy/api/gen2/__init__.py | 2 + ShellyPy/api/gen2/backends/__init__.py | 2 + ShellyPy/api/gen2/backends/json_rpc/__init__.py | 7 + ShellyPy/api/gen2/backends/json_rpc/light.py | 73 +++++++ ShellyPy/api/gen2/backends/json_rpc/meter.py | 24 +++ ShellyPy/api/gen2/backends/json_rpc/relay.py | 33 ++++ ShellyPy/api/gen2/backends/json_rpc/request.py | 136 +++++++++++++ ShellyPy/api/gen2/backends/json_rpc/roller.py | 45 +++++ ShellyPy/api/gen2/backends/json_rpc/settings.py | 25 +++ ShellyPy/api/gen2/device.py | 106 ++++++++++ ShellyPy/base/__init__.py | 8 + ShellyPy/base/device.py | 167 ++++++++++++++++ ShellyPy/base/hints.py | 17 ++ ShellyPy/base/light.py | 248 ++++++++++++++++++++++++ ShellyPy/base/meter.py | 53 +++++ ShellyPy/base/relay.py | 88 +++++++++ ShellyPy/base/roller.py | 128 ++++++++++++ ShellyPy/base/settings.py | 56 ++++++ ShellyPy/discovery/__init__.py | 0 ShellyPy/exceptions/__init__.py | 4 + ShellyPy/exceptions/backend.py | 6 + ShellyPy/exceptions/request.py | 9 + ShellyPy/exceptions/timer.py | 3 + ShellyPy/utils/__init__.py | 4 + ShellyPy/utils/attribute_list.py | 36 ++++ ShellyPy/utils/clamp.py | 13 ++ ShellyPy/utils/property_fetcher.py | 16 ++ examples/gen.py | 10 + examples/light.py | 27 +++ examples/relay.py | 15 ++ examples/roller.py | 18 ++ requirements.txt | 0 setup.py | 54 ++++++ 49 files changed, 1970 insertions(+) create mode 100644 .gitignore create mode 100644 LICENSE create mode 100644 README.md create mode 100644 ShellyPy/__init__.py create mode 100644 ShellyPy/api/__init__.py create mode 100644 ShellyPy/api/device.py create mode 100644 ShellyPy/api/gen1/__init__.py create mode 100644 ShellyPy/api/gen1/backends/__init__.py create mode 100644 ShellyPy/api/gen1/backends/http/__init__.py create mode 100644 ShellyPy/api/gen1/backends/http/light.py create mode 100644 ShellyPy/api/gen1/backends/http/meter.py create mode 100644 ShellyPy/api/gen1/backends/http/relay.py create mode 100644 ShellyPy/api/gen1/backends/http/request.py create mode 100644 ShellyPy/api/gen1/backends/http/roller.py create mode 100644 ShellyPy/api/gen1/backends/http/settings.py create mode 100644 ShellyPy/api/gen1/device.py create mode 100644 ShellyPy/api/gen2/__init__.py create mode 100644 ShellyPy/api/gen2/backends/__init__.py create mode 100644 ShellyPy/api/gen2/backends/json_rpc/__init__.py create mode 100644 ShellyPy/api/gen2/backends/json_rpc/light.py create mode 100644 ShellyPy/api/gen2/backends/json_rpc/meter.py create mode 100644 ShellyPy/api/gen2/backends/json_rpc/relay.py create mode 100644 ShellyPy/api/gen2/backends/json_rpc/request.py create mode 100644 ShellyPy/api/gen2/backends/json_rpc/roller.py create mode 100644 ShellyPy/api/gen2/backends/json_rpc/settings.py create mode 100644 ShellyPy/api/gen2/device.py create mode 100644 ShellyPy/base/__init__.py create mode 100644 ShellyPy/base/device.py create mode 100644 ShellyPy/base/hints.py create mode 100644 ShellyPy/base/light.py create mode 100644 ShellyPy/base/meter.py create mode 100644 ShellyPy/base/relay.py create mode 100644 ShellyPy/base/roller.py create mode 100644 ShellyPy/base/settings.py create mode 100644 ShellyPy/discovery/__init__.py create mode 100644 ShellyPy/exceptions/__init__.py create mode 100644 ShellyPy/exceptions/backend.py create mode 100644 ShellyPy/exceptions/request.py create mode 100644 ShellyPy/exceptions/timer.py create mode 100644 ShellyPy/utils/__init__.py create mode 100644 ShellyPy/utils/attribute_list.py create mode 100644 ShellyPy/utils/clamp.py create mode 100644 ShellyPy/utils/property_fetcher.py create mode 100644 examples/gen.py create mode 100644 examples/light.py create mode 100644 examples/relay.py create mode 100644 examples/roller.py create mode 100644 requirements.txt create mode 100755 setup.py 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 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 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", + ], +) -- cgit v1.2.3