Merge pull request #176 from plugwise/set-return

Set-functions: rework to raise instead of return

Author: bouwew

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+# v0.17.5: Smile: rework to raise instead of return
+- raise in error-cases, move LOGGER.debug messages into raise
+- clean up code
+
+# v0.17.4 - Smile: improve typing hints, implement mypy testing
+
 # v0.17.3 - Smile Adam: add support for heater_electric type Plugs
 
 # v0.17.2 - Smile Adam: more bugfixes, improvementds

plugwise/__init__.py

@@ -1,6 +1,6 @@
 """Plugwise module."""
 
-__version__ = "0.17.3"
+__version__ = "0.17.5"
 
 from plugwise.smile import Smile
 from plugwise.stick import Stick

plugwise/exceptions.py

@@ -59,6 +59,10 @@ class InvalidSetupError(PlugwiseException):
     """Raised when adding an Anna while an Adam exists."""
 
 
+class PlugwiseError(PlugwiseException):
+    """Raise when a non-specific error happens."""
+
+
 class UnsupportedDeviceError(PlugwiseException):
     """Raised when device is not supported."""
 

plugwise/helper.py

@@ -227,18 +227,18 @@ async def _request_validate(self, resp: ClientResponse, method: str) -> etree:
             return
 
         if resp.status == 401:
-            raise InvalidAuthentication
+            raise InvalidAuthentication(
+                "Invalid login, please retry with the correct credentials."
+            )
 
         if not (result := await resp.text()) or "<error>" in result:
-            LOGGER.error("Smile response empty or error in %s", result)
-            raise ResponseError
+            raise ResponseError(f"Smile response empty or error in {result}.")
 
         try:
             # Encode to ensure utf8 parsing
             xml = etree.XML(escape_illegal_xml_characters(result).encode())
         except etree.ParseError:
-            LOGGER.error("Smile returns invalid XML for %s", self._endpoint)
-            raise InvalidXMLError
+            raise InvalidXMLError(f"Smile returns invalid XML for {self._endpoint}.")
 
         return xml
 
@@ -271,8 +271,9 @@ async def _request(
                 )
         except ServerTimeoutError:
             if retry < 1:
-                LOGGER.error("Timed out sending %s command to Plugwise", command)
-                raise DeviceTimeoutError
+                raise DeviceTimeoutError(
+                    f"Timed out sending {command} command to Plugwise"
+                )
             return await self._request(command, retry - 1)
 
         return await self._request_validate(resp, method)

plugwise/smile.py

@@ -34,7 +34,7 @@
 from .exceptions import (
     ConnectionFailedError,
     InvalidSetupError,
-    InvalidXMLError,
+    PlugwiseError,
     UnsupportedDeviceError,
 )
 from .helper import SmileComm, SmileHelper, update_helper
@@ -275,21 +275,18 @@ async def connect(self) -> bool:
         dsmrmain = result.find("./module/protocols/dsmrmain")
         if "Plugwise" not in names:
             if dsmrmain is None:  # pragma: no cover
