From 7d27503b3b68791c718690732e9dff7a5d2cdc40 Mon Sep 17 00:00:00 2001 From: Sebastian Wolf Date: Sun, 8 Sep 2024 19:16:56 +0200 Subject: [PATCH 01/18] Remove version check for python 2 as its EOL is long ago. Add some of first type hints. Make Methods of ShellyBase abstract methods instead of raising an exception. This should be done in the child implementations. Small format fixes. --- ShellyPy/base.py | 41 +++++++++++++++++++++++------------------ ShellyPy/gen1.py | 10 ++-------- ShellyPy/gen2.py | 9 ++------- ShellyPy/wrapper.py | 35 +++++++++++++++++------------------ 4 files changed, 44 insertions(+), 51 deletions(-) diff --git a/ShellyPy/base.py b/ShellyPy/base.py index 6c34be0..5f430e1 100644 --- a/ShellyPy/base.py +++ b/ShellyPy/base.py @@ -1,13 +1,9 @@ -from sys import version_info +from abc import abstractmethod -if version_info.major == 3: - from json.decoder import JSONDecodeError -else: - JSONDecodeError = ValueError class ShellyBase: - def __init__(self, ip, port = "80", *args, **kwargs): + def __init__(self, ip: str, port: int = 80, *args, **kwargs) -> None: """ @param ip the target IP of the shelly device. Can be a string, list of strings or list of integers @param port target port, may be useful for non Shelly devices that have the same HTTP Api @@ -80,29 +76,38 @@ def __clamp_kalvin__(val): val = 3000 return val - def update(self): - raise NotImplementedError("Base Class") + @abstractmethod + def update(self) -> None: + ... + @abstractmethod def post(self, page, values = None): - raise NotImplementedError("Base Class") + ... + @abstractmethod def status(self): - raise NotImplementedError("Base Class") + ... + @abstractmethod def settings(self, subpage = None): - raise NotImplementedError("Base Class") + ... + @abstractmethod def meter(self, index): - raise NotImplementedError("Base Class") - + ... + + @abstractmethod def relay(self, index, *args, **kwargs): - raise NotImplementedError("Base Class") + ... + @abstractmethod def roller(self, index, *args, **kwargs): - raise NotImplementedError("Base Class") + ... + @abstractmethod def light(self, index, *args, **kwargs): - raise NotImplementedError("Base Class") - + ... + + @abstractmethod def emeter(self, index, *args, **kwargs): - raise NotImplementedError("Base Class") + ... diff --git a/ShellyPy/gen1.py b/ShellyPy/gen1.py index dea4650..dbf104d 100644 --- a/ShellyPy/gen1.py +++ b/ShellyPy/gen1.py @@ -1,10 +1,4 @@ -from sys import version_info - -if version_info.major == 3: - from json.decoder import JSONDecodeError -else: - JSONDecodeError = ValueError - +from json.decoder import JSONDecodeError from requests import post from requests.auth import HTTPBasicAuth @@ -15,7 +9,7 @@ class ShellyGen1(ShellyBase): - def __init__(self, ip, port = "80", *args, **kwargs): + def __init__(self, ip: str, port: int = 80, *args, **kwargs) -> None: """ @param ip the target IP of the shelly device. Can be a string, list of strings or list of integers @param port target port, may be useful for non Shelly devices that have the same HTTP Api diff --git a/ShellyPy/gen2.py b/ShellyPy/gen2.py index 199fb84..2c42295 100644 --- a/ShellyPy/gen2.py +++ b/ShellyPy/gen2.py @@ -1,9 +1,4 @@ -from sys import version_info - -if version_info.major == 3: - from json.decoder import JSONDecodeError -else: - JSONDecodeError = ValueError +from json.decoder import JSONDecodeError from requests import post from requests.auth import HTTPDigestAuth @@ -14,7 +9,7 @@ class ShellyGen2(ShellyBase): - def __init__(self, ip, port = "80", *args, **kwargs): + def __init__(self, ip: str, port: int = 80, *args, **kwargs) -> None: """ @param ip the target IP of the shelly device. Can be a string, list of strings or list of integers @param port target port, may be useful for non Shelly devices that have the same HTTP Api diff --git a/ShellyPy/wrapper.py b/ShellyPy/wrapper.py index 2715ce0..d9dce48 100644 --- a/ShellyPy/wrapper.py +++ b/ShellyPy/wrapper.py @@ -1,35 +1,34 @@ -from sys import version_info - -if version_info.major == 3: - from json.decoder import JSONDecodeError -else: - JSONDecodeError = ValueError +from json.decoder import JSONDecodeError +from typing import Any, Type from requests import get +from requests import Response +from .base import ShellyBase from .error import BadLogin, NotFound, BadResponse from .gen1 import ShellyGen1 from .gen2 import ShellyGen2 -class Shelly(): +class Shelly: - def __init__(self, ip, port = "80", *args, **kwargs): + def __init__(self, ip: str, port: int = 80, *args, **kwargs) -> None: """ @param ip the target IP of the shelly device. Can be a string, list of strings or list of integers - @param port target port, may be useful for non Shelly devices that have the same HTTP Api + @param port target port, may be useful for non Shelly devices that have the same HTTP Api @param login dict of login credentials. Keys needed are "username" and "password" @param timeout specify the amount of time until requests are aborted. @param debug enable debug printing @param init calls the update method on init """ - self._instance = self.__detect__(ip, port)(ip, port, *args, **kwargs) + self._instance = self.__detect(ip, port)(ip, port, *args, **kwargs) - def __detect__(self, ip, port): - url = "{}://{}:{}/shelly".format("http", ip, port) + @staticmethod + def __detect(ip: str, port: int, proto: str = 'http') -> Type[ShellyBase]: + url: str = f"{proto}://{ip}:{port}/shelly" - response = get(url, timeout=5) + response: Response = get(url, timeout=5) if response.status_code == 401: raise BadLogin() @@ -41,20 +40,20 @@ def __detect__(self, ip, port): except JSONDecodeError: raise BadResponse("Bad JSON") - gen = response_data.get("gen", 1) + gen: int = response_data.get("gen", 1) if gen == 1: return ShellyGen1 elif gen == 2: return ShellyGen2 else: - raise ValueError("Generation {} not supported".format(gen)) + raise ValueError(f"Generation {gen} not supported") - def __repr__(self): + def __repr__(self) -> str: return self.__getattr__("__repr__")() - def __str__(self): + def __str__(self) -> str: return self.__getattr__("__str__")() - def __getattr__(self, name): + def __getattr__(self, name: str) -> Any: return getattr(self._instance, name) From 30387ace2e6a6393fa69aee3306f294d9fc03453 Mon Sep 17 00:00:00 2001 From: Sebastian Wolf Date: Sun, 8 Sep 2024 20:37:21 +0200 Subject: [PATCH 02/18] Remove python2 from setup --- setup.py | 1 - 1 file changed, 1 deletion(-) diff --git a/setup.py b/setup.py index 6f3708e..aea22a1 100755 --- a/setup.py +++ b/setup.py @@ -32,7 +32,6 @@ install_requires=requirements, include_package_data=True, classifiers=[ - "Programming Language :: Python :: 2", "Programming Language :: Python :: 3", "License :: OSI Approved :: MIT License", "Operating System :: OS Independent", From 241e3f833e02b35c243266597ffcff8cbc5359b9 Mon Sep 17 00:00:00 2001 From: Sebastian Wolf Date: Sun, 8 Sep 2024 21:19:55 +0200 Subject: [PATCH 03/18] Cleanup docstring and reformat strings using f-string. --- ShellyPy/base.py | 1 + ShellyPy/gen1.py | 32 ++++++++++++++------------------ ShellyPy/gen2.py | 15 ++++++++------- ShellyPy/wrapper.py | 4 ++-- 4 files changed, 25 insertions(+), 27 deletions(-) diff --git a/ShellyPy/base.py b/ShellyPy/base.py index 5f430e1..5abe016 100644 --- a/ShellyPy/base.py +++ b/ShellyPy/base.py @@ -1,3 +1,4 @@ +from typing import Optional from abc import abstractmethod diff --git a/ShellyPy/gen1.py b/ShellyPy/gen1.py index dbf104d..c5ee8df 100644 --- a/ShellyPy/gen1.py +++ b/ShellyPy/gen1.py @@ -1,12 +1,13 @@ +from typing import Optional from json.decoder import JSONDecodeError from requests import post from requests.auth import HTTPBasicAuth from .error import BadLogin, NotFound, BadResponse - from .base import ShellyBase + class ShellyGen1(ShellyBase): def __init__(self, ip: str, port: int = 80, *args, **kwargs) -> None: @@ -22,11 +23,9 @@ def __init__(self, ip: str, port: int = 80, *args, **kwargs) -> None: super().__init__(ip, port, *args, **kwargs) self.__generation__ = 1 - def update(self): + def update(self) -> None: """ @brief update the Shelly attributes - - @param index index of the relay """ status = self.settings() @@ -51,28 +50,25 @@ def update(self): # Request meter information self.meter.append(self.meter(meter_index)) meter_index += 1 - except: + except Exception: break def post(self, page, values = None): """ @brief returns settings - @param page page to be accesed. Use the Shelly HTTP API Reference to see whats possible + @param page page to be accessed. Use the Shelly HTTP API Reference to see what's possible @return returns json response """ - url = "{}://{}:{}/{}?".format(self.__PROTOCOL__, self.__ip__, self.__port__, page) + url = f"{self.__PROTOCOL__}://{self.__ip__}:{self.__port__}/{page}?" if values: - url += "&".join(["{}={}".format(key, value) for key, value in values.items()]) + url += "&".join([f"{key}={value}" for key, value in values.items()]) if self.__debugging__: - print("Target Adress: {}\n" - "Authentication: {}\n" - "Timeout: {}" - "".format(url, any(self.__credentials__), self.__timeout__)) + print(f"Target Address: {url}\nAuthentication: {any(self.__credentials__)}\nTimeout: {self.__timeout__}") credentials = HTTPBasicAuth(*self.__credentials__) @@ -101,7 +97,7 @@ def settings(self, subpage = None): """ @brief returns settings - @param page page to be accesed. Use the Shelly HTTP API Reference to see whats possible + @param subpage page to be accessed. Use the Shelly HTTP API Reference to see what's possible @return returns settings as a dict """ @@ -120,7 +116,7 @@ def meter(self, index): @return returns attributes of meter: power, overpower, is_valid, timestamp, counters, total """ - return self.post("meter/{}".format(index)) + return self.post(f"meter/{index}") def relay(self, index, *args, **kwargs): """ @@ -145,7 +141,7 @@ def relay(self, index, *args, **kwargs): if timer: values["timer"] = timer - return self.post("relay/{}".format(index), values) + return self.post(f"relay/{index}", values) def roller(self, index, *args, **kwargs): """ @@ -173,7 +169,7 @@ def roller(self, index, *args, **kwargs): if duration is not None: values["duration"] = duration - return self.post("roller/{}".format(index), values) + return self.post(f"roller/{index}", values) def light(self, index, *args, **kwargs): """ @@ -237,11 +233,11 @@ def light(self, index, *args, **kwargs): if brightness is not None: values["brightness"] = self.__clamp_percentage__(brightness) - return self.post("light/{}".format(index), values) + return self.post(f"light/{index}", values) def emeter(self, index, *args, **kwargs): - return self.post("emeter/{}".format(index)) + return self.post(f"emeter/{index}") # backwards compatibility with old code Shelly = ShellyGen1 diff --git a/ShellyPy/gen2.py b/ShellyPy/gen2.py index 2c42295..02071d7 100644 --- a/ShellyPy/gen2.py +++ b/ShellyPy/gen2.py @@ -1,12 +1,13 @@ +from typing import Optional from json.decoder import JSONDecodeError from requests import post from requests.auth import HTTPDigestAuth from .error import BadLogin, NotFound, BadResponse - from .base import ShellyBase + class ShellyGen2(ShellyBase): def __init__(self, ip: str, port: int = 80, *args, **kwargs) -> None: @@ -19,18 +20,18 @@ def __init__(self, ip: str, port: int = 80, *args, **kwargs) -> None: @param init calls the update method on init """ - self.payload_id = 1 super().__init__(ip, port, *args, **kwargs) + self.payload_id = 1 self.__generation__ = 2 - def update(self): + def update(self) -> None: status = self.settings() self.__name__ = status["device"].get("name", self.__name__) self.__type__ = status["device"].get("mac", self.__type__) def post(self, page, values = None): - url = "{}://{}:{}/rpc".format(self.__PROTOCOL__, self.__ip__, self.__port__) + url = f"{self.__PROTOCOL__}://{self.__ip__}:{self.__port__}/rpc" # increment payload id globally self.payload_id += 1 @@ -48,7 +49,7 @@ def post(self, page, values = None): credentials = None try: - credentials = auth=HTTPDigestAuth('admin', self.__credentials__[1]) + credentials = HTTPDigestAuth('admin', self.__credentials__[1]) except IndexError: pass @@ -75,7 +76,7 @@ def post(self, page, values = None): elif error_code == 404: raise NotFound(error_message) else: - raise BadResponse("{}: {}".format(error_code, error_message)) + raise BadResponse(f"{error_code}: {error_message}") if response_data["id"] != payload_id: raise BadResponse("invalid payload id was returned") @@ -148,4 +149,4 @@ def light(self, index, *args, **kwargs): raise NotImplementedError("Unavailable") def emeter(self, index, *args, **kwargs): - raise NotImplementedError("Unavailable") \ No newline at end of file + raise NotImplementedError("Unavailable") diff --git a/ShellyPy/wrapper.py b/ShellyPy/wrapper.py index d9dce48..de8291f 100644 --- a/ShellyPy/wrapper.py +++ b/ShellyPy/wrapper.py @@ -1,15 +1,15 @@ +from typing import Any, Optional, Type from json.decoder import JSONDecodeError -from typing import Any, Type from requests import get from requests import Response from .base import ShellyBase from .error import BadLogin, NotFound, BadResponse - from .gen1 import ShellyGen1 from .gen2 import ShellyGen2 + class Shelly: def __init__(self, ip: str, port: int = 80, *args, **kwargs) -> None: From a23343778ae5f71d438753f01ed0881c5cdf0af5 Mon Sep 17 00:00:00 2001 From: Sebastian Wolf Date: Sun, 8 Sep 2024 21:30:55 +0200 Subject: [PATCH 04/18] Rename private properties using regular naming convention instead of some dunder-style --- ShellyPy/base.py | 26 +++++++++++++------------- ShellyPy/gen1.py | 16 ++++++++-------- ShellyPy/gen2.py | 10 +++++----- 3 files changed, 26 insertions(+), 26 deletions(-) diff --git a/ShellyPy/base.py b/ShellyPy/base.py index 5abe016..2c27bc3 100644 --- a/ShellyPy/base.py +++ b/ShellyPy/base.py @@ -14,36 +14,36 @@ def __init__(self, ip: str, port: int = 80, *args, **kwargs) -> None: @param init calls the update method on init """ - self.__name__ = "Unknown" - self.__type__ = "Unknown" - self.__generation__ = 0 + self._name = "Unknown" + self._type = "Unknown" + self._generation = 0 - self.__debugging__ = kwargs.get("debug", None) + self._debugging = kwargs.get("debug", None) - self.__PROTOCOL__ = "http" + self._proto = "http" login = kwargs.get("login", {}) # hostname would be more fitting, # but backwards compatibility - self.__ip__ = ip + self._hostname = ip - self.__port__ = port + self._port = port - self.__timeout__ = kwargs.get("timeout", 5) + self._timeout = kwargs.get("timeout", 5) - self.__credentials__ = ( + self._credentials = ( login.get("username", ""), login.get("password", "") ) if kwargs.get("init"): self.update() - def __repr__(self): - return "<{} {} Gen {} ({})>".format(self.__name__, self.__type__, self.__generation__, self.__ip__) + def __repr__(self) -> str: + return f"<{self._name} {self._type} Gen {self._generation} ({self._hostname})>" - def __str__(self): - return str(self.__name__) + def __str__(self) -> str: + return self._name @staticmethod def __clamp__(val): diff --git a/ShellyPy/gen1.py b/ShellyPy/gen1.py index c5ee8df..634bdd6 100644 --- a/ShellyPy/gen1.py +++ b/ShellyPy/gen1.py @@ -21,7 +21,7 @@ def __init__(self, ip: str, port: int = 80, *args, **kwargs) -> None: """ super().__init__(ip, port, *args, **kwargs) - self.__generation__ = 1 + self._generation = 1 def update(self) -> None: """ @@ -29,8 +29,8 @@ def update(self) -> None: """ status = self.settings() - self.__type__ = status['device'].get("type", self.__type__) - self.__name__ = status['device'].get("hostname", self.__name__) + self._type = status['device'].get("type", self._type) + self._name = status['device'].get("hostname", self._name) # Settings are already fetched to get device information might as well put the list of things the device has somewhere self.relays = status.get("relays", []) @@ -62,18 +62,18 @@ def post(self, page, values = None): @return returns json response """ - url = f"{self.__PROTOCOL__}://{self.__ip__}:{self.__port__}/{page}?" + url = f"{self._proto}://{self._hostname}:{self._port}/{page}?" if values: url += "&".join([f"{key}={value}" for key, value in values.items()]) - if self.__debugging__: - print(f"Target Address: {url}\nAuthentication: {any(self.__credentials__)}\nTimeout: {self.__timeout__}") + if self._debugging: + print(f"Target Address: {url}\nAuthentication: {any(self._credentials)}\nTimeout: {self._timeout}") - credentials = HTTPBasicAuth(*self.__credentials__) + credentials = HTTPBasicAuth(*self._credentials) response = post(url, auth=credentials, - timeout=self.__timeout__) + timeout=self._timeout) if response.status_code == 401: raise BadLogin() diff --git a/ShellyPy/gen2.py b/ShellyPy/gen2.py index 02071d7..bdc419e 100644 --- a/ShellyPy/gen2.py +++ b/ShellyPy/gen2.py @@ -27,11 +27,11 @@ def __init__(self, ip: str, port: int = 80, *args, **kwargs) -> None: def update(self) -> None: status = self.settings() - self.__name__ = status["device"].get("name", self.__name__) - self.__type__ = status["device"].get("mac", self.__type__) + self._name = status["device"].get("name", self._name) + self._type = status["device"].get("mac", self._type) def post(self, page, values = None): - url = f"{self.__PROTOCOL__}://{self.__ip__}:{self.__port__}/rpc" + url = f"{self._proto}://{self._hostname}:{self._port}/rpc" # increment payload id globally self.payload_id += 1 @@ -49,13 +49,13 @@ def post(self, page, values = None): credentials = None try: - credentials = HTTPDigestAuth('admin', self.__credentials__[1]) + credentials = HTTPDigestAuth('admin', self._credentials[1]) except IndexError: pass response = post(url, auth=credentials, json=payload, - timeout=self.__timeout__) + timeout=self._timeout) if response.status_code == 401: raise BadLogin() From 750ba600571c329078363ca703b1a365a12993ac Mon Sep 17 00:00:00 2001 From: Sebastian Wolf Date: Sun, 8 Sep 2024 21:40:28 +0200 Subject: [PATCH 05/18] Rename private methods using regular naming convention instead of pseudo dunder-methods. Apply type hints for those. Fix typo kalvin -> kelvin --- ShellyPy/base.py | 8 ++++---- ShellyPy/gen1.py | 16 ++++++++-------- ShellyPy/gen2.py | 2 +- 3 files changed, 13 insertions(+), 13 deletions(-) diff --git a/ShellyPy/base.py b/ShellyPy/base.py index 2c27bc3..749d50e 100644 --- a/ShellyPy/base.py +++ b/ShellyPy/base.py @@ -46,7 +46,7 @@ def __str__(self) -> str: return self._name @staticmethod - def __clamp__(val): + def _clamp(val: int) -> int: """clamp any number to 8 bit""" if val > 255: val = 255 @@ -56,7 +56,7 @@ def __clamp__(val): return val @staticmethod - def __clamp_percentage__(val): + def _clamp_percentage(val: int) -> int: """ clamp given percentage to a range from 0 to 100 """ @@ -67,9 +67,9 @@ def __clamp_percentage__(val): return val @staticmethod - def __clamp_kalvin__(val): + def _clamp_kelvin(val: int) -> int: """ - clamp given kalvin values for a range from 3000..6500 + clamp given kelvin values for a range from 3000..6500 """ if val > 6500: val = 6500 diff --git a/ShellyPy/gen1.py b/ShellyPy/gen1.py index 634bdd6..94a3513 100644 --- a/ShellyPy/gen1.py +++ b/ShellyPy/gen1.py @@ -164,7 +164,7 @@ def roller(self, index, *args, **kwargs): values["go"] = go if roller_pos is not None: - values["roller_pos"] = self.__clamp_percentage__(roller_pos) + values["roller_pos"] = self._clamp_percentage(roller_pos) if duration is not None: values["duration"] = duration @@ -213,25 +213,25 @@ def light(self, index, *args, **kwargs): values["turn"] = "off" if red is not None: - values["red"] = self.__clamp__(red) + values["red"] = self._clamp(red) if green is not None: - values["green"] = self.__clamp__(green) + values["green"] = self._clamp(green) if blue is not None: - values["blue"] = self.__clamp__(blue) + values["blue"] = self._clamp(blue) if white is not None: - values["white"] = self.__clamp__(white) + values["white"] = self._clamp(white) if gain is not None: - values["gain"] = self.__clamp_percentage__(gain) + values["gain"] = self._clamp_percentage(gain) if temp is not None: - values["temp"] = self.__clamp_kalvin__(temp) + values["temp"] = self._clamp_kelvin(temp) if brightness is not None: - values["brightness"] = self.__clamp_percentage__(brightness) + values["brightness"] = self._clamp_percentage(brightness) return self.post(f"light/{index}", values) diff --git a/ShellyPy/gen2.py b/ShellyPy/gen2.py index bdc419e..ca3ac17 100644 --- a/ShellyPy/gen2.py +++ b/ShellyPy/gen2.py @@ -138,7 +138,7 @@ def roller(self, index, *args, **kwargs): if roller_pos is not None: method = "Cover.GoToPosition" - values["pos"] = self.__clamp_percentage__(roller_pos) + values["pos"] = self._clamp_percentage(roller_pos) if duration is not None: values["duration"] = duration From 02012a259a10807569e23e9e6fd051494215a353 Mon Sep 17 00:00:00 2001 From: Sebastian Wolf Date: Sun, 8 Sep 2024 23:21:15 +0200 Subject: [PATCH 06/18] Change __init__ signatures using keywords instead of kwargs. Add type hints in initializers. Initialize some properties of gen1 that were not initialized before. Add types-request to requirements for better type hints. --- ShellyPy/base.py | 39 +++++++++++++++------------------------ ShellyPy/gen1.py | 22 ++++++++++++++++------ ShellyPy/gen2.py | 18 ++++++++++-------- ShellyPy/wrapper.py | 8 +++++--- requirements.txt | 1 + 5 files changed, 47 insertions(+), 41 deletions(-) diff --git a/ShellyPy/base.py b/ShellyPy/base.py index 749d50e..6700b92 100644 --- a/ShellyPy/base.py +++ b/ShellyPy/base.py @@ -4,7 +4,8 @@ class ShellyBase: - def __init__(self, ip: str, port: int = 80, *args, **kwargs) -> None: + def __init__(self, ip: str, port: int = 80, timeout: int = 5, + login: Optional[dict[str, str]] = None, debug: bool = False, init: bool = False) -> None: """ @param ip the target IP of the shelly device. Can be a string, list of strings or list of integers @param port target port, may be useful for non Shelly devices that have the same HTTP Api @@ -14,29 +15,19 @@ def __init__(self, ip: str, port: int = 80, *args, **kwargs) -> None: @param init calls the update method on init """ - self._name = "Unknown" - self._type = "Unknown" - self._generation = 0 + self._name: str = "Unknown" + self._type: str = "Unknown" + self._generation: int = 0 - self._debugging = kwargs.get("debug", None) + self._proto: str = "http" + self._hostname: str = ip # hostname would be more fitting, but backwards compatibility + self._port: int = port + self._timeout: int = timeout + self._credentials: tuple[str, str] = (login.get("username", ""), + login.get("password", "")) if login is not None else ("", "") - self._proto = "http" - - login = kwargs.get("login", {}) - - # hostname would be more fitting, - # but backwards compatibility - self._hostname = ip - - self._port = port - - self._timeout = kwargs.get("timeout", 5) - - self._credentials = ( - login.get("username", ""), login.get("password", "") - ) - - if kwargs.get("init"): + self._debugging: bool = debug + if init: self.update() def __repr__(self) -> str: @@ -82,7 +73,7 @@ def update(self) -> None: ... @abstractmethod - def post(self, page, values = None): + def post(self, page: str, values = None): ... @abstractmethod @@ -94,7 +85,7 @@ def settings(self, subpage = None): ... @abstractmethod - def meter(self, index): + def meter(self, index: int): ... @abstractmethod diff --git a/ShellyPy/gen1.py b/ShellyPy/gen1.py index 94a3513..76c0aa1 100644 --- a/ShellyPy/gen1.py +++ b/ShellyPy/gen1.py @@ -10,7 +10,8 @@ class ShellyGen1(ShellyBase): - def __init__(self, ip: str, port: int = 80, *args, **kwargs) -> None: + def __init__(self, ip: str, port: int = 80, timeout: int = 5, + login: Optional[dict[str, str]] = None, debug: bool = False, init: bool = False) -> None: """ @param ip the target IP of the shelly device. Can be a string, list of strings or list of integers @param port target port, may be useful for non Shelly devices that have the same HTTP Api @@ -20,8 +21,17 @@ def __init__(self, ip: str, port: int = 80, *args, **kwargs) -> None: @param init calls the update method on init """ - super().__init__(ip, port, *args, **kwargs) - self._generation = 1 + super().__init__(ip=ip, port=port, timeout=timeout, login=login, debug=debug, init=init) + self._generation: int = 1 + + self.relays: list = [] + self.rollers: list = [] + self.lights: list = [] + + self.irs = None + + self.emeters: list = [] + self.meters: list[dict] = [] def update(self) -> None: """ @@ -48,12 +58,12 @@ def update(self) -> None: while True: try: # Request meter information - self.meter.append(self.meter(meter_index)) + self.meters.append(self.meter(meter_index)) meter_index += 1 except Exception: break - def post(self, page, values = None): + def post(self, page: str, values = None): """ @brief returns settings @@ -108,7 +118,7 @@ def settings(self, subpage = None): return self.post(page) - def meter(self, index): + def meter(self, index: int): """ @brief Get meter information from a relay at the given index diff --git a/ShellyPy/gen2.py b/ShellyPy/gen2.py index ca3ac17..c7eab51 100644 --- a/ShellyPy/gen2.py +++ b/ShellyPy/gen2.py @@ -10,7 +10,8 @@ class ShellyGen2(ShellyBase): - def __init__(self, ip: str, port: int = 80, *args, **kwargs) -> None: + def __init__(self, ip: str, port: int = 80, timeout: int = 5, + login: Optional[dict[str, str]] = None, debug: bool = False, init: bool = False) -> None: """ @param ip the target IP of the shelly device. Can be a string, list of strings or list of integers @param port target port, may be useful for non Shelly devices that have the same HTTP Api @@ -20,9 +21,9 @@ def __init__(self, ip: str, port: int = 80, *args, **kwargs) -> None: @param init calls the update method on init """ - super().__init__(ip, port, *args, **kwargs) - self.payload_id = 1 - self.__generation__ = 2 + super().__init__(ip=ip, port=port, timeout=timeout, login=login, debug=debug, init=init) + self.payload_id: int = 1 + self._generation: int = 2 def update(self) -> None: status = self.settings() @@ -30,7 +31,7 @@ def update(self) -> None: self._name = status["device"].get("name", self._name) self._type = status["device"].get("mac", self._type) - def post(self, page, values = None): + def post(self, page: str, values = None): url = f"{self._proto}://{self._hostname}:{self._port}/rpc" # increment payload id globally @@ -89,7 +90,7 @@ def status(self): def settings(self, subpage = None): return self.post("Sys.GetConfig") - def meter(self, index): + def meter(self, index: int): raise NotImplementedError("Unavailable") def relay(self, index, *args, **kwargs): @@ -122,7 +123,8 @@ def roller(self, index, *args, **kwargs): roller_pos = kwargs.get("roller_pos", None) duration = kwargs.get("duration", None) - values = { + method: str = "" + values: dict[str, Any] = { "id": index } @@ -147,6 +149,6 @@ def roller(self, index, *args, **kwargs): def light(self, index, *args, **kwargs): raise NotImplementedError("Unavailable") - + def emeter(self, index, *args, **kwargs): raise NotImplementedError("Unavailable") diff --git a/ShellyPy/wrapper.py b/ShellyPy/wrapper.py index de8291f..87384e5 100644 --- a/ShellyPy/wrapper.py +++ b/ShellyPy/wrapper.py @@ -12,7 +12,8 @@ class Shelly: - def __init__(self, ip: str, port: int = 80, *args, **kwargs) -> None: + def __init__(self, ip: str, port: int = 80, timeout: int = 5, + login: Optional[dict[str, str]] = None, debug: bool = False, init: bool = False) -> None: """ @param ip the target IP of the shelly device. Can be a string, list of strings or list of integers @param port target port, may be useful for non Shelly devices that have the same HTTP Api @@ -22,7 +23,8 @@ def __init__(self, ip: str, port: int = 80, *args, **kwargs) -> None: @param init calls the update method on init """ - self._instance = self.__detect(ip, port)(ip, port, *args, **kwargs) + self._instance: ShellyBase = self.__detect(ip, port)(ip=ip, port=port, timeout=timeout, + login=login, debug=debug, init=init) @staticmethod def __detect(ip: str, port: int, proto: str = 'http') -> Type[ShellyBase]: @@ -36,7 +38,7 @@ def __detect(ip: str, port: int, proto: str = 'http') -> Type[ShellyBase]: raise NotFound("Not Found") try: - response_data = response.json() + response_data: dict[str, Any] = response.json() except JSONDecodeError: raise BadResponse("Bad JSON") diff --git a/requirements.txt b/requirements.txt index f229360..8dee024 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1 +1,2 @@ requests +types-requests From 4adae76e1a4f0a1b2668d47346c7c80e6f478742 Mon Sep 17 00:00:00 2001 From: Sebastian Wolf Date: Mon, 9 Sep 2024 07:32:22 +0200 Subject: [PATCH 07/18] Remove any appearance of **kwarg arguments with known keywords that are later extracted with those keywords in signature Only relay() method is an exception for that. Is signature is not final for now --- .gitignore | 1 + ShellyPy/base.py | 12 ++++++++---- ShellyPy/gen1.py | 47 +++++++++++++++++------------------------------ ShellyPy/gen2.py | 26 ++++++++++---------------- 4 files changed, 36 insertions(+), 50 deletions(-) diff --git a/.gitignore b/.gitignore index 7236703..069f1b4 100644 --- a/.gitignore +++ b/.gitignore @@ -4,3 +4,4 @@ dist *.egg-info __pycache__ *.pyc +.idea diff --git a/ShellyPy/base.py b/ShellyPy/base.py index 6700b92..9318b59 100644 --- a/ShellyPy/base.py +++ b/ShellyPy/base.py @@ -73,7 +73,7 @@ def update(self) -> None: ... @abstractmethod - def post(self, page: str, values = None): + def post(self, page: str, values: Optional[dict[str, Any]] = None): ... @abstractmethod @@ -93,13 +93,17 @@ def relay(self, index, *args, **kwargs): ... @abstractmethod - def roller(self, index, *args, **kwargs): + def roller(self, index: int, go: Optional[str] = None, + roller_pos: Optional[int] = None, duration: Optional[int] = None): ... @abstractmethod - def light(self, index, *args, **kwargs): + def light(self, index: int, mode: Optional[str] = None, timer: Optional[int] = None, turn: Optional[bool] = None, + red: Optional[int] = None, green: Optional[int] = None, blue: Optional[int] = None, + white: Optional[int] = None, gain: Optional[int] = None, temp: Optional[int] = None, + brightness: Optional[int] = None): ... @abstractmethod - def emeter(self, index, *args, **kwargs): + def emeter(self, index: int): ... diff --git a/ShellyPy/gen1.py b/ShellyPy/gen1.py index 76c0aa1..3b62da7 100644 --- a/ShellyPy/gen1.py +++ b/ShellyPy/gen1.py @@ -2,6 +2,7 @@ from json.decoder import JSONDecodeError from requests import post +from requests import Response from requests.auth import HTTPBasicAuth from .error import BadLogin, NotFound, BadResponse @@ -60,15 +61,14 @@ def update(self) -> None: # Request meter information self.meters.append(self.meter(meter_index)) meter_index += 1 - except Exception: + except (BadLogin, NotFound, BadResponse): break - def post(self, page: str, values = None): + def post(self, page: str, values: Optional[dict[str, Any]] = None): """ @brief returns settings @param page page to be accessed. Use the Shelly HTTP API Reference to see what's possible - @return returns json response """ @@ -82,8 +82,7 @@ def post(self, page: str, values = None): credentials = HTTPBasicAuth(*self._credentials) - response = post(url, auth=credentials, - timeout=self._timeout) + response: Response = post(url, auth=credentials, timeout=self._timeout) if response.status_code == 401: raise BadLogin() @@ -108,13 +107,12 @@ def settings(self, subpage = None): @brief returns settings @param subpage page to be accessed. Use the Shelly HTTP API Reference to see what's possible - @return returns settings as a dict """ page = "settings" if subpage: - page += "/" + subpage + page += f"/{subpage}" return self.post(page) @@ -137,7 +135,7 @@ def relay(self, index, *args, **kwargs): @param timer a one-shot flip-back timer in seconds """ - values = {} + values: dict[str, Any] = {} turn = kwargs.get("turn", None) timer = kwargs.get("timer", None) @@ -153,7 +151,8 @@ def relay(self, index, *args, **kwargs): return self.post(f"relay/{index}", values) - def roller(self, index, *args, **kwargs): + def roller(self, index: int, go: Optional[str] = None, + roller_pos: Optional[int] = None, duration: Optional[int] = None): """ @brief Interacts with a roller at a given index @@ -164,13 +163,9 @@ def roller(self, index, *args, **kwargs): @param duration how long it will take to get to that position """ - go = kwargs.get("go", None) - roller_pos = kwargs.get("roller_pos", None) - duration = kwargs.get("duration", None) + values: dict[str, Any] = {} - values = {} - - if go: + if go is not None: values["go"] = go if roller_pos is not None: @@ -181,12 +176,15 @@ def roller(self, index, *args, **kwargs): return self.post(f"roller/{index}", values) - def light(self, index, *args, **kwargs): + def light(self, index: int, mode: Optional[str] = None, timer: Optional[int] = None, turn: Optional[bool] = None, + red: Optional[int] = None, green: Optional[int] = None, blue: Optional[int] = None, + white: Optional[int] = None, gain: Optional[int] = None, temp: Optional[int] = None, + brightness: Optional[int] = None): """ @brief Interacts with lights at a given index - @param mode Accepts "white" and "color" as possible modes @param index index of the light. When in doubt use 0 + @param mode Accepts "white" and "color" as possible modes @param timer a one-shot flip-back timer in seconds @param turn Will turn the lights on or off @param red Red brightness, 0..255, only works if mode="color" @@ -197,18 +195,8 @@ def light(self, index, *args, **kwargs): @param temp Color temperature in K, 3000..6500, only works if mode="white" @param brightness Brightness, 0..100, only works if mode="white" """ - mode = kwargs.get("mode", None) - timer = kwargs.get("timer", None) - turn = kwargs.get("turn", None) - red = kwargs.get("red", None) - green = kwargs.get("green", None) - blue = kwargs.get("blue", None) - white = kwargs.get("white", None) - gain = kwargs.get("gain", None) - temp = kwargs.get("temp", None) - brightness = kwargs.get("brightness", None) - values = {} + values: dict[str, Any] = {} if mode: values["mode"] = mode @@ -245,8 +233,7 @@ def light(self, index, *args, **kwargs): return self.post(f"light/{index}", values) - def emeter(self, index, *args, **kwargs): - + def emeter(self, index: int): return self.post(f"emeter/{index}") # backwards compatibility with old code diff --git a/ShellyPy/gen2.py b/ShellyPy/gen2.py index c7eab51..29f8a38 100644 --- a/ShellyPy/gen2.py +++ b/ShellyPy/gen2.py @@ -31,7 +31,7 @@ def update(self) -> None: self._name = status["device"].get("name", self._name) self._type = status["device"].get("mac", self._type) - def post(self, page: str, values = None): + def post(self, page: str, values: Optional[dict[str, Any]] = None): url = f"{self._proto}://{self._hostname}:{self._port}/rpc" # increment payload id globally @@ -39,11 +39,7 @@ def post(self, page: str, values = None): # but keep a local copy around so we face no race conditions payload_id = self.payload_id - payload = { - "jsonrpc": "2.0", - "id": payload_id, - "method": page, - } + payload: dict[str, Any] = {"jsonrpc": "2.0", "id": payload_id, "method": page} if values: payload["params"] = values @@ -54,9 +50,7 @@ def post(self, page: str, values = None): except IndexError: pass - response = post(url, auth=credentials, - json=payload, - timeout=self._timeout) + response: Response = post(url, auth=credentials, json=payload, timeout=self._timeout) if response.status_code == 401: raise BadLogin() @@ -117,11 +111,8 @@ def relay(self, index, *args, **kwargs): return self.post(method, values) - def roller(self, index, *args, **kwargs): - - go = kwargs.get("go", None) - roller_pos = kwargs.get("roller_pos", None) - duration = kwargs.get("duration", None) + def roller(self, index: int, go: Optional[str] = None, + roller_pos: Optional[int] = None, duration: Optional[int] = None): method: str = "" values: dict[str, Any] = { @@ -147,8 +138,11 @@ def roller(self, index, *args, **kwargs): return self.post(method, values) - def light(self, index, *args, **kwargs): + def light(self, index: int, mode: Optional[str] = None, timer: Optional[int] = None, turn: Optional[bool] = None, + red: Optional[int] = None, green: Optional[int] = None, blue: Optional[int] = None, + white: Optional[int] = None, gain: Optional[int] = None, temp: Optional[int] = None, + brightness: Optional[int] = None): raise NotImplementedError("Unavailable") - def emeter(self, index, *args, **kwargs): + def emeter(self, index: int): raise NotImplementedError("Unavailable") From 54dd70fee9a9dfd211db9ab85700e8fdbb9ba1a3 Mon Sep 17 00:00:00 2001 From: Sebastian Wolf Date: Mon, 9 Sep 2024 20:12:17 +0200 Subject: [PATCH 08/18] Improve wrapper dunder methods --- ShellyPy/wrapper.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/ShellyPy/wrapper.py b/ShellyPy/wrapper.py index 87384e5..2135390 100644 --- a/ShellyPy/wrapper.py +++ b/ShellyPy/wrapper.py @@ -43,7 +43,7 @@ def __detect(ip: str, port: int, proto: str = 'http') -> Type[ShellyBase]: raise BadResponse("Bad JSON") gen: int = response_data.get("gen", 1) - + if gen == 1: return ShellyGen1 elif gen == 2: @@ -52,10 +52,10 @@ def __detect(ip: str, port: int, proto: str = 'http') -> Type[ShellyBase]: raise ValueError(f"Generation {gen} not supported") def __repr__(self) -> str: - return self.__getattr__("__repr__")() + return self._instance.__repr__() def __str__(self) -> str: - return self.__getattr__("__str__")() + return self._instance.__str__() def __getattr__(self, name: str) -> Any: - return getattr(self._instance, name) + return self._instance.__getattribute__(name) From c7cb1c2b3bc265d9eff68e8fbb773c04fda8c91b Mon Sep 17 00:00:00 2001 From: Sebastian Wolf Date: Mon, 9 Sep 2024 20:50:48 +0200 Subject: [PATCH 09/18] Fix device name not set returning null (None), preventing __str__ method to throw exception --- ShellyPy/gen2.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/ShellyPy/gen2.py b/ShellyPy/gen2.py index 29f8a38..ca84e47 100644 --- a/ShellyPy/gen2.py +++ b/ShellyPy/gen2.py @@ -28,7 +28,8 @@ def __init__(self, ip: str, port: int = 80, timeout: int = 5, def update(self) -> None: status = self.settings() - self._name = status["device"].get("name", self._name) + name: Optional[str] = status["device"].get("name", self._name) + self._name = name if name is not None else "Device name not yet set!" self._type = status["device"].get("mac", self._type) def post(self, page: str, values: Optional[dict[str, Any]] = None): From 4ced1852a00cff1b8eb349f5a2a91c6ccd80fe19 Mon Sep 17 00:00:00 2001 From: Sebastian Wolf Date: Mon, 9 Sep 2024 21:03:48 +0200 Subject: [PATCH 10/18] Import Any type and Response for working type hints --- ShellyPy/base.py | 2 +- ShellyPy/gen1.py | 4 ++-- ShellyPy/gen2.py | 7 ++++--- 3 files changed, 7 insertions(+), 6 deletions(-) diff --git a/ShellyPy/base.py b/ShellyPy/base.py index 9318b59..211b5c5 100644 --- a/ShellyPy/base.py +++ b/ShellyPy/base.py @@ -1,4 +1,4 @@ -from typing import Optional +from typing import Any, Optional from abc import abstractmethod diff --git a/ShellyPy/gen1.py b/ShellyPy/gen1.py index 3b62da7..3599041 100644 --- a/ShellyPy/gen1.py +++ b/ShellyPy/gen1.py @@ -1,4 +1,4 @@ -from typing import Optional +from typing import Any, Optional from json.decoder import JSONDecodeError from requests import post @@ -40,8 +40,8 @@ def update(self) -> None: """ status = self.settings() - self._type = status['device'].get("type", self._type) self._name = status['device'].get("hostname", self._name) + self._type = status['device'].get("type", self._type) # Settings are already fetched to get device information might as well put the list of things the device has somewhere self.relays = status.get("relays", []) diff --git a/ShellyPy/gen2.py b/ShellyPy/gen2.py index ca84e47..565ef53 100644 --- a/ShellyPy/gen2.py +++ b/ShellyPy/gen2.py @@ -1,7 +1,8 @@ -from typing import Optional +from typing import Any, Optional from json.decoder import JSONDecodeError from requests import post +from requests import Response from requests.auth import HTTPDigestAuth from .error import BadLogin, NotFound, BadResponse @@ -29,8 +30,8 @@ def update(self) -> None: status = self.settings() name: Optional[str] = status["device"].get("name", self._name) - self._name = name if name is not None else "Device name not yet set!" - self._type = status["device"].get("mac", self._type) + self._name: str = name if name is not None else "Device name not yet set!" + self._type: str = status["device"].get("mac", self._type) def post(self, page: str, values: Optional[dict[str, Any]] = None): url = f"{self._proto}://{self._hostname}:{self._port}/rpc" From e707dbb4f9efe0cb275620c2ad9b2891ba1070ae Mon Sep 17 00:00:00 2001 From: Sebastian Wolf Date: Mon, 9 Sep 2024 21:12:34 +0200 Subject: [PATCH 11/18] Set return types for all methods --- ShellyPy/base.py | 16 ++++++++-------- ShellyPy/gen1.py | 16 ++++++++-------- ShellyPy/gen2.py | 16 ++++++++-------- 3 files changed, 24 insertions(+), 24 deletions(-) diff --git a/ShellyPy/base.py b/ShellyPy/base.py index 211b5c5..5c4d6fa 100644 --- a/ShellyPy/base.py +++ b/ShellyPy/base.py @@ -73,37 +73,37 @@ def update(self) -> None: ... @abstractmethod - def post(self, page: str, values: Optional[dict[str, Any]] = None): + def post(self, page: str, values: Optional[dict[str, Any]] = None) -> dict[str, Any]: ... @abstractmethod - def status(self): + def status(self) -> dict[str, Any]: ... @abstractmethod - def settings(self, subpage = None): + def settings(self, subpage = None) -> dict[str, Any]: ... @abstractmethod - def meter(self, index: int): + def meter(self, index: int) -> dict[str, Any]: ... @abstractmethod - def relay(self, index, *args, **kwargs): + def relay(self, index, *args, **kwargs) -> dict[str, Any]: ... @abstractmethod def roller(self, index: int, go: Optional[str] = None, - roller_pos: Optional[int] = None, duration: Optional[int] = None): + roller_pos: Optional[int] = None, duration: Optional[int] = None) -> dict[str, Any]: ... @abstractmethod def light(self, index: int, mode: Optional[str] = None, timer: Optional[int] = None, turn: Optional[bool] = None, red: Optional[int] = None, green: Optional[int] = None, blue: Optional[int] = None, white: Optional[int] = None, gain: Optional[int] = None, temp: Optional[int] = None, - brightness: Optional[int] = None): + brightness: Optional[int] = None) -> dict[str, Any]: ... @abstractmethod - def emeter(self, index: int): + def emeter(self, index: int) -> dict[str, Any]: ... diff --git a/ShellyPy/gen1.py b/ShellyPy/gen1.py index 3599041..aa079e8 100644 --- a/ShellyPy/gen1.py +++ b/ShellyPy/gen1.py @@ -64,7 +64,7 @@ def update(self) -> None: except (BadLogin, NotFound, BadResponse): break - def post(self, page: str, values: Optional[dict[str, Any]] = None): + def post(self, page: str, values: Optional[dict[str, Any]] = None) -> dict[str, Any]: """ @brief returns settings @@ -94,7 +94,7 @@ def post(self, page: str, values: Optional[dict[str, Any]] = None): except JSONDecodeError: raise BadResponse("Bad JSON") - def status(self): + def status(self) -> dict[str, Any]: """ @brief returns status response @@ -102,7 +102,7 @@ def status(self): """ return self.post("status") - def settings(self, subpage = None): + def settings(self, subpage = None) -> dict[str, Any]: """ @brief returns settings @@ -116,7 +116,7 @@ def settings(self, subpage = None): return self.post(page) - def meter(self, index: int): + def meter(self, index: int) -> dict[str, Any]: """ @brief Get meter information from a relay at the given index @@ -126,7 +126,7 @@ def meter(self, index: int): return self.post(f"meter/{index}") - def relay(self, index, *args, **kwargs): + def relay(self, index, *args, **kwargs) -> dict[str, Any]: """ @brief Interacts with a relay at the given index @@ -152,7 +152,7 @@ def relay(self, index, *args, **kwargs): return self.post(f"relay/{index}", values) def roller(self, index: int, go: Optional[str] = None, - roller_pos: Optional[int] = None, duration: Optional[int] = None): + roller_pos: Optional[int] = None, duration: Optional[int] = None) -> dict[str, Any]: """ @brief Interacts with a roller at a given index @@ -179,7 +179,7 @@ def roller(self, index: int, go: Optional[str] = None, def light(self, index: int, mode: Optional[str] = None, timer: Optional[int] = None, turn: Optional[bool] = None, red: Optional[int] = None, green: Optional[int] = None, blue: Optional[int] = None, white: Optional[int] = None, gain: Optional[int] = None, temp: Optional[int] = None, - brightness: Optional[int] = None): + brightness: Optional[int] = None) -> dict[str, Any]: """ @brief Interacts with lights at a given index @@ -233,7 +233,7 @@ def light(self, index: int, mode: Optional[str] = None, timer: Optional[int] = N return self.post(f"light/{index}", values) - def emeter(self, index: int): + def emeter(self, index: int) -> dict[str, Any]: return self.post(f"emeter/{index}") # backwards compatibility with old code diff --git a/ShellyPy/gen2.py b/ShellyPy/gen2.py index 565ef53..a5f0550 100644 --- a/ShellyPy/gen2.py +++ b/ShellyPy/gen2.py @@ -33,7 +33,7 @@ def update(self) -> None: self._name: str = name if name is not None else "Device name not yet set!" self._type: str = status["device"].get("mac", self._type) - def post(self, page: str, values: Optional[dict[str, Any]] = None): + def post(self, page: str, values: Optional[dict[str, Any]] = None) -> dict[str, Any]: url = f"{self._proto}://{self._hostname}:{self._port}/rpc" # increment payload id globally @@ -80,16 +80,16 @@ def post(self, page: str, values: Optional[dict[str, Any]] = None): return response_data.get("result", {}) - def status(self): + def status(self) -> dict[str, Any]: return self.post("Sys.GetStatus") - def settings(self, subpage = None): + def settings(self, subpage = None) -> dict[str, Any]: return self.post("Sys.GetConfig") - def meter(self, index: int): + def meter(self, index: int) -> dict[str, Any]: raise NotImplementedError("Unavailable") - def relay(self, index, *args, **kwargs): + def relay(self, index, *args, **kwargs) -> dict[str, Any]: values = { "id": index @@ -114,7 +114,7 @@ def relay(self, index, *args, **kwargs): return self.post(method, values) def roller(self, index: int, go: Optional[str] = None, - roller_pos: Optional[int] = None, duration: Optional[int] = None): + roller_pos: Optional[int] = None, duration: Optional[int] = None) -> dict[str, Any]: method: str = "" values: dict[str, Any] = { @@ -143,8 +143,8 @@ def roller(self, index: int, go: Optional[str] = None, def light(self, index: int, mode: Optional[str] = None, timer: Optional[int] = None, turn: Optional[bool] = None, red: Optional[int] = None, green: Optional[int] = None, blue: Optional[int] = None, white: Optional[int] = None, gain: Optional[int] = None, temp: Optional[int] = None, - brightness: Optional[int] = None): + brightness: Optional[int] = None) -> dict[str, Any]: raise NotImplementedError("Unavailable") - def emeter(self, index: int): + def emeter(self, index: int) -> dict[str, Any]: raise NotImplementedError("Unavailable") From e3c99d8113acbe2c25f3810e9e301ad5824da6a9 Mon Sep 17 00:00:00 2001 From: Sebastian Wolf Date: Mon, 9 Sep 2024 21:30:36 +0200 Subject: [PATCH 12/18] Complete and fix existing docstrings --- ShellyPy/gen1.py | 31 ++++++++++++++++--------------- 1 file changed, 16 insertions(+), 15 deletions(-) diff --git a/ShellyPy/gen1.py b/ShellyPy/gen1.py index aa079e8..158192f 100644 --- a/ShellyPy/gen1.py +++ b/ShellyPy/gen1.py @@ -68,7 +68,8 @@ def post(self, page: str, values: Optional[dict[str, Any]] = None) -> dict[str, """ @brief returns settings - @param page page to be accessed. Use the Shelly HTTP API Reference to see what's possible + @param page the page to be accessed. Use the Shelly HTTP API Reference to see what's possible + @param values dict of values to send in the body of the request @return returns json response """ @@ -120,7 +121,7 @@ def meter(self, index: int) -> dict[str, Any]: """ @brief Get meter information from a relay at the given index - @param index index of the relay + @param index the index of the relay @return returns attributes of meter: power, overpower, is_valid, timestamp, counters, total """ @@ -130,7 +131,7 @@ def relay(self, index, *args, **kwargs) -> dict[str, Any]: """ @brief Interacts with a relay at the given index - @param index index of the relay + @param index the index of the relay @param turn Will turn the relay on or off @param timer a one-shot flip-back timer in seconds """ @@ -156,8 +157,8 @@ def roller(self, index: int, go: Optional[str] = None, """ @brief Interacts with a roller at a given index - @param self The object - @param index index of the roller. When in doubt use 0 + @param self the object + @param index the index of the roller. When in doubt use 0 @param go way of the roller to go. Accepted are "open", "close", "stop", "to_pos" @param roller_pos the wanted position in percent @param duration how long it will take to get to that position @@ -183,17 +184,17 @@ def light(self, index: int, mode: Optional[str] = None, timer: Optional[int] = N """ @brief Interacts with lights at a given index - @param index index of the light. When in doubt use 0 - @param mode Accepts "white" and "color" as possible modes + @param index the index of the light. When in doubt use 0 + @param mode accepts "white" and "color" as possible modes @param timer a one-shot flip-back timer in seconds - @param turn Will turn the lights on or off - @param red Red brightness, 0..255, only works if mode="color" - @param green Green brightness, 0..255, only works if mode="color" - @param blue Blue brightness, 0..255, only works if mode="color" - @param white White brightness, 0..255, only works if mode="color" - @param gain Gain for all channels, 0...100, only works if mode="color" - @param temp Color temperature in K, 3000..6500, only works if mode="white" - @param brightness Brightness, 0..100, only works if mode="white" + @param turn will turn the lights on or off + @param red brightness of red, 0..255, only works if mode="color" + @param green brightness of green, 0..255, only works if mode="color" + @param blue brightness of blue, 0..255, only works if mode="color" + @param white brightness of white, 0..255, only works if mode="color" + @param gain the gain for all channels, 0...100, only works if mode="color" + @param temp color temperature in K, 3000..6500, only works if mode="white" + @param brightness the brightness, 0..100, only works if mode="white" """ values: dict[str, Any] = {} From 2bf835055bc09777a3b31f500a0dfd6087e30651 Mon Sep 17 00:00:00 2001 From: Sebastian Wolf Date: Mon, 9 Sep 2024 21:49:04 +0200 Subject: [PATCH 13/18] Define proper signature for relay methods with proper type hints --- ShellyPy/base.py | 2 +- ShellyPy/gen1.py | 9 +++------ ShellyPy/gen2.py | 14 +++++--------- 3 files changed, 9 insertions(+), 16 deletions(-) diff --git a/ShellyPy/base.py b/ShellyPy/base.py index 5c4d6fa..7efdee4 100644 --- a/ShellyPy/base.py +++ b/ShellyPy/base.py @@ -89,7 +89,7 @@ def meter(self, index: int) -> dict[str, Any]: ... @abstractmethod - def relay(self, index, *args, **kwargs) -> dict[str, Any]: + def relay(self, index: int, timer: float = 0.0, turn: Optional[bool] = None) -> dict[str, Any]: ... @abstractmethod diff --git a/ShellyPy/gen1.py b/ShellyPy/gen1.py index 158192f..b4aa2e6 100644 --- a/ShellyPy/gen1.py +++ b/ShellyPy/gen1.py @@ -73,7 +73,7 @@ def post(self, page: str, values: Optional[dict[str, Any]] = None) -> dict[str, @return returns json response """ - url = f"{self._proto}://{self._hostname}:{self._port}/{page}?" + url: str = f"{self._proto}://{self._hostname}:{self._port}/{page}?" if values: url += "&".join([f"{key}={value}" for key, value in values.items()]) @@ -127,7 +127,7 @@ def meter(self, index: int) -> dict[str, Any]: return self.post(f"meter/{index}") - def relay(self, index, *args, **kwargs) -> dict[str, Any]: + def relay(self, index: int, timer: float = 0.0, turn: Optional[bool] = None) -> dict[str, Any]: """ @brief Interacts with a relay at the given index @@ -138,16 +138,13 @@ def relay(self, index, *args, **kwargs) -> dict[str, Any]: values: dict[str, Any] = {} - turn = kwargs.get("turn", None) - timer = kwargs.get("timer", None) - if turn is not None: if turn: values["turn"] = "on" else: values["turn"] = "off" - if timer: + if timer > 0.0: values["timer"] = timer return self.post(f"relay/{index}", values) diff --git a/ShellyPy/gen2.py b/ShellyPy/gen2.py index a5f0550..b6ca875 100644 --- a/ShellyPy/gen2.py +++ b/ShellyPy/gen2.py @@ -89,14 +89,12 @@ def settings(self, subpage = None) -> dict[str, Any]: def meter(self, index: int) -> dict[str, Any]: raise NotImplementedError("Unavailable") - def relay(self, index, *args, **kwargs) -> dict[str, Any]: + def relay(self, index: int, timer: float = 0.0, turn: Optional[bool] = None) -> dict[str, Any]: - values = { - "id": index - } + values: dict[str, Any] = {"id": index} - turn = kwargs.get("turn", None) - timer = kwargs.get("timer", None) + if timer > 0.0: + values["toggle_after"] = timer if turn is not None: method = "Switch.Set" @@ -106,8 +104,6 @@ def relay(self, index, *args, **kwargs) -> dict[str, Any]: else: values["on"] = False - if timer: - values["toggle_after"] = timer else: method = "Switch.GetStatus" @@ -121,7 +117,7 @@ def roller(self, index: int, go: Optional[str] = None, "id": index } - if go: + if go is not None: if go == "open": method = "Cover.Open" elif go == "close": From 156bda993b0d78386b7ec795449e6c59ac42e7b7 Mon Sep 17 00:00:00 2001 From: Sebastian Wolf Date: Mon, 9 Sep 2024 22:04:21 +0200 Subject: [PATCH 14/18] Add missing type hint for settings method. Add some other type hints --- ShellyPy/base.py | 2 +- ShellyPy/gen1.py | 8 ++++---- ShellyPy/gen2.py | 2 +- 3 files changed, 6 insertions(+), 6 deletions(-) diff --git a/ShellyPy/base.py b/ShellyPy/base.py index 7efdee4..6d10c12 100644 --- a/ShellyPy/base.py +++ b/ShellyPy/base.py @@ -81,7 +81,7 @@ def status(self) -> dict[str, Any]: ... @abstractmethod - def settings(self, subpage = None) -> dict[str, Any]: + def settings(self, subpage: Optional[str] = None) -> dict[str, Any]: ... @abstractmethod diff --git a/ShellyPy/gen1.py b/ShellyPy/gen1.py index b4aa2e6..a0f228a 100644 --- a/ShellyPy/gen1.py +++ b/ShellyPy/gen1.py @@ -38,10 +38,10 @@ def update(self) -> None: """ @brief update the Shelly attributes """ - status = self.settings() + status: dict[str, Any] = self.settings() - self._name = status['device'].get("hostname", self._name) - self._type = status['device'].get("type", self._type) + self._name: str = status['device'].get("hostname", self._name) + self._type: str = status['device'].get("type", self._type) # Settings are already fetched to get device information might as well put the list of things the device has somewhere self.relays = status.get("relays", []) @@ -103,7 +103,7 @@ def status(self) -> dict[str, Any]: """ return self.post("status") - def settings(self, subpage = None) -> dict[str, Any]: + def settings(self, subpage: Optional[str] = None) -> dict[str, Any]: """ @brief returns settings diff --git a/ShellyPy/gen2.py b/ShellyPy/gen2.py index b6ca875..42323b1 100644 --- a/ShellyPy/gen2.py +++ b/ShellyPy/gen2.py @@ -83,7 +83,7 @@ def post(self, page: str, values: Optional[dict[str, Any]] = None) -> dict[str, def status(self) -> dict[str, Any]: return self.post("Sys.GetStatus") - def settings(self, subpage = None) -> dict[str, Any]: + def settings(self, subpage: Optional[str] = None) -> dict[str, Any]: return self.post("Sys.GetConfig") def meter(self, index: int) -> dict[str, Any]: From f95f701676f04c3fe98b8b600aaa66b765bbe4f9 Mon Sep 17 00:00:00 2001 From: Sebastian Wolf Date: Mon, 9 Sep 2024 22:15:52 +0200 Subject: [PATCH 15/18] Add all relevant type hints in gen2 implementation --- ShellyPy/gen2.py | 17 +++++++---------- 1 file changed, 7 insertions(+), 10 deletions(-) diff --git a/ShellyPy/gen2.py b/ShellyPy/gen2.py index 42323b1..2512ba4 100644 --- a/ShellyPy/gen2.py +++ b/ShellyPy/gen2.py @@ -34,19 +34,19 @@ def update(self) -> None: self._type: str = status["device"].get("mac", self._type) def post(self, page: str, values: Optional[dict[str, Any]] = None) -> dict[str, Any]: - url = f"{self._proto}://{self._hostname}:{self._port}/rpc" + url: str = f"{self._proto}://{self._hostname}:{self._port}/rpc" # increment payload id globally self.payload_id += 1 # but keep a local copy around so we face no race conditions - payload_id = self.payload_id + payload_id: int = self.payload_id payload: dict[str, Any] = {"jsonrpc": "2.0", "id": payload_id, "method": page} if values: payload["params"] = values - credentials = None + credentials: Optional[HTTPDigestAuth] = None try: credentials = HTTPDigestAuth('admin', self._credentials[1]) except IndexError: @@ -60,13 +60,13 @@ def post(self, page: str, values: Optional[dict[str, Any]] = None) -> dict[str, raise NotFound("Not Found") try: - response_data = response.json() + response_data: dict[str, Any] = response.json() except JSONDecodeError: raise BadResponse("Bad JSON") if "error" in response_data: - error_code = response_data["error"].get("code", None) - error_message = response_data["error"].get("message", "") + error_code: Optional[int] = response_data["error"].get("code", None) + error_message: str = response_data["error"].get("message", "") if error_code == 401: raise BadLogin(error_message) @@ -92,7 +92,7 @@ def meter(self, index: int) -> dict[str, Any]: def relay(self, index: int, timer: float = 0.0, turn: Optional[bool] = None) -> dict[str, Any]: values: dict[str, Any] = {"id": index} - + method: str = "Switch.GetStatus" if timer > 0.0: values["toggle_after"] = timer @@ -104,9 +104,6 @@ def relay(self, index: int, timer: float = 0.0, turn: Optional[bool] = None) -> else: values["on"] = False - else: - method = "Switch.GetStatus" - return self.post(method, values) def roller(self, index: int, go: Optional[str] = None, From 93f63b6035e4c0df0c21903de6a2992e81d352f7 Mon Sep 17 00:00:00 2001 From: Sebastian Wolf Date: Mon, 9 Sep 2024 22:35:13 +0200 Subject: [PATCH 16/18] Add types for properties of gen1 --- ShellyPy/gen1.py | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/ShellyPy/gen1.py b/ShellyPy/gen1.py index a0f228a..3c96d42 100644 --- a/ShellyPy/gen1.py +++ b/ShellyPy/gen1.py @@ -25,14 +25,14 @@ def __init__(self, ip: str, port: int = 80, timeout: int = 5, super().__init__(ip=ip, port=port, timeout=timeout, login=login, debug=debug, init=init) self._generation: int = 1 - self.relays: list = [] - self.rollers: list = [] - self.lights: list = [] + self.relays: list[dict[str, Any]] = [] + self.rollers: list[dict[str, Any]] = [] + self.lights: list[dict[str, Any]] = [] - self.irs = None + self.ir_sense: str = "Unknown" - self.emeters: list = [] - self.meters: list[dict] = [] + self.emeters: list[dict[str, Any]] = [] + self.meters: list[dict[str, Any]] = [] def update(self) -> None: """ @@ -49,7 +49,7 @@ def update(self) -> None: # RGBW reuses the same lights array self.lights = status.get("lights", []) - self.irs = status.get("light_sensor", None) + self.ir_sense = status.get("light_sensor", "Unknown") self.emeters = status.get("emeter", []) From 1642db26e73ddbc6b25d2de6e8eb22e2a2f2f269 Mon Sep 17 00:00:00 2001 From: Sebastian Wolf Date: Mon, 9 Sep 2024 22:35:13 +0200 Subject: [PATCH 17/18] Add types for properties of gen1 --- ShellyPy/gen1.py | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/ShellyPy/gen1.py b/ShellyPy/gen1.py index a0f228a..c0ae6f8 100644 --- a/ShellyPy/gen1.py +++ b/ShellyPy/gen1.py @@ -25,14 +25,14 @@ def __init__(self, ip: str, port: int = 80, timeout: int = 5, super().__init__(ip=ip, port=port, timeout=timeout, login=login, debug=debug, init=init) self._generation: int = 1 - self.relays: list = [] - self.rollers: list = [] - self.lights: list = [] + self.relays: list[dict[str, Any]] = [] + self.rollers: list[dict[str, Any]] = [] + self.lights: list[dict[str, Any]] = [] - self.irs = None + self.ir_sense: str = "Unknown" - self.emeters: list = [] - self.meters: list[dict] = [] + self.emeters: list[dict[str, Any]] = [] + self.meters: list[dict[str, Any]] = [] def update(self) -> None: """ @@ -49,7 +49,7 @@ def update(self) -> None: # RGBW reuses the same lights array self.lights = status.get("lights", []) - self.irs = status.get("light_sensor", None) + self.ir_sense = status.get("light_sensor", "Unknown") self.emeters = status.get("emeter", []) @@ -111,7 +111,7 @@ def settings(self, subpage: Optional[str] = None) -> dict[str, Any]: @return returns settings as a dict """ - page = "settings" + page: str = "settings" if subpage: page += f"/{subpage}" From 5086face43189b6f0d24a6c1f997a10af3101bec Mon Sep 17 00:00:00 2001 From: Sebastian Wolf Date: Fri, 13 Sep 2024 20:40:05 +0200 Subject: [PATCH 18/18] Make wrapper derive from base class to implement all its methods and have propper doc and autofill in ides like vscode and pycharm. Put all docstrings to base class. Make base class private (_ShellyBase). Add some missing initial descriptions (should be improved later) --- ShellyPy/base.py | 84 +++++++++++++++++++++++++++++++++++++++++++-- ShellyPy/gen1.py | 75 ++++------------------------------------ ShellyPy/gen2.py | 17 ++++----- ShellyPy/wrapper.py | 56 ++++++++++++++++++++++-------- requirements.txt | 1 + 5 files changed, 136 insertions(+), 97 deletions(-) diff --git a/ShellyPy/base.py b/ShellyPy/base.py index 6d10c12..7874e42 100644 --- a/ShellyPy/base.py +++ b/ShellyPy/base.py @@ -2,13 +2,16 @@ from abc import abstractmethod -class ShellyBase: +class _ShellyBase: + """ + Base or parent class for Shelly device implementation + """ def __init__(self, ip: str, port: int = 80, timeout: int = 5, login: Optional[dict[str, str]] = None, debug: bool = False, init: bool = False) -> None: """ @param ip the target IP of the shelly device. Can be a string, list of strings or list of integers - @param port target port, may be useful for non Shelly devices that have the same HTTP Api + @param port target port, may be useful for non Shelly devices that have the same HTTP Api @param login dict of login credentials. Keys needed are "username" and "password" @param timeout specify the amount of time until requests are aborted. @param debug enable debug printing @@ -38,7 +41,11 @@ def __str__(self) -> str: @staticmethod def _clamp(val: int) -> int: - """clamp any number to 8 bit""" + """ + @brief clamp any number to 8 bit + @param val any number to clamp + @return clamped number + """ if val > 255: val = 255 elif val < 0: @@ -50,6 +57,8 @@ def _clamp(val: int) -> int: def _clamp_percentage(val: int) -> int: """ clamp given percentage to a range from 0 to 100 + @param val percentage to clamp + @return clamped percentage """ if val > 100: val = 100 @@ -61,6 +70,8 @@ def _clamp_percentage(val: int) -> int: def _clamp_kelvin(val: int) -> int: """ clamp given kelvin values for a range from 3000..6500 + @param val kelvin values to clamp + @return clamped kelvin value """ if val > 6500: val = 6500 @@ -70,31 +81,76 @@ def _clamp_kelvin(val: int) -> int: @abstractmethod def update(self) -> None: + """ + @brief update the Shelly attributes + """ ... @abstractmethod def post(self, page: str, values: Optional[dict[str, Any]] = None) -> dict[str, Any]: + """ + @brief returns settings of shelly device + + @param page the page to be accessed. Use the Shelly HTTP API Reference to see what's possible + @param values dict of values to send in the body of the request + @return returns json response + """ ... @abstractmethod def status(self) -> dict[str, Any]: + """ + @brief returns status response of shelly device + + @return status dict + """ ... @abstractmethod def settings(self, subpage: Optional[str] = None) -> dict[str, Any]: + """ + @brief returns settings of shelly device or a specific subset + + @param subpage page to be accessed. Use the Shelly HTTP API Reference to see what's possible + @return returns settings as a dict + """ ... @abstractmethod def meter(self, index: int) -> dict[str, Any]: + """ + @brief Get meter information from a relay at the given index + + @param index the index of the relay + @return returns attributes of meter: power, overpower, is_valid, timestamp, counters, total + """ ... @abstractmethod def relay(self, index: int, timer: float = 0.0, turn: Optional[bool] = None) -> dict[str, Any]: + """ + @brief Interacts with a relay at the given index + + @param index the index of the relay + @param turn Will turn the relay on or off + @param timer a one-shot flip-back timer in seconds + @return returns attributes dict of relay: was_on + """ ... @abstractmethod def roller(self, index: int, go: Optional[str] = None, roller_pos: Optional[int] = None, duration: Optional[int] = None) -> dict[str, Any]: + """ + @brief Interacts with a roller at a given index + + @param self the object + @param index the index of the roller. When in doubt use 0 + @param go way of the roller to go. Accepted are "open", "close", "stop", "to_pos" + @param roller_pos the wanted position in percent + @param duration how long it will take to get to that position + @return returns attributes dict of roller + """ ... @abstractmethod @@ -102,8 +158,30 @@ def light(self, index: int, mode: Optional[str] = None, timer: Optional[int] = N red: Optional[int] = None, green: Optional[int] = None, blue: Optional[int] = None, white: Optional[int] = None, gain: Optional[int] = None, temp: Optional[int] = None, brightness: Optional[int] = None) -> dict[str, Any]: + """ + @brief Interacts with lights at a given index + + @param index the index of the light. When in doubt use 0 + @param mode accepts "white" and "color" as possible modes + @param timer a one-shot flip-back timer in seconds + @param turn will turn the lights on or off + @param red brightness of red, 0..255, only works if mode="color" + @param green brightness of green, 0..255, only works if mode="color" + @param blue brightness of blue, 0..255, only works if mode="color" + @param white brightness of white, 0..255, only works if mode="color" + @param gain the gain for all channels, 0...100, only works if mode="color" + @param temp color temperature in K, 3000..6500, only works if mode="white" + @param brightness the brightness, 0..100, only works if mode="white" + @return returns json response as dict + """ ... @abstractmethod def emeter(self, index: int) -> dict[str, Any]: + """ + @brief Get emeter information + + @param index the index of the emeter. When in doubt use 0 + @return returns attributes dict of emeter + """ ... diff --git a/ShellyPy/gen1.py b/ShellyPy/gen1.py index c0ae6f8..fe5238e 100644 --- a/ShellyPy/gen1.py +++ b/ShellyPy/gen1.py @@ -6,21 +6,16 @@ from requests.auth import HTTPBasicAuth from .error import BadLogin, NotFound, BadResponse -from .base import ShellyBase +from .base import _ShellyBase -class ShellyGen1(ShellyBase): +class ShellyGen1(_ShellyBase): + """ + Implements a general Class for interaction with Shelly devices of generation 1 + """ def __init__(self, ip: str, port: int = 80, timeout: int = 5, login: Optional[dict[str, str]] = None, debug: bool = False, init: bool = False) -> None: - """ - @param ip the target IP of the shelly device. Can be a string, list of strings or list of integers - @param port target port, may be useful for non Shelly devices that have the same HTTP Api - @param login dict of login credentials. Keys needed are "username" and "password" - @param timeout specify the amount of time until requests are aborted. - @param debug enable debug printing - @param init calls the update method on init - """ super().__init__(ip=ip, port=port, timeout=timeout, login=login, debug=debug, init=init) self._generation: int = 1 @@ -35,9 +30,7 @@ def __init__(self, ip: str, port: int = 80, timeout: int = 5, self.meters: list[dict[str, Any]] = [] def update(self) -> None: - """ - @brief update the Shelly attributes - """ + status: dict[str, Any] = self.settings() self._name: str = status['device'].get("hostname", self._name) @@ -65,13 +58,6 @@ def update(self) -> None: break def post(self, page: str, values: Optional[dict[str, Any]] = None) -> dict[str, Any]: - """ - @brief returns settings - - @param page the page to be accessed. Use the Shelly HTTP API Reference to see what's possible - @param values dict of values to send in the body of the request - @return returns json response - """ url: str = f"{self._proto}://{self._hostname}:{self._port}/{page}?" @@ -96,20 +82,9 @@ def post(self, page: str, values: Optional[dict[str, Any]] = None) -> dict[str, raise BadResponse("Bad JSON") def status(self) -> dict[str, Any]: - """ - @brief returns status response - - @return status dict - """ return self.post("status") def settings(self, subpage: Optional[str] = None) -> dict[str, Any]: - """ - @brief returns settings - - @param subpage page to be accessed. Use the Shelly HTTP API Reference to see what's possible - @return returns settings as a dict - """ page: str = "settings" if subpage: @@ -118,23 +93,9 @@ def settings(self, subpage: Optional[str] = None) -> dict[str, Any]: return self.post(page) def meter(self, index: int) -> dict[str, Any]: - """ - @brief Get meter information from a relay at the given index - - @param index the index of the relay - @return returns attributes of meter: power, overpower, is_valid, timestamp, counters, total - """ - return self.post(f"meter/{index}") def relay(self, index: int, timer: float = 0.0, turn: Optional[bool] = None) -> dict[str, Any]: - """ - @brief Interacts with a relay at the given index - - @param index the index of the relay - @param turn Will turn the relay on or off - @param timer a one-shot flip-back timer in seconds - """ values: dict[str, Any] = {} @@ -151,15 +112,6 @@ def relay(self, index: int, timer: float = 0.0, turn: Optional[bool] = None) -> def roller(self, index: int, go: Optional[str] = None, roller_pos: Optional[int] = None, duration: Optional[int] = None) -> dict[str, Any]: - """ - @brief Interacts with a roller at a given index - - @param self the object - @param index the index of the roller. When in doubt use 0 - @param go way of the roller to go. Accepted are "open", "close", "stop", "to_pos" - @param roller_pos the wanted position in percent - @param duration how long it will take to get to that position - """ values: dict[str, Any] = {} @@ -178,21 +130,6 @@ def light(self, index: int, mode: Optional[str] = None, timer: Optional[int] = N red: Optional[int] = None, green: Optional[int] = None, blue: Optional[int] = None, white: Optional[int] = None, gain: Optional[int] = None, temp: Optional[int] = None, brightness: Optional[int] = None) -> dict[str, Any]: - """ - @brief Interacts with lights at a given index - - @param index the index of the light. When in doubt use 0 - @param mode accepts "white" and "color" as possible modes - @param timer a one-shot flip-back timer in seconds - @param turn will turn the lights on or off - @param red brightness of red, 0..255, only works if mode="color" - @param green brightness of green, 0..255, only works if mode="color" - @param blue brightness of blue, 0..255, only works if mode="color" - @param white brightness of white, 0..255, only works if mode="color" - @param gain the gain for all channels, 0...100, only works if mode="color" - @param temp color temperature in K, 3000..6500, only works if mode="white" - @param brightness the brightness, 0..100, only works if mode="white" - """ values: dict[str, Any] = {} diff --git a/ShellyPy/gen2.py b/ShellyPy/gen2.py index 2512ba4..c13a1dd 100644 --- a/ShellyPy/gen2.py +++ b/ShellyPy/gen2.py @@ -6,27 +6,23 @@ from requests.auth import HTTPDigestAuth from .error import BadLogin, NotFound, BadResponse -from .base import ShellyBase +from .base import _ShellyBase -class ShellyGen2(ShellyBase): +class ShellyGen2(_ShellyBase): + """ + Implements a general Class for interaction with Shelly devices of generation 2 + """ def __init__(self, ip: str, port: int = 80, timeout: int = 5, login: Optional[dict[str, str]] = None, debug: bool = False, init: bool = False) -> None: - """ - @param ip the target IP of the shelly device. Can be a string, list of strings or list of integers - @param port target port, may be useful for non Shelly devices that have the same HTTP Api - @param login dict of login credentials. Keys needed are "username" and "password" - @param timeout specify the amount of time until requests are aborted. - @param debug enable debug printing - @param init calls the update method on init - """ super().__init__(ip=ip, port=port, timeout=timeout, login=login, debug=debug, init=init) self.payload_id: int = 1 self._generation: int = 2 def update(self) -> None: + status = self.settings() name: Optional[str] = status["device"].get("name", self._name) @@ -34,6 +30,7 @@ def update(self) -> None: self._type: str = status["device"].get("mac", self._type) def post(self, page: str, values: Optional[dict[str, Any]] = None) -> dict[str, Any]: + url: str = f"{self._proto}://{self._hostname}:{self._port}/rpc" # increment payload id globally diff --git a/ShellyPy/wrapper.py b/ShellyPy/wrapper.py index 2135390..2ea5ad1 100644 --- a/ShellyPy/wrapper.py +++ b/ShellyPy/wrapper.py @@ -4,30 +4,25 @@ from requests import get from requests import Response -from .base import ShellyBase +from .base import _ShellyBase from .error import BadLogin, NotFound, BadResponse from .gen1 import ShellyGen1 from .gen2 import ShellyGen2 -class Shelly: - +class Shelly(_ShellyBase): + """ + General wrapper class for shelly devices of all generations automatically using methods of proper one generation + """ def __init__(self, ip: str, port: int = 80, timeout: int = 5, login: Optional[dict[str, str]] = None, debug: bool = False, init: bool = False) -> None: - """ - @param ip the target IP of the shelly device. Can be a string, list of strings or list of integers - @param port target port, may be useful for non Shelly devices that have the same HTTP Api - @param login dict of login credentials. Keys needed are "username" and "password" - @param timeout specify the amount of time until requests are aborted. - @param debug enable debug printing - @param init calls the update method on init - """ - - self._instance: ShellyBase = self.__detect(ip, port)(ip=ip, port=port, timeout=timeout, - login=login, debug=debug, init=init) + + super().__init__(ip=ip, port=port, timeout=timeout, login=login, debug=debug, init=init) + self._instance: _ShellyBase = self.__detect(ip, port)(ip=ip, port=port, timeout=timeout, + login=login, debug=debug, init=init) @staticmethod - def __detect(ip: str, port: int, proto: str = 'http') -> Type[ShellyBase]: + def __detect(ip: str, port: int, proto: str = 'http') -> Type[_ShellyBase]: url: str = f"{proto}://{ip}:{port}/shelly" response: Response = get(url, timeout=5) @@ -59,3 +54,34 @@ def __str__(self) -> str: def __getattr__(self, name: str) -> Any: return self._instance.__getattribute__(name) + + def update(self) -> None: + self._instance.update() + + def post(self, page: str, values: Optional[dict[str, Any]] = None) -> dict[str, Any]: + return self._instance.post(page, values) + + def status(self) -> dict[str, Any]: + return self._instance.status() + + def settings(self, subpage: Optional[str] = None) -> dict[str, Any]: + return self._instance.settings(subpage) + + def meter(self, index: int) -> dict[str, Any]: + return self._instance.meter(index) + + def relay(self, index: int, timer: float = 0.0, turn: Optional[bool] = None) -> dict[str, Any]: + return self._instance.relay(index, timer, turn) + + def roller(self, index: int, go: Optional[str] = None, + roller_pos: Optional[int] = None, duration: Optional[int] = None) -> dict[str, Any]: + return self._instance.roller(index, go, roller_pos, duration) + + def light(self, index: int, mode: Optional[str] = None, timer: Optional[int] = None, turn: Optional[bool] = None, + red: Optional[int] = None, green: Optional[int] = None, blue: Optional[int] = None, + white: Optional[int] = None, gain: Optional[int] = None, temp: Optional[int] = None, + brightness: Optional[int] = None) -> dict[str, Any]: + return self._instance.light(index, mode, timer, turn, red, green, blue, white, gain, temp, brightness) + + def emeter(self, index: int) -> dict[str, Any]: + return self._instance.emeter(index) diff --git a/requirements.txt b/requirements.txt index 8dee024..4c738f3 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,3 @@ requests types-requests +