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 6c34be0..7874e42 100644 --- a/ShellyPy/base.py +++ b/ShellyPy/base.py @@ -1,56 +1,51 @@ -from sys import version_info +from typing import Any, Optional +from abc import abstractmethod -if version_info.major == 3: - from json.decoder import JSONDecodeError -else: - JSONDecodeError = ValueError -class ShellyBase: +class _ShellyBase: + """ + Base or parent class for Shelly device implementation + """ - def __init__(self, ip, port = "80", *args, **kwargs): + 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 @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.__PROTOCOL__ = "http" - - login = kwargs.get("login", {}) - - # hostname would be more fitting, - # but backwards compatibility - self.__ip__ = 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): - 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): - """clamp any number to 8 bit""" + def _clamp(val: int) -> int: + """ + @brief clamp any number to 8 bit + @param val any number to clamp + @return clamped number + """ if val > 255: val = 255 elif val < 0: @@ -59,9 +54,11 @@ 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 + @param val percentage to clamp + @return clamped percentage """ if val > 100: val = 100 @@ -70,9 +67,11 @@ 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 + @param val kelvin values to clamp + @return clamped kelvin value """ if val > 6500: val = 6500 @@ -80,29 +79,109 @@ def __clamp_kalvin__(val): val = 3000 return val - def update(self): - raise NotImplementedError("Base Class") + @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 + """ + ... - def post(self, page, values = None): - raise NotImplementedError("Base Class") + @abstractmethod + def settings(self, subpage: Optional[str] = None) -> dict[str, Any]: + """ + @brief returns settings of shelly device or a specific subset - def status(self): - raise NotImplementedError("Base Class") + @param subpage page to be accessed. Use the Shelly HTTP API Reference to see what's possible + @return returns settings as a dict + """ + ... - def settings(self, subpage = None): - raise NotImplementedError("Base Class") + @abstractmethod + def meter(self, index: int) -> dict[str, Any]: + """ + @brief Get meter information from a relay at the given index - def meter(self, index): - raise NotImplementedError("Base Class") - - def relay(self, index, *args, **kwargs): - raise NotImplementedError("Base Class") + @param index the index of the relay + @return returns attributes of meter: power, overpower, is_valid, timestamp, counters, total + """ + ... - def roller(self, index, *args, **kwargs): - raise NotImplementedError("Base Class") + @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 + """ + ... - def light(self, index, *args, **kwargs): - raise NotImplementedError("Base Class") - - def emeter(self, index, *args, **kwargs): - raise NotImplementedError("Base Class") + @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 + 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]: + """ + @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 dea4650..fe5238e 100644 --- a/ShellyPy/gen1.py +++ b/ShellyPy/gen1.py @@ -1,43 +1,40 @@ -from sys import version_info - -if version_info.major == 3: - from json.decoder import JSONDecodeError -else: - JSONDecodeError = ValueError - +from typing import Any, Optional +from json.decoder import JSONDecodeError from requests import post +from requests import Response from requests.auth import HTTPBasicAuth from .error import BadLogin, NotFound, BadResponse +from .base import _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: -from .base import ShellyBase + super().__init__(ip=ip, port=port, timeout=timeout, login=login, debug=debug, init=init) + self._generation: int = 1 -class ShellyGen1(ShellyBase): + self.relays: list[dict[str, Any]] = [] + self.rollers: list[dict[str, Any]] = [] + self.lights: list[dict[str, Any]] = [] - def __init__(self, ip, port = "80", *args, **kwargs): - """ - @param ip the target IP of the shelly device. Can be a string, list of strings or list of integers - @param port target port, may be useful for non Shelly devices that have the same HTTP Api - @param login dict of login credentials. Keys needed are "username" and "password" - @param timeout specify the amount of time until requests are aborted. - @param debug enable debug printing - @param init calls the update method on init - """ + self.ir_sense: str = "Unknown" - super().__init__(ip, port, *args, **kwargs) - self.__generation__ = 1 + self.emeters: list[dict[str, Any]] = [] + self.meters: list[dict[str, Any]] = [] - def update(self): - """ - @brief update the Shelly attributes + def update(self) -> None: - @param index index of the relay - """ - status = self.settings() + status: dict[str, Any] = self.settings() - self.__type__ = status['device'].get("type", self.__type__) - self.__name__ = status['device'].get("hostname", self.__name__) + 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", []) @@ -45,7 +42,7 @@ def update(self): # 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", []) @@ -55,35 +52,24 @@ def update(self): while True: try: # Request meter information - self.meter.append(self.meter(meter_index)) + self.meters.append(self.meter(meter_index)) meter_index += 1 - except: + except (BadLogin, NotFound, BadResponse): break - def post(self, page, values = None): - """ - @brief returns settings + def post(self, page: str, values: Optional[dict[str, Any]] = None) -> dict[str, Any]: - @param page page to be accesed. Use the Shelly HTTP API Reference to see whats possible - - @return returns json response - """ - - url = "{}://{}:{}/{}?".format(self.__PROTOCOL__, self.__ip__, self.__port__, page) + url: str = f"{self._proto}://{self._hostname}:{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__)) + 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__) + response: Response = post(url, auth=credentials, timeout=self._timeout) if response.status_code == 401: raise BadLogin() @@ -95,52 +81,23 @@ def post(self, page, values = None): except JSONDecodeError: raise BadResponse("Bad JSON") - def status(self): - """ - @brief returns status response - - @return status dict - """ + def status(self) -> dict[str, Any]: return self.post("status") - def settings(self, subpage = None): - """ - @brief returns settings - - @param page page to be accesed. Use the Shelly HTTP API Reference to see whats possible - - @return returns settings as a dict - """ + def settings(self, subpage: Optional[str] = None) -> dict[str, Any]: - page = "settings" + page: str = "settings" if subpage: - page += "/" + subpage + page += f"/{subpage}" return self.post(page) - def meter(self, index): - """ - @brief Get meter information from a relay at the given index + def meter(self, index: int) -> dict[str, Any]: + return self.post(f"meter/{index}") - @param index index of the relay - @return returns attributes of meter: power, overpower, is_valid, timestamp, counters, total - """ + def relay(self, index: int, timer: float = 0.0, turn: Optional[bool] = None) -> dict[str, Any]: - return self.post("meter/{}".format(index)) - - def relay(self, index, *args, **kwargs): - """ - @brief Interacts with a relay at the given index - - @param index index of the relay - @param turn Will turn the relay on or off - @param timer a one-shot flip-back timer in seconds - """ - - values = {} - - turn = kwargs.get("turn", None) - timer = kwargs.get("timer", None) + values: dict[str, Any] = {} if turn is not None: if turn: @@ -148,67 +105,33 @@ def relay(self, index, *args, **kwargs): else: values["turn"] = "off" - if timer: + if timer > 0.0: values["timer"] = timer - return self.post("relay/{}".format(index), values) - - def roller(self, index, *args, **kwargs): - """ - @brief Interacts with a roller at a given index + return self.post(f"relay/{index}", values) - @param self The object - @param index index of the roller. When in doubt use 0 - @param go way of the roller to go. Accepted are "open", "close", "stop", "to_pos" - @param roller_pos the wanted position in percent - @param duration how long it will take to get to that position - """ + def roller(self, index: int, go: Optional[str] = None, + roller_pos: Optional[int] = None, duration: Optional[int] = None) -> dict[str, Any]: - 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: - values["roller_pos"] = self.__clamp_percentage__(roller_pos) + values["roller_pos"] = self._clamp_percentage(roller_pos) if duration is not None: values["duration"] = duration - return self.post("roller/{}".format(index), values) - - def light(self, index, *args, **kwargs): - """ - @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 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" - """ - 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 = {} + return self.post(f"roller/{index}", values) + + 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]: + + values: dict[str, Any] = {} if mode: values["mode"] = mode @@ -223,31 +146,30 @@ 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) - - return self.post("light/{}".format(index), values) + values["brightness"] = self._clamp_percentage(brightness) - def emeter(self, index, *args, **kwargs): + return self.post(f"light/{index}", values) - return self.post("emeter/{}".format(index)) + def emeter(self, index: int) -> dict[str, Any]: + return self.post(f"emeter/{index}") # backwards compatibility with old code Shelly = ShellyGen1 diff --git a/ShellyPy/gen2.py b/ShellyPy/gen2.py index 199fb84..c13a1dd 100644 --- a/ShellyPy/gen2.py +++ b/ShellyPy/gen2.py @@ -1,65 +1,55 @@ -from sys import version_info - -if version_info.major == 3: - from json.decoder import JSONDecodeError -else: - JSONDecodeError = ValueError +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 +from .base import _ShellyBase + -from .base import ShellyBase +class ShellyGen2(_ShellyBase): + """ + Implements a general Class for interaction with Shelly devices of generation 2 + """ -class ShellyGen2(ShellyBase): + def __init__(self, ip: str, port: int = 80, timeout: int = 5, + login: Optional[dict[str, str]] = None, debug: bool = False, init: bool = False) -> None: - def __init__(self, ip, port = "80", *args, **kwargs): - """ - @param ip the target IP of the shelly device. Can be a string, list of strings or list of integers - @param port target port, may be useful for non Shelly devices that have the same HTTP Api - @param login dict of login credentials. Keys needed are "username" and "password" - @param timeout specify the amount of time until requests are aborted. - @param debug enable debug printing - @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 - self.payload_id = 1 - super().__init__(ip, port, *args, **kwargs) - self.__generation__ = 2 + def update(self) -> None: - def update(self): status = self.settings() - self.__name__ = status["device"].get("name", self.__name__) - self.__type__ = status["device"].get("mac", self.__type__) + name: Optional[str] = status["device"].get("name", self._name) + 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) -> dict[str, Any]: - def post(self, page, values = None): - url = "{}://{}:{}/rpc".format(self.__PROTOCOL__, self.__ip__, self.__port__) + 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 = { - "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 - credentials = None + credentials: Optional[HTTPDigestAuth] = None try: - credentials = auth=HTTPDigestAuth('admin', self.__credentials__[1]) + credentials = HTTPDigestAuth('admin', self._credentials[1]) 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() @@ -67,43 +57,41 @@ def post(self, page, values = None): 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) 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") 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: Optional[str] = None) -> dict[str, Any]: return self.post("Sys.GetConfig") - def meter(self, index): + def meter(self, index: int) -> dict[str, Any]: raise NotImplementedError("Unavailable") - def relay(self, index, *args, **kwargs): + def relay(self, index: int, timer: float = 0.0, turn: Optional[bool] = None) -> dict[str, Any]: - values = { - "id": index - } - - turn = kwargs.get("turn", None) - timer = kwargs.get("timer", None) + values: dict[str, Any] = {"id": index} + method: str = "Switch.GetStatus" + if timer > 0.0: + values["toggle_after"] = timer if turn is not None: method = "Switch.Set" @@ -113,24 +101,17 @@ def relay(self, index, *args, **kwargs): else: values["on"] = False - if timer: - values["toggle_after"] = timer - else: - method = "Switch.GetStatus" - 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) -> dict[str, Any]: - values = { + method: str = "" + values: dict[str, Any] = { "id": index } - if go: + if go is not None: if go == "open": method = "Cover.Open" elif go == "close": @@ -142,15 +123,18 @@ 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 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) -> dict[str, Any]: + raise NotImplementedError("Unavailable") + + def emeter(self, index: int) -> dict[str, Any]: raise NotImplementedError("Unavailable") - - def emeter(self, index, *args, **kwargs): - raise NotImplementedError("Unavailable") \ No newline at end of file diff --git a/ShellyPy/wrapper.py b/ShellyPy/wrapper.py index 2715ce0..2ea5ad1 100644 --- a/ShellyPy/wrapper.py +++ b/ShellyPy/wrapper.py @@ -1,35 +1,31 @@ -from sys import version_info - -if version_info.major == 3: - from json.decoder import JSONDecodeError -else: - JSONDecodeError = ValueError +from typing import Any, Optional, Type +from json.decoder import JSONDecodeError 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, port = "80", *args, **kwargs): - """ - @param ip the target IP of the shelly device. Can be a string, list of strings or list of integers - @param port target port, may be useful for non Shelly devices that have the same HTTP Api - @param login dict of login credentials. Keys needed are "username" and "password" - @param timeout specify the amount of time until requests are aborted. - @param debug enable debug printing - @param init calls the update method on init - """ +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: - self._instance = self.__detect__(ip, port)(ip, port, *args, **kwargs) + 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) - 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() @@ -37,24 +33,55 @@ def __detect__(self, ip, port): raise NotFound("Not Found") try: - response_data = response.json() + response_data: dict[str, Any] = response.json() 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) -> str: + return self._instance.__repr__() + + def __str__(self) -> str: + return self._instance.__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 __repr__(self): - return self.__getattr__("__repr__")() + 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 __str__(self): - return self.__getattr__("__str__")() + 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 __getattr__(self, name): - return getattr(self._instance, name) + def emeter(self, index: int) -> dict[str, Any]: + return self._instance.emeter(index) diff --git a/requirements.txt b/requirements.txt index f229360..4c738f3 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1 +1,3 @@ requests +types-requests + 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",