-                LOGGER.error(
-                    "Connected but expected text not returned, \
-                    we got %s. Please create an issue on \
-                    http://github.com/plugwise/python-plugwise",
-                    result,
+                raise ConnectionFailedError(
+                    f"Connected but expected text not returned, \
+                    we got {result}. Please create an issue on \
+                    http://github.com/plugwise/python-plugwise"
                 )
-                raise ConnectionFailedError
 
         # Check if Anna is connected to an Adam
         if "159.2" in models:
-            LOGGER.error(
+            raise InvalidSetupError(
                 "Your Anna is connected to an Adam, make \
-                sure to only add the Adam as integration.",
+                sure to only add the Adam as integration."
             )
-            raise InvalidSetupError
 
         # Determine smile specifics
         await self._smile_detect(result, dsmrmain)
@@ -320,38 +317,31 @@ async def _smile_detect_legacy(self, result: etree, dsmrmain: etree) -> str:
         if result.find('./appliance[type="thermostat"]') is None:
             # It's a P1 legacy:
             if dsmrmain is not None:
-                try:
-                    status = await self._request(STATUS)
-                    self.smile_fw_version = status.find("./system/version").text
-                    model = status.find("./system/product").text
-                    self.smile_hostname = status.find("./network/hostname").text
-                    self.smile_mac_address = status.find("./network/mac_address").text
-                except InvalidXMLError:  # pragma: no cover
-                    # Corner case check
-                    raise ConnectionFailedError
+                status = await self._request(STATUS)
+                self.smile_fw_version = status.find("./system/version").text
+                model = status.find("./system/product").text
+                self.smile_hostname = status.find("./network/hostname").text
+                self.smile_mac_address = status.find("./network/mac_address").text
 
             # Or a legacy Stretch:
             elif network is not None:
-                try:
-                    system = await self._request(SYSTEM)
-                    self.smile_fw_version = system.find("./gateway/firmware").text
-                    model = system.find("./gateway/product").text
-                    self.smile_hostname = system.find("./gateway/hostname").text
-                    # If wlan0 contains data it's active, so eth0 should be checked last
-                    for network in ["wlan0", "eth0"]:
-                        locator = f"./{network}/mac"
-                        if (net_locator := system.find(locator)) is not None:
-                            self.smile_mac_address = net_locator.text
-                except InvalidXMLError:  # pragma: no cover
-                    # Corner case check
-                    raise ConnectionFailedError
+                system = await self._request(SYSTEM)
+                self.smile_fw_version = system.find("./gateway/firmware").text
+                model = system.find("./gateway/product").text
+                self.smile_hostname = system.find("./gateway/hostname").text
+                # If wlan0 contains data it's active, so eth0 should be checked last
+                for network in ["wlan0", "eth0"]:
+                    locator = f"./{network}/mac"
+                    if (net_locator := system.find(locator)) is not None:
+                        self.smile_mac_address = net_locator.text
+
             else:  # pragma: no cover
                 # No cornercase, just end of the line
-                LOGGER.error(
+                raise ConnectionFailedError(
                     "Connected but no gateway device information found, please create \
                      an issue on http://github.com/plugwise/python-plugwise"
                 )
-                raise ConnectionFailedError
+
         return model
 
     async def _smile_detect(self, result: etree, dsmrmain: etree) -> None:
@@ -370,23 +360,20 @@ async def _smile_detect(self, result: etree, dsmrmain: etree) -> None:
 
         if model is None or self.smile_fw_version is None:  # pragma: no cover
             # Corner case check
-            LOGGER.error(
+            raise UnsupportedDeviceError(
                 "Unable to find model or version information, please create \
                  an issue on http://github.com/plugwise/python-plugwise"
             )
-            raise UnsupportedDeviceError
 
         ver = semver.VersionInfo.parse(self.smile_fw_version)
         target_smile = f"{model}_v{ver.major}"
         LOGGER.debug("Plugwise identified as %s", target_smile)
         if target_smile not in SMILES:
-            LOGGER.error(
-                'Your version Smile identified as "%s" seems\
+            raise UnsupportedDeviceError(
+                "Your version Smile identified as {target_smile} seems\
                  unsupported by our plugin, please create an issue\
-                 on http://github.com/plugwise/python-plugwise',
-                target_smile,
+                 on http://github.com/plugwise/python-plugwise"
             )
-            raise UnsupportedDeviceError
 
         self.smile_name = SMILES[target_smile]["friendly_name"]
         self.smile_type = SMILES[target_smile]["type"]
@@ -472,15 +459,15 @@ async def async_update(self) -> list[dict[str, Any]]:
 
         return [self.gw_data, self.gw_devices]
 
-    async def _set_schedule_state_legacy(self, name: str, status: str) -> bool:
+    async def _set_schedule_state_legacy(self, name: str, status: str) -> None:
         """Helper-function for set_schedule_state()."""
         schedule_rule_id: str | None = None
         for rule in self._domain_objects.findall("rule"):
             if rule.find("name").text == name:
                 schedule_rule_id = rule.attrib["id"]
 
         if schedule_rule_id is None:
-            return False
+            raise PlugwiseError("No schedule available.")
 
         state = "false"
         if status == "on":
@@ -497,18 +484,18 @@ async def _set_schedule_state_legacy(self, name: str, status: str) -> bool:
         )
 
         await self._request(uri, method="put", data=data)
-        return True
 
-    async def set_schedule_state(self, loc_id: str, name: str, state: str) -> bool:
+    async def set_schedule_state(self, loc_id: str, name: str, state: str) -> None:
         """Set the Schedule, with the given name, on the relevant Thermostat.
         Determined from - DOMAIN_OBJECTS.
         """
         if self._smile_legacy:
-            return await self._set_schedule_state_legacy(name, state)
+            await self._set_schedule_state_legacy(name, state)
+            return
 
         schedule_rule = self._rule_ids_by_name(name, loc_id)
         if not schedule_rule or schedule_rule is None:
-            return False
+            raise PlugwiseError("No schedule with this name available.")
 
         schedule_rule_id: str = next(iter(schedule_rule))
 
@@ -543,31 +530,29 @@ async def set_schedule_state(self, loc_id: str, name: str, state: str) -> bool:
         )
         await self._request(uri, method="put", data=data)
 
-        return True
-
-    async def _set_preset_legacy(self, preset: str) -> bool:
+    async def _set_preset_legacy(self, preset: str) -> None:
         """Set the given Preset on the relevant Thermostat - from DOMAIN_OBJECTS."""
         locator = f'rule/directives/when/then[@icon="{preset}"].../.../...'
         if (rule := self._domain_objects.find(locator)) is None:
-            return False
+            raise PlugwiseError("Invalid preset.")
 
         uri = RULES
         data = f'<rules><rule id="{rule.attrib["id"]}"><active>true</active></rule></rules>'
 
         await self._request(uri, method="put", data=data)
-        return True
 
-    async def set_preset(self, loc_id: str, preset: str) -> bool:
+    async def set_preset(self, loc_id: str, preset: str) -> None:
         """Set the given Preset on the relevant Thermostat - from LOCATIONS."""
         if self._smile_legacy:
-            return await self._set_preset_legacy(preset)
+            await self._set_preset_legacy(preset)
+            return
 
         current_location = self._locations.find(f'location[@id="{loc_id}"]')
         location_name = current_location.find("name").text
         location_type = current_location.find("type").text
 
         if preset not in self._presets(loc_id):
-            return False
+            raise PlugwiseError("Invalid preset.")
 
         uri = f"{LOCATIONS};id={loc_id}"
         data = (
@@ -577,9 +562,8 @@ async def set_preset(self, loc_id: str, preset: str) -> bool:
         )
 
         await self._request(uri, method="put", data=data)
-        return True
 
-    async def set_temperature(self, loc_id: str, temperature: str) -> bool:
+    async def set_temperature(self, loc_id: str, temperature: str) -> None:
         """Set the given Temperature on the relevant Thermostat."""
         uri = self._thermostat_uri(loc_id)
         data = (
@@ -588,9 +572,8 @@ async def set_temperature(self, loc_id: str, temperature: str) -> bool:
         )
 
         await self._request(uri, method="put", data=data)
-        return True
 
-    async def set_max_boiler_temperature(self, temperature: str) -> bool:
+    async def set_max_boiler_temperature(self, temperature: str) -> None:
         """Set the max. Boiler Temperature on the Central heating boiler."""
         locator = f'appliance[@id="{self._heater_id}"]/actuator_functionalities/thermostat_functionality'
         th_func = self._appliances.find(locator)
@@ -601,11 +584,10 @@ async def set_max_boiler_temperature(self, temperature: str) -> bool:
         data = f"<thermostat_functionality><setpoint>{temperature}</setpoint></thermostat_functionality>"
 
         await self._request(uri, method="put", data=data)
-        return True
 
     async def _set_groupswitch_member_state(
         self, members: list[str], state: str, switch: Munch
-    ) -> bool:
+    ) -> None:
         """Helper-function for set_switch_state() .
         Set the given State of the relevant Switch within a group of members.
         """
@@ -619,11 +601,9 @@ async def _set_groupswitch_member_state(
 
             await self._request(uri, method="put", data=data)
 
-        return True
-
     async def set_switch_state(
         self, appl_id: str, members: list[str] | None, model: str, state: str
-    ) -> bool:
+    ) -> None:
         """Set the given State of the relevant Switch."""
         switch = Munch()
         switch.actuator = "actuator_functionalities"
@@ -659,15 +639,14 @@ async def set_switch_state(
             lock_state: str = self._appliances.find(locator).text
             # Don't bother switching a relay when the corresponding lock-state is true
             if lock_state == "true":
-                return False
+                raise PlugwiseError("Cannot switch a locked Relay.")
 
         await self._request(uri, method="put", data=data)
-        return True
 
-    async def set_regulation_mode(self, mode: str) -> bool:
+    async def set_regulation_mode(self, mode: str) -> None:
         """Set the heating regulation mode."""
         if mode not in self._allowed_modes:
-            return False
+            raise PlugwiseError("Invalid regulation mode.")
 
         uri = f"{APPLIANCES};type=gateway/regulation_mode_control"
         duration = ""
@@ -676,11 +655,9 @@ async def set_regulation_mode(self, mode: str) -> bool:
         data = f"<regulation_mode_control_functionality>{duration}<mode>{mode}</mode></regulation_mode_control_functionality>"
 
         await self._request(uri, method="put", data=data)
-        return True
 
-    async def delete_notification(self) -> bool:
+    async def delete_notification(self) -> None:
         """Delete the active Plugwise Notification."""
         uri = NOTIFICATIONS
 
         await self._request(uri, method="delete")
-        return True